summaryrefslogtreecommitdiffstats
path: root/modules/queue/manager_test.go
diff options
context:
space:
mode:
authorwxiaoguang <wxiaoguang@gmail.com>2023-05-08 19:49:59 +0800
committerGitHub <noreply@github.com>2023-05-08 19:49:59 +0800
commit6f9c278559789066aa831c1df25b0d866103d02d (patch)
treee3a1880e0d4cf88916f9d1b65d82fbd4c41ea47f /modules/queue/manager_test.go
parentcb700aedd1e670fb47b8cf0cd67fb117a1ad88a2 (diff)
downloadgitea-6f9c278559789066aa831c1df25b0d866103d02d.tar.gz
gitea-6f9c278559789066aa831c1df25b0d866103d02d.zip
Rewrite queue (#24505)
# ⚠️ Breaking Many deprecated queue config options are removed (actually, they should have been removed in 1.18/1.19). If you see the fatal message when starting Gitea: "Please update your app.ini to remove deprecated config options", please follow the error messages to remove these options from your app.ini. Example: ``` 2023/05/06 19:39:22 [E] Removed queue option: `[indexer].ISSUE_INDEXER_QUEUE_TYPE`. Use new options in `[queue.issue_indexer]` 2023/05/06 19:39:22 [E] Removed queue option: `[indexer].UPDATE_BUFFER_LEN`. Use new options in `[queue.issue_indexer]` 2023/05/06 19:39:22 [F] Please update your app.ini to remove deprecated config options ``` Many options in `[queue]` are are dropped, including: `WRAP_IF_NECESSARY`, `MAX_ATTEMPTS`, `TIMEOUT`, `WORKERS`, `BLOCK_TIMEOUT`, `BOOST_TIMEOUT`, `BOOST_WORKERS`, they can be removed from app.ini. # The problem The old queue package has some legacy problems: * complexity: I doubt few people could tell how it works. * maintainability: Too many channels and mutex/cond are mixed together, too many different structs/interfaces depends each other. * stability: due to the complexity & maintainability, sometimes there are strange bugs and difficult to debug, and some code doesn't have test (indeed some code is difficult to test because a lot of things are mixed together). * general applicability: although it is called "queue", its behavior is not a well-known queue. * scalability: it doesn't seem easy to make it work with a cluster without breaking its behaviors. It came from some very old code to "avoid breaking", however, its technical debt is too heavy now. It's a good time to introduce a better "queue" package. # The new queue package It keeps using old config and concept as much as possible. * It only contains two major kinds of concepts: * The "base queue": channel, levelqueue, redis * They have the same abstraction, the same interface, and they are tested by the same testing code. * The "WokerPoolQueue", it uses the "base queue" to provide "worker pool" function, calls the "handler" to process the data in the base queue. * The new code doesn't do "PushBack" * Think about a queue with many workers, the "PushBack" can't guarantee the order for re-queued unhandled items, so in new code it just does "normal push" * The new code doesn't do "pause/resume" * The "pause/resume" was designed to handle some handler's failure: eg: document indexer (elasticsearch) is down * If a queue is paused for long time, either the producers blocks or the new items are dropped. * The new code doesn't do such "pause/resume" trick, it's not a common queue's behavior and it doesn't help much. * If there are unhandled items, the "push" function just blocks for a few seconds and then re-queue them and retry. * The new code doesn't do "worker booster" * Gitea's queue's handlers are light functions, the cost is only the go-routine, so it doesn't make sense to "boost" them. * The new code only use "max worker number" to limit the concurrent workers. * The new "Push" never blocks forever * Instead of creating more and more blocking goroutines, return an error is more friendly to the server and to the end user. There are more details in code comments: eg: the "Flush" problem, the strange "code.index" hanging problem, the "immediate" queue problem. Almost ready for review. TODO: * [x] add some necessary comments during review * [x] add some more tests if necessary * [x] update documents and config options * [x] test max worker / active worker * [x] re-run the CI tasks to see whether any test is flaky * [x] improve the `handleOldLengthConfiguration` to provide more friendly messages * [x] fine tune default config values (eg: length?) ## Code coverage: ![image](https://user-images.githubusercontent.com/2114189/236620635-55576955-f95d-4810-b12f-879026a3afdf.png)
Diffstat (limited to 'modules/queue/manager_test.go')
-rw-r--r--modules/queue/manager_test.go124
1 files changed, 124 insertions, 0 deletions
diff --git a/modules/queue/manager_test.go b/modules/queue/manager_test.go
new file mode 100644
index 0000000000..50265e27b6
--- /dev/null
+++ b/modules/queue/manager_test.go
@@ -0,0 +1,124 @@
+// Copyright 2023 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package queue
+
+import (
+ "context"
+ "path/filepath"
+ "testing"
+
+ "code.gitea.io/gitea/modules/setting"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestManager(t *testing.T) {
+ oldAppDataPath := setting.AppDataPath
+ setting.AppDataPath = t.TempDir()
+ defer func() {
+ setting.AppDataPath = oldAppDataPath
+ }()
+
+ newQueueFromConfig := func(name, cfg string) (*WorkerPoolQueue[int], error) {
+ cfgProvider, err := setting.NewConfigProviderFromData(cfg)
+ if err != nil {
+ return nil, err
+ }
+ qs, err := setting.GetQueueSettings(cfgProvider, name)
+ if err != nil {
+ return nil, err
+ }
+ return NewWorkerPoolQueueBySetting(name, qs, func(s ...int) (unhandled []int) { return nil }, false)
+ }
+
+ // test invalid CONN_STR
+ _, err := newQueueFromConfig("default", `
+[queue]
+DATADIR = temp-dir
+CONN_STR = redis://
+`)
+ assert.ErrorContains(t, err, "invalid leveldb connection string")
+
+ // test default config
+ q, err := newQueueFromConfig("default", "")
+ assert.NoError(t, err)
+ assert.Equal(t, "default", q.GetName())
+ assert.Equal(t, "level", q.GetType())
+ assert.Equal(t, filepath.Join(setting.AppDataPath, "queues/common"), q.baseConfig.DataFullDir)
+ assert.Equal(t, 100, q.baseConfig.Length)
+ assert.Equal(t, 20, q.batchLength)
+ assert.Equal(t, "", q.baseConfig.ConnStr)
+ assert.Equal(t, "default_queue", q.baseConfig.QueueFullName)
+ assert.Equal(t, "default_queue_unique", q.baseConfig.SetFullName)
+ assert.Equal(t, 10, q.GetWorkerMaxNumber())
+ assert.Equal(t, 0, q.GetWorkerNumber())
+ assert.Equal(t, 0, q.GetWorkerActiveNumber())
+ assert.Equal(t, 0, q.GetQueueItemNumber())
+ assert.Equal(t, "int", q.GetItemTypeName())
+
+ // test inherited config
+ cfgProvider, err := setting.NewConfigProviderFromData(`
+[queue]
+TYPE = channel
+DATADIR = queues/dir1
+LENGTH = 100
+BATCH_LENGTH = 20
+CONN_STR = "addrs=127.0.0.1:6379 db=0"
+QUEUE_NAME = _queue1
+
+[queue.sub]
+TYPE = level
+DATADIR = queues/dir2
+LENGTH = 102
+BATCH_LENGTH = 22
+CONN_STR =
+QUEUE_NAME = _q2
+SET_NAME = _u2
+MAX_WORKERS = 2
+`)
+
+ assert.NoError(t, err)
+
+ q1 := createWorkerPoolQueue[string]("no-such", cfgProvider, nil, false)
+ assert.Equal(t, "no-such", q1.GetName())
+ assert.Equal(t, "dummy", q1.GetType()) // no handler, so it becomes dummy
+ assert.Equal(t, filepath.Join(setting.AppDataPath, "queues/dir1"), q1.baseConfig.DataFullDir)
+ assert.Equal(t, 100, q1.baseConfig.Length)
+ assert.Equal(t, 20, q1.batchLength)
+ assert.Equal(t, "addrs=127.0.0.1:6379 db=0", q1.baseConfig.ConnStr)
+ assert.Equal(t, "no-such_queue1", q1.baseConfig.QueueFullName)
+ assert.Equal(t, "no-such_queue1_unique", q1.baseConfig.SetFullName)
+ assert.Equal(t, 10, q1.GetWorkerMaxNumber())
+ assert.Equal(t, 0, q1.GetWorkerNumber())
+ assert.Equal(t, 0, q1.GetWorkerActiveNumber())
+ assert.Equal(t, 0, q1.GetQueueItemNumber())
+ assert.Equal(t, "string", q1.GetItemTypeName())
+ qid1 := GetManager().qidCounter
+
+ q2 := createWorkerPoolQueue("sub", cfgProvider, func(s ...int) (unhandled []int) { return nil }, false)
+ assert.Equal(t, "sub", q2.GetName())
+ assert.Equal(t, "level", q2.GetType())
+ assert.Equal(t, filepath.Join(setting.AppDataPath, "queues/dir2"), q2.baseConfig.DataFullDir)
+ assert.Equal(t, 102, q2.baseConfig.Length)
+ assert.Equal(t, 22, q2.batchLength)
+ assert.Equal(t, "", q2.baseConfig.ConnStr)
+ assert.Equal(t, "sub_q2", q2.baseConfig.QueueFullName)
+ assert.Equal(t, "sub_q2_u2", q2.baseConfig.SetFullName)
+ assert.Equal(t, 2, q2.GetWorkerMaxNumber())
+ assert.Equal(t, 0, q2.GetWorkerNumber())
+ assert.Equal(t, 0, q2.GetWorkerActiveNumber())
+ assert.Equal(t, 0, q2.GetQueueItemNumber())
+ assert.Equal(t, "int", q2.GetItemTypeName())
+ qid2 := GetManager().qidCounter
+
+ assert.Equal(t, q1, GetManager().ManagedQueues()[qid1])
+
+ GetManager().GetManagedQueue(qid1).SetWorkerMaxNumber(120)
+ assert.Equal(t, 120, q1.workerMaxNum)
+
+ stop := runWorkerPoolQueue(q2)
+ assert.NoError(t, GetManager().GetManagedQueue(qid2).FlushWithContext(context.Background(), 0))
+ assert.NoError(t, GetManager().FlushAll(context.Background(), 0))
+ stop()
+}