back out locking added in c78ada09f230a3c66cd934860700f93ff31a93eb

util/execpool/backlog.go

@@ -19,23 +19,19 @@ package execpool
 import (
 	"context"
 	"sync"
-
-	"github.com/algorand/go-deadlock"
 )
 
 // A backlog for an execution pool. The typical usage of this is to
 // create non-blocking queue which would get executed once the execution pool is ready to accept new
 // tasks.
 type backlog struct {
-	mu        deadlock.Mutex
 	pool      ExecutionPool
 	wg        sync.WaitGroup
 	buffer    chan backlogItemTask
 	ctx       context.Context
 	ctxCancel context.CancelFunc
 	owner     interface{}
 	priority  Priority
-	quit      bool
 }
 
 type backlogItemTask struct {
@@ -82,25 +78,11 @@ func (b *backlog) GetParallelism() int {
 
 // IsFull test to see if the input buffer is full.
 func (b *backlog) IsFull() bool {
-	b.mu.Lock()
-	defer b.mu.Unlock()
 	return len(b.buffer) == cap(b.buffer)
 }
 
 // Enqueue enqueues a single task into the backlog
 func (b *backlog) Enqueue(enqueueCtx context.Context, t ExecFunc, arg interface{}, priority Priority, out chan interface{}) error {
-	b.mu.Lock()
-	defer b.mu.Unlock()
-	if b.quit {
-		select {
-		case <-enqueueCtx.Done():
-			return enqueueCtx.Err()
-		case <-b.ctx.Done():
-			return b.ctx.Err()
-		default:
-			return nil
-		}
-	}
 	select {
 	case b.buffer <- backlogItemTask{
 		enqueuedTask: enqueuedTask{
@@ -120,18 +102,6 @@ func (b *backlog) Enqueue(enqueueCtx context.Context, t ExecFunc, arg interface{
 
 // Enqueue enqueues a single task into the backlog
 func (b *backlog) EnqueueBacklog(enqueueCtx context.Context, t ExecFunc, arg interface{}, out chan interface{}) error {
-	b.mu.Lock()
-	defer b.mu.Unlock()
-	if b.quit {
-		select {
-		case <-enqueueCtx.Done():
-			return enqueueCtx.Err()
-		case <-b.ctx.Done():
-			return b.ctx.Err()
-		default:
-			return nil
-		}
-	}
 	select {
 	case b.buffer <- backlogItemTask{
 		enqueuedTask: enqueuedTask{
@@ -151,9 +121,6 @@ func (b *backlog) EnqueueBacklog(enqueueCtx context.Context, t ExecFunc, arg int
 
 // Shutdown shuts down the backlog.
 func (b *backlog) Shutdown() {
-	b.mu.Lock()
-	defer b.mu.Unlock()
-	b.quit = true
 	b.ctxCancel()
 	close(b.buffer)
 	b.wg.Wait()