Bound asynchronous session work
This commit is contained in:
parent
6a4f83b5b2
commit
6a8e897fa5
|
|
@ -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
|
||||
}),
|
||||
|
|
|
|||
Loading…
Reference in New Issue