scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 09:05:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

cleanly shut down the executor

Browse source
This commit is contained in:
Jaden Weiss 2020-06-05 15:25:35 -04:00
parent 33728e65d9
commit b0a0524ffe
No known key found for this signature in database
GPG key ID: 177F065773634B67
1 changed files with 13 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
executor.go
View file

@ -59,6 +59,8 @@ type executor struct {
// Maximum number of Set() or Clear() commands per request.
MaxWritesPerRequest int
shutdown bool
workMu sync.RWMutex
workersWG sync.WaitGroup
workerPoolSize int
work chan job
@ -115,6 +117,9 @@ func newExecutor(opts ...executorOption) *executor {
}
func (e *executor) Close() error {
e.workMu.Lock()
defer e.workMu.Unlock()
e.shutdown = true
close(e.work)
e.workersWG.Wait()
return nil
@ -3768,6 +3773,8 @@ func worker(work chan job) {
}
}
var errShutdown = errors.New("executor has shut down")
// 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")
@ -3775,6 +3782,12 @@ 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()
if e.shutdown {
return nil, errShutdown
}
ch := make(chan mapResponse, len(shards))