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] 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.