use task pool for executor workers

This adopts the task pool functionality to let us spawn new worker
threads when worker threads are blocked. The underlying reason for
this is the same as the reason for the previous worker-pool-growing
strategy; while our design persistently has at least one thing which
can proceed, it can be the case that there are N things blocked,
where N is the size of our worker pool. Blocked workers shouldn't
count against our desired number of workers.

Originally, the intent was to thread this into RBF, and provide
backpressure from RBF on the pool when blocking on writes. Unfortunately,
that's not good enough, because while a write is blocked, the Qcx
calling it is *also* holding the Qcx's mutex, which means that any other
NewTx on that Qcx will *also* block. So we need to block for the
entire time of the NewTx.

Removing the existing worker spawning code resulted in a subtle
and maybe-harmless change; prior to this, each invocation of `mapperLocal`
would hold a lock, which meant that all the tasks for a given local mapper
would be put in the queue *sequentially*, ensuring that they'd all be
picked up by workers before things from later workers.

With the new pushback, that's not, strictly, necessary. Also, if you
disable it, you can end up with 300,000 goroutines at once, most of them
blocked.

A smallish run does, in fact, eventually complete anyway -- it will
indeed keep making workers until everything gets one. However, while
it's *correct*, it's also noticably *slower*. The same test workload
goes from around 33 seconds to a bit over 40 seconds when that lock
isn't present. (But that's with an extremely small WAL write cap
introduced to make the previous deadlock possible.)

With large numbers of shards, the practical impact is that you can
have quite a lot of things in process, with hundreds of goroutines
each, all blocked waiting for one writer. If we force them to all be
processed at the same time, all the reads that are connected to
each other are much more likely to get all processed at once, before
something new comes along.

In short, that lock isn't strictly necessary but it seems to help
noticably with performance and reduce simultaneous goroutines
significantly.

executor.go

@@ -22,6 +22,7 @@ import (
 	"github.com/molecula/featurebase/v3/proto"
 	"github.com/molecula/featurebase/v3/roaring"
 	"github.com/molecula/featurebase/v3/shardwidth"
+	"github.com/molecula/featurebase/v3/task"
 	"github.com/molecula/featurebase/v3/testhook"
 	"github.com/molecula/featurebase/v3/topology"
 	"github.com/molecula/featurebase/v3/tracing"
@@ -51,9 +52,6 @@ type executor struct {
 	Node    *topology.Node
 	Cluster *cluster
 
-	// how many jobs the work queue has seen
-	workCounter uint64
-
 	// Client used for remote requests.
 	client *InternalClient
 
@@ -61,10 +59,9 @@ type executor struct {
 	MaxWritesPerRequest int
 
 	shutdown       bool
-	workMu         sync.RWMutex
-	workersWG      sync.WaitGroup
+	workers        *task.Pool
+	workerPoolMu   sync.Mutex
 	workerPoolSize int
-	currentWorkers int64
 	work           chan job
 
 	// Maximum per-request memory usage (Extract() only)
@@ -130,72 +127,11 @@ func newExecutor(opts ...executorOption) *executor {
 	// workloads. Possible that it could be smaller.
 	e.work = make(chan job, e.workerPoolSize)
 	_ = testhook.Opened(NewAuditor(), e, nil)
-	for i := 0; i < e.workerPoolSize; i++ {
-		e.addWorker()
-	}
-	go func() {
-		// background task: every so often, check to see whether we have
-		// work in the queue but none has been taken for a while. if so, we
-		// need more workers.
-		prev := atomic.LoadUint64(&e.workCounter)
-		periodic := time.NewTicker(50 * time.Millisecond)
-		defer periodic.Stop()
-		running := true
-		idle := 0
-		for running {
-			<-periodic.C
-			func() {
-				e.workMu.RLock()
-				defer e.workMu.RUnlock()
-				if e.shutdown {
-					running = false
-					return
-				}
-				if len(e.work) == 0 {
-					idle++
-					if idle > 10 && atomic.LoadInt64(&e.currentWorkers) > int64(e.workerPoolSize*2) {
-						select {
-						case e.work <- job{idleHands: true}:
-							// we closed an excess worker
-						default:
-							// somehow between our test above and now the work
-							// queue FILLED UP and we stoically accept this
-						}
-						idle = 0
-					}
-					return
-				}
-				next := atomic.LoadUint64(&e.workCounter)
-				if next == prev {
-					e.addWorker()
-				}
-				prev = next
-			}()
-		}
-	}()
+	e.workers = task.NewPool(e.workerPoolSize, e.doOneJob, e)
 	return e
 }
 
-func (e *executor) addWorker() {
-	e.workersWG.Add(1)
-	n := atomic.AddInt64(&e.currentWorkers, 1)
-	if e.Holder != nil {
-		e.Holder.Stats.Gauge("worker_total", float64(n), 0)
-	}
-
-	go func() {
-		defer e.workersWG.Done()
-		e.worker(e.work)
-		n := atomic.AddInt64(&e.currentWorkers, -1)
-		if e.Holder != nil {
-			e.Holder.Stats.Gauge("worker_total", float64(n), 0)
-		}
-	}()
-}
-
 func (e *executor) Close() error {
-	e.workMu.Lock()
-	defer e.workMu.Unlock()
 	if e.shutdown {
 		// otherwise close(e.work) can result in
 		// panic: close of closed channel.
@@ -206,15 +142,23 @@ func (e *executor) Close() error {
 	e.shutdown = true
 	_ = testhook.Closed(NewAuditor(), e, nil)
 	close(e.work)
-	e.workersWG.Wait()
+	e.workers.Close()
 	return nil
 }
 
+// PoolSize is exported to let the task pool update us
+func (e *executor) PoolSize(n int) {
+	if e.Holder != nil {
+		e.Holder.Stats.Gauge("worker_total", float64(n), 0)
+	}
+}
+
 // InitStats initializes stats counters. Must be called after Holder set.
 func (e *executor) InitStats() {
 	if e.Holder != nil {
 		e.Holder.Stats.Count("job_total", 0, 0)
-		e.Holder.Stats.Gauge("worker_total", float64(atomic.LoadInt64(&e.currentWorkers)), 0)
+		l, _, _ := e.workers.Stats()
+		e.Holder.Stats.Gauge("worker_total", float64(l), 0)
 	}
 }
 
@@ -6068,9 +6012,25 @@ type job struct {
 	idleHands       bool
 }
 
+// doOneJob had one job. *disappointed sigh*
+func (e *executor) doOneJob() {
+	j, ok := <-e.work
+	if !ok {
+		return
+	}
+	// Skip out early if the context is done, but still send
+	// an ack so mapperLocal can be sure we aren't about to
+	// work on something it sent us.
+	if err := j.ctx.Err(); err != nil {
+		j.resultChan <- mapResponse{result: nil, err: err}
+		return
+	}
+	result, err := j.mapFn(j.ctx, j.shard, &mapOptions{memoryAvailable: j.memoryAvailable})
+	j.resultChan <- mapResponse{result: result, err: err}
+}
+
 func (e *executor) worker(work chan job) {
 	for j := range work {
-		atomic.AddUint64(&e.workCounter, 1)
 		e.Holder.Stats.Count("job_total", 1, 0)
 		if j.idleHands {
 			return
@@ -6096,9 +6056,8 @@ func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFu
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	done := ctx.Done()
-	e.workMu.RLock()
-	defer e.workMu.RUnlock()
-
+	e.workerPoolMu.Lock()
+	defer e.workerPoolMu.Unlock()
 	if e.shutdown {
 		return nil, errShutdown
 	}

holder.go

@@ -70,6 +70,9 @@ type Holder struct {
 	sharder     disco.Sharder
 	serializer  Serializer
 
+	// executor, which we use only to get access to its worker pool
+	executor *executor
+
 	// Close management
 	wg      sync.WaitGroup
 	closing chan struct{}

server.go

@@ -494,6 +494,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 	s.holder.Stats = s.holder.Stats.WithTags(fmt.Sprintf("node_id:%s", s.nodeID))
 
 	s.executor.Holder = s.holder
+	s.holder.executor = s.executor
 	s.executor.Cluster = s.cluster
 	s.executor.MaxWritesPerRequest = s.maxWritesPerRequest
 	s.cluster.broadcaster = s

txfactory.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/molecula/featurebase/v3/task"
 	"github.com/molecula/featurebase/v3/testhook"
 	"github.com/molecula/featurebase/v3/vprint"
 	"github.com/pkg/errors"
@@ -83,8 +84,9 @@ var sep = string(os.PathSeparator)
 // See also the Qcx.GetTx() example and the TxGroup description below.
 //
 type Qcx struct {
-	Grp *TxGroup
-	Txf *TxFactory
+	Grp     *TxGroup
+	Txf     *TxFactory
+	workers *task.Pool
 
 	// if we go back to using Qcx values, this must become a pointer,
 	// or otherwise be dealt with because copies of Mutex are a no-no.
@@ -178,6 +180,9 @@ func (f *TxFactory) NewQcx() (qcx *Qcx) {
 		Grp: f.NewTxGroup(),
 		Txf: f,
 	}
+	if f.holder != nil && f.holder.executor != nil {
+		qcx.workers = f.holder.executor.workers
+	}
 	if f.typeOfTx == "roaring" {
 		qcx.isRoaring = true
 	}
@@ -223,6 +228,10 @@ var ErrQcxDone = fmt.Errorf("Qcx already Aborted or Finished, so must call reset
 // to make it clear we are referring to the first and final error.
 //
 func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) {
+	if qcx.workers != nil {
+		qcx.workers.Block()
+		defer qcx.workers.Unblock()
+	}
 	qcx.mu.Lock()
 	defer qcx.mu.Unlock()