eventwait: redesign as global /eventwait with structured delta events
BREAKING CHANGE: /session/eventwait is removed. Use /eventwait instead.
The old event system had several problems:
1. eventwait lived inside /session/ (per-session) but events are global
2. It returned the entire session index on every event, growing unboundedly
3. No structured event data — consumers couldn't tell WHAT happened
New design:
- Global /eventwait at the root of the 9P namespace
- Events are structured delta lines: '<scope> <action> <path>'
e.g. 'S new session/foo', 'A kill session/foo/agent/bar'
- Ring buffer capped at 100 events (bounded memory)
- Offset-based blocking reads (consumers track position)
- Per-event granularity: S=new/kill/rename, A=new/kill/rename
Call sites updated:
- Session create → 'S new session/{name}'
- Agent create → 'A new session/{name}/agent/{aid}'
- Session kill → 'A kill ...' per agent + 'S kill session/{name}'
- Session rename → 'S rename session/{old} session/{new}'
- Agent rename → 'A rename session/{name}/agent/{old} ...'
KDE GUI updated to use ollie-9p read /eventwait instead of
plan9port 9p read session/eventwait.
This commit is contained in:
parent
8d89ac31eb
commit
fd5b215268
|
|
@ -377,11 +377,12 @@ func NewRootStore(reg *toolsrv.Registry) *fs.Tree {
|
|||
}
|
||||
return "", fmt.Errorf("%s: not found", name)
|
||||
}),
|
||||
fs.WithLister(func(_ []string) ([]os.DirEntry, error) {
|
||||
fs.WithLister(func(_ []string) ([]os.DirEntry, error) {
|
||||
return []os.DirEntry{
|
||||
fs.FileEntry("backends", fs.Perms[fs.PathRoot].FileMode),
|
||||
fs.FileEntry("ctl", 0666),
|
||||
fs.DirEntry("elevate", 0755),
|
||||
fs.FileEntry("eventwait", fs.Perms[fs.PathRoot].FileMode),
|
||||
fs.FileEntry("generate", 0666),
|
||||
fs.FileEntry("models", fs.Perms[fs.PathRoot].FileMode),
|
||||
fs.FileEntry("help", fs.Perms[fs.PathRoot].FileMode),
|
||||
|
|
@ -389,6 +390,19 @@ func NewRootStore(reg *toolsrv.Registry) *fs.Tree {
|
|||
}, nil
|
||||
}),
|
||||
fs.WithOpener(func(_ []string, name string) (fs.File, error) {
|
||||
if name == "eventwait" {
|
||||
return &fs.FileConfig{
|
||||
StatFn: func() (os.FileInfo, error) {
|
||||
return &fs.SyntheticFileInfo{Name_: "eventwait", Mode_: fs.Perms[fs.PathRoot].Files["eventwait"]}, nil
|
||||
},
|
||||
ReadFn: func() ([]byte, error) {
|
||||
return nil, fmt.Errorf("eventwait: use blocking read")
|
||||
},
|
||||
WriteFn: readOnly,
|
||||
BlockingReadFn: fs.WaitEvent,
|
||||
IsBlocking_: true,
|
||||
}, nil
|
||||
}
|
||||
if name == "complete" {
|
||||
return &fs.FileConfig{
|
||||
StatFn: func() (os.FileInfo, error) { return &fs.SyntheticFileInfo{Name_: "complete", Mode_: 0666}, nil },
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ func Create(rs *rootState, args []string) (string, error) {
|
|||
rs.mu.Unlock()
|
||||
|
||||
rs.cfg.Log.Info("created empty session %s (id=%s)", name, sessID)
|
||||
broadcastEvents()
|
||||
PostEvent("S new session/" + name)
|
||||
return name, nil
|
||||
}
|
||||
|
||||
|
|
@ -323,7 +323,7 @@ func CreateAgent(rs *rootState, sessName string, args []string) error {
|
|||
|
||||
rs.cfg.Log.Info("created agent in session %s (backend=%s model=%s agent=%s)",
|
||||
sessName, ag.BackendName(), ag.ModelName(), ag.Name())
|
||||
broadcastEvents()
|
||||
PostEvent("A new session/" + sessName + "/agent/" + ag.Name())
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -41,8 +41,11 @@ var Perms = map[string]Perm{
|
|||
DirMode: 0755,
|
||||
FileMode: 0444,
|
||||
Files: map[string]os.FileMode{
|
||||
"backends": 0444,
|
||||
"help": 0444,
|
||||
"backends": 0444,
|
||||
"help": 0444,
|
||||
"models": 0444,
|
||||
"agents": 0444,
|
||||
"eventwait": 0444,
|
||||
},
|
||||
},
|
||||
PathElevate: {
|
||||
|
|
@ -57,9 +60,8 @@ var Perms = map[string]Perm{
|
|||
DirMode: 0755,
|
||||
FileMode: 0444,
|
||||
Files: map[string]os.FileMode{
|
||||
"new": 0666,
|
||||
"idx": 0444,
|
||||
"eventwait": 0444,
|
||||
"new": 0666,
|
||||
"idx": 0444,
|
||||
},
|
||||
},
|
||||
PathSessionDir: {DirMode: 0755},
|
||||
|
|
|
|||
|
|
@ -23,17 +23,25 @@ func Kill(rs *rootState, id string) {
|
|||
delete(rs.sessions, id)
|
||||
rs.mu.Unlock()
|
||||
if sess != nil {
|
||||
sess.Cancel()
|
||||
// Capture agent names before cleanup for events.
|
||||
var agentNames []string
|
||||
if sess.Core != nil {
|
||||
for _, ag := range sess.Core.Agents() {
|
||||
agentNames = append(agentNames, ag.Name())
|
||||
}
|
||||
sess.Core.Close()
|
||||
}
|
||||
sess.Cancel()
|
||||
if sess.proc != nil {
|
||||
sess.proc.Close()
|
||||
}
|
||||
removePersistedAgent(rs, sess.id)
|
||||
rs.cfg.Log.Info("killed session %s", id)
|
||||
for _, name := range agentNames {
|
||||
PostEvent("A kill session/" + id + "/agent/" + name)
|
||||
}
|
||||
PostEvent("S kill session/" + id)
|
||||
}
|
||||
broadcastEvents()
|
||||
}
|
||||
|
||||
// KillFromRoot is the public API for callers with *Tree.
|
||||
|
|
@ -65,7 +73,7 @@ func Rename(rs *rootState, old, new string) error {
|
|||
rs.mu.Unlock()
|
||||
|
||||
rs.cfg.Log.Info("renamed session %s -> %s", old, new)
|
||||
broadcastEvents()
|
||||
PostEvent("S rename session/" + old + " session/" + new)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -17,45 +17,81 @@ import (
|
|||
"ollie/toolsrv"
|
||||
)
|
||||
|
||||
// eventBroadcaster manages notification of session index changes.
|
||||
// The GUI can tail session/eventwait to detect creates, deletes, and renames.
|
||||
var (
|
||||
evMu sync.Mutex
|
||||
evChans []chan struct{}
|
||||
)
|
||||
// Event ring buffer — global lifecycle events for session/agent changes.
|
||||
// Consumers read from /eventwait (blocking, offset-based).
|
||||
const eventRingCap = 100
|
||||
|
||||
func broadcastEvents() {
|
||||
evMu.Lock()
|
||||
for _, ch := range evChans {
|
||||
select {
|
||||
case ch <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
type eventRing struct {
|
||||
mu sync.Mutex
|
||||
cond sync.Cond
|
||||
buf []string // circular: buf[start%cap]..buf[(start+count-1)%cap]
|
||||
cap int
|
||||
start int
|
||||
count int
|
||||
total int64 // monotonic counter; doubles as stable offset
|
||||
}
|
||||
|
||||
var evRing = func() eventRing {
|
||||
e := eventRing{cap: eventRingCap, buf: make([]string, eventRingCap)}
|
||||
e.cond.L = &e.mu
|
||||
return e
|
||||
}()
|
||||
|
||||
// PostEvent appends a structured event line and wakes all waiters.
|
||||
// Event format: "<scope> <action> <path>" e.g. "S new session/foo"
|
||||
func PostEvent(line string) {
|
||||
evRing.mu.Lock()
|
||||
idx := (evRing.start + evRing.count) % evRing.cap
|
||||
evRing.buf[idx] = line
|
||||
if evRing.count < evRing.cap {
|
||||
evRing.count++
|
||||
} else {
|
||||
evRing.start = (evRing.start + 1) % evRing.cap
|
||||
}
|
||||
evMu.Unlock()
|
||||
evRing.total++
|
||||
evRing.mu.Unlock()
|
||||
evRing.cond.Broadcast()
|
||||
}
|
||||
|
||||
// eventRegister returns a channel that fires on the next session index change.
|
||||
func eventRegister() chan struct{} {
|
||||
ch := make(chan struct{}, 1)
|
||||
evMu.Lock()
|
||||
evChans = append(evChans, ch)
|
||||
evMu.Unlock()
|
||||
return ch
|
||||
}
|
||||
// WaitEvent blocks until events are available since the given offset (string-encoded int64).
|
||||
// Returns new event lines and the updated offset for the next call.
|
||||
func WaitEvent(ctx context.Context, offsetStr string) ([]byte, string, error) {
|
||||
evRing.mu.Lock()
|
||||
defer evRing.mu.Unlock()
|
||||
|
||||
func eventRelease(ch chan struct{}) {
|
||||
evMu.Lock()
|
||||
for i, c := range evChans {
|
||||
if c == ch {
|
||||
evChans = append(evChans[:i], evChans[i+1:]...)
|
||||
break
|
||||
}
|
||||
var offset int64
|
||||
if offsetStr != "" {
|
||||
fmt.Sscanf(offsetStr, "%d", &offset)
|
||||
} else {
|
||||
offset = evRing.total
|
||||
}
|
||||
evMu.Unlock()
|
||||
|
||||
// Block until new events or context cancelled.
|
||||
for evRing.total <= offset {
|
||||
if ctx.Err() != nil {
|
||||
return nil, "", ctx.Err()
|
||||
}
|
||||
evRing.cond.Wait()
|
||||
}
|
||||
|
||||
// Calculate the index range for events after `offset`.
|
||||
// offset is the total count at last read. We need events
|
||||
// from position (evRing.total - evRing.count) to evRing.total.
|
||||
base := evRing.total - int64(evRing.count)
|
||||
if offset < base {
|
||||
offset = base // consumer fell behind — give them everything we have
|
||||
}
|
||||
skip := int(offset - base)
|
||||
var sb strings.Builder
|
||||
for i := 0; i < evRing.count-skip; i++ {
|
||||
idx := (evRing.start + skip + i) % evRing.cap
|
||||
sb.WriteString(evRing.buf[idx])
|
||||
sb.WriteByte('\n')
|
||||
}
|
||||
return []byte(sb.String()), fmt.Sprintf("%d", evRing.total), nil
|
||||
}
|
||||
|
||||
var rootFileOrder = []string{"new", "idx", "eventwait"}
|
||||
var rootFileOrder = []string{"new", "idx"}
|
||||
|
||||
// FileMode returns the mode for a fixed session file,
|
||||
// or 0 and false if the name is not a fixed file.
|
||||
|
|
@ -157,9 +193,9 @@ func NewRoot(cfg Config) *Tree {
|
|||
} else {
|
||||
sess.Core.Agent().SetName(new)
|
||||
}
|
||||
rs.cfg.Log.Info("renamed agent in session %s -> %s", sessName, new)
|
||||
broadcastEvents()
|
||||
return nil
|
||||
rs.cfg.Log.Info("renamed agent in session %s -> %s", sessName, new)
|
||||
PostEvent("A rename session/" + sessName + "/agent/" + aid + " session/" + sessName + "/agent/" + new)
|
||||
return nil
|
||||
}
|
||||
return Rename(rs, old, new)
|
||||
}),
|
||||
|
|
@ -426,30 +462,9 @@ func rootOpen(rs *rootState, name string) (File, error) {
|
|||
}, nil
|
||||
}
|
||||
|
||||
switch name {
|
||||
case "eventwait":
|
||||
return &FileConfig{
|
||||
StatFn: func() (os.FileInfo, error) {
|
||||
return &SyntheticFileInfo{Name_: "eventwait", Mode_: Perms[PathSessions].Files["eventwait"]}, nil
|
||||
},
|
||||
ReadFn: func() ([]byte, error) {
|
||||
return []byte(buildIndex(rs)), nil
|
||||
},
|
||||
WriteFn: func([]byte) error { return fmt.Errorf("eventwait: read-only") },
|
||||
BlockingReadFn: func(ctx context.Context, _ string) ([]byte, string, error) {
|
||||
ch := eventRegister()
|
||||
defer eventRelease(ch)
|
||||
select {
|
||||
case <-ch:
|
||||
return []byte(buildIndex(rs)), "", nil
|
||||
case <-ctx.Done():
|
||||
return nil, "", ctx.Err()
|
||||
}
|
||||
},
|
||||
IsBlocking_: true,
|
||||
}, nil
|
||||
case "new":
|
||||
return &FileConfig{
|
||||
switch name {
|
||||
case "new":
|
||||
return &FileConfig{
|
||||
StatFn: func() (os.FileInfo, error) {
|
||||
return &SyntheticFileInfo{Name_: "new", Mode_: Perms[PathSessions].Files["new"]}, nil
|
||||
},
|
||||
|
|
|
|||
2
kde
2
kde
|
|
@ -1 +1 @@
|
|||
Subproject commit cd7bb9e272abe857d78fd984c84d90f475b17b1f
|
||||
Subproject commit bfac41b2aec18abe8808fd710587c93d58ed309b
|
||||
Loading…
Reference in New Issue