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/stack-go.go b/x/xnet/stack-go.go index 6d863cd..9e82dcf 100644 --- a/x/xnet/stack-go.go +++ b/x/xnet/stack-go.go @@ -176,13 +176,19 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype if isDial { var conn tcp.Conn // DIAL TCP: active connection a.k.a TCP Client branch. - err = conn.Configure(tcp.ConnConfig{ + conncfg := tcp.ConnConfig{ // TODO(pato): Eventually add UDP configuration. we use TCP for now for simplicity's sake. TxBuf: make([]byte, s.plcfg.TxBufSize), RxBuf: make([]byte, s.plcfg.RxBufSize), TxPacketQueueSize: s.plcfg.QueueSize, RWBackoff: s.plcfg.NewBackoff(), - }) + } + if s.plcfg.NewPolicy != nil { + // A dialed connection needs loss recovery as much as a pooled + // one. See [TCPPoolConfig.NewPolicy]. + conncfg.Policy = s.plcfg.NewPolicy() + } + err = conn.Configure(conncfg) if err != nil { return nil, err } diff --git a/x/xnet/tcppool.go b/x/xnet/tcppool.go index 0013a3d..be92bf5 100644 --- a/x/xnet/tcppool.go +++ b/x/xnet/tcppool.go @@ -42,8 +42,9 @@ type TCPPoolConfig struct { ConnLogger *slog.Logger // NanoTime returns the current monotonic time in nanoseconds. - // Used for pool timeout tracking and passed to each [tcp.Conn] for - // retransmission timing (RFC 6298). If nil, defaults to time.Now().UnixNano(). + // Used for pool timeout tracking. If nil, defaults to time.Now().UnixNano(). + // Retransmission timing is not driven by this clock: a [tcp.Policy] carries + // its own. See NewPolicy. NanoTime func() int64 // EstablishedTimeout sets the timeout for a TCP connection since it is acquired until it is established. // If the connection does not establish in this time it will be closed by the pool. @@ -56,6 +57,9 @@ type TCPPoolConfig struct { // NewBackoff returns the backoff to use for every newly configured TCP connection. Must be non-nil. // This should always return a static(non-method) function unless you know what you are doing. NewBackoff func() lneto.BackoffStrategy + // NewPolicy if non-nil creates a [tcp.Policy] for each [tcp.Conn] used by the configured Listener. + // NewPolicy should not return reused policies unless the algorithm is stateless. See [tcp.Policy] for more information. + NewPolicy func() tcp.Policy } func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) { @@ -88,6 +92,11 @@ func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) { Logger: cfg.ConnLogger, RWBackoff: cfg.NewBackoff(), } + if cfg.NewPolicy != nil { + // One Policy per connection: it shadows that connection's send + // sequence space and so cannot be shared. + conncfg.Policy = cfg.NewPolicy() + } err := pool.conns[i].Configure(conncfg) if err != nil { return nil, err diff --git a/x/xnet/xnet_listener_test.go b/x/xnet/xnet_listener_test.go index 931ff41..b1cb5bb 100644 --- a/x/xnet/xnet_listener_test.go +++ b/x/xnet/xnet_listener_test.go @@ -1,12 +1,19 @@ package xnet import ( + "context" + "fmt" + "net" "net/netip" + "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" ) func TestStackAsyncListener_SingleConnection(t *testing.T) { @@ -306,3 +313,191 @@ func TestListener_ResetAfterClose(t *testing.T) { t.Fatalf("expected port %d after re-Reset, got %d", svPort, listener.LocalPort()) } } + +// TestTCPRetransmitsLostSegment drops one data segment and requires bytes to arrive anyway. +// This in particular tests the RTO [tcp.Policy] since tcp +// package by itself will not trigger a retransmission unless dupacks are received. +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{ + Hostname: "rtx-client", + RandSeed: 11, + StaticAddress4: [4]byte{10, 0, 0, 90}, + MaxActiveTCPPorts: 2, + HardwareAddress: [6]byte{0xbe, 0xef, 0, 0, 0, 90}, + MTU: MTU, + ICMPQueueLimit: 2, + }); err != nil { + t.Fatal(err) + } + if err := sv.Reset(StackConfig{ + Hostname: "rtx-server", + RandSeed: ^int64(11), + StaticAddress4: [4]byte{10, 0, 0, 91}, + MaxActiveTCPPorts: 2, + HardwareAddress: [6]byte{0xbe, 0xef, 0, 0, 0, 91}, + MTU: MTU, + ICMPQueueLimit: 2, + }); err != nil { + t.Fatal(err) + } + client.SetGatewayHardwareAddr(sv.HardwareAddr()) + sv.SetGatewayHardwareAddr(client.HardwareAddr()) + + 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(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{}) + if err != nil { + t.Fatal(err) + } + listener := lsAny.(net.Listener) + defer listener.Close() + + // 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 + + go func() { + c, err := listener.Accept() + if err != nil { + svGoro.FinishWithErr(err) + return + } + 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) + go func() { + cAny, err := clGo.SocketNetip(context.Background(), "tcp", syscall.AF_INET, sockSTREAM, + netip.AddrPort{}, raddr) + if err != nil { + 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 + } + + 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 { + t.Fatal("no frame was dropped, so the test did not exercise retransmission") + } +}