From 0790fbe866e0137d8b265072920d92060b23ec15 Mon Sep 17 00:00:00 2001 From: Seebs Date: Fri, 12 Feb 2021 15:12:30 -0600 Subject: [PATCH 1/4] only persist views to etcd when they're not already known Persisting views to etcd every time we check for them causes what ends up being about a factor of 60 slowdown. Let's do it a little less. --- field.go | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/field.go b/field.go index 34a6aa72e..0bfdede5d 100644 --- a/field.go +++ b/field.go @@ -1187,6 +1187,12 @@ func (f *Field) createViewIfNotExistsBase(cvm *CreateViewMessage) (*view, bool, f.mu.Lock() defer f.mu.Unlock() + // If we already have this view, we can probably assume etcd already + // has it. + if view := f.viewMap[cvm.View]; view != nil { + return view, false, nil + } + // Create the view in etcd as the system of record. // Don't persist views related to the existence field. if f.name != existenceFieldName { @@ -1195,9 +1201,6 @@ func (f *Field) createViewIfNotExistsBase(cvm *CreateViewMessage) (*view, bool, } } - if view := f.viewMap[cvm.View]; view != nil { - return view, false, nil - } view := f.newView(f.viewPath(cvm.View), cvm.View) if err := view.openEmpty(); err != nil { From e2b6912d1cde480005123c0d6df633f72a8abbe2 Mon Sep 17 00:00:00 2001 From: Seebs Date: Fri, 12 Feb 2021 15:30:54 -0600 Subject: [PATCH 2/4] uninvert test for coordinator node / primary field translation node If we're the primary field translation node, we don't need to set up translation replication; we only need that if we're *not*. So it makes sense to test if !IsPrimaryFieldTranslationNode... except that the test is to determine whether to return early. So it should not be inverted. --- holder.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/holder.go b/holder.go index 91d0e1211..c6c9977c7 100644 --- a/holder.go +++ b/holder.go @@ -1947,7 +1947,7 @@ func (s *holderSyncer) initializeIndexTranslateReplication(snap *topology.Cluste // initializeFieldTranslateReplication connects the coordinator to stream field data. func (s *holderSyncer) initializeFieldTranslateReplication(snap *topology.ClusterSnapshot) error { // Skip if coordinator. - if !snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID) { + if snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID) { return nil } From 07a014e9523fdd51a9096a25e519c0c16cf76910 Mon Sep 17 00:00:00 2001 From: Seebs Date: Fri, 12 Feb 2021 15:46:43 -0600 Subject: [PATCH 3/4] cache Nodes calls in EtcdWithCache The Peers() data is cached, but then every call still has to unmarshal JSON and that's stunningly expensive. Let's not! --- etcd/cache.go | 36 ++++++++++++++---------------------- 1 file changed, 14 insertions(+), 22 deletions(-) diff --git a/etcd/cache.go b/etcd/cache.go index 47543779a..dc7b8a539 100644 --- a/etcd/cache.go +++ b/etcd/cache.go @@ -16,9 +16,6 @@ package etcd import ( "context" - "encoding/json" - "log" - "sort" "sync" "time" @@ -35,8 +32,12 @@ type EtcdWithCache struct { peerMetadataMu sync.RWMutex peerMetadata map[string][]byte - stateMu sync.Mutex + stateMu sync.Mutex // cluster state cache updates + peersMu sync.Mutex // peer-list cache updates + nodes []*topology.Node // unmarshalled Node data + nodesTTL int // seconds + nodesLastRequest time.Time // last time requested nodeStates map[string]nodeState nodeStateTTL int // seconds nodeStateFrequency int // max requests per second allowed before using the cache @@ -63,6 +64,7 @@ func NewEtcdWithCache(opt Options, replicas int) *EtcdWithCache { nodeStateFrequency: 1, clusterStateTTL: 6, clusterStateFrequency: 1, + nodesTTL: 6, peerMetadata: make(map[string][]byte), nodeStates: make(map[string]nodeState), @@ -151,27 +153,17 @@ func (c *EtcdWithCache) NodeState(ctx context.Context, peerID string) (disco.Nod return ns.val, nil } -// Nodes implements the Noder interface. +// Nodes caches the result of the underlying implementation's node list. func (c *EtcdWithCache) Nodes() []*topology.Node { - peers := c.Peers() - nodes := make([]*topology.Node, len(peers)) - for i, peer := range peers { - node := &topology.Node{} - if meta, err := c.Metadata(context.Background(), peer.ID); err != nil { - log.Println(err, "getting metadata") // TODO: handle this with a logger - } else if err := json.Unmarshal(meta, node); err != nil { - log.Println(err, "unmarshaling json metadata") - } + c.peersMu.Lock() + defer c.peersMu.Unlock() - node.ID = peer.ID - - nodes[i] = node + now := time.Now() + if now.Sub(c.nodesLastRequest) > (time.Duration(c.nodesTTL) * time.Second) { + c.nodes = c.Etcd.Nodes() + c.nodesLastRequest = now } - - // Nodes must be sorted. - sort.Sort(topology.ByID(nodes)) - - return nodes + return c.nodes } // SetNodes implements the Noder interface as NOP From 82a975fd2a5f89e36484e5c3414ede4e1b52dffc Mon Sep 17 00:00:00 2001 From: Seebs Date: Fri, 12 Feb 2021 17:20:59 -0600 Subject: [PATCH 4/4] 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