diff --git a/pkg/tasks/run_task_matrix/task.go b/pkg/tasks/run_task_matrix/task.go index 1be318eb..88a51eeb 100644 --- a/pkg/tasks/run_task_matrix/task.go +++ b/pkg/tasks/run_task_matrix/task.go @@ -123,6 +123,13 @@ func (t *Task) Execute(ctx context.Context) error { var currentTaskWaitChan, previousTaskWaitChan chan bool + // populate the task index map before spawning any child goroutines. + // the watcher goroutines read this map, so writing it here first avoids + // a concurrent map read/write against the spawn loop below. + for i := range t.tasks { + t.taskIdxMap[t.tasks[i]] = i + } + // start child tasks for i := range t.tasks { taskWaitGroup.Add(1) @@ -132,8 +139,6 @@ func (t *Task) Execute(ctx context.Context) error { currentTaskWaitChan = make(chan bool) } - t.taskIdxMap[t.tasks[i]] = i - go func(i int, taskWaitChan, prevTaskWaitChan chan bool) { defer taskWaitGroup.Done() diff --git a/pkg/tasks/run_task_matrix/task_test.go b/pkg/tasks/run_task_matrix/task_test.go new file mode 100644 index 00000000..ce1f442a --- /dev/null +++ b/pkg/tasks/run_task_matrix/task_test.go @@ -0,0 +1,124 @@ +package runtaskmatrix + +import ( + "context" + "io" + "testing" + "time" + + "github.com/ethpandaops/assertoor/pkg/helper" + "github.com/ethpandaops/assertoor/pkg/types" + "github.com/sirupsen/logrus" +) + +// stubTaskState reports a finished, successful child task so the watcher reads +// the task index map once and exits. +type stubTaskState struct { + index types.TaskIndex +} + +func (s *stubTaskState) Index() types.TaskIndex { return s.index } +func (s *stubTaskState) ParentIndex() types.TaskIndex { return 0 } +func (s *stubTaskState) ID() string { return "" } +func (s *stubTaskState) Name() string { return "stub" } +func (s *stubTaskState) Title() string { return "stub" } +func (s *stubTaskState) Description() string { return "" } +func (s *stubTaskState) Config() any { return nil } +func (s *stubTaskState) Timeout() time.Duration { return 0 } +func (s *stubTaskState) GetTaskStatusVars() types.Variables { return nil } +func (s *stubTaskState) GetScopeOwner() types.TaskIndex { return 0 } + +func (s *stubTaskState) GetTaskStatus() *types.TaskStatus { + return &types.TaskStatus{Index: s.index, IsRunning: false, Result: types.TaskResultSuccess} +} + +func (s *stubTaskState) GetTaskResultUpdateChan(oldResult types.TaskResult) <-chan bool { + return nil +} + +// stubScheduler runs the supplied watch function in its own goroutine, which is +// what concurrently reads the task index map. +type stubScheduler struct{} + +func (s *stubScheduler) GetTaskState(taskIndex types.TaskIndex) types.TaskState { + return &stubTaskState{index: taskIndex} +} +func (s *stubScheduler) GetTaskCount() uint64 { return 0 } +func (s *stubScheduler) GetAllTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetRootTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetAllCleanupTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetRootCleanupTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetServices() types.TaskServices { return nil } +func (s *stubScheduler) GetTestRunID() uint64 { return 0 } +func (s *stubScheduler) GetTestRunCtx() context.Context { return context.Background() } +func (s *stubScheduler) TestResultPath() (string, error) { return "", nil } + +func (s *stubScheduler) ParseTaskOptions(_ helper.IRawMessage) (*types.TaskOptions, error) { + return nil, nil +} + +func (s *stubScheduler) ExecuteTask(ctx context.Context, taskIndex types.TaskIndex, taskWatchFn func(ctx context.Context, cancelFn context.CancelFunc, taskIndex types.TaskIndex)) error { + if taskWatchFn != nil { + watchCtx, cancel := context.WithCancel(ctx) + go taskWatchFn(watchCtx, cancel, taskIndex) + } + + return nil +} + +var ( + _ types.TaskSchedulerRunner = (*stubScheduler)(nil) + _ types.TaskState = (*stubTaskState)(nil) +) + +// TestExecuteTaskIndexMapNoRace runs the matrix executor with many children so the +// per-child watcher goroutines read taskIdxMap while the executor spawns the +// remaining children. The index map must be fully written before any goroutine +// starts, otherwise this races (run with -race). +func TestExecuteTaskIndexMapNoRace(t *testing.T) { + const childCount = 64 + + logger := logrus.New() + logger.SetOutput(io.Discard) + logger.SetLevel(logrus.PanicLevel) + + tasks := make([]types.TaskIndex, childCount) + for i := range tasks { + tasks[i] = types.TaskIndex(i + 1) + } + + config := DefaultConfig() + config.RunConcurrent = true + + task := &Task{ + config: config, + logger: logger, + tasks: tasks, + taskIdxMap: map[types.TaskIndex]int{}, + resultNotifyChan: make(chan taskResultUpdate, childCount), + } + task.ctx = &types.TaskContext{ + Scheduler: &stubScheduler{}, + Index: types.TaskIndex(1000), + SetResult: func(types.TaskResult) {}, + ReportProgress: func(float64, string) {}, + } + + done := make(chan error, 1) + go func() { + done <- task.Execute(context.Background()) + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("Execute returned error: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("Execute did not complete in time") + } + + if len(task.taskIdxMap) != childCount { + t.Fatalf("taskIdxMap has %d entries, want %d", len(task.taskIdxMap), childCount) + } +} diff --git a/pkg/tasks/run_tasks_concurrent/task.go b/pkg/tasks/run_tasks_concurrent/task.go index b8049cde..6a1f4499 100644 --- a/pkg/tasks/run_tasks_concurrent/task.go +++ b/pkg/tasks/run_tasks_concurrent/task.go @@ -124,12 +124,17 @@ func (t *Task) Execute(ctx context.Context) error { t.taskCtx = taskCtx + // populate the task index map before spawning any child goroutines. + // the watcher goroutines read this map, so writing it here first avoids + // a concurrent map read/write against the spawn loop below. + for i := range t.tasks { + t.taskIdxMap[t.tasks[i]] = i + } + // start child tasks for i := range t.tasks { taskWaitGroup.Add(1) - t.taskIdxMap[t.tasks[i]] = i - go func(i int) { defer taskWaitGroup.Done() diff --git a/pkg/tasks/run_tasks_concurrent/task_test.go b/pkg/tasks/run_tasks_concurrent/task_test.go new file mode 100644 index 00000000..e5ad89e2 --- /dev/null +++ b/pkg/tasks/run_tasks_concurrent/task_test.go @@ -0,0 +1,122 @@ +package runtasksconcurrent + +import ( + "context" + "io" + "testing" + "time" + + "github.com/ethpandaops/assertoor/pkg/helper" + "github.com/ethpandaops/assertoor/pkg/types" + "github.com/sirupsen/logrus" +) + +// stubTaskState reports a finished, successful child task so the watcher reads +// the task index map once and exits. +type stubTaskState struct { + index types.TaskIndex +} + +func (s *stubTaskState) Index() types.TaskIndex { return s.index } +func (s *stubTaskState) ParentIndex() types.TaskIndex { return 0 } +func (s *stubTaskState) ID() string { return "" } +func (s *stubTaskState) Name() string { return "stub" } +func (s *stubTaskState) Title() string { return "stub" } +func (s *stubTaskState) Description() string { return "" } +func (s *stubTaskState) Config() any { return nil } +func (s *stubTaskState) Timeout() time.Duration { return 0 } +func (s *stubTaskState) GetTaskStatusVars() types.Variables { return nil } +func (s *stubTaskState) GetScopeOwner() types.TaskIndex { return 0 } + +func (s *stubTaskState) GetTaskStatus() *types.TaskStatus { + return &types.TaskStatus{Index: s.index, IsRunning: false, Result: types.TaskResultSuccess} +} + +func (s *stubTaskState) GetTaskResultUpdateChan(oldResult types.TaskResult) <-chan bool { + return nil +} + +// stubScheduler mirrors the real scheduler's contract that matters for this test: +// ExecuteTask runs the supplied watch function in its own goroutine, which is what +// concurrently reads the task index map. +type stubScheduler struct{} + +func (s *stubScheduler) GetTaskState(taskIndex types.TaskIndex) types.TaskState { + return &stubTaskState{index: taskIndex} +} +func (s *stubScheduler) GetTaskCount() uint64 { return 0 } +func (s *stubScheduler) GetAllTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetRootTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetAllCleanupTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetRootCleanupTasks() []types.TaskIndex { return nil } +func (s *stubScheduler) GetServices() types.TaskServices { return nil } +func (s *stubScheduler) GetTestRunID() uint64 { return 0 } +func (s *stubScheduler) GetTestRunCtx() context.Context { return context.Background() } +func (s *stubScheduler) TestResultPath() (string, error) { return "", nil } + +func (s *stubScheduler) ParseTaskOptions(_ helper.IRawMessage) (*types.TaskOptions, error) { + return nil, nil +} + +func (s *stubScheduler) ExecuteTask(ctx context.Context, taskIndex types.TaskIndex, taskWatchFn func(ctx context.Context, cancelFn context.CancelFunc, taskIndex types.TaskIndex)) error { + if taskWatchFn != nil { + watchCtx, cancel := context.WithCancel(ctx) + go taskWatchFn(watchCtx, cancel, taskIndex) + } + + return nil +} + +var ( + _ types.TaskSchedulerRunner = (*stubScheduler)(nil) + _ types.TaskState = (*stubTaskState)(nil) +) + +// TestExecuteTaskIndexMapNoRace runs the concurrent executor with many children so +// the per-child watcher goroutines read taskIdxMap while the executor spawns the +// remaining children. The index map must be fully written before any goroutine +// starts, otherwise this races (run with -race). +func TestExecuteTaskIndexMapNoRace(t *testing.T) { + const childCount = 64 + + logger := logrus.New() + logger.SetOutput(io.Discard) + logger.SetLevel(logrus.PanicLevel) + + tasks := make([]types.TaskIndex, childCount) + for i := range tasks { + tasks[i] = types.TaskIndex(i + 1) + } + + task := &Task{ + config: DefaultConfig(), + logger: logger, + tasks: tasks, + taskIdxMap: map[types.TaskIndex]int{}, + resultNotifyChan: make(chan taskResultUpdate, childCount), + } + task.ctx = &types.TaskContext{ + Scheduler: &stubScheduler{}, + Index: types.TaskIndex(1000), + SetResult: func(types.TaskResult) {}, + ReportProgress: func(float64, string) {}, + } + + done := make(chan error, 1) + go func() { + done <- task.Execute(context.Background()) + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("Execute returned error: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("Execute did not complete in time") + } + + if len(task.taskIdxMap) != childCount { + t.Fatalf("taskIdxMap has %d entries, want %d", len(task.taskIdxMap), childCount) + } +}