Merge pull request #1419 from seebs/discoFever

Performance improvements for disco branch

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)
 	}
 

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

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

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

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 {

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
 	}
 

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))]
+}

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