Multiple Queue improvements: LevelDB Wait on empty, shutdown empty shadow level queue, reduce goroutines etc (#15693)

* move shutdownfns, terminatefns and hammerfns out of separate goroutines

Coalesce the shutdownfns etc into a list of functions that get run at shutdown
rather then have them run at goroutines blocked on selects.

This may help reduce the background select/poll load in certain
configurations.

* The LevelDB queues can actually wait on empty instead of polling

Slight refactor to cause leveldb queues to wait on empty instead of polling.

* Shutdown the shadow level queue once it is empty

* Remove bytefifo additional goroutine for readToChan as it can just be run in run

* Remove additional removeWorkers goroutine for workers

* Simplify the AtShutdown and AtTerminate functions and add Channel Flusher

* Add shutdown flusher to CUQ

* move persistable channel shutdown stuff to Shutdown Fn

* Ensure that UPCQ has the correct config

* handle shutdown during the flushing

* reduce risk of race between zeroBoost and addWorkers

* prevent double shutdown

Signed-off-by: Andrew Thornton <art27@cantab.net>

1 files changed, 3 insertions, 20 deletions

diff --git a/modules/graceful/context.go b/modules/graceful/context.go
index 1ad1109b4e..9d955329a4 100644
--- a/modules/graceful/context.go
+++ b/modules/graceful/context.go
@@ -6,17 +6,9 @@ package graceful
 
 import (
 	"context"
-	"fmt"
 	"time"
 )
 
-// Errors for context.Err()
-var (
-	ErrShutdown  = fmt.Errorf("Graceful Manager called Shutdown")
-	ErrHammer    = fmt.Errorf("Graceful Manager called Hammer")
-	ErrTerminate = fmt.Errorf("Graceful Manager called Terminate")
-)
-
 // ChannelContext is a context that wraps a channel and error as a context
 type ChannelContext struct {
 	done <-chan struct{}
@@ -63,28 +55,19 @@ func (ctx *ChannelContext) Value(key interface{}) interface{} {
 // Callers using this context should ensure that they are registered as a running server
 // in order that they are waited for.
 func (g *Manager) ShutdownContext() context.Context {
-	return &ChannelContext{
-		done: g.IsShutdown(),
-		err:  ErrShutdown,
-	}
+	return g.shutdownCtx
 }
 
 // HammerContext returns a context.Context that is Done at hammer
 // Callers using this context should ensure that they are registered as a running server
 // in order that they are waited for.
 func (g *Manager) HammerContext() context.Context {
-	return &ChannelContext{
-		done: g.IsHammer(),
-		err:  ErrHammer,
-	}
+	return g.hammerCtx
 }
 
 // TerminateContext returns a context.Context that is Done at terminate
 // Callers using this context should ensure that they are registered as a terminating server
 // in order that they are waited for.
 func (g *Manager) TerminateContext() context.Context {
-	return &ChannelContext{
-		done: g.IsTerminate(),
-		err:  ErrTerminate,
-	}
+	return g.terminateCtx
 }