diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..1a80735 --- /dev/null +++ b/go.mod @@ -0,0 +1,3 @@ +module github.com/mmccl5/go-task-task + +go 1.26.5 diff --git a/workspace/task/runner.go b/workspace/task/runner.go index 0ba04c6..79ebaa3 100644 --- a/workspace/task/runner.go +++ b/workspace/task/runner.go @@ -3,9 +3,9 @@ package task import ( "context" "sync" - "time" ) +// Runner executes tasks concurrently. type Runner struct { mu sync.RWMutex tasks []*Task @@ -23,6 +23,34 @@ func (r *Runner) AddTask(t *Task) { r.tasks = append(r.tasks, t) } +// RunnerStatus contains a snapshot of the runner's task states. +type RunnerStatus struct { + TaskID string + State State + Err error +} + +// Status returns a thread-safe snapshot of every task's state and error. +func (r *Runner) Status() []RunnerStatus { + r.mu.RLock() + snap := make([]*Task, len(r.tasks)) + copy(snap, r.tasks) + r.mu.RUnlock() + + results := make([]RunnerStatus, len(snap)) + for i, t := range snap { + results[i] = RunnerStatus{ + TaskID: t.ID, + State: t.GetState(), + Err: t.GetErr(), + } + } + return results +} + +// Run executes all tasks concurrently. It uses the task's own public API +// for thread-safe state transitions so that concurrent readers (e.g. the +// race-detector test in runner_test.go) do not cause data races. func (r *Runner) Run(ctx context.Context) { r.mu.RLock() tasks := make([]*Task, len(r.tasks)) @@ -35,35 +63,22 @@ func (r *Runner) Run(ctx context.Context) { go func(task *Task) { defer wg.Done() + // Check whether the context has been cancelled before we start. if ctx.Err() != nil { - task.mu.Lock() - task.State = StateFailed - task.Err = ctx.Err() - task.History = append(task.History, StateFailed) - task.mu.Unlock() + task.TransitionTo(StateFailed, ctx.Err()) return } - task.mu.Lock() - task.State = StateRunning - task.StartedAt = time.Now() - task.History = append(task.History, StateRunning) - task.mu.Unlock() + task.TransitionTo(StateRunning, nil) err := task.Action(ctx) - task.mu.Lock() - task.FinishedAt = time.Now() if err != nil { - task.State = StateFailed - task.Err = err - task.History = append(task.History, StateFailed) + task.TransitionTo(StateFailed, err) } else { - task.State = StateCompleted - task.History = append(task.History, StateCompleted) + task.TransitionTo(StateCompleted, nil) } - task.mu.Unlock() }(t) } wg.Wait() -} \ No newline at end of file +} diff --git a/workspace/task/runner_test.go b/workspace/task/runner_test.go index d02f599..b643efd 100644 --- a/workspace/task/runner_test.go +++ b/workspace/task/runner_test.go @@ -26,11 +26,11 @@ func TestRunner_Run(t *testing.T) { runner.Run(context.Background()) - if t1.State != StateCompleted { - t.Errorf("expected t1 to be Completed, got %s", t1.State) + if s := t1.GetState(); s != StateCompleted { + t.Errorf("expected t1 to be Completed, got %s", s) } - if t2.State != StateFailed { - t.Errorf("expected t2 to be Failed, got %s", t2.State) + if s := t2.GetState(); s != StateFailed { + t.Errorf("expected t2 to be Failed, got %s", s) } } @@ -98,4 +98,204 @@ func TestRunner_Run_Cancel(t *testing.T) { if err := t1.GetErr(); err != context.Canceled { t.Errorf("expected t1 error to be context.Canceled, got %v", err) } -} \ No newline at end of file +} + +func TestRunner_Run_ConcurrentWrites(t *testing.T) { + // Create tasks that use SetMetadata/GetMetadata while running. + runner := NewRunner() + numTasks := 50 + + // Collect tasks for background reading. + taskList := make([]*Task, numTasks) + for i := 0; i < numTasks; i++ { + id := fmt.Sprintf("write-%d", i) + tk := NewTask(id, nil) + tk2 := tk + tk.Action = func(ctx context.Context) error { + for j := 0; j < 5; j++ { + tk2.SetMetadata(fmt.Sprintf("key-%d", j), j) + _, _ = tk2.GetMetadata(fmt.Sprintf("key-%d", j)) + } + return nil + } + runner.AddTask(tk) + taskList[i] = tk + } + + // Concurrently read Metadata from another goroutine while tasks run. + stopChan := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stopChan: + return + default: + for _, t := range taskList { + _ = t.GetState() + _, _ = t.GetMetadata("key-0") + } + time.Sleep(1 * time.Millisecond) + } + } + }() + + runner.Run(context.Background()) + close(stopChan) + wg.Wait() +} + +func TestRunner_Run_Timeout(t *testing.T) { + runner := NewRunner() + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + + t1 := NewTask("slow", func(ctx context.Context) error { + select { + case <-time.After(500 * time.Millisecond): + return nil + case <-ctx.Done(): + return ctx.Err() + } + }) + + runner.AddTask(t1) + runner.Run(ctx) + + if state := t1.GetState(); state != StateFailed { + t.Errorf("expected t1 to be Failed after timeout, got %s", state) + } +} + +func TestRunner_Run_Status(t *testing.T) { + runner := NewRunner() + t1 := NewTask("s1", func(ctx context.Context) error { return nil }) + t2 := NewTask("s2", func(ctx context.Context) error { return errors.New("err") }) + + runner.AddTask(t1) + runner.AddTask(t2) + runner.Run(context.Background()) + + st := runner.Status() + if len(st) != 2 { + t.Fatalf("expected 2 status entries, got %d", len(st)) + } + if st[0].State != StateCompleted && st[0].State != StateFailed { + t.Errorf("unexpected state for task s1: %s", st[0].State) + } + if st[1].State != StateFailed && st[1].State != StateCompleted { + t.Errorf("unexpected state for task s2: %s", st[1].State) + } +} + +func TestRunner_Run_Historical(t *testing.T) { + runner := NewRunner() + t1 := NewTask("hist", func(ctx context.Context) error { return nil }) + + runner.AddTask(t1) + runner.Run(context.Background()) + + hist := t1.GetHistory() + expectedStates := []State{StatePending, StateRunning, StateCompleted} + if len(hist) != len(expectedStates) { + t.Fatalf("expected %d history entries, got %d: %v", len(expectedStates), len(hist), hist) + } + for i, s := range expectedStates { + if hist[i] != s { + t.Errorf("history[%d] = %s, want %s", i, hist[i], s) + } + } +} + +func TestTask_ConcurrentAccessors(t *testing.T) { + task := NewTask("stress", func(ctx context.Context) error { + time.Sleep(20 * time.Millisecond) + return nil + }) + + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 100; j++ { + _ = task.GetState() + _ = task.GetErr() + _, _ = task.GetMetadata("any") + _ = task.GetHistory() + _ = task.GetStartedAt() + _ = task.GetFinishedAt() + } + }() + } + + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 50; j++ { + task.SetState(StateRunning) + task.SetErr(nil) + task.SetMetadata(fmt.Sprintf("k-%d", j), j) + } + task.TransitionTo(StateCompleted, nil) + }() + + wg.Wait() + + if s := task.GetState(); s != StateCompleted { + t.Errorf("expected final state Completed, got %s", s) + } +} + +func TestRunner_Run_MetadataRace(t *testing.T) { + runner := NewRunner() + numTasks := 50 + + var tasks []*Task + for i := 0; i < numTasks; i++ { + id := fmt.Sprintf("%d", i) + tk := NewTask(id, nil) + tk2 := tk + tk.Action = func(ctx context.Context) error { + tk2.SetMetadata("result", "ok") + return nil + } + runner.AddTask(tk) + tasks = append(tasks, tk) + } + + var wg sync.WaitGroup + stopChan := make(chan struct{}) + + for i := 0; i < 3; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stopChan: + return + default: + for _, t := range tasks { + _ = t.GetState() + _, _ = t.GetMetadata("result") + _ = t.GetErr() + } + time.Sleep(500 * time.Microsecond) + } + } + }() + } + + runner.Run(context.Background()) + close(stopChan) + wg.Wait() + + for i, task := range tasks { + if s := task.GetState(); s != StateCompleted { + t.Errorf("task %d had state %s, want Completed", i, s) + } + } +} diff --git a/workspace/task/task.go b/workspace/task/task.go index 8f403dc..f0927ad 100644 --- a/workspace/task/task.go +++ b/workspace/task/task.go @@ -21,7 +21,7 @@ type Task struct { State State Action func(ctx context.Context) error Err error - Metadata map[string]interface{ } + Metadata map[string]interface{} History []State StartedAt time.Time FinishedAt time.Time @@ -37,12 +37,14 @@ func NewTask(id string, action func(ctx context.Context) error) *Task { } } +// GetState returns the current state of the task. func (t *Task) GetState() State { t.mu.RLock() defer t.mu.RUnlock() return t.State } +// SetState updates the task's state and appends it to the history. func (t *Task) SetState(state State) { t.mu.Lock() defer t.mu.Unlock() @@ -50,6 +52,24 @@ func (t *Task) SetState(state State) { t.History = append(t.History, state) } +// TransitionTo is an atomic state-and-error transition. +// It sets the state, appends to history, records the error, +// and records FinishedAt if the state is a terminal one +// (Completed or Failed). +func (t *Task) TransitionTo(state State, err error) { + t.mu.Lock() + defer t.mu.Unlock() + t.State = state + t.History = append(t.History, state) + t.Err = err + if state == StateRunning { + t.StartedAt = time.Now() + } + if state == StateCompleted || state == StateFailed { + t.FinishedAt = time.Now() + } +} + func (t *Task) GetErr() error { t.mu.RLock() defer t.mu.RUnlock() @@ -93,4 +113,4 @@ func (t *Task) GetFinishedAt() time.Time { t.mu.RLock() defer t.mu.RUnlock() return t.FinishedAt -} \ No newline at end of file +}