Queue: Make WorkerPools and Queues flushable (#10001)

* Make WorkerPools and Queues flushable

Adds Flush methods to Queues and the WorkerPool
Further abstracts the WorkerPool
Adds a final step to Flush the queues in the defer from PrintCurrentTest
Fixes an issue with Settings inheritance in queues

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

* Change to for loop

* Add IsEmpty and begin just making the queues composed WorkerPools

* subsume workerpool into the queues and create a flushable interface

* Add manager command

* Move flushall to queue.Manager and add to testlogger

* As per @guillep2k

* as per @guillep2k

* Just make queues all implement flushable and clean up the wrapped queue flushes

* cope with no timeout

Co-authored-by: Lauris BH <lauris@nix.lv>

1 files changed, 28 insertions, 5 deletions

diff --git a/modules/graceful/manager_windows.go b/modules/graceful/manager_windows.go
index 526fc0bd24..cb4475ebb2 100644
--- a/modules/graceful/manager_windows.go
+++ b/modules/graceful/manager_windows.go
@@ -43,6 +43,7 @@ type Manager struct {
 	runningServerWaitGroup sync.WaitGroup
 	createServerWaitGroup  sync.WaitGroup
 	terminateWaitGroup     sync.WaitGroup
+	shutdownRequested	   chan struct{}
 }
 
 func newGracefulManager(ctx context.Context) *Manager {
@@ -62,6 +63,7 @@ func (g *Manager) start() {
 	g.shutdown = make(chan struct{})
 	g.hammer = make(chan struct{})
 	g.done = make(chan struct{})
+	g.shutdownRequested = make(chan struct{})
 
 	// Set the running state
 	g.setState(stateRunning)
@@ -107,7 +109,10 @@ loop:
 	for {
 		select {
 		case <-g.ctx.Done():
-			g.doShutdown()
+			g.DoGracefulShutdown()
+			waitTime += setting.GracefulHammerTime
+			break loop
+		case <-g.shutdownRequested:
 			waitTime += setting.GracefulHammerTime
 			break loop
 		case change := <-changes:
@@ -115,12 +120,12 @@ loop:
 			case svc.Interrogate:
 				status <- change.CurrentStatus
 			case svc.Stop, svc.Shutdown:
-				g.doShutdown()
+				g.DoGracefulShutdown()
 				waitTime += setting.GracefulHammerTime
 				break loop
 			case hammerCode:
-				g.doShutdown()
-				g.doHammerTime(0 * time.Second)
+				g.DoGracefulShutdown()
+				g.DoImmediateHammer()
 				break loop
 			default:
 				log.Debug("Unexpected control request: %v", change.Cmd)
@@ -140,7 +145,7 @@ hammerLoop:
 			case svc.Interrogate:
 				status <- change.CurrentStatus
 			case svc.Stop, svc.Shutdown, hammerCmd:
-				g.doHammerTime(0 * time.Second)
+				g.DoImmediateHammer()
 				break hammerLoop
 			default:
 				log.Debug("Unexpected control request: %v", change.Cmd)
@@ -152,6 +157,24 @@ hammerLoop:
 	return false, 0
 }
 
+// DoImmediateHammer causes an immediate hammer
+func (g *Manager) DoImmediateHammer() {
+	g.doHammerTime(0 * time.Second)
+}
+
+// DoGracefulShutdown causes a graceful shutdown
+func (g *Manager) DoGracefulShutdown() {
+	g.lock.Lock()
+	select {
+	case <-g.shutdownRequested:
+		g.lock.Unlock()
+	default:
+		close(g.shutdownRequested)
+		g.lock.Unlock()
+		g.doShutdown()
+	}
+}
+
 // RegisterServer registers the running of a listening server.
 // Any call to RegisterServer must be matched by a call to ServerDone
 func (g *Manager) RegisterServer() {