[WIP] Do not merge - collecting performance patches for scale testing

client/client.go

@@ -91,7 +91,7 @@ const (
 
 	// allocSyncIntv is the batching period of allocation updates before they
 	// are synced with the server.
-	allocSyncIntv = 200 * time.Millisecond
+	allocSyncIntv = time.Second
 
 	// allocSyncRetryIntv is the interval on which we retry updating
 	// the status of the allocation
@@ -1904,7 +1904,6 @@ func (c *Client) AllocStateUpdated(alloc *structs.Allocation) {
 // allocSync is a long lived function that batches allocation updates to the
 // server.
 func (c *Client) allocSync() {
-	staggered := false
 	syncTicker := time.NewTicker(allocSyncIntv)
 	updates := make(map[string]*structs.Allocation)
 	for {
@@ -1933,19 +1932,23 @@ func (c *Client) allocSync() {
 			}
 
 			var resp structs.GenericResponse
-			if err := c.RPC("Node.UpdateAlloc", &args, &resp); err != nil {
+			err := c.RPC("Node.UpdateAlloc", &args, &resp)
+			if err != nil {
+				// Error updating allocations, do *not* clear
+				// updates and retry after backoff
 				c.logger.Error("error updating allocations", "error", err)
-				syncTicker.Stop()
-				syncTicker = time.NewTicker(c.retryIntv(allocSyncRetryIntv))
-				staggered = true
-			} else {
-				updates = make(map[string]*structs.Allocation)
-				if staggered {
-					syncTicker.Stop()
-					syncTicker = time.NewTicker(allocSyncIntv)
-					staggered = false
-				}
+				syncTicker.Reset(c.retryIntv(allocSyncRetryIntv))
+				continue
 			}
+
+			// Successfully updated allocs, reset map and ticker.
+			// Always reset ticker to give loop time to receive
+			// alloc updates. If the RPC took the ticker interval
+			// we may call it in a tight loop before draining
+			// buffered updates.
+			updates = make(map[string]*structs.Allocation, len(updates))
+			syncTicker.Stop()
+			syncTicker = time.NewTicker(allocSyncIntv)
 		}
 	}
 }

nomad/deploymentwatcher/deployment_watcher.go

@@ -873,6 +873,10 @@ func (w *deploymentWatcher) getAllocsCh(index uint64) <-chan *allocUpdates {
 // getAllocs retrieves the allocations that are part of the deployment blocking
 // at the given index.
 func (w *deploymentWatcher) getAllocs(index uint64) ([]*structs.AllocListStub, uint64, error) {
+	if err := w.queryLimiter.Wait(w.ctx); err != nil {
+		return nil, 0, err
+	}
+
 	resp, index, err := w.state.BlockingQuery(w.getAllocsImpl, index, w.ctx)
 	if err != nil {
 		return nil, 0, err
@@ -886,10 +890,6 @@ func (w *deploymentWatcher) getAllocs(index uint64) ([]*structs.AllocListStub, u
 
 // getDeploysImpl retrieves all deployments from the passed state store.
 func (w *deploymentWatcher) getAllocsImpl(ws memdb.WatchSet, state *state.StateStore) (interface{}, uint64, error) {
-	if err := w.queryLimiter.Wait(w.ctx); err != nil {
-		return nil, 0, err
-	}
-
 	// Capture all the allocations
 	allocs, err := state.AllocsByDeployment(ws, w.deploymentID)
 	if err != nil {

nomad/node_endpoint.go

@@ -1144,8 +1144,13 @@ func (n *Node) UpdateAlloc(args *structs.AllocUpdateRequest, reply *structs.Gene
 			updates := n.updates
 			evals := n.evals
 			future := n.updateFuture
-			n.updates = nil
-			n.evals = nil
+
+			// Assume future update patterns will be similar to
+			// current batch and set cap appropriately to avoid
+			// slice resizing.
+			n.updates = make([]*structs.Allocation, 0, len(updates))
+			n.evals = make([]*structs.Evaluation, 0, len(evals))
+
 			n.updateFuture = nil
 			n.updateTimer = nil
 			n.updatesLock.Unlock()

nomad/structs/network.go

@@ -70,6 +70,9 @@ func (idx *NetworkIndex) getUsedPortsFor(ip string) Bitmap {
 // Release is called when the network index is no longer needed
 // to attempt to re-use some of the memory it has allocated
 func (idx *NetworkIndex) Release() {
+	if idx == nil {
+		return
+	}
 	for _, b := range idx.UsedPorts {
 		bitmapPool.Put(b)
 	}

scheduler/rank.go

@@ -3,6 +3,7 @@ package scheduler
 import (
 	"fmt"
 	"math"
+	"sync"
 
 	"github.com/hashicorp/nomad/nomad/structs"
 )
@@ -202,13 +203,21 @@ OUTER:
 		}
 
 		// Index the existing network usage
-		netIdx := structs.NewNetworkIndex()
-		netIdx.SetNode(option.Node)
-		netIdx.AddAllocs(proposed)
+		var netIdx *structs.NetworkIndex
+		initNetIdx := func() {
+			netIdx = structs.NewNetworkIndex()
+			netIdx.SetNode(option.Node)
+			netIdx.AddAllocs(proposed)
+		}
+		var initNetIdxOnce sync.Once
 
 		// Create a device allocator
-		devAllocator := newDeviceAllocator(iter.ctx, option.Node)
-		devAllocator.AddAllocs(proposed)
+		var devAllocator *deviceAllocator
+		initDevAllocator := func() {
+			devAllocator = newDeviceAllocator(iter.ctx, option.Node)
+			devAllocator.AddAllocs(proposed)
+		}
+		var initDevAllocatorOnce sync.Once
 
 		// Track the affinities of the devices
 		totalDeviceAffinityWeight := 0.0
@@ -241,6 +250,7 @@ OUTER:
 
 		// Check if we need task group network resource
 		if len(iter.taskGroup.Networks) > 0 {
+			initNetIdxOnce.Do(initNetIdx)
 			ask := iter.taskGroup.Networks[0].Copy()
 			offer, err := netIdx.AssignPorts(ask)
 			if err != nil {
@@ -308,6 +318,7 @@ OUTER:
 
 			// Check if we need a network resource
 			if len(task.Resources.Networks) > 0 {
+				initNetIdxOnce.Do(initNetIdx)
 				ask := task.Resources.Networks[0].Copy()
 				offer, err := netIdx.AssignNetwork(ask)
 				if offer == nil {
@@ -356,6 +367,7 @@ OUTER:
 
 			// Check if we need to assign devices
 			for _, req := range task.Resources.Devices {
+				initDevAllocatorOnce.Do(initDevAllocator)
 				offer, sumAffinities, err := devAllocator.AssignDevice(req)
 				if offer == nil {
 					// If eviction is not enabled, mark this node as exhausted and continue