lkn
/
ollie-9p
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

9p: fix *wait one-shot semantics and Tflush cancellation 

Two fixes:
- *wait files are one-shot per open: offset > 0 returns EOF immediately
  instead of blocking on the next state transition (was consuming two
  transitions per cat invocation)
- Tflush now cancels the in-flight request goroutine via per-tag contexts,
  so Ctrl+C on a blocking cat properly unblocks the read

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Levi Neely 2026-04-19 16:56:05 +02:00
parent 2bc30d275c
commit 2f3c11649b
1 changed files with 38 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
internal/p9/server.go
View File

@ -96,10 +96,11 @@ type fid struct {
// connState tracks all open fids for a single 9P connection. // connState tracks all open fids for a single 9P connection.
type connState struct { type connState struct {
mu sync.RWMutex mu sync.RWMutex
fids map[uint32]*fid fids map[uint32]*fid
ctx context.Context ctx context.Context
cancel context.CancelFunc cancel context.CancelFunc
pending map[uint16]context.CancelFunc // in-flight request cancels, keyed by tag
} }
// Server is the 9P server for ollie sessions. // Server is the 9P server for ollie sessions.
@ -200,9 +201,10 @@ func (s *Server) Serve(conn net.Conn) {
defer conn.Close() defer conn.Close()
connCtx, connCancel := context.WithCancel(context.Background()) connCtx, connCancel := context.WithCancel(context.Background())
cs := &connState{ cs := &connState{
fids: make(map[uint32]*fid), fids: make(map[uint32]*fid),
ctx: connCtx, ctx: connCtx,
cancel: connCancel, cancel: connCancel,
pending: make(map[uint16]context.CancelFunc),
} }
s.mu.Lock() s.mu.Lock()
s.conns = append(s.conns, cs) s.conns = append(s.conns, cs)
@ -237,11 +239,22 @@ func (s *Server) Serve(conn net.Conn) {
} }
break break
} }
reqCtx, reqCancel := context.WithCancel(connCtx)
cs.mu.Lock()
cs.pending[fc.Tag] = reqCancel
cs.mu.Unlock()
wg.Add(1) wg.Add(1)
go func(fc *plan9.Fcall) { go func(fc *plan9.Fcall, ctx context.Context) {
defer wg.Done() defer func() {
responses <- s.handle(cs, fc) reqCancel()
}(fc) cs.mu.Lock()
delete(cs.pending, fc.Tag)
cs.mu.Unlock()
wg.Done()
}()
responses <- s.handle(cs, fc, ctx)
}(fc, reqCtx)
} }
wg.Wait() wg.Wait()
@ -250,7 +263,7 @@ func (s *Server) Serve(conn net.Conn) {
var plog = olog.New("9p") var plog = olog.New("9p")
func (s *Server) handle(cs *connState, fc *plan9.Fcall) *plan9.Fcall { func (s *Server) handle(cs *connState, fc *plan9.Fcall, ctx context.Context) *plan9.Fcall {
switch fc.Type { switch fc.Type {
case plan9.Tversion: case plan9.Tversion:
msize := fc.Msize msize := fc.Msize
@ -271,7 +284,7 @@ func (s *Server) handle(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
case plan9.Tcreate: case plan9.Tcreate:
return s.create(cs, fc) return s.create(cs, fc)
case plan9.Tread: case plan9.Tread:
return s.read(cs, fc) return s.read(cs, fc, ctx)
case plan9.Twrite: case plan9.Twrite:
return s.write(cs, fc) return s.write(cs, fc)
case plan9.Tstat: case plan9.Tstat:
@ -279,7 +292,11 @@ func (s *Server) handle(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
case plan9.Twstat: case plan9.Twstat:
return s.wstat(cs, fc) return s.wstat(cs, fc)
case plan9.Tflush: case plan9.Tflush:
// No blocking reads to cancel; always succeed. cs.mu.Lock()
if cancel, ok := cs.pending[fc.Oldtag]; ok {
cancel()
}
cs.mu.Unlock()
return &plan9.Fcall{Type: plan9.Rflush, Tag: fc.Tag} return &plan9.Fcall{Type: plan9.Rflush, Tag: fc.Tag}
case plan9.Tclunk: case plan9.Tclunk:
return s.clunk(cs, fc) return s.clunk(cs, fc)
@ -576,7 +593,7 @@ func (s *Server) create(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
return &plan9.Fcall{Type: plan9.Rcreate, Tag: fc.Tag, Qid: qid} return &plan9.Fcall{Type: plan9.Rcreate, Tag: fc.Tag, Qid: qid}
} }
func (s *Server) read(cs *connState, fc *plan9.Fcall) *plan9.Fcall { func (s *Server) read(cs *connState, fc *plan9.Fcall, ctx context.Context) *plan9.Fcall {
cs.mu.RLock() cs.mu.RLock()
f, ok := cs.fids[fc.Fid] f, ok := cs.fids[fc.Fid]
if !ok { if !ok {
@ -762,7 +779,12 @@ func (s *Server) read(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: 0} return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: 0}
} }
// *wait files block until a value changes; use connection context. // *wait files block until a value changes; use connection context.
// One blocking read per open: a non-zero offset means the client
// already received data this open and is now polling for EOF.
if strings.HasSuffix(parts[2], "wait") { if strings.HasSuffix(parts[2], "wait") {
if fc.Offset > 0 {
return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: 0}
}
cs.mu.RLock() cs.mu.RLock()
f, fidOK := cs.fids[fc.Fid] f, fidOK := cs.fids[fc.Fid]
var base string var base string
@ -770,7 +792,7 @@ func (s *Server) read(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
base = f.waitBase base = f.waitBase
} }
cs.mu.RUnlock() cs.mu.RUnlock()
content, err := store.Wait(cs.ctx, parts[2], base) content, err := store.Wait(ctx, parts[2], base)
if err != nil { if err != nil {
return errFcall(fc, err.Error()) return errFcall(fc, err.Error())
} }