From 8b2d8295d1d8fa224f92bc68bd9236111d4f9cea Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 17 Jul 2018 08:08:51 -0500 Subject: [PATCH 1/9] rename cluster.coordinatorNode to unprotectedCoordinatorNode --- cluster.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/cluster.go b/cluster.go index 6699a5e56..c64ec3a7b 100644 --- a/cluster.go +++ b/cluster.go @@ -233,8 +233,8 @@ func newCluster() *cluster { } } -// coordinatorNode returns the coordinator node. -func (c *cluster) coordinatorNode() *Node { +// unprotectedCoordinatorNode returns the coordinator node. +func (c *cluster) unprotectedCoordinatorNode() *Node { return c.unprotectedNodeByID(c.Coordinator) } @@ -428,7 +428,7 @@ func (c *cluster) setNodeState(state string) error { // nolint: unparam } c.logger.Printf("Sending State %s (%s)", state, c.Coordinator) - if err := c.sendTo(c.coordinatorNode(), ns); err != nil { + if err := c.sendTo(c.unprotectedCoordinatorNode(), ns); err != nil { return fmt.Errorf("sending node state error: err=%s", err) } @@ -1101,7 +1101,7 @@ func (c *cluster) generateResizeJobByAction(nodeAction nodeAction) (*resizeJob, instr := &ResizeInstruction{ JobID: j.ID, Node: toCluster.unprotectedNodeByID(id), - Coordinator: c.coordinatorNode(), + Coordinator: c.unprotectedCoordinatorNode(), Sources: sources, Schema: &Schema{Indexes: c.holder.Schema()}, // Include the schema to ensure it's in sync on the receiving node. ClusterStatus: c.Status(), @@ -1629,7 +1629,7 @@ func (c *cluster) nodeJoin(node *Node) error { func (c *cluster) nodeLeave(node *Node) error { // Refuse the request if this is not the coordinator. if !c.isCoordinator() { - return fmt.Errorf("node removal requests are only valid on the coordinator node: %s", c.coordinatorNode().ID) + return fmt.Errorf("node removal requests are only valid on the coordinator node: %s", c.unprotectedCoordinatorNode().ID) } if c.State() != ClusterStateNormal { From 04bdc67d7fc4ba083703ea1d3b86a94a5aee6956 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 17 Jul 2018 11:58:19 -0500 Subject: [PATCH 2/9] rename a few things to unprotected* and use safe coordinatorNode in setNodeState --- cluster.go | 43 +++++++++++++++++++++++++------------------ 1 file changed, 25 insertions(+), 18 deletions(-) diff --git a/cluster.go b/cluster.go index c64ec3a7b..176f3e62e 100644 --- a/cluster.go +++ b/cluster.go @@ -233,6 +233,12 @@ func newCluster() *cluster { } } +func (c *cluster) coordinatorNode() *Node { + c.mu.RLock() + defer c.mu.RUnlock() + return c.unprotectedCoordinatorNode() +} + // unprotectedCoordinatorNode returns the coordinator node. func (c *cluster) unprotectedCoordinatorNode() *Node { return c.unprotectedNodeByID(c.Coordinator) @@ -428,7 +434,7 @@ func (c *cluster) setNodeState(state string) error { // nolint: unparam } c.logger.Printf("Sending State %s (%s)", state, c.Coordinator) - if err := c.sendTo(c.unprotectedCoordinatorNode(), ns); err != nil { + if err := c.sendTo(c.coordinatorNode(), ns); err != nil { return fmt.Errorf("sending node state error: err=%s", err) } @@ -453,7 +459,7 @@ func (c *cluster) receiveNodeState(nodeID string, state string) error { // Set cluster state to NORMAL. if c.haveTopologyAgreement() && c.allNodesReady() { - return c.setStateAndBroadcast(ClusterStateNormal) + return c.unprotectedSetStateAndBroadcast(ClusterStateNormal) } return nil @@ -914,7 +920,7 @@ func (c *cluster) handleNodeAction(nodeAction nodeAction) error { j, err := c.generateResizeJob(nodeAction) if err != nil { c.logger.Printf("generateResizeJob error: err=%s", err) - if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil { + if err := c.unprotectedSetStateAndBroadcast(ClusterStateNormal); err != nil { c.logger.Printf("setStateAndBroadcast error: err=%s", err) } return errors.Wrap(err, "setting state") @@ -957,14 +963,15 @@ func (c *cluster) handleNodeAction(nodeAction nodeAction) error { return nil } -func (c *cluster) setStateAndBroadcast(state string) error { - c.SetState(state) +func (c *cluster) unprotectedSetStateAndBroadcast(state string) error { + c.setState(state) if c.Static { return nil } // Broadcast cluster status changes to the cluster. c.logger.Printf("broadcasting ClusterStatus: %s", state) - return c.broadcaster.SendSync(c.Status()) + return c.broadcaster.SendSync(c.Status()) // TODO fix c.Status + } func (c *cluster) sendTo(node *Node, m Message) error { @@ -1005,7 +1012,7 @@ func (c *cluster) listenForJoins() { // Only change state to NORMAL if we have successfully added at least one host. if setNormal { // Put the cluster back to state NORMAL and broadcast. - if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil { + if err := c.unprotectedSetStateAndBroadcast(ClusterStateNormal); err != nil { c.logger.Printf("setStateAndBroadcast error: err=%s", err) } } @@ -1035,7 +1042,7 @@ func (c *cluster) generateResizeJob(nodeAction nodeAction) (*resizeJob, error) { c.mu.Lock() defer c.mu.Unlock() - j, err := c.generateResizeJobByAction(nodeAction) + j, err := c.unprotectedGenerateResizeJobByAction(nodeAction) if err != nil { return nil, errors.Wrap(err, "generating job") } @@ -1053,11 +1060,11 @@ func (c *cluster) generateResizeJob(nodeAction nodeAction) (*resizeJob, error) { return j, nil } -// generateResizeJobByAction returns a resizeJob with instructions based on +// unprotectedGenerateResizeJobByAction returns a resizeJob with instructions based on // the difference between Cluster and a new Cluster with/without uri. // Broadcaster is associated to the resizeJob here for use in broadcasting // the resize instructions to other nodes in the cluster. -func (c *cluster) generateResizeJobByAction(nodeAction nodeAction) (*resizeJob, error) { +func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*resizeJob, error) { j := newResizeJob(c.Nodes, nodeAction.node, nodeAction.action) j.Broadcaster = c.broadcaster @@ -1582,7 +1589,7 @@ func (c *cluster) nodeJoin(node *Node) error { // 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 c.setStateAndBroadcast(ClusterStateNormal) + return c.unprotectedSetStateAndBroadcast(ClusterStateNormal) } return nil } else if err != nil { @@ -1590,7 +1597,7 @@ func (c *cluster) nodeJoin(node *Node) error { } if c.haveTopologyAgreement() && c.allNodesReady() { - return c.setStateAndBroadcast(ClusterStateNormal) + return c.unprotectedSetStateAndBroadcast(ClusterStateNormal) } else { // Send the status to the remote node. This lets the remote node // know that it can proceed with opening its Holder. @@ -1610,14 +1617,14 @@ func (c *cluster) nodeJoin(node *Node) error { if err := c.addNode(node); err != nil { return errors.Wrap(err, "adding node") } - return c.setStateAndBroadcast(ClusterStateNormal) + return c.unprotectedSetStateAndBroadcast(ClusterStateNormal) } else if err != nil { return errors.Wrap(err, "checking if holder has data2") } // If the cluster has data, we need to change to RESIZING and // kick off the resizing process. - if err := c.setStateAndBroadcast(ClusterStateResizing); err != nil { + if err := c.unprotectedSetStateAndBroadcast(ClusterStateResizing); err != nil { return errors.Wrap(err, "broadcasting state") } c.joiningLeavingNodes <- nodeAction{node, resizeJobActionAdd} @@ -1628,7 +1635,7 @@ func (c *cluster) nodeJoin(node *Node) error { // nodeLeave initiates the removal of a node from the cluster. func (c *cluster) nodeLeave(node *Node) error { // Refuse the request if this is not the coordinator. - if !c.isCoordinator() { + if !c.unprotectedIsCoordinator() { return fmt.Errorf("node removal requests are only valid on the coordinator node: %s", c.unprotectedCoordinatorNode().ID) } @@ -1647,7 +1654,7 @@ func (c *cluster) nodeLeave(node *Node) error { } // See if resize job can be generated - if _, err := c.generateResizeJobByAction(nodeAction{c.unprotectedNodeByID(node.ID), resizeJobActionRemove}); err != nil { + if _, err := c.unprotectedGenerateResizeJobByAction(nodeAction{c.unprotectedNodeByID(node.ID), resizeJobActionRemove}); err != nil { return errors.Wrap(err, "generating job") } @@ -1664,14 +1671,14 @@ func (c *cluster) nodeLeave(node *Node) error { if err := c.removeNode(n); err != nil { return errors.Wrap(err, "removing node") } - return c.setStateAndBroadcast(ClusterStateNormal) + return c.unprotectedSetStateAndBroadcast(ClusterStateNormal) } else if err != nil { return errors.Wrap(err, "checking if holder has data") } // If the cluster has data then change state to RESIZING and // kick off the resizing process. - if err := c.setStateAndBroadcast(ClusterStateResizing); err != nil { + if err := c.unprotectedSetStateAndBroadcast(ClusterStateResizing); err != nil { return errors.Wrap(err, "broadcasting state") } c.joiningLeavingNodes <- nodeAction{n, resizeJobActionRemove} From 18321b88f9df7b0fb24ffd7c517d6fdb59db9fc7 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 17 Jul 2018 14:33:45 -0500 Subject: [PATCH 3/9] rename setID as unprotected and add some "unprotected" comments --- cluster.go | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/cluster.go b/cluster.go index 176f3e62e..06beebe9a 100644 --- a/cluster.go +++ b/cluster.go @@ -311,7 +311,7 @@ func (c *cluster) unprotectedUpdateCoordinator(n *Node) bool { } // addNode adds a node to the Cluster and updates and saves the -// new topology. +// new topology. unprotected. func (c *cluster) addNode(node *Node) error { c.logger.Printf("add node %s to cluster on %s", node, c.Node) @@ -338,7 +338,7 @@ func (c *cluster) addNode(node *Node) error { } // removeNode removes a node from the Cluster and updates and saves the -// new topology. +// new topology. unprotected. func (c *cluster) removeNode(node *Node) error { // remove from cluster if !c.removeNodeBasicSorted(node) { @@ -362,7 +362,7 @@ func (c *cluster) nodeIDs() []string { return Nodes(c.Nodes).IDs() } -func (c *cluster) setID(id string) { +func (c *cluster) unprotectedSetID(id string) { // Don't overwrite ClusterID. if c.id != "" { return @@ -500,8 +500,8 @@ func (c *cluster) nodePositionByID(nodeID string) int { return -1 } -// addNodeBasicSorted adds a node to the cluster, sorted by id. -// Returns a pointer to the node and true if the node was added. +// addNodeBasicSorted adds a node to the cluster, sorted by id. Returns a +// pointer to the node and true if the node was added. unprotected. func (c *cluster) addNodeBasicSorted(node *Node) bool { n := c.unprotectedNodeByID(node.ID) if n != nil { @@ -516,8 +516,8 @@ func (c *cluster) addNodeBasicSorted(node *Node) bool { return true } -// removeNodeBasicSorted removes a node from the cluster, maintaining -// the sort order. Returns true if the node was removed. +// removeNodeBasicSorted removes a node from the cluster, maintaining the sort +// order. Returns true if the node was removed. unprotected. func (c *cluster) removeNodeBasicSorted(node *Node) bool { i := c.nodePositionByID(node.ID) if i < 0 { @@ -1696,7 +1696,7 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error { } // Set ClusterID. - c.setID(cs.ClusterID) + c.unprotectedSetID(cs.ClusterID) officialNodes := cs.Nodes From 7efdacd0288a94889c8a9018dc7d3f5b3ab1497b Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 17 Jul 2018 14:34:36 -0500 Subject: [PATCH 4/9] rename setState to unprotected --- cluster.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/cluster.go b/cluster.go index 06beebe9a..16ff2fec2 100644 --- a/cluster.go +++ b/cluster.go @@ -381,11 +381,11 @@ func (c *cluster) State() string { func (c *cluster) SetState(state string) { c.mu.Lock() - c.setState(state) + c.unprotectedSetState(state) c.mu.Unlock() } -func (c *cluster) setState(state string) { +func (c *cluster) unprotectedSetState(state string) { // Ignore cases where the state hasn't changed. if state == c.state { return @@ -964,7 +964,7 @@ func (c *cluster) handleNodeAction(nodeAction nodeAction) error { } func (c *cluster) unprotectedSetStateAndBroadcast(state string) error { - c.setState(state) + c.unprotectedSetState(state) if c.Static { return nil } @@ -1728,7 +1728,7 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error { } } - c.setState(cs.State) + c.unprotectedSetState(cs.State) c.markAsJoined() From fa755fdd81d271318468e43255d44580828202e7 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 17 Jul 2018 16:32:17 -0500 Subject: [PATCH 5/9] fix ClusterCluster not to broadcast to self. stops deadlock when cluster has appropriate internal locking --- cluster.go | 14 ++++++++++++-- utils_internal_test.go | 42 +++++++++++++++++++++++++++--------------- 2 files changed, 39 insertions(+), 17 deletions(-) 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) } From e1e4df67bcec6eaed2a1541f7d804dbb0c0d8379 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Wed, 18 Jul 2018 16:47:24 -0500 Subject: [PATCH 6/9] fix race conds and add more locking/annotation --- cluster.go | 65 ++++++++++++++++++++++++++++++++++++------------------ 1 file changed, 44 insertions(+), 21 deletions(-) diff --git a/cluster.go b/cluster.go index f87632c97..85e5bddd5 100644 --- a/cluster.go +++ b/cluster.go @@ -281,7 +281,7 @@ func (c *cluster) setCoordinator(n *Node) error { } // Broadcast cluster status. - return c.broadcaster.SendSync(c.Status()) + return c.broadcaster.SendSync(c.status()) } // updateCoordinator updates this nodes Coordinator value as well as @@ -456,7 +456,9 @@ func (c *cluster) receiveNodeState(nodeID string, state string) error { return nil } + c.Topology.mu.Lock() c.Topology.nodeStates[nodeID] = state + c.Topology.mu.Unlock() c.logger.Printf("received state %s (%s)", state, nodeID) // Set cluster state to NORMAL. @@ -467,8 +469,14 @@ func (c *cluster) receiveNodeState(nodeID string, state string) error { return nil } -// Status returns the the cluster's status including what nodes it contains, its ID, and current state. -func (c *cluster) Status() *ClusterStatus { +func (c *cluster) status() *ClusterStatus { + c.mu.RLock() + defer c.mu.RUnlock() + return c.unprotectedStatus() +} + +// unprotectedStatus returns the the cluster's status including what nodes it contains, its ID, and current state. +func (c *cluster) unprotectedStatus() *ClusterStatus { return &ClusterStatus{ ClusterID: c.id, State: c.state, @@ -607,7 +615,7 @@ func (c *cluster) fragCombos(idx string, maxShard uint64, fieldViews viewsByFiel // diff compares c with another cluster and determines if a node is being // added or removed. An error is returned for any case other than where -// exactly one node is added or removed. +// exactly one node is added or removed. unprotected. func (c *cluster) diff(other *cluster) (action string, nodeID string, err error) { lenFrom := len(c.Nodes) lenTo := len(other.Nodes) @@ -646,7 +654,7 @@ func (c *cluster) diff(other *cluster) (action string, nodeID string, err error) } // fragSources returns a list of ResizeSources - for each node in the `to` cluster - -// required to move from cluster `c` to cluster `to`. +// required to move from cluster `c` to cluster `to`. unprotected. func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSource, error) { m := make(map[string][]*ResizeSource) @@ -745,7 +753,7 @@ func (c *cluster) partition(index string, shard uint64) int { return int(h.Sum64() % uint64(c.partitionN)) } -// shardNodes returns a list of nodes that own a fragment. +// shardNodes returns a list of nodes that own a fragment. unprotected func (c *cluster) shardNodes(index string, shard uint64) []*Node { return c.partitionNodes(c.partition(index, shard)) } @@ -755,7 +763,7 @@ func (c *cluster) ownsShard(nodeID string, index string, shard uint64) bool { return Nodes(c.shardNodes(index, shard)).ContainsID(nodeID) } -// partitionNodes returns a list of nodes that own a partition. +// partitionNodes returns a list of nodes that own a partition. unprotected. func (c *cluster) partitionNodes(partitionID int) []*Node { // Default replica count to between one and the number of nodes. // The replica count can be zero if there are no nodes. @@ -895,8 +903,9 @@ func (c *cluster) markAsJoined() { } } +// needTopologyAgreement is unprotected. func (c *cluster) needTopologyAgreement() bool { - return c.State() == ClusterStateStarting && !stringSlicesAreEqual(c.Topology.nodeIDs, c.nodeIDs()) + return c.state == ClusterStateStarting && !stringSlicesAreEqual(c.Topology.nodeIDs, c.nodeIDs()) } // haveTopologyAgreement is unprotected. @@ -921,10 +930,13 @@ func (c *cluster) allNodesReady() bool { } func (c *cluster) handleNodeAction(nodeAction nodeAction) error { - j, err := c.generateResizeJob(nodeAction) + + c.mu.Lock() + j, err := c.unprotectedGenerateResizeJob(nodeAction) + c.mu.Unlock() if err != nil { c.logger.Printf("generateResizeJob error: err=%s", err) - if err := c.unprotectedSetStateAndBroadcast(ClusterStateNormal); err != nil { + if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil { c.logger.Printf("setStateAndBroadcast error: err=%s", err) } return errors.Wrap(err, "setting state") @@ -955,8 +967,12 @@ func (c *cluster) handleNodeAction(nodeAction nodeAction) error { } // Add/remove uri to/from the cluster. if j.action == resizeJobActionRemove { + c.mu.Lock() + defer c.mu.Unlock() return c.removeNode(nodeAction.node) } else if j.action == resizeJobActionAdd { + c.mu.Lock() + defer c.mu.Unlock() return c.addNode(nodeAction.node) } case resizeJobStateAborted: @@ -980,7 +996,7 @@ func (c *cluster) unprotectedSetStateAndBroadcast(state string) error { } // Broadcast cluster status changes to the cluster. c.logger.Printf("broadcasting ClusterStatus: %s", state) - return c.broadcaster.SendSync(c.Status()) // TODO fix c.Status + return c.broadcaster.SendSync(c.unprotectedStatus()) // TODO fix c.Status } @@ -1022,7 +1038,7 @@ func (c *cluster) listenForJoins() { // Only change state to NORMAL if we have successfully added at least one host. if setNormal { // Put the cluster back to state NORMAL and broadcast. - if err := c.unprotectedSetStateAndBroadcast(ClusterStateNormal); err != nil { + if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil { c.logger.Printf("setStateAndBroadcast error: err=%s", err) } } @@ -1044,13 +1060,11 @@ func (c *cluster) listenForJoins() { }() } -// generateResizeJob creates a new resizeJob based on the new node being +// unprotectedGenerateResizeJob creates a new resizeJob based on the new node being // added/removed. It also saves a reference to the resizeJob in the `jobs` map // for future lookup by JobID. -func (c *cluster) generateResizeJob(nodeAction nodeAction) (*resizeJob, error) { +func (c *cluster) unprotectedGenerateResizeJob(nodeAction nodeAction) (*resizeJob, error) { c.logger.Printf("generateResizeJob: %v", nodeAction) - c.mu.Lock() - defer c.mu.Unlock() j, err := c.unprotectedGenerateResizeJobByAction(nodeAction) if err != nil { @@ -1121,7 +1135,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (* Coordinator: c.unprotectedCoordinatorNode(), Sources: sources, Schema: &Schema{Indexes: c.holder.Schema()}, // Include the schema to ensure it's in sync on the receiving node. - ClusterStatus: c.Status(), + ClusterStatus: c.unprotectedStatus(), } j.Instructions = append(j.Instructions, instr) } @@ -1134,6 +1148,10 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (* func (c *cluster) completeCurrentJob(state string) error { c.mu.Lock() defer c.mu.Unlock() + return c.unprotectedCompleteCurrentJob(state) +} + +func (c *cluster) unprotectedCompleteCurrentJob(state string) error { if !c.unprotectedIsCoordinator() { return ErrNodeNotCoordinator } @@ -1579,7 +1597,10 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) error { return nil } +// nodeJoin is unprotected. func (c *cluster) nodeJoin(node *Node) error { + c.mu.Lock() + defer c.mu.Unlock() 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) { @@ -1611,7 +1632,7 @@ func (c *cluster) nodeJoin(node *Node) error { } else { // 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.Status()) + return c.sendTo(node, c.unprotectedStatus()) } } @@ -1619,7 +1640,7 @@ func (c *cluster) nodeJoin(node *Node) error { // This is useful in the case where a node is restarted or temporarily leaves // the cluster. if node := c.unprotectedNodeByID(node.ID); node != nil { - return c.sendTo(node, c.Status()) + return c.sendTo(node, c.unprotectedStatus()) } // If the holder does not yet contain data, go ahead and add the node. @@ -1644,13 +1665,15 @@ func (c *cluster) nodeJoin(node *Node) error { // nodeLeave initiates the removal of a node from the cluster. func (c *cluster) nodeLeave(node *Node) error { + c.mu.Lock() + defer c.mu.Unlock() // Refuse the request if this is not the coordinator. if !c.unprotectedIsCoordinator() { return fmt.Errorf("node removal requests are only valid on the coordinator node: %s", c.unprotectedCoordinatorNode().ID) } - if c.State() != ClusterStateNormal { - return fmt.Errorf("Cluster must be in state %s to remove a node. Current state: %s", ClusterStateNormal, c.State()) + if c.state != ClusterStateNormal { + return fmt.Errorf("Cluster must be in state %s to remove a node. Current state: %s", ClusterStateNormal, c.state) } // Ensure that node is in the cluster. From 3b8b190849c5aa1e61f049c067fb08c323ae68e5 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 19 Jul 2018 09:17:36 -0500 Subject: [PATCH 7/9] add nolint unparam for setStateAndBroadcast --- cluster.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cluster.go b/cluster.go index f39924c59..7047a6d4a 100644 --- a/cluster.go +++ b/cluster.go @@ -980,7 +980,7 @@ func (c *cluster) handleNodeAction(nodeAction nodeAction) error { return nil } -func (c *cluster) setStateAndBroadcast(state string) error { +func (c *cluster) setStateAndBroadcast(state string) error { // nolint: unparam c.mu.Lock() defer c.mu.Unlock() return c.unprotectedSetStateAndBroadcast(state) From 774e91ad30053bf9c361ea623d9de25d82afafc9 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 19 Jul 2018 09:40:23 -0500 Subject: [PATCH 8/9] finish commenting methods as unprotected. --- cluster.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/cluster.go b/cluster.go index 7047a6d4a..1878e234c 100644 --- a/cluster.go +++ b/cluster.go @@ -1500,7 +1500,7 @@ func (t *Topology) encode() *internal.Topology { return encodeTopology(t) } -// loadTopology reads the topology for the node. +// loadTopology reads the topology for the node. unprotected. func (c *cluster) loadTopology() error { buf, err := ioutil.ReadFile(filepath.Join(c.Path, ".topology")) if os.IsNotExist(err) { @@ -1523,7 +1523,7 @@ func (c *cluster) loadTopology() error { return nil } -// saveTopology writes the current topology to disk. +// saveTopology writes the current topology to disk. unprotected. func (c *cluster) saveTopology() error { if err := os.MkdirAll(c.Path, 0777); err != nil { @@ -1765,6 +1765,8 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error { return nil } +// setStatic is unprotected, but only called before the cluster has been started +// (and therefore not concurrently). func (c *cluster) setStatic(hosts []string) error { c.Static = true c.Coordinator = c.Node.ID From 30e0d42c3d3ff89747403291eeaaa5433bf34fb3 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 19 Jul 2018 09:43:48 -0500 Subject: [PATCH 9/9] remove incorrect nodeJoin comment --- cluster.go | 1 - 1 file changed, 1 deletion(-) diff --git a/cluster.go b/cluster.go index 1878e234c..479047ca8 100644 --- a/cluster.go +++ b/cluster.go @@ -1594,7 +1594,6 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) error { return nil } -// nodeJoin is unprotected. func (c *cluster) nodeJoin(node *Node) error { c.mu.Lock() defer c.mu.Unlock()