From 48de260013693b7d44b63ec65a03f7304da29fea Mon Sep 17 00:00:00 2001 From: soypat Date: Mon, 7 Sep 2026 15:27:50 -0700 Subject: [PATCH] claude: fix up test to use ltesto.Sched and enable ltesto.Sched multigoro --- internal/ltesto/stackschduler.go | 148 ++++++++++++++++---- x/xnet/retransmit_test.go | 225 +++++++++++++++++++------------ 2 files changed, 258 insertions(+), 115 deletions(-) diff --git a/internal/ltesto/stackschduler.go b/internal/ltesto/stackschduler.go index c41b31f..0060fa3 100644 --- a/internal/ltesto/stackschduler.go +++ b/internal/ltesto/stackschduler.go @@ -15,33 +15,57 @@ import ( // from the test thread, or vice versa. func NewSched(t testing.TB) *Sched { return &Sched{ - t: t, - goroYieldSignal: make(chan struct{}), - goroContinueSignal: make(chan struct{}), - finishChan: make(chan error, 1), - timeout: time.Second, + t: t, + timeout: time.Second, } } // Sched is the shared state behind a [SchedGoro]/[SchedDriver] pair. It exposes no // handoff methods directly; obtain a handle with [Sched.Goro] (for the // scheduled goroutine) or [Sched.Driver] (for the test thread). +// +// A Sched may schedule more than one goroutine: call [Sched.Goro] once per +// goroutine and drive them as a barrier with [Sched.AwaitAllParked] and +// [Sched.YieldToAllParked]. The single-goroutine methods ([Sched.AwaitGoroYield], +// [Sched.AwaitGoroYieldOrDone], [Sched.YieldToGoro] and [Sched.Done]) address the +// first handle handed out and are the right tool when there is only one. type Sched struct { - t testing.TB + t testing.TB + goros []*schedGoro + finishcalled atomic.Bool + timeout time.Duration +} + +// schedGoro is the per-goroutine handoff state. The channels are shared with the +// scheduled goroutine; parked, finished and err are driver-side bookkeeping and +// must only ever be touched from the test thread. +type schedGoro struct { // when stack backs off it signals here and waits until channel read or timeout. - goroYieldSignal chan struct{} + yieldSignal chan struct{} // when main goroutine is ready for more information this channel is written to to signal waiting on stack activity. - goroContinueSignal chan struct{} - finishChan chan error - finishcalled atomic.Bool - coroCalls atomic.Int32 - timeout time.Duration + continueSignal chan struct{} + finishChan chan error + + parked bool // goroutine is suspended inside Yield, awaiting a continue. + finished bool // goroutine terminated via FinishWithErr. + err error +} + +// goro0 returns the first handed-out goroutine state, which the single-goroutine +// driver methods address. +func (ss *Sched) goro0() *schedGoro { + if len(ss.goros) == 0 { + panic("Sched.Goro must be called before driving the scheduler") + } + return ss.goros[0] } // AwaitGoroYield blocks until the coroutine suspends itself via [SchedGoro.Yield]. func (ss *Sched) AwaitGoroYield() { + g := ss.goro0() select { - case <-ss.goroYieldSignal: + case <-g.yieldSignal: + g.parked = true case <-time.After(ss.timeout): ss.t.Fatal("timeout waiting for stack to backoff") } @@ -54,10 +78,13 @@ func (ss *Sched) AwaitGoroYield() { // the same select, avoiding the deadlock of guessing whether the goroutine will yield // again. Do not mix with [Sched.Done] on the same scheduler. func (ss *Sched) AwaitGoroYieldOrDone() (done bool, err error) { + g := ss.goro0() select { - case <-ss.goroYieldSignal: + case <-g.yieldSignal: + g.parked = true return false, nil - case err = <-ss.finishChan: + case err = <-g.finishChan: + g.finished, g.err = true, err return true, err case <-time.After(ss.timeout): ss.t.Fatal("timeout waiting for stack to yield or finish") @@ -67,34 +94,103 @@ func (ss *Sched) AwaitGoroYieldOrDone() (done bool, err error) { // YieldToGoro wakes a coroutine parked in [SchedGoro.Yield], letting the goroutine run on. func (ss *Sched) YieldToGoro() { + g := ss.goro0() select { - case ss.goroContinueSignal <- struct{}{}: + case g.continueSignal <- struct{}{}: + g.parked = false case <-time.After(ss.timeout): ss.t.Fatal("timeout while trying to yield to stack") } } +// AwaitAllParked blocks until every scheduled goroutine has either suspended +// itself in [SchedGoro.Yield] or terminated via [SchedGoro.FinishWithErr]. Once it +// returns, no scheduled goroutine is runnable, so the driver may touch state they +// share — pumping frames between stacks, advancing a simulated clock — without +// racing them. Pair it with [Sched.YieldToAllParked] to step the whole set. +// +// allFinished reports that every goroutine has terminated, which is the loop's +// exit condition; err is the first non-nil terminal error handed over so far. +func (ss *Sched) AwaitAllParked() (allFinished bool, err error) { + ss.goro0() // Panics if the scheduler has no goroutines to drive. + for _, g := range ss.goros { + if g.parked || g.finished { + continue // Already accounted for; waiting again would deadlock. + } + select { + case <-g.yieldSignal: + g.parked = true + case gerr := <-g.finishChan: + g.finished, g.err = true, gerr + case <-time.After(ss.timeout): + ss.t.Fatal("timeout waiting for scheduled goroutines to park or finish") + return true, nil + } + } + allFinished = true + for _, g := range ss.goros { + if !g.finished { + allFinished = false + } + if err == nil { + err = g.err + } + } + return allFinished, err +} + +// YieldToAllParked wakes every goroutine currently parked in [SchedGoro.Yield], +// letting them all run on until they park again. Goroutines that have already +// terminated are skipped, so it is safe to call until [Sched.AwaitAllParked] +// reports every goroutine finished. +func (ss *Sched) YieldToAllParked() { + ss.goro0() // Panics if the scheduler has no goroutines to drive. + for _, g := range ss.goros { + if !g.parked { + continue + } + select { + case g.continueSignal <- struct{}{}: + g.parked = false + case <-time.After(ss.timeout): + ss.t.Fatal("timeout while trying to yield to scheduled goroutine") + } + } +} + // Done returns the channel that receives the coroutine's terminal error from // [SchedGoro.FinishWithErr]. It may only be called once. func (ss *Sched) Done() <-chan error { + g := ss.goro0() if ss.finishcalled.CompareAndSwap(false, true) { - return ss.finishChan + return g.finishChan } panic("Done called twice") } // Goro returns the handle whose methods must be called from inside the -// scheduled (stack) goroutine. +// scheduled (stack) goroutine. Call it once per goroutine to be scheduled, from +// the test thread and before those goroutines start: the handles are handed out +// unsynchronized. The first handle is the one the single-goroutine driver methods +// address; drive two or more with [Sched.AwaitAllParked] and [Sched.YieldToAllParked]. func (ss *Sched) Goro() SchedGoro { - if !ss.coroCalls.CompareAndSwap(0, 1) { - panic("only one goroutine supported for now") + g := &schedGoro{ + yieldSignal: make(chan struct{}), + continueSignal: make(chan struct{}), + finishChan: make(chan error, 1), } - return SchedGoro{ss: ss} + ss.goros = append(ss.goros, g) + return SchedGoro{ss: ss, g: g} } // SchedGoro is the coroutine-side handle of a [Sched]. Every method MUST be // called from inside the scheduled goroutine and never from the test thread. -type SchedGoro struct{ ss *Sched } +// It holds its own handoff state directly so the goroutine never reads the +// scheduler's handle list, which the test thread may still be appending to. +type SchedGoro struct { + ss *Sched + g *schedGoro +} // Yield suspends the goroutine at a backoff point and parks until the driver // calls [SchedDriver.YieldToGoro]. Its signature satisfies [lneto.BackoffStrategy] so it @@ -103,12 +199,12 @@ func (c SchedGoro) Yield(consecutiveBackoffs uint) time.Duration { ss := c.ss timeout := time.After(ss.timeout) select { - case ss.goroYieldSignal <- struct{}{}: + case c.g.yieldSignal <- struct{}{}: case <-timeout: ss.t.Fatal("timeout backing off, possible race condition? Multiple stacks using same backoff is unexpected pattern") } select { - case <-ss.goroContinueSignal: + case <-c.g.continueSignal: case <-timeout: ss.t.Fatal("timeout waiting for continue") } @@ -119,10 +215,10 @@ func (c SchedGoro) Yield(consecutiveBackoffs uint) time.Duration { // channel. It must be called at most once. func (c SchedGoro) FinishWithErr(err error) { ss := c.ss - if len(ss.finishChan) != 0 { + if len(c.g.finishChan) != 0 { ss.t.Fatal("Coro.FinishWithErr can be called once only") } - ss.finishChan <- err + c.g.finishChan <- err } // Finish is just shorthand for c.FinishWithErr(nil). diff --git a/x/xnet/retransmit_test.go b/x/xnet/retransmit_test.go index d9270da..3635f07 100644 --- a/x/xnet/retransmit_test.go +++ b/x/xnet/retransmit_test.go @@ -2,29 +2,44 @@ package xnet import ( "context" + "fmt" "net" "net/netip" - "runtime" - "sync/atomic" "syscall" "testing" "time" "github.com/soypat/lneto" "github.com/soypat/lneto/ethernet" + "github.com/soypat/lneto/internal/ltesto" "github.com/soypat/lneto/tcp" "github.com/soypat/lneto/tcp/rto" ) // TestTCPRetransmitsLostSegment drops exactly one data segment and requires the -// bytes to arrive anyway. It covers [TCPPoolConfig.NewPolicy] reaching the -// pooled and dialed connections alike: without a [tcp.Policy] installed the -// loss is terminal, which is what this test asserts against. +// bytes to arrive anyway. It covers [TCPPoolConfig.NewPolicy] reaching the pooled +// and dialed connections alike: with no [tcp.Policy] installed nothing notices the +// loss, no retransmission is ever sent and the read below never completes. +// +// The server and the client each get an [ltesto.Sched] goroutine and the test +// thread drives them as a barrier: it only moves frames or advances the clock +// once both are parked, so the stacks are never touched concurrently. Time is +// simulated, so waiting out the one-second initial RTO (RFC 6298 §2.1) costs +// nothing and the outcome does not depend on how fast the machine is. func TestTCPRetransmitsLostSegment(t *testing.T) { const ( MTU = ethernet.MaxMTU svPort = 80 bufSize = 2 << 10 + want = "this segment is lost in transit" + // A quiet round means both sides are waiting on the network, which is + // what a lost segment looks like: only then does the clock move, so the + // RTO expires in a bounded number of rounds instead of in real time. + quietStep = 100 * time.Millisecond + maxRounds = 600 + // Headers total 54 bytes, so a larger frame carries payload. Dropping a + // bare ACK would exercise the other direction's recovery instead. + minDataFrame = 14 + 20 + 20 + 8 ) client, sv := new(StackAsync), new(StackAsync) if err := client.Reset(StackConfig{ @@ -52,26 +67,45 @@ func TestTCPRetransmitsLostSegment(t *testing.T) { client.SetGatewayHardwareAddr(sv.HardwareAddr()) sv.SetGatewayHardwareAddr(client.HardwareAddr()) - pool := TCPPoolConfig{ - PoolSize: 2, QueueSize: 4, - TxBufSize: bufSize, RxBufSize: bufSize, - EstablishedTimeout: 30 * time.Second, - ClosingTimeout: 30 * time.Second, - NewBackoff: func() lneto.BackoffStrategy { return backoffYield }, - NewPolicy: func() tcp.Policy { - timer := new(rto.Timer) - if err := timer.Configure(func() int64 { return time.Now().UnixNano() }); err != nil { - t.Fatal(err) - } - return timer - }, + tsched := ltesto.NewSched(t) + svGoro, clGoro := tsched.Goro(), tsched.Goro() + + // Simulated monotonic clock. Only the driver writes it, and only while every + // scheduled goroutine is parked, so it needs no synchronization of its own. + var now int64 + nanotime := func() int64 { return now } + + // Each side backs off into its own scheduler handle, so the driver can park + // and resume the two independently. + newPool := func(yield lneto.BackoffStrategy) TCPPoolConfig { + return TCPPoolConfig{ + PoolSize: 2, QueueSize: 4, + TxBufSize: bufSize, RxBufSize: bufSize, + // Well past the simulated time this test spends, so the pool never + // reaps a connection out from under the retransmission. + EstablishedTimeout: 120 * time.Second, + ClosingTimeout: 120 * time.Second, + NanoTime: nanotime, + NewBackoff: func() lneto.BackoffStrategy { return yield }, + NewPolicy: func() tcp.Policy { + timer := new(rto.Timer) + if err := timer.Configure(nanotime); err != nil { + t.Error(err) + } + return timer + }, + } } - svGo := sv.StackBlocking(backoffYield).StackGo(StackGoConfig{ListenerPoolConfig: pool}) - clGo := client.StackBlocking(backoffYield).StackGo(StackGoConfig{ - ListenerPoolConfig: pool, - TCPDialTimeout: 2 * time.Second, - TCPDialRetries: 2, + svGo := sv.StackBlocking(svGoro.Yield).StackGo(StackGoConfig{ + ListenerPoolConfig: newPool(svGoro.Yield), }) + clGo := client.StackBlocking(clGoro.Yield).StackGo(StackGoConfig{ + ListenerPoolConfig: newPool(clGoro.Yield), + TCPDialTimeout: 60 * time.Second, + TCPDialRetries: 1, + }) + svGo.blk._nanotime = nanotime + clGo.blk._nanotime = nanotime lsAny, err := svGo.SocketNetip(context.Background(), "tcp", syscall.AF_INET, sockSTREAM, netip.AddrPortFrom(netip.AddrFrom4(sv.Addr4()), svPort), netip.AddrPort{}) @@ -81,85 +115,98 @@ func TestTCPRetransmitsLostSegment(t *testing.T) { listener := lsAny.(net.Listener) defer listener.Close() - // dropNext arms the pump to swallow the next server→client data frame. - var dropNext, dropped atomic.Bool - stopPump := make(chan struct{}) - defer close(stopPump) - go func() { - buf := make([]byte, MTU+ethernet.MaxOverheadSize) - for { - select { - case <-stopPump: - return - default: - } - moved := false - if n, err := client.EgressEthernet(buf); err == nil && n > 0 { - sv.IngressEthernet(buf[:n]) - moved = true - } - if n, err := sv.EgressEthernet(buf); err == nil && n > 0 { - // Drop only a data-carrying frame: headers total 54 bytes, so - // anything larger has payload. Dropping a bare ACK would test the - // other direction's recovery instead. - if dropNext.Load() && n > 14+20+20+8 { - dropNext.Store(false) - dropped.Store(true) - } else { - client.IngressEthernet(buf[:n]) - } - moved = true - } - if !moved { - runtime.Gosched() - } - } - }() + // dropNext arms the driver to swallow the next server→client data frame. It + // is handed between the server goroutine and the driver by the scheduler + // handoff, which orders every access to it. + var dropNext, dropped bool - served := make(chan error, 1) go func() { c, err := listener.Accept() if err != nil { - served <- err + svGoro.FinishWithErr(err) return } - defer c.Close() - c.SetDeadline(time.Now().Add(30 * time.Second)) - dropNext.Store(true) // the very next data frame is lost - _, err = c.Write([]byte("this segment is lost in transit")) - served <- err + dropNext = true // The very next data frame is lost in transit. + _, err = c.Write([]byte(want)) + c.Close() // Closing here is what makes #182's FIN-WAIT-1 retransmit matter. + svGoro.FinishWithErr(err) }() raddr := netip.AddrPortFrom(netip.AddrFrom4(sv.Addr4()), svPort) - cAny, err := clGo.SocketNetip(context.Background(), "tcp", syscall.AF_INET, sockSTREAM, - netip.AddrPort{}, raddr) - if err != nil { - t.Fatal(err) - } - conn := cAny.(net.Conn) - defer conn.Close() - // Generous on purpose: the first RTO is one second (RFC 6298 §2.1) and may - // back off once. What is under test is that recovery happens at all. - conn.SetDeadline(time.Now().Add(15 * time.Second)) - - want := "this segment is lost in transit" - got := make([]byte, 0, len(want)) - rb := make([]byte, 64) - for len(got) < len(want) { - n, err := conn.Read(rb) - got = append(got, rb[:n]...) + go func() { + cAny, err := clGo.SocketNetip(context.Background(), "tcp", syscall.AF_INET, sockSTREAM, + netip.AddrPort{}, raddr) if err != nil { - t.Fatalf("read %d/%d bytes after losing one segment (dropped=%v): %v (no retransmission timer?)", - len(got), len(want), dropped.Load(), err) + clGoro.FinishWithErr(err) + return } + conn := cAny.(net.Conn) + got := make([]byte, 0, len(want)) + rb := make([]byte, 64) + for len(got) < len(want) { + n, err := conn.Read(rb) + got = append(got, rb[:n]...) + if err != nil { + clGoro.FinishWithErr(fmt.Errorf("read %d/%d bytes: %w", len(got), len(want), err)) + return + } + } + if string(got) != want { + clGoro.FinishWithErr(fmt.Errorf("read %q, want %q", got, want)) + return + } + // Closed before finishing: a Yield after FinishWithErr would never be + // serviced, since the driver stops resuming a goroutine it has reaped. + conn.Close() + clGoro.Finish() + }() + + var buf [MTU + ethernet.MaxOverheadSize]byte + // pump moves one frame each way, dropping the armed one. Only ever called + // with both goroutines parked. + // Ingress errors are not fatal here: once a segment is dropped the frames + // behind it arrive past rcv.nxt and are rejected, which is precisely the + // stall the retransmission has to break. Egress errors are real faults. + pump := func() (moved bool) { + n, err := client.EgressEthernet(buf[:]) + if err != nil { + t.Fatal("client egress:", err) + } else if n > 0 { + sv.IngressEthernet(buf[:n]) + moved = true + } + n, err = sv.EgressEthernet(buf[:]) + if err != nil { + t.Fatal("server egress:", err) + } else if n > 0 { + if dropNext && n > minDataFrame { + dropNext, dropped = false, true + } else { + client.IngressEthernet(buf[:n]) + } + moved = true + } + return moved } - if string(got) != want { - t.Fatalf("read %q, want %q", got, want) + + for round := 0; ; round++ { + if round == maxRounds { + t.Fatalf("no retransmission after %d rounds and %v of simulated time (dropped=%v): is a Policy installed?", + maxRounds, time.Duration(now), dropped) + } + allFinished, err := tsched.AwaitAllParked() + if err != nil { + t.Fatalf("after losing one segment (dropped=%v): %v", dropped, err) + } + if allFinished { + break + } + if !pump() { + now += int64(quietStep) // Both sides idle: let the RTO age. + } + tsched.YieldToAllParked() } - if !dropped.Load() { + if !dropped { t.Fatal("no frame was dropped, so the test did not exercise retransmission") } - if err := <-served; err != nil { - t.Fatalf("server side: %v", err) - } }