Files
mesh-host/cmd/mesh-host/marker.go
T
jschoubben c1b8b9d115 The composition pass never waits on marking haves, and the compose verb answers while a pass runs
The marks of the haves (a refusal, the words a module keeps) read and write each module's have in turn, and each write
walks every have: on a slow bus the pass waited minutes on them (shanks, 2026-11-05 22:22). They now run off the pass,
newest word per kind, each bounded; an ask of compose during a pass is answered at once and queues one more pass.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-05 22:33:21 +01:00

77 lines
2.0 KiB
Go

package main
import (
"context"
"sync"
"time"
)
// markWithin bounds one marking of the haves (what the composer tells its agent: a refusal, the words it keeps). The marks
// read and write each module's have over the bus one after the other; a bus that answers slowly or not at all must hold
// the marking, never the composition pass (issue 1965, shanks 2026-11-05 22:22: a pass waited minutes on them).
var markWithin = 45 * time.Second
// marker runs the composer's marks of the haves off the composition pass: set returns at once, one worker runs what was
// set last under each name, in the order the names were first set, each within markWithin. A mark set while one runs
// replaces the pending mark of its name, so the haves are marked once for the newest word, not once for every pass.
type marker struct {
run context.Context
mu sync.Mutex
pending map[string]func(context.Context)
order []string
working bool
idle *sync.Cond
}
func newMarker(run context.Context) *marker {
m := &marker{run: run, pending: map[string]func(context.Context){}}
m.idle = sync.NewCond(&m.mu)
return m
}
// set takes the mark name should carry; it never waits on the bus.
func (m *marker) set(name string, mark func(context.Context)) {
m.mu.Lock()
defer m.mu.Unlock()
if _, queued := m.pending[name]; !queued {
m.order = append(m.order, name)
}
m.pending[name] = mark
if m.working {
return
}
m.working = true
go m.work()
}
func (m *marker) work() {
for {
m.mu.Lock()
if len(m.order) == 0 || m.run.Err() != nil {
m.order, m.pending = nil, map[string]func(context.Context){}
m.working = false
m.idle.Broadcast()
m.mu.Unlock()
return
}
name := m.order[0]
m.order = m.order[1:]
mark := m.pending[name]
delete(m.pending, name)
m.mu.Unlock()
ctx, cancel := context.WithTimeout(m.run, markWithin)
mark(ctx)
cancel()
}
}
// wait returns once no mark is pending or running (for tests).
func (m *marker) wait() {
m.mu.Lock()
defer m.mu.Unlock()
for m.working {
m.idle.Wait()
}
}