diff --git a/cmd/olliesrv/internal/fs/spec.go b/cmd/olliesrv/internal/fs/spec.go index 1d9f995..ae22776 100644 --- a/cmd/olliesrv/internal/fs/spec.go +++ b/cmd/olliesrv/internal/fs/spec.go @@ -40,9 +40,28 @@ const ( var treeSpecUID, treeSpecGID string +const asyncWorkLimit = 64 + +var asyncWorkSlots = make(chan struct{}, asyncWorkLimit) + +func startAsync(ctx context.Context, fn func()) bool { + select { + case asyncWorkSlots <- struct{}{}: + case <-ctx.Done(): + return false + default: + return false + } + go func() { + defer func() { <-asyncWorkSlots }() + fn() + }() + return true +} + // runWorkflow executes a workflow script in the background. func runWorkflow(s *session.Session, workflow string) { - go func() { + startAsync(s.Ctx, func() { scriptPath := util.CfgDir() + "/workflows/" + workflow if _, err := os.Stat(scriptPath); err != nil { s.SetGoalStatus("error: workflow not found: " + workflow) @@ -65,7 +84,7 @@ func runWorkflow(s *session.Session, workflow string) { s.SetGoalStatus("error: " + msg) } // Script exited — don't auto-set complete; the conductor agent handles that. - }() + }) } // buildTreeSpec constructs the full 9P namespace. @@ -607,10 +626,10 @@ func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl s.InvalidateModelsCache() return nil } - go func() { + startAsync(s.Ctx, func() { a.Submit(s.Ctx, input) a.EnsureTrailingNewline() - }() + }) return nil }), ), @@ -623,12 +642,14 @@ func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl a.Queue(input) // If the agent is idle, trigger processing. if !a.IsRunning() { - go func() { + if !startAsync(s.Ctx, func() { if next, ok := a.PopQueue(); ok { a.Submit(s.Ctx, next) a.EnsureTrailingNewline() } - }() + }) { + return fmt.Errorf("async work limit reached") + } } return nil }),