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

@@ -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()

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)

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.

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

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