agent: replace sync.Cond with close-channel signal for WaitChange
sync.Cond is the wrong primitive for Go — it requires holding a mutex during Wait and needs context.AfterFunc hacks to integrate with ctx cancellation. Replace with a simple signal channel: notifyChange() closes it and swaps in a fresh one; WaitChange() selects on the channel and ctx.Done(). Simpler, idiomatic, no spurious wakeups.
This commit is contained in:
parent
5b11e8afcb
commit
57984dfa3f
|
|
@ -41,10 +41,10 @@ type Agent struct {
|
|||
fifo Fifo // prompt queue
|
||||
toolCallCount atomic.Int64
|
||||
pendingInject atomic.Pointer[string]
|
||||
submitMu sync.Mutex // serializes Submit calls (commands + turns)
|
||||
stateMu sync.RWMutex
|
||||
changeMu sync.Mutex
|
||||
changeCond *sync.Cond
|
||||
submitMu sync.Mutex // serializes Submit calls (commands + turns)
|
||||
stateMu sync.RWMutex
|
||||
signalMu sync.Mutex
|
||||
signalCh chan struct{} // closed on state change; replaced with a fresh channel
|
||||
|
||||
// Event handler — receives events from this agent.
|
||||
output EventHandler
|
||||
|
|
@ -171,43 +171,36 @@ func (ag *Agent) setReply(reply string) {
|
|||
|
||||
// notifyChange wakes all goroutines waiting on state changes.
|
||||
func (ag *Agent) notifyChange() {
|
||||
ag.changeMu.Lock()
|
||||
ag.changeCond.Broadcast()
|
||||
ag.changeMu.Unlock()
|
||||
ag.signalMu.Lock()
|
||||
close(ag.signalCh)
|
||||
ag.signalCh = make(chan struct{})
|
||||
ag.signalMu.Unlock()
|
||||
}
|
||||
|
||||
// WaitChange blocks until the agent's state differs from current.
|
||||
// Returns the new value and true, or ("", false) if ctx is cancelled.
|
||||
func (ag *Agent) WaitChange(ctx context.Context, field, current string) (string, bool) {
|
||||
done := make(chan struct{})
|
||||
context.AfterFunc(ctx, func() {
|
||||
ag.changeMu.Lock()
|
||||
ag.changeCond.Broadcast()
|
||||
ag.changeMu.Unlock()
|
||||
close(done)
|
||||
})
|
||||
|
||||
ag.changeMu.Lock()
|
||||
for {
|
||||
// Snapshot the signal channel before reading state.
|
||||
ag.signalMu.Lock()
|
||||
ch := ag.signalCh
|
||||
ag.signalMu.Unlock()
|
||||
|
||||
var val string
|
||||
switch field {
|
||||
case WatchState:
|
||||
val = ag.State()
|
||||
default:
|
||||
ag.changeMu.Unlock()
|
||||
return "", false
|
||||
}
|
||||
if val != current {
|
||||
ag.changeMu.Unlock()
|
||||
return val, true
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
ag.changeMu.Unlock()
|
||||
return "", false
|
||||
}
|
||||
ag.changeCond.Wait()
|
||||
if ctx.Err() != nil {
|
||||
ag.changeMu.Unlock()
|
||||
// Wait for either a state change or context cancellation.
|
||||
select {
|
||||
case <-ch:
|
||||
// State changed — loop to re-check.
|
||||
case <-ctx.Done():
|
||||
return "", false
|
||||
}
|
||||
}
|
||||
|
|
@ -588,7 +581,7 @@ func NewAgent(cfg AgentCfg) *Agent {
|
|||
flush: cfg.Flush,
|
||||
state: "idle",
|
||||
}
|
||||
ag.changeCond = sync.NewCond(&ag.changeMu)
|
||||
ag.signalCh = make(chan struct{})
|
||||
ag.wireToolsChanged()
|
||||
return ag
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue