From ec09582f44da6df9fe2c706bccb6502149a829e8 Mon Sep 17 00:00:00 2001
From: Matt Jaffee <matthew.jaffee@gmail.com>
Date: Wed, 10 Jul 2019 09:39:02 -0500
Subject: [PATCH 1/5] get read lock only where possible in Holder

---
 holder.go | 16 +++++++++++-----
 1 file changed, 11 insertions(+), 5 deletions(-)

diff --git a/holder.go b/holder.go
index 692868efe..ad26838d1 100644
--- a/holder.go
+++ b/holder.go
@@ -234,8 +234,8 @@ func (h *Holder) Close() error {
 // This is used to determine if the rebalancing of data is necessary
 // when a node joins the cluster.
 func (h *Holder) HasData() (bool, error) {
-	h.mu.Lock()
-	defer h.mu.Unlock()
+	h.mu.RLock()
+	defer h.mu.RUnlock()
 	if len(h.indexes) > 0 {
 		return true, nil
 	}
@@ -385,15 +385,21 @@ func (h *Holder) CreateIndex(name string, opt IndexOptions) (*Index, error) {
 // CreateIndexIfNotExists returns an index by name.
 // The index is created if it does not already exist.
 func (h *Holder) CreateIndexIfNotExists(name string, opt IndexOptions) (*Index, error) {
-	h.mu.Lock()
-	defer h.mu.Unlock()
+	h.mu.RLock()
 
 	// Find index in cache first.
 	if index := h.indexes[name]; index != nil {
+		h.mu.RUnlock()
 		return index, nil
 	}
 
-	return h.createIndex(name, opt)
+	h.mu.RUnlock()
+
+	index, err := h.CreateIndex(name, opt)
+	if _, ok := err.(ConflictError); err != nil && !ok {
+		return nil, err
+	}
+	return index, nil
 }
 
 func (h *Holder) createIndex(name string, opt IndexOptions) (*Index, error) {

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 2/5] 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

From 7d7a5539cacdbdabdad44576a683f8f2a883f32f Mon Sep 17 00:00:00 2001
From: Matt Jaffee <matthew.jaffee@gmail.com>
Date: Mon, 15 Jul 2019 07:56:57 -0500
Subject: [PATCH 3/5] make executor work chan smaller, add executor.Close

the size of the work chan probably doesn't matter... there is some
discussion of this on the associated PR
https://github.com/pilosa/pilosa/pull/2034
may test with an unbuffered channel as well.

Closing the executor avoids leaking goroutines which seems to be an
issue while running the test suite.
---
 executor.go | 14 ++++++++++++--
 server.go   |  8 +++++++-
 2 files changed, 19 insertions(+), 3 deletions(-)

diff --git a/executor.go b/executor.go
index 3b0270ebd..6131a8d71 100644
--- a/executor.go
+++ b/executor.go
@@ -72,9 +72,14 @@ func optExecutorInternalQueryClient(c InternalQueryClient) executorOption {
 
 // newExecutor returns a new instance of Executor.
 func newExecutor(opts ...executorOption) *executor {
+	// this is somewhat arbitrary, though going less than
+	// runtime.NumCPU() would likely result in a loss of throughput.
+	workerPoolSize := runtime.NumCPU() + 8
 	e := &executor{
 		client: newNopInternalQueryClient(),
-		work:   make(chan job, 2000),
+
+		// capacity of this channel is unlikely to affect much
+		work: make(chan job, workerPoolSize),
 	}
 	for _, opt := range opts {
 		err := opt(e)
@@ -82,12 +87,17 @@ func newExecutor(opts ...executorOption) *executor {
 			panic(err)
 		}
 	}
-	for i := 0; i < runtime.NumCPU()+8; i++ {
+	for i := 0; i < workerPoolSize; i++ {
 		go worker(e.work)
 	}
 	return e
 }
 
+func (e *executor) Close() error {
+	close(e.work)
+	return nil
+}
+
 // Execute executes a PQL query.
 func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *execOptions) (QueryResponse, error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.Execute")
diff --git a/server.go b/server.go
index ad81aa1b3..2b67c2b78 100644
--- a/server.go
+++ b/server.go
@@ -426,6 +426,8 @@ func (s *Server) Open() error {
 
 // Close closes the server and waits for it to shutdown.
 func (s *Server) Close() error {
+	errE := s.executor.Close()
+
 	// Notify goroutines to stop.
 	close(s.closing)
 	s.wg.Wait()
@@ -445,7 +447,11 @@ func (s *Server) Close() error {
 	if errh != nil {
 		return errors.Wrap(errh, "closing holder")
 	}
-	return errors.Wrap(errc, "closing cluster")
+	if errc != nil {
+		return errors.Wrap(errc, "closing cluster")
+	}
+	return errors.Wrap(errE, "closing executor")
+
 }
 
 // loadNodeID gets NodeID from disk, or creates a new value.

From a7d9b0a5ae87a4fcedce0790ebad2f2a29af5056 Mon Sep 17 00:00:00 2001
From: Matt Jaffee <matthew.jaffee@gmail.com>
Date: Mon, 15 Jul 2019 08:36:22 -0500
Subject: [PATCH 4/5] make sure workers are done when closing via a WaitGroup

still running out of goroutines in race tests in CI, so hopefully this
fixes that.
---
 executor.go | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/executor.go b/executor.go
index 6131a8d71..bd48033da 100644
--- a/executor.go
+++ b/executor.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"runtime"
 	"sort"
+	"sync"
 	"time"
 
 	"github.com/pilosa/pilosa/pql"
@@ -57,7 +58,8 @@ type executor struct {
 	// Stores key/id translation data.
 	TranslateStore TranslateStore
 
-	work chan job
+	workersWG sync.WaitGroup
+	work      chan job
 }
 
 // executorOption is a functional option type for pilosa.Executor
@@ -88,13 +90,18 @@ func newExecutor(opts ...executorOption) *executor {
 		}
 	}
 	for i := 0; i < workerPoolSize; i++ {
-		go worker(e.work)
+		e.workersWG.Add(1)
+		go func() {
+			defer e.workersWG.Done()
+			worker(e.work)
+		}()
 	}
 	return e
 }
 
 func (e *executor) Close() error {
 	close(e.work)
+	e.workersWG.Wait()
 	return nil
 }
 

From 9a453ef51ab6db68312eaadc3ee6883e0644edbf Mon Sep 17 00:00:00 2001
From: Matt Jaffee <matthew.jaffee@gmail.com>
Date: Mon, 15 Jul 2019 13:55:44 -0500
Subject: [PATCH 5/5] expose worker pool size to config, so we can set it lower
 in tests

we are experiencing issues with CI where it fails with race: limit on
8128 simultaneously alive goroutines is exceeded, dying

this, despite the fact that closing the executor should clean up all
worker goroutines. Apparently in CircleCI runtime.NumCPU() reports 36,
so the goroutines added up quickly.
---
 executor.go      | 29 ++++++++++++++++++-----------
 server.go        | 32 +++++++++++++++++++++++---------
 server/config.go | 13 ++++++++++++-
 server/server.go |  1 +
 test/pilosa.go   |  1 +
 5 files changed, 55 insertions(+), 21 deletions(-)

diff --git a/executor.go b/executor.go
index bd48033da..2110a92c6 100644
--- a/executor.go
+++ b/executor.go
@@ -18,7 +18,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"runtime"
 	"sort"
 	"sync"
 	"time"
@@ -58,8 +57,9 @@ type executor struct {
 	// Stores key/id translation data.
 	TranslateStore TranslateStore
 
-	workersWG sync.WaitGroup
-	work      chan job
+	workersWG      sync.WaitGroup
+	workerPoolSize int
+	work           chan job
 }
 
 // executorOption is a functional option type for pilosa.Executor
@@ -72,16 +72,18 @@ func optExecutorInternalQueryClient(c InternalQueryClient) executorOption {
 	}
 }
 
+func optExecutorWorkerPoolSize(size int) executorOption {
+	return func(e *executor) error {
+		e.workerPoolSize = size
+		return nil
+	}
+}
+
 // newExecutor returns a new instance of Executor.
 func newExecutor(opts ...executorOption) *executor {
-	// this is somewhat arbitrary, though going less than
-	// runtime.NumCPU() would likely result in a loss of throughput.
-	workerPoolSize := runtime.NumCPU() + 8
 	e := &executor{
-		client: newNopInternalQueryClient(),
-
-		// capacity of this channel is unlikely to affect much
-		work: make(chan job, workerPoolSize),
+		client:         newNopInternalQueryClient(),
+		workerPoolSize: 2,
 	}
 	for _, opt := range opts {
 		err := opt(e)
@@ -89,7 +91,12 @@ func newExecutor(opts ...executorOption) *executor {
 			panic(err)
 		}
 	}
-	for i := 0; i < workerPoolSize; i++ {
+	// this channel cap doesn't necessarily have to be the same as
+	// workerPoolSize... any larger doesn't seem to have an effect in
+	// the few tests we've done at scale with concurrent query
+	// workloads. Possible that it could be smaller.
+	e.work = make(chan job, e.workerPoolSize)
+	for i := 0; i < e.workerPoolSize; i++ {
 		e.workersWG.Add(1)
 		go func() {
 			defer e.workersWG.Done()
diff --git a/server.go b/server.go
index 2b67c2b78..6eee374d1 100644
--- a/server.go
+++ b/server.go
@@ -49,13 +49,14 @@ type Server struct { // nolint: maligned
 	closing chan struct{}
 
 	// Internal
-	holder          *Holder
-	cluster         *cluster
-	diagnostics     *diagnosticsCollector
-	executor        *executor
-	hosts           []string
-	clusterDisabled bool
-	serializer      Serializer
+	holder           *Holder
+	cluster          *cluster
+	diagnostics      *diagnosticsCollector
+	executor         *executor
+	executorPoolSize int
+	hosts            []string
+	clusterDisabled  bool
+	serializer       Serializer
 
 	// External
 	systemInfo SystemInfo
@@ -179,13 +180,19 @@ func OptServerGCNotifier(gcn GCNotifier) ServerOption {
 // used to set the implementation of InternalClient.
 func OptServerInternalClient(c InternalClient) ServerOption {
 	return func(s *Server) error {
-		s.executor = newExecutor(optExecutorInternalQueryClient(c))
 		s.defaultClient = c
 		s.cluster.InternalClient = c
 		return nil
 	}
 }
 
+func OptServerExecutorPoolSize(size int) ServerOption {
+	return func(s *Server) error {
+		s.executorPoolSize = size
+		return nil
+	}
+}
+
 // OptServerPrimaryTranslateStore has been deprecated.
 func OptServerPrimaryTranslateStore(store TranslateStore) ServerOption {
 	return func(s *Server) error {
@@ -306,7 +313,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 
 		logger: logger.NopLogger,
 	}
-	s.executor = newExecutor(optExecutorInternalQueryClient(s.defaultClient))
 	s.cluster.InternalClient = s.defaultClient
 
 	s.diagnostics.server = s
@@ -317,6 +323,14 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 			return nil, errors.Wrap(err, "applying option")
 		}
 	}
+
+	// set up executor after server opts have been processed
+	executorOpts := []executorOption{optExecutorInternalQueryClient(s.defaultClient)}
+	if s.executorPoolSize > 0 {
+		executorOpts = append(executorOpts, optExecutorWorkerPoolSize(s.executorPoolSize))
+	}
+	s.executor = newExecutor(executorOpts...)
+
 	s.holder.translateFile.logger = s.logger
 
 	path, err := expandDirName(s.dataDir)
diff --git a/server/config.go b/server/config.go
index c42d41b6c..b0db09ba7 100644
--- a/server/config.go
+++ b/server/config.go
@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"log"
 	"net"
+	"runtime"
 	"strconv"
 	"strings"
 	"time"
@@ -85,6 +86,13 @@ type Config struct {
 	// TLS
 	TLS TLSConfig `toml:"tls"`
 
+	// WorkerPoolSize controls how many goroutines are created for
+	// processing queries. Defaults to runtime.NumCPU(). It is
+	// intentionally not defined as a flag... only exposed here so
+	// that we can limit the size while running tests in CI so we
+	// don't exhaust the goroutine limit.
+	WorkerPoolSize int
+
 	Cluster struct {
 		// Disabled controls whether clustering functionality is enabled.
 		Disabled    bool     `toml:"disabled"`
@@ -151,7 +159,10 @@ func NewConfig() *Config {
 		// a bit below your system limits.
 		MaxMapCount:  1000000,
 		MaxFileCount: 1000000,
-		TLS:          TLSConfig{},
+
+		TLS: TLSConfig{},
+
+		WorkerPoolSize: runtime.NumCPU(),
 	}
 
 	// Cluster config.
diff --git a/server/server.go b/server/server.go
index 743accf95..db367f418 100644
--- a/server/server.go
+++ b/server/server.go
@@ -284,6 +284,7 @@ func (m *Command) SetupServer() error {
 		pilosa.OptServerMaxWritesPerRequest(m.Config.MaxWritesPerRequest),
 		pilosa.OptServerMetricInterval(time.Duration(m.Config.Metric.PollInterval)),
 		pilosa.OptServerDiagnosticsInterval(diagnosticsInterval),
+		pilosa.OptServerExecutorPoolSize(m.Config.WorkerPoolSize),
 
 		pilosa.OptServerLogger(m.logger),
 		pilosa.OptServerAttrStoreFunc(boltdb.NewAttrStore),
diff --git a/test/pilosa.go b/test/pilosa.go
index 31e93bb1d..4a2cd2973 100644
--- a/test/pilosa.go
+++ b/test/pilosa.go
@@ -72,6 +72,7 @@ func newCommand(opts ...server.CommandOption) *Command {
 	m.Config.Bind = "http://localhost:0"
 	m.Config.Cluster.Disabled = true
 	m.Config.Translation.MapSize = 140000
+	m.Config.WorkerPoolSize = 2
 
 	if testing.Verbose() {
 		m.Command.Stdout = os.Stdout