claude: fix up test to use ltesto.Sched and enable ltesto.Sched multigoro

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,
+		t:       t,
-		goroYieldSignal:    make(chan struct{}),
+		timeout: time.Second,
-		goroContinueSignal: make(chan struct{}),
-		finishChan:         make(chan error, 1),
-		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{}
+	continueSignal chan struct{}
-	finishChan         chan error
+	finishChan     chan error
-	finishcalled       atomic.Bool
+
-	coroCalls          atomic.Int32
+	parked   bool // goroutine is suspended inside Yield, awaiting a continue.
-	timeout            time.Duration
+	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) {
+	g := &schedGoro{
-		panic("only one goroutine supported for now")
+		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).

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
+// bytes to arrive anyway. It covers [TCPPoolConfig.NewPolicy] reaching the pooled
-// pooled and dialed connections alike: without a [tcp.Policy] installed the
+// and dialed connections alike: with no [tcp.Policy] installed nothing notices the
-// loss is terminal, which is what this test asserts against.
+// 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{
+	tsched := ltesto.NewSched(t)
-		PoolSize: 2, QueueSize: 4,
+	svGoro, clGoro := tsched.Goro(), tsched.Goro()
-		TxBufSize: bufSize, RxBufSize: bufSize,
+
-		EstablishedTimeout: 30 * time.Second,
+	// Simulated monotonic clock. Only the driver writes it, and only while every
-		ClosingTimeout:     30 * time.Second,
+	// scheduled goroutine is parked, so it needs no synchronization of its own.
-		NewBackoff:         func() lneto.BackoffStrategy { return backoffYield },
+	var now int64
-		NewPolicy: func() tcp.Policy {
+	nanotime := func() int64 { return now }
-			timer := new(rto.Timer)
+
-			if err := timer.Configure(func() int64 { return time.Now().UnixNano() }); err != nil {
+	// Each side backs off into its own scheduler handle, so the driver can park
-				t.Fatal(err)
+	// and resume the two independently.
-			}
+	newPool := func(yield lneto.BackoffStrategy) TCPPoolConfig {
-			return timer
+		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})
+	svGo := sv.StackBlocking(svGoro.Yield).StackGo(StackGoConfig{
-	clGo := client.StackBlocking(backoffYield).StackGo(StackGoConfig{
+		ListenerPoolConfig: newPool(svGoro.Yield),
-		ListenerPoolConfig: pool,
-		TCPDialTimeout:     2 * time.Second,
-		TCPDialRetries:     2,
 	})
+	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.
+	// dropNext arms the driver to swallow the next server→client data frame. It
-	var dropNext, dropped atomic.Bool
+	// is handed between the server goroutine and the driver by the scheduler
-	stopPump := make(chan struct{})
+	// handoff, which orders every access to it.
-	defer close(stopPump)
+	var dropNext, dropped bool
-	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()
-			}
-		}
-	}()
 
-	served := make(chan error, 1)
 	go func() {
 		c, err := listener.Accept()
 		if err != nil {
-			served <- err
+			svGoro.FinishWithErr(err)
 			return
 		}
-		defer c.Close()
+		dropNext = true // The very next data frame is lost in transit.
-		c.SetDeadline(time.Now().Add(30 * time.Second))
+		_, err = c.Write([]byte(want))
-		dropNext.Store(true) // the very next data frame is lost
+		c.Close() // Closing here is what makes #182's FIN-WAIT-1 retransmit matter.
-		_, err = c.Write([]byte("this segment is lost in transit"))
+		svGoro.FinishWithErr(err)
-		served <- err
 	}()
 
 	raddr := netip.AddrPortFrom(netip.AddrFrom4(sv.Addr4()), svPort)
-	cAny, err := clGo.SocketNetip(context.Background(), "tcp", syscall.AF_INET, sockSTREAM,
+	go func() {
-		netip.AddrPort{}, raddr)
+		cAny, err := clGo.SocketNetip(context.Background(), "tcp", syscall.AF_INET, sockSTREAM,
-	if err != nil {
+			netip.AddrPort{}, raddr)
-		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]...)
 		if err != nil {
-			t.Fatalf("read %d/%d bytes after losing one segment (dropped=%v): %v (no retransmission timer?)",
+			clGoro.FinishWithErr(err)
-				len(got), len(want), dropped.Load(), 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)
-	}
 }