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, 32 insertions, 17 deletions

diff --git a/modules/queue/queue_disk_channel_test.go b/modules/queue/queue_disk_channel_test.go
index 93061bffc6..561f98ca90 100644
--- a/modules/queue/queue_disk_channel_test.go
+++ b/modules/queue/queue_disk_channel_test.go
@@ -5,10 +5,8 @@
 package queue
 
 import (
-	"context"
 	"io/ioutil"
 	"testing"
-	"time"
 
 	"code.gitea.io/gitea/modules/util"
 	"github.com/stretchr/testify/assert"
@@ -32,17 +30,19 @@ func TestPersistableChannelQueue(t *testing.T) {
 	defer util.RemoveAll(tmpDir)
 
 	queue, err := NewPersistableChannelQueue(handle, PersistableChannelQueueConfiguration{
-		DataDir:     tmpDir,
-		BatchLength: 2,
-		QueueLength: 20,
-		Workers:     1,
-		MaxWorkers:  10,
+		DataDir:      tmpDir,
+		BatchLength:  2,
+		QueueLength:  20,
+		Workers:      1,
+		BoostWorkers: 0,
+		MaxWorkers:   10,
+		Name:         "first",
 	}, &testData{})
 	assert.NoError(t, err)
 
-	go queue.Run(func(_ context.Context, shutdown func()) {
+	go queue.Run(func(shutdown func()) {
 		queueShutdown = append(queueShutdown, shutdown)
-	}, func(_ context.Context, terminate func()) {
+	}, func(terminate func()) {
 		queueTerminate = append(queueTerminate, terminate)
 	})
 
@@ -64,13 +64,18 @@ func TestPersistableChannelQueue(t *testing.T) {
 	assert.Equal(t, test2.TestString, result2.TestString)
 	assert.Equal(t, test2.TestInt, result2.TestInt)
 
+	// test1 is a testData not a *testData so will be rejected
 	err = queue.Push(test1)
 	assert.Error(t, err)
 
+	// Now shutdown the queue
 	for _, callback := range queueShutdown {
 		callback()
 	}
-	time.Sleep(200 * time.Millisecond)
+
+	// Wait til it is closed
+	<-queue.(*PersistableChannelQueue).closed
+
 	err = queue.Push(&test1)
 	assert.NoError(t, err)
 	err = queue.Push(&test2)
@@ -80,23 +85,33 @@ func TestPersistableChannelQueue(t *testing.T) {
 		assert.Fail(t, "Handler processing should have stopped")
 	default:
 	}
+
+	// terminate the queue
 	for _, callback := range queueTerminate {
 		callback()
 	}
 
+	select {
+	case <-handleChan:
+		assert.Fail(t, "Handler processing should have stopped")
+	default:
+	}
+
 	// Reopen queue
 	queue, err = NewPersistableChannelQueue(handle, PersistableChannelQueueConfiguration{
-		DataDir:     tmpDir,
-		BatchLength: 2,
-		QueueLength: 20,
-		Workers:     1,
-		MaxWorkers:  10,
+		DataDir:      tmpDir,
+		BatchLength:  2,
+		QueueLength:  20,
+		Workers:      1,
+		BoostWorkers: 0,
+		MaxWorkers:   10,
+		Name:         "second",
 	}, &testData{})
 	assert.NoError(t, err)
 
-	go queue.Run(func(_ context.Context, shutdown func()) {
+	go queue.Run(func(shutdown func()) {
 		queueShutdown = append(queueShutdown, shutdown)
-	}, func(_ context.Context, terminate func()) {
+	}, func(terminate func()) {
 		queueTerminate = append(queueTerminate, terminate)
 	})