mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-30 18:41:49 -07:00
Fix SequentialScheduler test deadlock from bounded test task queue (#11772)
## What changed? Back `testSequentialTaskQueue` with an unbounded slice instead of a 3000-capacity channel, so `Add` no longer blocks. ## Why? `common/tasks` deadlocked and timed out after 15m on main ([run 32773080605](https://github.com/temporalio/temporal/actions/runs/32773080605)). `SequentialScheduler.Submit` calls `Add` from inside `PutOrDo`'s callback, which runs under the shard write lock. Once the test queue's channel filled, the submitter blocked in `Add` while holding that lock, and the only worker was blocked on the same lock in `RemoveIf`. Holding the lock across the add is deliberate — it keeps the add atomic against `RemoveIf`'s empty-check — so the invariant is that `Add` must not block, and only the test queue violated it. ## How did you test it? - [x] built - [x] run locally and tested manually - [x] covered by existing tests
This commit is contained in:
@@ -385,7 +385,7 @@ func (s *sequentialSchedulerSuite) newTestProcessor() *SequentialScheduler[*Mock
|
||||
return 1
|
||||
}
|
||||
factory := func(task *MockTask) SequentialTaskQueue[*MockTask] {
|
||||
return newTestSequentialTaskQueue[*MockTask](1, 3000)
|
||||
return newTestSequentialTaskQueue[*MockTask](1)
|
||||
}
|
||||
return NewSequentialScheduler[*MockTask](
|
||||
&SequentialSchedulerOptions{
|
||||
@@ -405,7 +405,7 @@ func (s *sequentialSchedulerSuite) newTestProcessorWithQueueSize(queueSize int)
|
||||
return 1
|
||||
}
|
||||
factory := func(task *MockTask) SequentialTaskQueue[*MockTask] {
|
||||
return newTestSequentialTaskQueue[*MockTask](1, 3000)
|
||||
return newTestSequentialTaskQueue[*MockTask](1)
|
||||
}
|
||||
return NewSequentialScheduler[*MockTask](
|
||||
&SequentialSchedulerOptions{
|
||||
|
||||
@@ -1,13 +1,15 @@
|
||||
package tasks
|
||||
|
||||
import "sync"
|
||||
|
||||
type testSequentialTaskQueue[T Task] struct {
|
||||
q chan T
|
||||
id int
|
||||
sync.Mutex
|
||||
tasks []T
|
||||
id int
|
||||
}
|
||||
|
||||
func newTestSequentialTaskQueue[T Task](id, capacity int) SequentialTaskQueue[T] {
|
||||
func newTestSequentialTaskQueue[T Task](id int) SequentialTaskQueue[T] {
|
||||
return &testSequentialTaskQueue[T]{
|
||||
q: make(chan T, capacity),
|
||||
id: id,
|
||||
}
|
||||
}
|
||||
@@ -17,23 +19,30 @@ func (s *testSequentialTaskQueue[T]) ID() any {
|
||||
}
|
||||
|
||||
func (s *testSequentialTaskQueue[T]) Add(task T) {
|
||||
s.q <- task
|
||||
s.Lock()
|
||||
defer s.Unlock()
|
||||
s.tasks = append(s.tasks, task)
|
||||
}
|
||||
|
||||
func (s *testSequentialTaskQueue[T]) Remove() T {
|
||||
select {
|
||||
case t := <-s.q:
|
||||
return t
|
||||
default:
|
||||
var emptyT T
|
||||
return emptyT
|
||||
s.Lock()
|
||||
defer s.Unlock()
|
||||
var task T
|
||||
if len(s.tasks) == 0 {
|
||||
return task
|
||||
}
|
||||
task, s.tasks = s.tasks[0], s.tasks[1:]
|
||||
return task
|
||||
}
|
||||
|
||||
func (s *testSequentialTaskQueue[T]) IsEmpty() bool {
|
||||
return len(s.q) == 0
|
||||
s.Lock()
|
||||
defer s.Unlock()
|
||||
return len(s.tasks) == 0
|
||||
}
|
||||
|
||||
func (s *testSequentialTaskQueue[T]) Len() int {
|
||||
return len(s.q)
|
||||
s.Lock()
|
||||
defer s.Unlock()
|
||||
return len(s.tasks)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user