mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 17:15:56 +00:00
remove ReceiveEvent
This commit is contained in:
Travis
parent
652014539c
commit
afc53e1163
4 changed files with 0 additions and 215 deletions
152
cluster.go
152
cluster.go
|
|
@ -923,37 +923,6 @@ func (c *cluster) markAsJoined() {
|
|||
}
|
||||
}
|
||||
|
||||
// needTopologyAgreement is unprotected.
|
||||
func (c *cluster) needTopologyAgreement() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// haveTopologyAgreement is unprotected.
|
||||
func (c *cluster) haveTopologyAgreement() bool {
|
||||
if c.Static {
|
||||
return true
|
||||
}
|
||||
return stringSlicesAreEqual(c.Topology.nodeIDs, c.nodeIDs())
|
||||
}
|
||||
|
||||
// allNodesReady is unprotected.
|
||||
func (c *cluster) allNodesReady() (ret bool) {
|
||||
if c.Static {
|
||||
return true
|
||||
}
|
||||
nodeStates, err := c.stator.NodeStates(context.TODO())
|
||||
if err != nil {
|
||||
c.logger.Printf("getting node states error: %v", err)
|
||||
return false
|
||||
}
|
||||
for _, s := range nodeStates {
|
||||
if s != disco.NodeStateStarted {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (c *cluster) sendTo(node *topology.Node, m Message) error {
|
||||
if err := c.broadcaster.SendTo(node, m); err != nil {
|
||||
return errors.Wrap(err, "sending")
|
||||
|
|
@ -1659,127 +1628,6 @@ func (c *cluster) confirmNodeDown(uri pnet.URI) bool {
|
|||
return true
|
||||
}
|
||||
|
||||
// ReceiveEvent represents an implementation of EventHandler.
|
||||
func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) {
|
||||
// Ignore events sent from this node.
|
||||
if e.Node.ID == c.Node.ID {
|
||||
return nil
|
||||
}
|
||||
switch e.Event {
|
||||
case NodeJoin:
|
||||
// Ignore the event if this is not the coordinator.
|
||||
if !c.isCoordinator() {
|
||||
return nil
|
||||
}
|
||||
return c.nodeJoin(e.Node)
|
||||
case NodeLeave:
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if c.unprotectedIsCoordinator() {
|
||||
c.logger.Printf("received node leave: %v", e.Node)
|
||||
// if removeNodeBasicSorted succeeds, that means that the node was
|
||||
// not already removed by a removeNode request. We treat this as the
|
||||
// host being temporarily unavailable, and expect it to come back
|
||||
// up.
|
||||
if c.confirmNodeDown(e.Node.URI) {
|
||||
if c.removeNodeBasicSorted(e.Node.ID) {
|
||||
c.Topology.nodeStates[e.Node.ID] = nodeStateDown
|
||||
// put the cluster into STARTING if we've lost a number of nodes
|
||||
// equal to or greater than ReplicaN
|
||||
}
|
||||
} else {
|
||||
c.logger.Printf("ignored received node leave: %v", e.Node)
|
||||
}
|
||||
}
|
||||
case NodeUpdate:
|
||||
c.logger.Printf("received node update event: id: %v, string: %v, uri: %v", e.Node.ID, e.Node.String(), e.Node.URI)
|
||||
// NodeUpdate is intentionally not implemented.
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// nodeJoin should only be called by the coordinator.
|
||||
func (c *cluster) nodeJoin(node *topology.Node) error {
|
||||
c.abortAntiEntropy()
|
||||
// Technically there is a race condition here which could
|
||||
// allow the anti-entropy process to re-start (and acquire
|
||||
// the lock) before this lock has time to succeed. In that
|
||||
// case, the user would have to wait through an entire
|
||||
// anti-entropy cycle. We decided it wasn't worth the
|
||||
// complexity (of, for example, implementing this with
|
||||
// channels) to avoid that rare case.
|
||||
c.muAntiEntropy.Lock()
|
||||
defer c.muAntiEntropy.Unlock()
|
||||
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.logger.Printf("node join event on coordinator, node: %s, id: %s", node.URI, node.ID)
|
||||
if c.needTopologyAgreement() {
|
||||
// A host that is not part of the topology can't be added to the STARTING cluster.
|
||||
if !c.Topology.ContainsID(node.ID) {
|
||||
err := fmt.Sprintf("host is not in topology: %s", node.ID)
|
||||
c.logger.Printf("%v", err)
|
||||
return errors.New(err)
|
||||
}
|
||||
|
||||
if err := c.addNode(node); err != nil {
|
||||
return errors.Wrap(err, "adding node for agreement")
|
||||
}
|
||||
|
||||
// Only change to normal if there is no existing data. Otherwise,
|
||||
// the coordinator needs to wait to receive READY messages (nodeStates)
|
||||
// from remote nodes before setting the cluster to state NORMAL.
|
||||
if ok, err := c.holder.HasData(); !ok && err == nil {
|
||||
// If the result of the previous AddNode completed the joining of nodes
|
||||
// in the topology, then change the state to NORMAL.
|
||||
if c.haveTopologyAgreement() {
|
||||
return nil
|
||||
}
|
||||
// This lets the remote node to proceed with opening its holder,
|
||||
// instead of waiting in DOWN state because cluster is in STARTING state.
|
||||
return c.sendTo(node, c.unprotectedStatus())
|
||||
} else if err != nil {
|
||||
return errors.Wrap(err, "checking if holder has data")
|
||||
}
|
||||
|
||||
if c.haveTopologyAgreement() && c.allNodesReady() {
|
||||
return nil
|
||||
}
|
||||
// Send the status to the remote node. This lets the remote node
|
||||
// know that it can proceed with opening its Holder.
|
||||
return c.sendTo(node, c.unprotectedStatus())
|
||||
}
|
||||
|
||||
// If the cluster already contains the node, just send it the cluster status.
|
||||
// This is useful in the case where a node is restarted or temporarily leaves
|
||||
// the cluster.
|
||||
if cnode := c.unprotectedNodeByID(node.ID); cnode != nil {
|
||||
if cnode.URI != node.URI {
|
||||
c.logger.Printf("node: %v changed URI from %s to %s", cnode.ID, cnode.URI, node.URI)
|
||||
cnode.URI = node.URI
|
||||
}
|
||||
if cnode.GRPCURI != node.GRPCURI {
|
||||
cnode.GRPCURI = node.GRPCURI
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// If the holder does not yet contain data, go ahead and add the node.
|
||||
if ok, err := c.holder.HasData(); !ok && err == nil {
|
||||
if err := c.addNode(node); err != nil {
|
||||
return errors.Wrap(err, "adding node")
|
||||
}
|
||||
return nil
|
||||
} else if err != nil {
|
||||
return errors.Wrap(err, "checking if holder has data2")
|
||||
}
|
||||
|
||||
c.joiningLeavingNodes <- nodeAction{node, resizeJobActionAdd}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// nodeLeave initiates the removal of a node from the cluster.
|
||||
func (c *cluster) nodeLeave(nodeID string) error {
|
||||
c.abortAntiEntropy()
|
||||
|
|
|
|||
24
pilosa.go
24
pilosa.go
|
|
@ -181,30 +181,6 @@ func validateName(name string) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// stringSlicesAreEqual determines if two string slices are equal.
|
||||
func stringSlicesAreEqual(a, b []string) bool {
|
||||
|
||||
if a == nil && b == nil {
|
||||
return true
|
||||
}
|
||||
|
||||
if a == nil || b == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
if len(a) != len(b) {
|
||||
return false
|
||||
}
|
||||
|
||||
for i := range a {
|
||||
if a[i] != b[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func timestamp() int64 {
|
||||
return time.Now().UnixNano()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -863,11 +863,6 @@ func (s *Server) receiveMessage(m Message) error {
|
|||
}
|
||||
case *RecalculateCaches:
|
||||
s.holder.recalculateCaches()
|
||||
case *NodeEvent:
|
||||
err := s.cluster.ReceiveEvent(obj)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "cluster receiving NodeEvent %v", obj)
|
||||
}
|
||||
case *NodeStatus:
|
||||
s.handleRemoteStatus(obj)
|
||||
case *TransactionMessage:
|
||||
|
|
|
|||
|
|
@ -211,40 +211,6 @@ func (t *ClusterCluster) clusterByID(id string) *cluster {
|
|||
|
||||
// addNode adds a node to the cluster and (potentially) starts a resize job.
|
||||
func (t *ClusterCluster) addNode() error {
|
||||
id := len(t.Clusters)
|
||||
|
||||
c, err := t.addCluster(id, false)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Send NodeJoin event to coordinator.
|
||||
if id > 0 {
|
||||
coord := t.Clusters[0]
|
||||
ev := &NodeEvent{
|
||||
Event: NodeJoin,
|
||||
Node: c.Node,
|
||||
}
|
||||
|
||||
if err := coord.ReceiveEvent(ev); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
state, err := coord.State()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Wait for the AddNode job to finish.
|
||||
if state != string(ClusterStateNormal) {
|
||||
t.resizeDone = make(chan struct{})
|
||||
t.mu.Lock()
|
||||
t.resizing = true
|
||||
t.mu.Unlock()
|
||||
<-t.resizeDone
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue