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, 31 insertions, 15 deletions

diff --git a/modules/graceful/manager_unix.go b/modules/graceful/manager_unix.go
index 323c6a4111..68aa724264 100644
--- a/modules/graceful/manager_unix.go
+++ b/modules/graceful/manager_unix.go
@@ -110,28 +110,19 @@ func (g *Manager) handleSignals(ctx context.Context) {
 		case sig := <-signalChannel:
 			switch sig {
 			case syscall.SIGHUP:
-				if setting.GracefulRestartable {
-					log.Info("PID: %d. Received SIGHUP. Forking...", pid)
-					err := g.doFork()
-					if err != nil && err.Error() != "another process already forked. Ignoring this one" {
-						log.Error("Error whilst forking from PID: %d : %v", pid, err)
-					}
-				} else {
-					log.Info("PID: %d. Received SIGHUP. Not set restartable. Shutting down...", pid)
-
-					g.doShutdown()
-				}
+				log.Info("PID: %d. Received SIGHUP. Attempting GracefulShutdown...", pid)
+				g.DoGracefulShutdown()
 			case syscall.SIGUSR1:
 				log.Info("PID %d. Received SIGUSR1.", pid)
 			case syscall.SIGUSR2:
 				log.Warn("PID %d. Received SIGUSR2. Hammering...", pid)
-				g.doHammerTime(0 * time.Second)
+				g.DoImmediateHammer()
 			case syscall.SIGINT:
 				log.Warn("PID %d. Received SIGINT. Shutting down...", pid)
-				g.doShutdown()
+				g.DoGracefulShutdown()
 			case syscall.SIGTERM:
 				log.Warn("PID %d. Received SIGTERM. Shutting down...", pid)
-				g.doShutdown()
+				g.DoGracefulShutdown()
 			case syscall.SIGTSTP:
 				log.Info("PID %d. Received SIGTSTP.", pid)
 			default:
@@ -139,7 +130,7 @@ func (g *Manager) handleSignals(ctx context.Context) {
 			}
 		case <-ctx.Done():
 			log.Warn("PID: %d. Background context for manager closed - %v - Shutting down...", pid, ctx.Err())
-			g.doShutdown()
+			g.DoGracefulShutdown()
 		}
 	}
 }
@@ -160,6 +151,31 @@ func (g *Manager) doFork() error {
 	return err
 }
 
+// DoGracefulRestart causes a graceful restart
+func (g *Manager) DoGracefulRestart() {
+	if setting.GracefulRestartable {
+		log.Info("PID: %d. Forking...", os.Getpid())
+		err := g.doFork()
+		if err != nil && err.Error() != "another process already forked. Ignoring this one" {
+			log.Error("Error whilst forking from PID: %d : %v", os.Getpid(), err)
+		}
+	} else {
+		log.Info("PID: %d. Not set restartable. Shutting down...", os.Getpid())
+
+		g.doShutdown()
+	}
+}
+
+// DoImmediateHammer causes an immediate hammer
+func (g *Manager) DoImmediateHammer() {
+	g.doHammerTime(0 * time.Second)
+}
+
+// DoGracefulShutdown causes a graceful shutdown
+func (g *Manager) DoGracefulShutdown() {
+	g.doShutdown()
+}
+
 // RegisterServer registers the running of a listening server, in the case of unix this means that the parent process can now die.
 // Any call to RegisterServer must be matched by a call to ServerDone
 func (g *Manager) RegisterServer() {