diff options
Diffstat (limited to 'internal/executor/executor.go')
| -rw-r--r-- | internal/executor/executor.go | 33 |
1 files changed, 24 insertions, 9 deletions
diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 2f01b93..0ad9da9 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -273,29 +273,42 @@ func (p *Pool) getRunner(t *task.Task) (Runner, error) { return runner, nil } -// decActiveAgent decrements the active counters for a finished task. Safe to -// call multiple times — subsequent calls are no-ops via the cleaned flag. -// Always call this before sending on resultCh so consumers observing a result -// see the accounting already settled (no zero-count map entries lingering). +// releaseDispatchSlot decrements the pool-level active counter and unblocks +// the dispatch goroutine. It must be deferred at the top of execute() and +// executeResume() so that every exit path — including early returns before +// the per-agent slot is taken — releases the slot acquired by dispatch(). +func (p *Pool) releaseDispatchSlot() { + p.mu.Lock() + p.active-- + p.mu.Unlock() + select { + case p.doneCh <- struct{}{}: + default: + } +} + +// decActiveAgent decrements the per-agent active counter for a finished task. +// Safe to call multiple times — subsequent calls are no-ops via the cleaned +// flag. Always call this before sending on resultCh so consumers observing a +// result see the accounting already settled (no zero-count map entries +// lingering). The pool-level p.active slot is handled separately by +// releaseDispatchSlot, which is deferred at the top of execute/executeResume. func (p *Pool) decActiveAgent(agentType string, cleaned *bool) { if *cleaned { return } *cleaned = true p.mu.Lock() - p.active-- p.activePerAgent[agentType]-- if p.activePerAgent[agentType] == 0 { delete(p.activePerAgent, agentType) } p.mu.Unlock() - select { - case p.doneCh <- struct{}{}: - default: - } } func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Execution) { + defer p.releaseDispatchSlot() + agentType := t.Agent.Type if agentType == "" { agentType = "claude" @@ -892,6 +905,8 @@ func pickAgent(status SystemStatus) string { } func (p *Pool) execute(ctx context.Context, t *task.Task) { + defer p.releaseDispatchSlot() + // 1. Load-balanced agent selection + model classification. p.mu.Lock() activeTasks := make(map[string]int) |
