From 82a975fd2a5f89e36484e5c3414ede4e1b52dffc Mon Sep 17 00:00:00 2001 From: Seebs Date: Fri, 12 Feb 2021 17:20:59 -0600 Subject: [PATCH] avoid race conditions on nodes Turns out we sometimes modify returned nodes. Handle this better, but also fix up some cases where we were generating node lists we didn't really need to answer simple questions. --- api.go | 22 ++++++++++++++-------- cluster.go | 22 +++++++++++++--------- etcd/embed.go | 4 +++- topology/hasher.go | 10 ++++++++++ topology/snapshot.go | 21 +++++++++++++++------ 5 files changed, 55 insertions(+), 24 deletions(-) diff --git a/api.go b/api.go index 417b45dfc..cf5431198 100644 --- a/api.go +++ b/api.go @@ -605,8 +605,8 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) // Validate that this handler owns the shard. - if !snap.OwnsShard(api.Node().ID, indexName, shard) { - api.server.logger.Printf("node %s does not own shard %d of index %s", api.Node().ID, shard, indexName) + if !snap.OwnsShard(api.NodeID(), indexName, shard) { + api.server.logger.Printf("node %s does not own shard %d of index %s", api.NodeID(), shard, indexName) return ErrClusterDoesNotOwnShard } @@ -807,6 +807,12 @@ func (api *API) Node() *topology.Node { return api.server.node() } +// NodeID gets the ID alone, so it doesn't have to do a complete lookup +// of the node, searching by its ID, to return the ID it searched for. +func (api *API) NodeID() string { + return api.server.nodeID +} + // PrimaryNode returns the coordinator node for the cluster. func (api *API) PrimaryNode() *topology.Node { // Create a snapshot of the cluster to use for node/partition calculations. @@ -1735,8 +1741,8 @@ func (api *API) validateShardOwnership(indexName string, shard uint64) error { // Create a snapshot of the cluster to use for node/partition calculations. snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) // Validate that this handler owns the shard. - if !snap.OwnsShard(api.Node().ID, indexName, shard) { - api.server.logger.Printf("node %s does not own shard %d of index %s", api.Node().ID, shard, indexName) + if !snap.OwnsShard(api.NodeID(), indexName, shard) { + api.server.logger.Printf("node %s does not own shard %d of index %s", api.NodeID(), shard, indexName) return ErrClusterDoesNotOwnShard } return nil @@ -2026,7 +2032,7 @@ func (api *API) PrimaryReplicaNodeURL() url.URL { // Create a snapshot of the cluster to use for node/partition calculations. snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) - node := snap.PrimaryReplicaNode(api.Node().ID) + node := snap.PrimaryReplicaNode(api.NodeID()) if node == nil { return url.URL{} } @@ -2137,7 +2143,7 @@ func (api *API) ReserveIDs(key IDAllocKey, session [32]byte, offset uint64, coun // Create a snapshot of the cluster to use for node/partition calculations. snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) - if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) { + if !snap.IsPrimaryFieldTranslationNode(api.NodeID()) { return api.holder.ida.reserve(key, session, offset, count) } @@ -2152,7 +2158,7 @@ func (api *API) CommitIDs(key IDAllocKey, session [32]byte, count uint64) error // Create a snapshot of the cluster to use for node/partition calculations. snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) - if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) { + if !snap.IsPrimaryFieldTranslationNode(api.NodeID()) { return api.holder.ida.commit(key, session, count) } @@ -2167,7 +2173,7 @@ func (api *API) ResetIDAlloc(index string) error { // Create a snapshot of the cluster to use for node/partition calculations. snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) - if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) { + if !snap.IsPrimaryFieldTranslationNode(api.NodeID()) { return api.holder.ida.reset(index) } diff --git a/cluster.go b/cluster.go index 715bcea86..37ba01b5f 100644 --- a/cluster.go +++ b/cluster.go @@ -709,25 +709,29 @@ func (c *cluster) addNodeBasicSorted(node *topology.Node) bool { // concurrent use, result may be modified. func (c *cluster) Nodes() []*topology.Node { nodes := c.noder.Nodes() + // duplicate the nodes since we're going to be altering them + copiedNodes := make([]topology.Node, len(nodes)) + result := make([]*topology.Node, len(nodes)) // Create a snapshot of the cluster to use for node/partition calculations. - snap := topology.NewClusterSnapshot(topology.NewLocalNoder(nodes), c.Hasher, c.ReplicaN) - primaryNode := snap.PrimaryFieldTranslationNode() + primary := topology.PrimaryNode(nodes, c.Hasher) // Set node states and IsPrimary. - for _, node := range nodes { - node.IsPrimary = node.ID == primaryNode.ID - + for i, node := range nodes { + copiedNodes[i] = *node + result[i] = &copiedNodes[i] + if node == primary { + copiedNodes[i].IsPrimary = true + } s, err := c.stator.NodeState(context.Background(), node.ID) if err != nil { // TODO should we delete this? - node.State = string(disco.NodeStateUnknown) + copiedNodes[i].State = string(disco.NodeStateUnknown) continue } - node.State = string(s) + copiedNodes[i].State = string(s) } - - return nodes + return result } // removeNodeBasicSorted removes a node from the cluster, maintaining the sort diff --git a/etcd/embed.go b/etcd/embed.go index 09bd82608..dd1062285 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -1127,9 +1127,11 @@ func (e *Etcd) RemoveShard(ctx context.Context, index, field string, shard uint6 // based on the etcd peers. func (e *Etcd) Nodes() []*topology.Node { peers := e.Peers() + // For N>1, this might actually reduce GC load. Maybe. + nodeData := make([]topology.Node, len(peers)) nodes := make([]*topology.Node, len(peers)) for i, peer := range peers { - node := &topology.Node{} + node := &nodeData[i] if meta, err := e.Metadata(context.Background(), peer.ID); err != nil { log.Println(err, "getting metadata") // TODO: handle this with a logger diff --git a/topology/hasher.go b/topology/hasher.go index a5c3f5964..41cd36b95 100644 --- a/topology/hasher.go +++ b/topology/hasher.go @@ -39,3 +39,13 @@ func (h *Jmphasher) Hash(key uint64, n int) int { func (h *Jmphasher) Name() string { return "jump-hash" } + +// PrimaryNode yields the node that would be selected as the primary from +// a list, for a given ID. It assumes the list is already in the +// expected order, as from Noder.Nodes(). +func PrimaryNode(nodes []*Node, hasher Hasher) *Node { + if len(nodes) == 0 { + return nil + } + return nodes[hasher.Hash(0, len(nodes))] +} diff --git a/topology/snapshot.go b/topology/snapshot.go index b7d865335..8ad6305c6 100644 --- a/topology/snapshot.go +++ b/topology/snapshot.go @@ -107,8 +107,14 @@ func (c *ClusterSnapshot) ShardNodes(index string, shard uint64) []*Node { } // OwnsShard returns true if a host owns a fragment. -func (c *ClusterSnapshot) OwnsShard(nodeID string, index string, shard uint64) bool { - return Nodes(c.ShardNodes(index, shard)).ContainsID(nodeID) +func (c *ClusterSnapshot) OwnsShard(nodeID string, index string, shard uint64) (ret bool) { + idx := c.Hasher.Hash(uint64(c.ShardToShardPartition(index, shard)), len(c.Nodes)) + for i := 0; i < c.ReplicaN; i++ { + if c.Nodes[(idx+i)%len(c.Nodes)].ID == nodeID { + return true + } + } + return false } // KeyNodes returns a list of nodes that own a key. @@ -147,11 +153,14 @@ func (c *ClusterSnapshot) IsPrimaryFieldTranslationNode(nodeID string) bool { } // PrimaryPartitionNode returns the primary node of the given partition. -func (c *ClusterSnapshot) PrimaryPartitionNode(partition int) *Node { - if nodes := c.PartitionNodes(partition); len(nodes) > 0 { - return nodes[0] +func (c *ClusterSnapshot) PrimaryPartitionNode(partitionID int) *Node { + // Determine primary owner node. + nodeIndex := c.PrimaryNodeIndex(partitionID) + if nodeIndex < 0 { + // no nodes anyway + return nil } - return nil + return c.Nodes[nodeIndex] } // IsPrimary returns true if the given node is the primary for the given