Skip to content

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

.

parent 9afb634d
Expand all Show whitespace changes
Inline Side-by-side
Showing with 92 additions and 91 deletions
go/neo/master.go
View file @ b9a07f51
......@@ -141,6 +141,7 @@ func (m *Master) ServeLink(ctx context.Context, link *NodeLink) {
nodeCh, nodeUnsubscribe := m.nodeTab.SubscribeBuffered()
_ = nodeUnsubscribe
//partCh, partUnsubscribe := m.partTab.SubscribeBuffered()
// TODO cluster subscribe
//clusterCh := make(chan ClusterState)
......@@ -168,9 +169,10 @@ func (m *Master) ServeLink(ctx context.Context, link *NodeLink) {
case nodeUpdateV := <-nodeCh:
// TODO
_ = nodeUpdateV
case clusterState = <-clusterCh:
changed = true
//case clusterState = <-clusterCh:
// changed = true
}
}
}()
......
go/neo/nodetab.go
View file @ b9a07f51
......@@ -73,7 +73,7 @@ type NodeTable struct {
sync.RWMutex
nodev []*Node
subscribev []chan *Node
notifyv []chan NodeInfo // subscribers
ver int // ↑ for versioning XXX do we need this?
}
......@@ -91,14 +91,14 @@ type Node struct {
// it returns a channel via which updates will be delivered and unsubscribe function
//
// XXX locking: client for subscribe/unsubscribe XXX ok?
func (nt *NodeTable) Subscribe() (ch chan *Node, unsubscribe func()) {
ch = make(chan *Node) // XXX how to specify ch buf size if needed ?
nt.subscribev = append(nt.subscribev, ch)
func (nt *NodeTable) Subscribe() (ch chan NodeInfo, unsubscribe func()) {
ch = make(chan NodeInfo) // XXX how to specify ch buf size if needed ?
nt.notifyv = append(nt.notifyv, ch)
unsubscribe = func() {
for i, c := range nt.subscribev {
for i, c := range nt.notifyv {
if c == ch {
nt.subscribev = append(nt.subscribev[:i], nt.subscribev[i+1:]...)
nt.notifyv = append(nt.notifyv[:i], nt.notifyv[i+1:]...)
close(ch)
return
}
......@@ -117,12 +117,12 @@ func (nt *NodeTable) Subscribe() (ch chan *Node, unsubscribe func()) {
// to infinity - via e.g. detecting stuck connections and unsubscribing on shutdown
//
// XXX locking: client for subscribe/unsubscribe XXX ok?
func (nt *NodeTable) SubscribeBuffered() (ch chan []*Node, unsubscribe func()) {
func (nt *NodeTable) SubscribeBuffered() (ch chan []NodeInfo, unsubscribe func()) {
in, unsubscribe := nt.Subscribe()
ch = make(chan []*Node)
ch = make(chan []NodeInfo)
go func() {
var updatev []*Node
var updatev []NodeInfo
shutdown := false
for {
......@@ -175,10 +175,10 @@ func (nt *NodeTable) Add(node *Node) {
// Lookup finds node by nodeID
func (nt *NodeTable) Lookup(nodeID NodeID) *Node {
func (nt *NodeTable) Lookup(uuid NodeUUID) *Node {
// FIXME linear scan
for _, node := range nt.nodev {
if node.Info.NodeID == nodeID {
if node.Info.NodeUUID == uuid {
return node
}
}
......@@ -197,7 +197,7 @@ func (nt *NodeTable) String() string {
for _, node := range nt.nodev {
// XXX recheck output
i := node.Info
fmt.Fprintf(&buf, "%s (%s)\t%s\t%s\n", i.NodeID, i.NodeType, i.NodeState, i.Address)
fmt.Fprintf(&buf, "%s (%s)\t%s\t%s\n", i.NodeUUID, i.NodeType, i.NodeState, i.Address)
}
return buf.String()
......
go/neo/parttab.go
View file @ b9a07f51
......@@ -114,7 +114,7 @@ type PartitionTable struct {
// PartitionCell describes one storage in a ptid entry in partition table
type PartitionCell struct {
NodeID
NodeUUID
CellState
// XXX ? + .haveUpToTid associated node has data up to such tid
......@@ -135,7 +135,7 @@ type PartitionCell struct {
// for cluster to be really operational it has to be checked whether
// nodes referenced by pt are up and running
//
// XXX or keep not only NodeID in PartitionCell - add *Node ?
// XXX or keep not only NodeUUID in PartitionCell - add *Node ?
func (pt *PartitionTable) Operational() bool {
for _, ptEntry := range pt.ptTab {
if len(ptEntry) == 0 {
......
go/neo/proto-marshal.go
View file @ b9a07f51
This diff is collapsed.
go/neo/proto-str.go
View file @ b9a07f51
......@@ -20,14 +20,14 @@ func (e *Error) Error() string {
const nodeTypeChar = "MSCA4567"
func (nid NodeID) String() string {
func (nodeUUID NodeUUID) String() string {
// return ex 'S1', 'M2', ...
if nid == 0 {
if nodeUUID == 0 {
return "?0"
}
typ := nid >> 24
num := nid & (1<<24 - 1)
typ := nodeUUID >> 24
num := nodeUUID & (1<<24 - 1)
temp := typ&(1 << 7) != 0
typ &= 1<<7 - 1
......
go/neo/proto.go
View file @ b9a07f51
......@@ -118,7 +118,7 @@ const (
CORRUPTED //short: C
)
// NodeID is a node identifier, 4-bytes signed integer
// NodeUUID is a node identifier, 4-bytes signed integer
//
// High-order byte:
// 7 6 5 4 3 2 1 0
......@@ -131,9 +131,9 @@ const (
// Extra namespace information and non-randomness of 3 LOB help to read logs.
//
// TODO -> back to 16-bytes randomly generated UUID
type NodeID int32
type NodeUUID int32
// TODO NodeType -> base NodeID
// TODO NodeType -> base NodeUUID
var ErrDecodeOverflow = errors.New("decode: bufer overflow")
......@@ -232,18 +232,17 @@ func float64_NEODecode(b []byte) float64 {
type NodeInfo struct {
NodeType
Address // serving address
NodeID
NodeUUID
NodeState
IdTimestamp float64
}
// XXX -> parttab.go ?
type CellInfo struct {
NodeID
NodeUUID
CellState
}
//type RowList []struct {
type RowInfo struct {
Offset uint32 // PNumber
CellList []CellInfo
......@@ -272,7 +271,7 @@ type CloseClient struct {
// connection. Any -> Any.
type RequestIdentification struct {
NodeType NodeType // XXX name
NodeID NodeID
NodeUUID NodeUUID
Address Address // where requesting node is also accepting connections
Name string // XXX -> ClusterName
IdTimestamp float64
......@@ -281,10 +280,10 @@ type RequestIdentification struct {
// XXX -> ReplyIdentification? RequestIdentification.Answer somehow ?
type AcceptIdentification struct {
NodeType NodeType // XXX name
MyNodeID NodeID
MyNodeUUID NodeUUID
NumPartitions uint32 // PNumber
NumReplicas uint32 // PNumber
YourNodeID NodeID
YourNodeUUID NodeUUID
}
// Ask current primary master's uuid. CTL -> A.
......@@ -292,12 +291,12 @@ type PrimaryMaster struct {
}
type AnswerPrimary struct {
PrimaryNodeID NodeID
PrimaryNodeUUID NodeUUID
}
// Send list of known master nodes. SM -> Any.
type NotPrimaryMaster struct {
Primary NodeID // XXX PSignedNull in py
Primary NodeUUID // XXX PSignedNull in py
KnownMasterList []struct {
Address
}
......@@ -348,7 +347,7 @@ type PartitionChanges struct {
CellList []struct {
// XXX does below correlate with Cell inside top-level CellList ?
Offset uint32 // PNumber
NodeID NodeID
NodeUUID NodeUUID
CellState CellState
}
}
......@@ -421,7 +420,7 @@ type AnswerBeginTransaction struct {
// True is returned if it's still possible to finish the transaction.
type FailedVote struct {
Tid zodb.Tid
NodeList []NodeID
NodeList []NodeUUID
// XXX _answer = Error
}
......@@ -536,7 +535,7 @@ type AnswerStoreObject struct {
// Abort a transaction. C -> S and C -> PM -> S.
type AbortTransaction struct {
Tid zodb.Tid
NodeList []NodeID // unused for * -> S
NodeList []NodeUUID // unused for * -> S
}
// Ask to store a transaction. C -> S.
......@@ -645,7 +644,7 @@ type AnswerObjectHistory struct {
type PartitionList struct {
MinOffset uint32 // PNumber
MaxOffset uint32 // PNumber
NodeID NodeID
NodeUUID NodeUUID
}
type AnswerPartitionList struct {
......@@ -667,7 +666,7 @@ type AnswerNodeList struct {
// Set the node state
type SetNodeState struct {
NodeID
NodeUUID
NodeState
// XXX _answer = Error ?
......@@ -675,14 +674,14 @@ type SetNodeState struct {
// Ask the primary to include some pending node in the partition table
type AddPendingNodes struct {
NodeList []NodeID
NodeList []NodeUUID
// XXX _answer = Error
}
// Ask the primary to optimize the partition table. A -> PM.
type TweakPartitionTable struct {
NodeList []NodeID
NodeList []NodeUUID
// XXX _answer = Error
}
......@@ -715,7 +714,7 @@ type repairFlags struct {
// Ask storage nodes to repair their databases. ctl -> A -> M
type Repair struct {
NodeList []NodeID
NodeList []NodeUUID
repairFlags
}
......@@ -803,7 +802,7 @@ type AnswerPack struct {
// ctl -> A
// A -> M
type CheckReplicas struct {
PartitionDict map[uint32]NodeID // partition -> source (PNumber)
PartitionDict map[uint32]NodeUUID // partition -> source (PNumber)
MinTID zodb.Tid
MaxTID zodb.Tid
......@@ -870,7 +869,7 @@ type AnswerCheckSerialRange struct {
// S -> M
type PartitionCorrupted struct {
Partition uint32 // PNumber
CellList []NodeID
CellList []NodeUUID
}
......
go/neo/server.go
View file @ b9a07f51
......@@ -139,10 +139,10 @@ func IdentifyPeer(link *NodeLink, myNodeType NodeType) (nodeInfo RequestIdentifi
err = EncodeAndSend(conn, &AcceptIdentification{
NodeType: myNodeType,
MyNodeID: 0, // XXX
MyNodeUUID: 0, // XXX
NumPartitions: 0, // XXX
NumReplicas: 0, // XXX
YourNodeID: pkt.NodeID,
YourNodeUUID: pkt.NodeUUID,
})
if err != nil {
......@@ -181,7 +181,7 @@ func IdentifyMe(link *NodeLink, nodeType NodeType /*XXX*/) (peerType NodeType, e
err = EncodeAndSend(conn, &RequestIdentification{
NodeType: nodeType,
NodeID: 0, // XXX
NodeUUID: 0, // XXX
Address: Address{}, // XXX
Name: "", // XXX cluster name ?
IdTimestamp: 0, // XXX
......
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