1
0
mirror of https://github.com/go-gitea/gitea.git synced 2024-12-22 17:34:03 +03:00

Fix queue worker incorrectly stopped when there are still more items in the queue (#29532)

Without `case <-t.C`, the workers would stop incorrectly, the test won't
pass. For the worse case, there might be only one running worker
processing the queue items for long time because other workers are
stopped. The root cause is related to the logic of doDispatchBatchToWorker.
It isn't a serious problem at the moment, so keep it as-is.
This commit is contained in:
wxiaoguang 2024-03-03 00:07:54 +08:00 committed by GitHub
parent a3f05d0d98
commit 6465f94a2d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 42 additions and 9 deletions

View File

@ -60,6 +60,9 @@ func (q *WorkerPoolQueue[T]) doDispatchBatchToWorker(wg *workerGroup[T], flushCh
full = true full = true
} }
// TODO: the logic could be improved in the future, to avoid a data-race between "doStartNewWorker" and "workerNum"
// The root problem is that if we skip "doStartNewWorker" here, the "workerNum" might be decreased by other workers later
// So ideally, it should check whether there are enough workers by some approaches, and start new workers if necessary.
q.workerNumMu.Lock() q.workerNumMu.Lock()
noWorker := q.workerNum == 0 noWorker := q.workerNum == 0
if full || noWorker { if full || noWorker {
@ -143,7 +146,11 @@ func (q *WorkerPoolQueue[T]) doStartNewWorker(wp *workerGroup[T]) {
log.Debug("Queue %q starts new worker", q.GetName()) log.Debug("Queue %q starts new worker", q.GetName())
defer log.Debug("Queue %q stops idle worker", q.GetName()) defer log.Debug("Queue %q stops idle worker", q.GetName())
atomic.AddInt32(&q.workerStartedCounter, 1) // Only increase counter, used for debugging
t := time.NewTicker(workerIdleDuration) t := time.NewTicker(workerIdleDuration)
defer t.Stop()
keepWorking := true keepWorking := true
stopWorking := func() { stopWorking := func() {
q.workerNumMu.Lock() q.workerNumMu.Lock()
@ -158,13 +165,18 @@ func (q *WorkerPoolQueue[T]) doStartNewWorker(wp *workerGroup[T]) {
case batch, ok := <-q.batchChan: case batch, ok := <-q.batchChan:
if !ok { if !ok {
stopWorking() stopWorking()
} else { continue
q.doWorkerHandle(batch) }
t.Reset(workerIdleDuration) q.doWorkerHandle(batch)
// reset the idle ticker, and drain the tick after reset in case a tick is already triggered
t.Reset(workerIdleDuration)
select {
case <-t.C:
default:
} }
case <-t.C: case <-t.C:
q.workerNumMu.Lock() q.workerNumMu.Lock()
keepWorking = q.workerNum <= 1 keepWorking = q.workerNum <= 1 // keep the last worker running
if !keepWorking { if !keepWorking {
q.workerNum-- q.workerNum--
} }

View File

@ -40,6 +40,8 @@ type WorkerPoolQueue[T any] struct {
workerMaxNum int workerMaxNum int
workerActiveNum int workerActiveNum int
workerNumMu sync.Mutex workerNumMu sync.Mutex
workerStartedCounter int32
} }
type flushType chan struct{} type flushType chan struct{}

View File

@ -11,6 +11,7 @@ import (
"time" "time"
"code.gitea.io/gitea/modules/setting" "code.gitea.io/gitea/modules/setting"
"code.gitea.io/gitea/modules/test"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
) )
@ -175,11 +176,7 @@ func testWorkerPoolQueuePersistence(t *testing.T, queueSetting setting.QueueSett
} }
func TestWorkerPoolQueueActiveWorkers(t *testing.T) { func TestWorkerPoolQueueActiveWorkers(t *testing.T) {
oldWorkerIdleDuration := workerIdleDuration defer test.MockVariableValue(&workerIdleDuration, 300*time.Millisecond)()
workerIdleDuration = 300 * time.Millisecond
defer func() {
workerIdleDuration = oldWorkerIdleDuration
}()
handler := func(items ...int) (unhandled []int) { handler := func(items ...int) (unhandled []int) {
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
@ -250,3 +247,25 @@ func TestWorkerPoolQueueShutdown(t *testing.T) {
q, _ = newWorkerPoolQueueForTest("test-workpoolqueue", qs, handler, false) q, _ = newWorkerPoolQueueForTest("test-workpoolqueue", qs, handler, false)
assert.EqualValues(t, 20, q.GetQueueItemNumber()) assert.EqualValues(t, 20, q.GetQueueItemNumber())
} }
func TestWorkerPoolQueueWorkerIdleReset(t *testing.T) {
defer test.MockVariableValue(&workerIdleDuration, 10*time.Millisecond)()
handler := func(items ...int) (unhandled []int) {
time.Sleep(50 * time.Millisecond)
return nil
}
q, _ := newWorkerPoolQueueForTest("test-workpoolqueue", setting.QueueSettings{Type: "channel", BatchLength: 1, MaxWorkers: 2, Length: 100}, handler, false)
stop := runWorkerPoolQueue(q)
for i := 0; i < 20; i++ {
assert.NoError(t, q.Push(i))
}
time.Sleep(500 * time.Millisecond)
assert.EqualValues(t, 2, q.GetWorkerNumber())
assert.EqualValues(t, 2, q.GetWorkerActiveNumber())
// when the queue never becomes empty, the existing workers should keep working
assert.EqualValues(t, 2, q.workerStartedCounter)
stop()
}