From 4e55a1fd7325513f3fdf84665852a5349c47a50f Mon Sep 17 00:00:00 2001
From: Matt Jaffee <matthew.jaffee@gmail.com>
Date: Thu, 11 Jul 2019 12:44:18 -0500
Subject: [PATCH] add worker pool to executor for local query processing

Pilosa previously spawned a goroutine for each remote node that a
query needed to be forwarded to, and then forwarded a single request
containing all the shards that the query should operate on. It then
spawned a goroutine *per local shard* to process the query
locally. This was fine if there weren't too many shards, or too many
queries coming in concurrently, but we found that it created issues
when there were 100s or 1000s of shards per node, and dozens of
queries arriving concurrently.

Specifically, the memberlist "hiccup" issue is highly correlated with
many goroutine scenarios, and after applying this patch, memberlist
complaints in the logs were much decreased, and nodeLeave events under
concurrent query load almost entirely eliminated.

This patch creates a fixed size pool of goroutines to do local shard
processing, and passes work to them through a channel, one job per
query per shard. Handling of remote requests (forwarding queries) is
unchanged.

We set the pool size to NumCPU()+8 somewhat arbitrarily, but this
seemed to work pretty well in our testing on 32 core machines. It's a
pretty big improvement over launching a goroutine per shard per query
which is what we were doing previously, so we can tune it more later
if necessary.
---
 executor.go | 40 +++++++++++++++++++++++++++++++---------
 1 file changed, 31 insertions(+), 9 deletions(-)

diff --git a/executor.go b/executor.go
index 0052e4705..3b0270ebd 100644
--- a/executor.go
+++ b/executor.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"runtime"
 	"sort"
 	"time"
 
@@ -55,6 +56,8 @@ type executor struct {
 
 	// Stores key/id translation data.
 	TranslateStore TranslateStore
+
+	work chan job
 }
 
 // executorOption is a functional option type for pilosa.Executor
@@ -71,6 +74,7 @@ func optExecutorInternalQueryClient(c InternalQueryClient) executorOption {
 func newExecutor(opts ...executorOption) *executor {
 	e := &executor{
 		client: newNopInternalQueryClient(),
+		work:   make(chan job, 2000),
 	}
 	for _, opt := range opts {
 		err := opt(e)
@@ -78,6 +82,9 @@ func newExecutor(opts ...executorOption) *executor {
 			panic(err)
 		}
 	}
+	for i := 0; i < runtime.NumCPU()+8; i++ {
+		go worker(e.work)
+	}
 	return e
 }
 
@@ -2516,6 +2523,24 @@ func (e *executor) mapper(ctx context.Context, ch chan mapResponse, nodes []*Nod
 	return nil
 }
 
+type job struct {
+	shard      uint64
+	mapFn      mapFunc
+	ctx        context.Context
+	resultChan chan mapResponse
+}
+
+func worker(work chan job) {
+	for j := range work {
+		result, err := j.mapFn(j.shard)
+
+		select {
+		case <-j.ctx.Done():
+		case j.resultChan <- mapResponse{result: result, err: err}:
+		}
+	}
+}
+
 // mapperLocal performs map & reduce entirely on the local node.
 func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFunc, reduceFn reduceFunc) (interface{}, error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapperLocal")
@@ -2524,15 +2549,12 @@ func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFu
 	ch := make(chan mapResponse, len(shards))
 
 	for _, shard := range shards {
-		go func(shard uint64) {
-			result, err := mapFn(shard)
-
-			// Return response to the channel.
-			select {
-			case <-ctx.Done():
-			case ch <- mapResponse{result: result, err: err}:
-			}
-		}(shard)
+		e.work <- job{
+			shard:      shard,
+			mapFn:      mapFn,
+			ctx:        ctx,
+			resultChan: ch,
+		}
 	}
 
 	// Reduce results