@ -39,21 +39,21 @@ func (q *ExecQueue) Add(f func()) {
// RunSync waits for the queue to be drained and then synchronously runs f.
// RunSync waits for the queue to be drained and then synchronously runs f.
// It returns an error if the queue is closed before f is run or ctx expires.
// It returns an error if the queue is closed before f is run or ctx expires.
func ( q * ExecQueue ) RunSync ( ctx context . Context , f func ( ) ) error {
func ( q * ExecQueue ) RunSync ( ctx context . Context , f func ( ) ) error {
for {
q . mu . Lock ( )
if err := q . Wait ( ctx ) ; err != nil {
q . initCtxLocked ( )
return err
shutdownCtx := q . ctx
}
q . mu . Unlock ( )
q . mu . Lock ( )
if q . inFlight {
ch := make ( chan struct { } )
q . mu . Unlock ( )
q . Add ( f )
continue
q . Add ( func ( ) { close ( ch ) } )
}
select {
defer q . mu . Unlock ( )
case <- ch :
if q . closed {
return errors . New ( "closed" )
}
f ( )
return nil
return nil
case <- ctx . Done ( ) :
return ctx . Err ( )
case <- shutdownCtx . Done ( ) :
return errExecQueueShutdown
}
}
}
}
@ -94,6 +94,8 @@ func (q *ExecQueue) initCtxLocked() {
}
}
}
}
var errExecQueueShutdown = errors . New ( "execqueue shut down" )
// Wait waits for the queue to be empty or shut down.
// Wait waits for the queue to be empty or shut down.
func ( q * ExecQueue ) Wait ( ctx context . Context ) error {
func ( q * ExecQueue ) Wait ( ctx context . Context ) error {
q . mu . Lock ( )
q . mu . Lock ( )
@ -104,10 +106,11 @@ func (q *ExecQueue) Wait(ctx context.Context) error {
q . doneWaiter = waitCh
q . doneWaiter = waitCh
}
}
closed := q . closed
closed := q . closed
shutdownCtx := q . ctx
q . mu . Unlock ( )
q . mu . Unlock ( )
if closed {
if closed {
return err ors. New ( "execqueue shut down" )
return err ExecQueueShutdown
}
}
if waitCh == nil {
if waitCh == nil {
return nil
return nil
@ -116,8 +119,8 @@ func (q *ExecQueue) Wait(ctx context.Context) error {
select {
select {
case <- waitCh :
case <- waitCh :
return nil
return nil
case <- q. c tx. Done ( ) :
case <- shutdownC tx. Done ( ) :
return err ors. New ( "execqueue shut down" )
return err ExecQueueShutdown
case <- ctx . Done ( ) :
case <- ctx . Done ( ) :
return ctx . Err ( )
return ctx . Err ( )
}
}