Skip to content

N
neoppod
Commit 3689c816 authored by Kirill Smelkov's avatar Kirill Smelkov
Browse files
Options

X splitting into packages in progress

parent 9888b7dc
Show whitespace changes
Inline Side-by-side
Showing with 119 additions and 96 deletions
go/neo/client.go → go/neo/client/client.go
View file @ 3689c816
......@@ -22,23 +22,24 @@ import (
"context"
"net/url"
"../zodb"
"../../neo"
"../../zodb"
)
// Client talks to NEO cluster and exposes access it via ZODB interfaces
type Client struct {
// XXX move -> nodeCommon?
// ---- 8< ----
myInfo NodeInfo // XXX -> only NodeUUID
myInfo neo.NodeInfo // XXX -> only NodeUUID
clusterName string
net Network // network we are sending/receiving on
net neo.Network // network we are sending/receiving on
masterAddr string // address of master XXX -> Address ?
// ---- 8< ----
storLink *NodeLink // link to storage node
storConn *Conn // XXX main connection to storage
storLink *neo.NodeLink // link to storage node
storConn *neo.Conn // XXX main connection to storage
}
var _ zodb.IStorage = (*Client)(nil)
......@@ -57,8 +58,8 @@ func (c *Client) Close() error {
func (c *Client) LastTid() (zodb.Tid, error) {
// FIXME do not use global conn (see comment in openClientByURL)
// XXX open new conn for this particular req/reply ?
reply := AnswerLastTransaction{}
err := Ask(c.storConn, &LastTransaction{}, &reply)
reply := neo.AnswerLastTransaction{}
err := neo.Ask(c.storConn, &neo.LastTransaction{}, &reply)
if err != nil {
return 0, err // XXX err ctx
}
......@@ -67,17 +68,17 @@ func (c *Client) LastTid() (zodb.Tid, error) {
func (c *Client) Load(xid zodb.Xid) (data []byte, tid zodb.Tid, err error) {
// FIXME do not use global conn (see comment in openClientByURL)
req := GetObject{Oid: xid.Oid}
req := neo.GetObject{Oid: xid.Oid}
if xid.TidBefore {
req.Serial = INVALID_TID
req.Serial = neo.INVALID_TID
req.Tid = xid.Tid
} else {
req.Serial = xid.Tid
req.Tid = INVALID_TID
req.Tid = neo.INVALID_TID
}
resp := AnswerGetObject{}
err = Ask(c.storConn, &req, &resp)
resp := neo.AnswerGetObject{}
err = neo.Ask(c.storConn, &req, &resp)
if err != nil {
return nil, 0, err // XXX err context
}
......@@ -97,7 +98,7 @@ func (c *Client) Iterate(tidMin, tidMax zodb.Tid) zodb.IStorageIterator {
// NewClient creates and identifies new client connected to storage over storLink
func NewClient(storLink *NodeLink) (*Client, error) {
func NewClient(storLink *neo.NodeLink) (*Client, error) {
// TODO .myInfo.NodeType = CLIENT
// .clusterName = clusterName
// .net = ...
......@@ -106,7 +107,7 @@ func NewClient(storLink *NodeLink) (*Client, error) {
// XXX move -> Run?
// first identify ourselves to peer
accept, err := IdentifyWith(STORAGE, storLink, cli.myInfo, cli.clusterName)
accept, err := neo.IdentifyWith(neo.STORAGE, storLink, cli.myInfo, cli.clusterName)
if err != nil {
return nil, err
}
......@@ -133,8 +134,8 @@ func NewClient(storLink *NodeLink) (*Client, error) {
func openClientByURL(ctx context.Context, u *url.URL) (zodb.IStorage, error) {
// XXX for now url is treated as storage node URL
// XXX check/use other url fields
net := NetPlain("tcp") // TODO + TLS; not only "tcp" ?
storLink, err := Dial(ctx, net, u.Host)
net := neo.NetPlain("tcp") // TODO + TLS; not only "tcp" ?
storLink, err := neo.Dial(ctx, net, u.Host)
if err != nil {
return nil, err
}
......
go/neo/command.go → go/neo/cmd/neo/command.go
View file @ 3689c816
......@@ -15,10 +15,10 @@
//
// See COPYING file for full licensing terms.
package neo
package main
// registry of all commands & help topics
import "../zodb/zodbtools"
import "../../../zodb/zodbtools"
var Commands = zodbtools.CommandRegistry{
{"master", masterSummary, masterUsage, masterMain},
......
go/neo/connection.go
View file @ 3689c816
......@@ -251,7 +251,7 @@ func (c *Conn) shutdown() {
}
// Close closes connection
// Any blocked Send() or Recv() will be unblocked and return error
// Any blocked Send*() or Recv*() will be unblocked and return error
//
// NOTE for Send() - once transmission was started - it will complete in the
// background on the wire not to break node-node link framing.
......
go/neo/neo.go
View file @ 3689c816
// TODO copyright / license
// Copyright (C) 2016-2017 Nexedi SA and Contributors.
// Kirill Smelkov <kirr@nexedi.com>
//
// This program is free software: you can Use, Study, Modify and Redistribute
// it under the terms of the GNU General Public License version 3, or (at your
// option) any later version, as published by the Free Software Foundation.
//
// You can also Link and Combine this program with other software covered by
// the terms of any of the Open Source Initiative approved licenses and Convey
// the resulting work. Corresponding source of such a combination shall include
// the source code for all other software used.
//
// This program is distributed WITHOUT ANY WARRANTY; without even the implied
// warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
//
// See COPYING file for full licensing terms.
// Package neo implements distributed object storage for ZODB
// TODO text
// Package neo and its children provide distributed object storage for ZODB
//
// Package neo itself provides protocol definition and common infrastructure.
// See packages neo.client and neo.server for client and server sides respectively.
// XXX text
package neo
import (
......
go/neo/proto.go
View file @ 3689c816
// NEO. Protocol description
//go:generate sh -c "go run protogen.go >proto-marshal.go"
package neo
// protocol definition
// NOTE for some packets it is possible to decode raw packet -> go version from
// PktBuf in place. E.g. for GetObject.
......@@ -139,9 +139,9 @@ type NodeUUID int32
// ErrDecodeOverflow is the error returned by NEOPktDecode when decoding hit buffer overflow
var ErrDecodeOverflow = errors.New("decode: bufer overflow")
// NEOPkt is the interface implemented by packets to marshal/unmarshal them into/from wire format
// Pkt is the interface implemented by NEO packets to marshal/unmarshal them into/from wire format
// XXX -> will be neo.Pkt after splitting into packages
type NEOPkt interface {
type Pkt interface {
// NEOPktMsgCode returns message code needed to be used for particular packet type
// on the wire
NEOPktMsgCode() uint16
......
go/neo/protogen.go
View file @ 3689c816
......@@ -21,7 +21,7 @@
NEO. Protocol module. Code generator
This program generates marshalling code for packet types defined in proto.go .
For every type 4 methods are generated in accordance with NEOPkt interface:
For every type 4 methods are generated in accordance with neo.Pkt interface:
NEOPktMsgCode() uint16
NEOPktEncodedLen() int
......@@ -469,7 +469,7 @@ type sizer struct {
// encoder generates code to encode a packet
//
// when type is recursively walked, for every case code to update `data[n:]` is generated.
// no overflow checks are generated as by NEOPkt interface provided data
// no overflow checks are generated as by neo.Pkt interface provided data
// buffer should have at least payloadLen length returned by NEOPktEncodedInfo()
// (the size computed by sizer).
//
......
go/neo/master.go → go/neo/server/master.go
View file @ 3689c816
......@@ -30,7 +30,8 @@ import (
"os"
"sync"
"../zodb"
"../../neo"
"../../zodb"
"lab.nexedi.com/kirr/go123/xerr"
)
......@@ -38,7 +39,7 @@ import (
// Master is a node overseeing and managing how whole NEO cluster works
type Master struct {
clusterName string
nodeUUID NodeUUID
nodeUUID neo.NodeUUID
// last allocated oid & tid
// XXX how to start allocating oid from 0, not 1 ?
......@@ -48,9 +49,9 @@ type Master struct {
// master manages node and partition tables and broadcast their updates
// to all nodes in cluster
stateMu sync.RWMutex // XXX recheck: needed ?
nodeTab NodeTable
partTab PartitionTable
clusterState ClusterState
nodeTab neo.NodeTable
partTab neo.PartitionTable
clusterState neo.ClusterState
// channels controlling main driver
ctlStart chan chan error // request to start cluster
......@@ -65,22 +66,22 @@ type Master struct {
// node connects
type nodeCome struct {
link *NodeLink
idReq RequestIdentification // we received this identification request
idResp chan NEOPkt // what we reply (AcceptIdentification | Error)
link *neo.NodeLink
idReq neo.RequestIdentification // we received this identification request
idResp chan neo.Pkt // what we reply (AcceptIdentification | Error)
}
// node disconnects
type nodeLeave struct {
link *NodeLink // XXX better use uuid allocated on nodeCome ?
link *neo.NodeLink // XXX better use uuid allocated on nodeCome ?
}
// NewMaster TODO ...
func NewMaster(clusterName string) *Master {
m := &Master{clusterName: clusterName}
m.nodeUUID = m.allocUUID(MASTER)
m.nodeUUID = m.allocUUID(neo.MASTER)
// TODO update nodeTab with self
m.clusterState = ClusterRecovering // XXX no elections - we are the only master
m.clusterState = neo.ClusterRecovering // XXX no elections - we are the only master
go m.run(context.TODO()) // XXX ctx
return m
......@@ -115,7 +116,7 @@ func (m *Master) Shutdown() error {
// setClusterState sets .clusterState and notifies subscribers
func (m *Master) setClusterState(state ClusterState) {
func (m *Master) setClusterState(state neo.ClusterState) {
if state == m.clusterState { // <- XXX do we really need this ?
return
}
......@@ -179,7 +180,7 @@ func (m *Master) recovery(ctx context.Context) (err error) {
fmt.Println("master: recovery")
defer xerr.Context(&err, "master: recovery")
m.setClusterState(ClusterRecovering)
m.setClusterState(neo.ClusterRecovering)
rctx, rcancel := context.WithCancel(ctx)
defer rcancel()
......@@ -188,7 +189,7 @@ func (m *Master) recovery(ctx context.Context) (err error) {
// start recovery on all storages we are currently in touch with
for _, stor := range m.nodeTab.StorageList() {
if stor.NodeState > DOWN { // XXX state cmp ok ? XXX or stor.Link != nil ?
if stor.NodeState > neo.DOWN { // XXX state cmp ok ? XXX or stor.Link != nil ?
inprogress++
go storCtlRecovery(rctx, stor.Link, recovery)
}
......@@ -274,7 +275,7 @@ loop:
// storRecovery is result of a storage node passing recovery phase
type storRecovery struct {
partTab PartitionTable
partTab neo.PartitionTable
// XXX + backup_tid, truncate_tid ?
err error
......@@ -282,7 +283,7 @@ type storRecovery struct {
// storCtlRecovery drives a storage node during cluster recovering state
// it retrieves various ids and partition table from as stored on the storage
func storCtlRecovery(ctx context.Context, link *NodeLink, res chan storRecovery) {
func storCtlRecovery(ctx context.Context, link *neo.NodeLink, res chan storRecovery) {
var err error
// XXX where this close link on error should be ?
defer func() {
......@@ -309,30 +310,30 @@ func storCtlRecovery(ctx context.Context, link *NodeLink, res chan storRecovery)
}
// XXX cancel on ctx
recovery := AnswerRecovery{}
err = Ask(conn, &Recovery{}, &recovery)
recovery := neo.AnswerRecovery{}
err = neo.Ask(conn, &neo.Recovery{}, &recovery)
if err != nil {
return
}
resp := AnswerPartitionTable{}
err = Ask(conn, &X_PartitionTable{}, &resp)
resp := neo.AnswerPartitionTable{}
err = neo.Ask(conn, &neo.X_PartitionTable{}, &resp)
if err != nil {
return
}
// reconstruct partition table from response
pt := PartitionTable{}
pt := neo.PartitionTable{}
pt.ptid = resp.PTid
for _, row := range resp.RowList {
i := row.Offset
for i >= uint32(len(pt.ptTab)) {
pt.ptTab = append(pt.ptTab, []PartitionCell{})
pt.ptTab = append(pt.ptTab, []neo.PartitionCell{})
}
//pt.ptTab[i] = append(pt.ptTab[i], row.CellList...)
for _, cell := range row.CellList {
pt.ptTab[i] = append(pt.ptTab[i], PartitionCell{
pt.ptTab[i] = append(pt.ptTab[i], neo.PartitionCell{
NodeUUID: cell.NodeUUID,
CellState: cell.CellState,
})
......@@ -366,7 +367,7 @@ func (m *Master) verify(ctx context.Context) (err error) {
fmt.Println("master: verify")
defer xerr.Context(&err, "master: verify")
m.setClusterState(ClusterVerifying)
m.setClusterState(neo.ClusterVerifying)
vctx, vcancel := context.WithCancel(ctx)
defer vcancel()
......@@ -381,7 +382,7 @@ func (m *Master) verify(ctx context.Context) (err error) {
// start verification on all storages we are currently in touch with
for _, stor := range m.nodeTab.StorageList() {
if stor.NodeState > DOWN { // XXX state cmp ok ? XXX or stor.Link != nil ?
if stor.NodeState > neo.DOWN { // XXX state cmp ok ? XXX or stor.Link != nil ?
inprogress++
go storCtlVerify(vctx, stor.Link, verify)
}
......@@ -468,12 +469,12 @@ loop:
type storVerify struct {
lastOid zodb.Oid
lastTid zodb.Tid
link *NodeLink
link *neo.NodeLink
err error
}
// storCtlVerify drives a storage node during cluster verifying (= starting) state
func storCtlVerify(ctx context.Context, link *NodeLink, res chan storVerify) {
func storCtlVerify(ctx context.Context, link *neo.NodeLink, res chan storVerify) {
// XXX link.Close on err
// XXX cancel on ctx
......@@ -488,8 +489,8 @@ func storCtlVerify(ctx context.Context, link *NodeLink, res chan storVerify) {
// FIXME stub
conn, _ := link.NewConn()
locked := AnswerLockedTransactions{}
err = Ask(conn, &LockedTransactions{}, &locked)
locked := neo.AnswerLockedTransactions{}
err = neo.Ask(conn, &neo.LockedTransactions{}, &locked)
if err != nil {
return
}
......@@ -500,8 +501,8 @@ func storCtlVerify(ctx context.Context, link *NodeLink, res chan storVerify) {
return
}
last := AnswerLastIDs{}
err = Ask(conn, &LastIDs{}, &last)
last := neo.AnswerLastIDs{}
err = neo.Ask(conn, &neo.LastIDs{}, &last)
if err != nil {
return
}
......@@ -573,7 +574,7 @@ loop:
// accept processes identification request of just connected node and either accepts or declines it
// if node identification is accepted nodeTab is updated and corresponding node entry is returned
func (m *Master) accept(n nodeCome) (node *Node, ok bool) {
func (m *Master) accept(n nodeCome) (node *neo.Node, ok bool) {
// XXX also verify ? :
// - NodeType valid
// - IdTimestamp ?
......@@ -609,7 +610,7 @@ func (m *Master) accept(n nodeCome) (node *Node, ok bool) {
n.idResp <- &AcceptIdentification{
NodeType: MASTER,
NodeType: neo.MASTER,
MyNodeUUID: m.nodeUUID,
NumPartitions: 1, // FIXME hardcoded
NumReplicas: 1, // FIXME hardcoded
......@@ -642,7 +643,7 @@ func (m *Master) accept(n nodeCome) (node *Node, ok bool) {
// allocUUID allocates new node uuid for a node of kind nodeType
// XXX it is bad idea for master to assign uuid to coming node
// -> better nodes generate really unique UUID themselves and always show with them
func (m *Master) allocUUID(nodeType NodeType) NodeUUID {
func (m *Master) allocUUID(nodeType neo.NodeType) neo.NodeUUID {
// see NodeUUID & NodeUUID.String for details
// XXX better to keep this code near to ^^^ (e.g. attached to NodeType)
// XXX but since whole uuid assign idea is not good - let's keep it dirty here
......@@ -659,7 +660,7 @@ func (m *Master) allocUUID(nodeType NodeType) NodeUUID {
// ServeLink serves incoming node-node link connection
// XXX +error return?
func (m *Master) ServeLink(ctx context.Context, link *NodeLink) {
func (m *Master) ServeLink(ctx context.Context, link *neo.NodeLink) {
logf := func(format string, argv ...interface{}) {
fmt.Printf("master: %s: " + format + "\n", append([]interface{}{link}, argv...))
}
......@@ -704,7 +705,7 @@ func (m *Master) ServeLink(ctx context.Context, link *NodeLink) {
}
// convey identification request to master
idRespCh := make(chan NEOPkt)
idRespCh := make(chan neo.Pkt)
m.nodeCome <- nodeCome{link, idReq, idRespCh}
idResp := <-idRespCh
......@@ -760,7 +761,7 @@ func (m *Master) ServeLink(ctx context.Context, link *NodeLink) {
m.stateMu.Unlock()
go func() {
var pkt NEOPkt
var pkt neo.Pkt
for {
select {
......@@ -798,8 +799,8 @@ func (m *Master) ServeLink(ctx context.Context, link *NodeLink) {
// ServeClient serves incoming connection on which peer identified itself as client
// XXX +error return?
//func (m *Master) ServeClient(ctx context.Context, conn *Conn) {
func (m *Master) ServeClient(ctx context.Context, link *NodeLink) {
//func (m *Master) ServeClient(ctx context.Context, conn *neo.Conn) {
func (m *Master) ServeClient(ctx context.Context, link *neo.NodeLink) {
// TODO
}
......@@ -839,7 +840,7 @@ type storageStopOperation struct {
// with e.g. a command or request and expects corresponding answer
//
// XXX +error return?
func (m *Master) DriveStorage(ctx context.Context, link *NodeLink) {
func (m *Master) DriveStorage(ctx context.Context, link *neo.NodeLink) {
// ? >UnfinishedTransactions
// ? <AnswerUnfinishedTransactions (none currently)
......@@ -952,11 +953,11 @@ func (m *Master) DriveStorage(ctx context.Context, link *NodeLink) {
// StopOperation PM -> S
}
func (m *Master) ServeAdmin(ctx context.Context, conn *Conn) {
func (m *Master) ServeAdmin(ctx context.Context, conn *neo.Conn) {
// TODO
}
func (m *Master) ServeMaster(ctx context.Context, conn *Conn) {
func (m *Master) ServeMaster(ctx context.Context, conn *neo.Conn) {
// TODO (for elections)
}
......
go/neo/server.go → go/neo/server/server.go
View file @ 3689c816
......@@ -24,6 +24,8 @@ import (
"net"
"reflect"
"../../neo"
"lab.nexedi.com/kirr/go123/xerr"
)
......@@ -31,7 +33,7 @@ import (
type Server interface {
// ServeLink serves already established nodelink (connection) in a blocking way.
// ServeLink is usually run in separate goroutine
ServeLink(ctx context.Context, link *NodeLink)
ServeLink(ctx context.Context, link *neo.NodeLink)
}
// Serve runs service on a listener
......@@ -79,7 +81,7 @@ func Serve(ctx context.Context, l net.Listener, srv Server) error {
// ListenAndServe listens on network address and then calls Serve to handle incoming connections
// XXX unused -> goes away ?
func ListenAndServe(ctx context.Context, net Network, laddr string, srv Server) error {
func ListenAndServe(ctx context.Context, net neo.Network, laddr string, srv Server) error {
l, err := net.Listen(laddr)
if err != nil {
return err
......@@ -95,7 +97,7 @@ func ListenAndServe(ctx context.Context, net Network, laddr string, srv Server)
// it expects peer to send RequestIdentification packet and replies with AcceptIdentification if identification passes.
// returns information about identified node or error.
// XXX recheck identification logic here
func IdentifyPeer(link *NodeLink, myNodeType NodeType) (nodeInfo RequestIdentification /*TODO -> NodeInfo*/, err error) {
func IdentifyPeer(link *neo.NodeLink, myNodeType neo.NodeType) (nodeInfo neo.RequestIdentification, err error) {
defer xerr.Contextf(&err, "%s: identify", link)
// the first conn must come with RequestIdentification packet
......@@ -139,7 +141,7 @@ func IdentifyPeer(link *NodeLink, myNodeType NodeType) (nodeInfo RequestIdentifi
// IdentifyWith identifies local node with remote peer
// it also verifies peer's node type to what caller expects
// XXX place != ok (this is client, not server ?)
func IdentifyWith(expectPeerType NodeType, link *NodeLink, myInfo NodeInfo, clusterName string) (accept *AcceptIdentification, err error) {
func IdentifyWith(expectPeerType neo.NodeType, link *neo.NodeLink, myInfo neo.NodeInfo, clusterName string) (accept *neo.AcceptIdentification, err error) {
defer xerr.Contextf(&err, "%s: request identification", link)
conn, err := link.NewConn()
......@@ -153,7 +155,7 @@ func IdentifyWith(expectPeerType NodeType, link *NodeLink, myInfo NodeInfo, clus
}
}()
accept = &AcceptIdentification{}
accept = &neo.AcceptIdentification{}
err = Ask(conn, &RequestIdentification{
NodeType: myInfo.NodeType,
NodeUUID: myInfo.NodeUUID,
......@@ -178,7 +180,7 @@ func IdentifyWith(expectPeerType NodeType, link *NodeLink, myInfo NodeInfo, clus
// XXX naming for RecvAndDecode and EncodeAndSend
// RecvAndDecode receives packet from conn and decodes it
func RecvAndDecode(conn *Conn) (NEOPkt, error) {
func RecvAndDecode(conn *neo.Conn) (neo.Pkt, error) {
pkt, err := conn.Recv()
if err != nil {
return nil, err
......@@ -195,7 +197,7 @@ func RecvAndDecode(conn *Conn) (NEOPkt, error) {
}
// TODO use free-list for decoded packets + when possible decode in-place
pktObj := reflect.New(msgType).Interface().(NEOPkt)
pktObj := reflect.New(msgType).Interface().(neo.Pkt)
_, err = pktObj.NEOPktDecode(pkt.Payload())
if err != nil {
// XXX -> ProtoError ?
......@@ -206,7 +208,7 @@ func RecvAndDecode(conn *Conn) (NEOPkt, error) {
}
// EncodeAndSend encodes pkt and sends it to conn
func EncodeAndSend(conn *Conn, pkt NEOPkt) error {
func EncodeAndSend(conn *neo.Conn, pkt neo.Pkt) error {
l := pkt.NEOPktEncodedLen()
buf := PktBuf{make([]byte, PktHeadLen + l)} // XXX -> freelist
......@@ -222,7 +224,7 @@ func EncodeAndSend(conn *Conn, pkt NEOPkt) error {
// Ask does simple request/response protocol exchange
// It expects the answer to be exactly of resp type and errors otherwise
func Ask(conn *Conn, req NEOPkt, resp NEOPkt) error {
func Ask(conn *neo.Conn, req neo.Pkt, resp neo.Pkt) error {
err := EncodeAndSend(conn, req)
if err != nil {
return err
......@@ -237,7 +239,7 @@ func Ask(conn *Conn, req NEOPkt, resp NEOPkt) error {
// unexpected packet or packet with wrong header
// XXX -> ConnError{Op: "decode"} ?
type ProtoError struct {
Conn *Conn
Conn *neo.Conn
Err error
}
......@@ -247,7 +249,7 @@ func (e *ProtoError) Error() string {
// Expect receives 1 packet and expects it to be exactly of msg type
// XXX naming (-> Recv1 ?)
func Expect(conn *Conn, msg NEOPkt) (err error) {
func Expect(conn *neo.Conn, msg neo.Pkt) (err error) {
pkt, err := conn.Recv()
if err != nil {
return err
......
go/neo/storage.go → go/neo/server/storage.go
View file @ 3689c816
......@@ -28,8 +28,9 @@ import (
"strings"
"time"
"../zodb"
"../zodb/storage/fs1"
"../../neo"
"../../zodb"
"../../zodb/storage/fs1"
)
// XXX fmt -> log
......@@ -38,10 +39,10 @@ import (
type Storage struct {
// XXX move -> nodeCommon?
// ---- 8< ----
myInfo NodeInfo // XXX -> only Address + NodeUUID ?
myInfo neo.NodeInfo // XXX -> only Address + NodeUUID ?
clusterName string
net Network // network we are sending/receiving on
net neo.Network // network we are sending/receiving on
masterAddr string // address of master XXX -> Address ?
// ---- 8< ----
......@@ -51,7 +52,7 @@ type Storage struct {
// NewStorage creates new storage node that will listen on serveAddr and talk to master on masterAddr
// The storage uses zstor as underlying backend for storing data.
// To actually start running the node - call Run. XXX text
func NewStorage(cluster string, masterAddr string, serveAddr string, net Network, zstor zodb.IStorage) *Storage {
func NewStorage(cluster string, masterAddr string, serveAddr string, net neo.Network, zstor zodb.IStorage) *Storage {
// convert serveAddr into neo format
addr, err := ParseAddress(serveAddr)
if err != nil {
......@@ -176,7 +177,7 @@ func (stor *Storage) talkMaster1(ctx context.Context) error {
// ServeLink serves incoming node-node link connection
// XXX +error return?
func (stor *Storage) ServeLink(ctx context.Context, link *NodeLink) {
func (stor *Storage) ServeLink(ctx context.Context, link *neo.NodeLink) {
fmt.Printf("stor: %s: serving new node\n", link)
// close link when either cancelling or returning (e.g. due to an error)
......@@ -203,7 +204,7 @@ func (stor *Storage) ServeLink(ctx context.Context, link *NodeLink) {
return
}
var serveConn func(context.Context, *Conn)
var serveConn func(context.Context, *neo.Conn)
switch nodeInfo.NodeType {
case CLIENT:
serveConn = stor.ServeClient
......@@ -233,22 +234,22 @@ func (stor *Storage) ServeLink(ctx context.Context, link *NodeLink) {
// XXX move err{Encode,Decode} out of here
// errEncode translates an error into Error packet
func errEncode(err error) *Error {
func errEncode(err error) *neo.Error {
switch err := err.(type) {
case *Error:
return err
case *zodb.ErrXidMissing:
// XXX abusing message for xid
return &Error{Code: OID_NOT_FOUND, Message: err.Xid.String()}
return &neo.Error{Code: OID_NOT_FOUND, Message: err.Xid.String()}
default:
return &Error{Code: NOT_READY /* XXX how to report 503? was BROKEN_NODE */, Message: err.Error()}
return &neo.Error{Code: NOT_READY /* XXX how to report 503? was BROKEN_NODE */, Message: err.Error()}
}
}
// errDecode decodes error from Error packet
func errDecode(e *Error) error {
func errDecode(e *neo.Error) error {
switch e.Code {
case OID_NOT_FOUND:
xid, err := zodb.ParseXid(e.Message) // XXX abusing message for xid
......@@ -260,7 +261,7 @@ func errDecode(e *Error) error {
return e
}
func (stor *Storage) ServeMaster(ctx context.Context, conn *Conn) {
func (stor *Storage) ServeMaster(ctx context.Context, conn *neo.Conn) {
// state changes:
//
......@@ -275,7 +276,7 @@ func (stor *Storage) ServeMaster(ctx context.Context, conn *Conn) {
// ServeClient serves incoming connection on which peer identified itself as client
// XXX +error return?
func (stor *Storage) ServeClient(ctx context.Context, conn *Conn) {
func (stor *Storage) ServeClient(ctx context.Context, conn *neo.Conn) {
fmt.Printf("stor: %s: serving new client conn\n", conn)
// close connection when either cancelling or returning (e.g. due to an error)
......@@ -312,7 +313,7 @@ func (stor *Storage) ServeClient(ctx context.Context, conn *Conn) {
xid.TidBefore = true
}
var reply NEOPkt
var reply neo.Pkt
data, tid, err := stor.zstor.Load(xid)
if err != nil {
// TODO translate err to NEO protocol error codes
......@@ -334,7 +335,7 @@ func (stor *Storage) ServeClient(ctx context.Context, conn *Conn) {
EncodeAndSend(conn, reply) // XXX err
case *LastTransaction:
var reply NEOPkt
var reply neo.Pkt
lastTid, err := stor.zstor.LastTid()
if err != nil {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7