From 96ab9314d1deb934b3ac17d80401440efc1e823d Mon Sep 17 00:00:00 2001 From: Seebs Date: Thu, 10 Feb 2022 12:09:38 -0600 Subject: [PATCH] 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 | 107 ++++++++++++++++----------------------------------- holder.go | 3 ++ server.go | 1 + txfactory.go | 13 ++++++- 4 files changed, 48 insertions(+), 76 deletions(-) diff --git a/executor.go b/executor.go index 3b09cd92c..232c1984c 100644 --- a/executor.go +++ b/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 } diff --git a/holder.go b/holder.go index 18cd7154c..4ecb5ec6e 100644 --- a/holder.go +++ b/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{} diff --git a/server.go b/server.go index 3475b670a..531e56737 100644 --- a/server.go +++ b/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 diff --git a/txfactory.go b/txfactory.go index 58134a6bb..54dfa4390 100644 --- a/txfactory.go +++ b/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()