diff --git a/cluster.go b/cluster.go index 16ff2fec2..f87632c97 100644 --- a/cluster.go +++ b/cluster.go @@ -445,12 +445,14 @@ func (c *cluster) setNodeState(state string) error { // nolint: unparam // Coordinator to keep track of, during startup, which nodes have // finished opening their Holder. func (c *cluster) receiveNodeState(nodeID string, state string) error { - if !c.isCoordinator() { + c.mu.Lock() + defer c.mu.Unlock() + if !c.unprotectedIsCoordinator() { return nil } // This method is really only useful during initial startup. - if c.State() != ClusterStateStarting { + if c.state != ClusterStateStarting { return nil } @@ -897,6 +899,7 @@ func (c *cluster) needTopologyAgreement() bool { return c.State() == ClusterStateStarting && !stringSlicesAreEqual(c.Topology.nodeIDs, c.nodeIDs()) } +// haveTopologyAgreement is unprotected. func (c *cluster) haveTopologyAgreement() bool { if c.Static { return true @@ -904,6 +907,7 @@ func (c *cluster) haveTopologyAgreement() bool { return stringSlicesAreEqual(c.Topology.nodeIDs, c.nodeIDs()) } +// allNodesReady is unprotected. func (c *cluster) allNodesReady() bool { if c.Static { return true @@ -963,6 +967,12 @@ func (c *cluster) handleNodeAction(nodeAction nodeAction) error { return nil } +func (c *cluster) setStateAndBroadcast(state string) error { + c.mu.Lock() + defer c.mu.Unlock() + return c.unprotectedSetStateAndBroadcast(state) +} + func (c *cluster) unprotectedSetStateAndBroadcast(state string) error { c.unprotectedSetState(state) if c.Static { diff --git a/utils_internal_test.go b/utils_internal_test.go index bf2c4f01f..85a225c34 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -229,7 +229,7 @@ func (t *ClusterCluster) addCluster(i int, saveTopology bool) (*cluster, error) c.holder = h c.Node = node c.Coordinator = t.common.Nodes[0].ID // the first node is the coordinator - c.broadcaster = t + c.broadcaster = t.broadcaster(c) // add nodes if saveTopology { @@ -302,39 +302,51 @@ func (t *ClusterCluster) Close() error { return nil } -// SendSync is a test implemenetation of Broadcaster SendSync method. -func (t *ClusterCluster) SendSync(m Message) error { +type bcast struct { + t *ClusterCluster + c *cluster +} + +func (b bcast) SendSync(m Message) error { switch obj := m.(type) { case *ClusterStatus: // Apply the send message to all nodes (except the coordinator). - for _, c := range t.Clusters { - c.mergeClusterStatus(obj) + for _, c := range b.t.Clusters { + if c != b.c { + c.mergeClusterStatus(obj) + } } - t.mu.RLock() - if obj.State == ClusterStateNormal && t.resizing { - close(t.resizeDone) + b.t.mu.RLock() + if obj.State == ClusterStateNormal && b.t.resizing { + close(b.t.resizeDone) } - t.mu.RUnlock() + b.t.mu.RUnlock() } - return nil } +func (t *ClusterCluster) broadcaster(c *cluster) broadcaster { + return bcast{ + t: t, + c: c, + } +} + // SendAsync is a test implemenetation of Broadcaster SendAsync method. -func (t *ClusterCluster) SendAsync(Message) error { +func (bcast) SendAsync(Message) error { return nil } // SendTo is a test implemenetation of Broadcaster SendTo method. -func (t *ClusterCluster) SendTo(to *Node, m Message) error { +func (b bcast) SendTo(to *Node, m Message) error { switch obj := m.(type) { case *ResizeInstruction: - err := t.FollowResizeInstruction(obj) + err := b.t.FollowResizeInstruction(obj) if err != nil { return err } case *ResizeInstructionComplete: - coord := t.clusterByID(to.ID) + coord := b.t.clusterByID(to.ID) go coord.markResizeInstructionComplete(obj) } return nil @@ -404,5 +416,5 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *ResizeInstruction) error } node := instr.Coordinator - return t.SendTo(node, complete) + return bcast{t: t}.SendTo(node, complete) }