From b795d6a43778e4343e3c4beb24ea25d5d058ba3f Mon Sep 17 00:00:00 2001 From: Pat Whittingslow Date: Sat, 27 Jun 2026 10:37:46 -0300 Subject: [PATCH] use ltesto as package to contain scheduler/goroutine testing logic (#143) * use ltesto as package to contain scheduler/goroutine testing logic * add sched usage to another test * testCloseTransmitsPending rewrite * replace deadline with context --- internal/ltesto/stackschduler.go | 131 +++++++++++++++++++++++++++++++ x/xnet/xnet_concurrent_test.go | 86 +++++++++++--------- x/xnet/xnet_listener_test.go | 8 +- x/xnet/xnet_test.go | 119 +++++++++------------------- 4 files changed, 220 insertions(+), 124 deletions(-) create mode 100644 internal/ltesto/stackschduler.go diff --git a/internal/ltesto/stackschduler.go b/internal/ltesto/stackschduler.go new file mode 100644 index 0000000..c41b31f --- /dev/null +++ b/internal/ltesto/stackschduler.go @@ -0,0 +1,131 @@ +package ltesto + +import ( + "sync/atomic" + "testing" + "time" + + "github.com/soypat/lneto" +) + +// NewSched creates a cooperative two-goroutine scheduler modelling a +// coroutine handoff: the scheduled (stack) goroutine drives the [SchedGoro] handle +// while the controlling test thread drives the [SchedDriver] handle. Splitting the +// API across two handles makes it impossible to call a goroutine-side method +// 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, + } +} + +// 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). +type Sched struct { + t testing.TB + // when stack backs off it signals here and waits until channel read or timeout. + goroYieldSignal 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 +} + +// AwaitGoroYield blocks until the coroutine suspends itself via [SchedGoro.Yield]. +func (ss *Sched) AwaitGoroYield() { + select { + case <-ss.goroYieldSignal: + case <-time.After(ss.timeout): + ss.t.Fatal("timeout waiting for stack to backoff") + } +} + +// AwaitGoroYieldOrDone blocks until the coroutine either parks itself via +// [SchedGoro.Yield] (returning done=false) or terminates via [SchedGoro.FinishWithErr] +// /[SchedGoro.Finish] (returning done=true and the terminal error). It lets a driver +// loop service an a-priori-unknown number of yields and still observe completion in +// 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) { + select { + case <-ss.goroYieldSignal: + return false, nil + case err = <-ss.finishChan: + return true, err + case <-time.After(ss.timeout): + ss.t.Fatal("timeout waiting for stack to yield or finish") + return true, nil + } +} + +// YieldToGoro wakes a coroutine parked in [SchedGoro.Yield], letting the goroutine run on. +func (ss *Sched) YieldToGoro() { + select { + case ss.goroContinueSignal <- struct{}{}: + case <-time.After(ss.timeout): + ss.t.Fatal("timeout while trying to yield to stack") + } +} + +// 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 { + if ss.finishcalled.CompareAndSwap(false, true) { + return ss.finishChan + } + panic("Done called twice") +} + +// Goro returns the handle whose methods must be called from inside the +// scheduled (stack) goroutine. +func (ss *Sched) Goro() SchedGoro { + if !ss.coroCalls.CompareAndSwap(0, 1) { + panic("only one goroutine supported for now") + } + return SchedGoro{ss: ss} +} + +// 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 } + +// Yield suspends the goroutine at a backoff point and parks until the driver +// calls [SchedDriver.YieldToGoro]. Its signature satisfies [lneto.BackoffStrategy] so it +// can be passed directly as the stack's backoff strategy. +func (c SchedGoro) Yield(consecutiveBackoffs uint) time.Duration { + ss := c.ss + timeout := time.After(ss.timeout) + select { + case ss.goroYieldSignal <- 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 <-timeout: + ss.t.Fatal("timeout waiting for continue") + } + return lneto.BackoffFlagNop // backoff yield implemented on our side. +} + +// FinishWithErr terminates the coroutine, handing err to the driver's [SchedDriver.Done] +// channel. It must be called at most once. +func (c SchedGoro) FinishWithErr(err error) { + ss := c.ss + if len(ss.finishChan) != 0 { + ss.t.Fatal("Coro.FinishWithErr can be called once only") + } + ss.finishChan <- err +} + +// Finish is just shorthand for c.FinishWithErr(nil). +func (c SchedGoro) Finish() { + c.FinishWithErr(nil) +} diff --git a/x/xnet/xnet_concurrent_test.go b/x/xnet/xnet_concurrent_test.go index 8169b72..6838893 100644 --- a/x/xnet/xnet_concurrent_test.go +++ b/x/xnet/xnet_concurrent_test.go @@ -16,6 +16,7 @@ import ( "github.com/soypat/lneto" "github.com/soypat/lneto/ethernet" "github.com/soypat/lneto/internal" + "github.com/soypat/lneto/internal/ltesto" "github.com/soypat/lneto/ipv4" "github.com/soypat/lneto/tcp" ) @@ -120,7 +121,7 @@ func TestTCPListener_ConcurrentEcho(t *testing.T) { wg.Add(1) go func(clientID int) { defer wg.Done() - if runClient(t, clientID, &clientStacks[clientID], &clientConns[clientID], + if runClient(t, ctx, clientID, &clientStacks[clientID], &clientConns[clientID], serverIP, serverPort) { clientSuccess[clientID] = true } @@ -206,7 +207,7 @@ func echoServer(ctx context.Context, listener *tcp.Listener) { } if listener.NumberOfReadyToAccept() == 0 { - time.Sleep(time.Millisecond) + runtime.Gosched() continue } @@ -240,7 +241,7 @@ func echoServer(ctx context.Context, listener *tcp.Listener) { } } -func runClient(t *testing.T, id int, stack *StackAsync, conn *tcp.Conn, +func runClient(t *testing.T, ctx context.Context, id int, stack *StackAsync, conn *tcp.Conn, serverAddr netip.Addr, serverPort uint16) bool { // Dial server. clientPort := uint16(10000 + id) @@ -250,14 +251,8 @@ func runClient(t *testing.T, id int, stack *StackAsync, conn *tcp.Conn, return false } - // Wait for connection established (handshake via kernel loop). - deadline := time.Now().Add(5 * time.Second) - for conn.State() != tcp.StateEstablished { - if time.Now().After(deadline) { - t.Errorf("client %d: timeout waiting for established state, got %s", id, conn.State()) - return false - } - time.Sleep(time.Millisecond) + for conn.State() != tcp.StateEstablished && ctx.Err() == nil { + runtime.Gosched() } // Send test data. @@ -268,15 +263,11 @@ func runClient(t *testing.T, id int, stack *StackAsync, conn *tcp.Conn, return false } - // Read echo response. + // Read echo response. conn.Read yields (backoffYield) until data arrives, and the + // overall test timeout (ctx) backstops a hang. var buf [64]byte - deadline = time.Now().Add(5 * time.Second) var totalRead int - for totalRead < len(testData) { - if time.Now().After(deadline) { - t.Errorf("client %d: timeout waiting for echo response, got %d/%d bytes", id, totalRead, len(testData)) - return false - } + for totalRead < len(testData) && ctx.Err() == nil { n, err := conn.Read(buf[totalRead:]) if err != nil { t.Errorf("client %d read failed: %v", id, err) @@ -325,11 +316,24 @@ func testCloseTransmitsPending(tst *tester, s1, s2 *StackAsync, c1, c2 *tcp.Conn logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{ Level: slog.LevelDebug - 99, })) + // When the payload exceeds the Tx buffer, c1.Write must run in a background + // goroutine that blocks until the driver drains the buffer. The scheduler turns + // that blocking into a deterministic, sleep-free handoff: c1's backoff parks the + // writer and the driver releases it after freeing buffer space. + async := datalen > tx1Buf + var tsched *ltesto.Sched + var tgoro ltesto.SchedGoro + c1Backoff := backoffYield + if async { + tsched = ltesto.NewSched(t) + tgoro = tsched.Goro() + c1Backoff = tgoro.Yield + } err := c1.Configure(tcp.ConnConfig{ RxBuf: nil, TxBuf: make([]byte, tx1Buf), TxPacketQueueSize: queueSize, - RWBackoff: backoffYield, + RWBackoff: c1Backoff, Logger: logger, }) if err != nil { @@ -350,29 +354,19 @@ func testCloseTransmitsPending(tst *tester, s1, s2 *StackAsync, c1, c2 *tcp.Conn for i := range datalen { data[i] = byte(i) } - deadline := time.Now().Add(3600 * time.Second) - err = c1.SetDeadline(deadline) - err2 := c2.SetDeadline(deadline) - if err != nil || err2 != nil { - t.Fatal(err) - } - - async := datalen > tx1Buf if async { // Since data does not fit in TCP Tx buffer the test must be run asynchronously. c1.InternalHandler().SetLoggers(logger, logger) // c1.InternalHandler().SetLoggers(nil, nil) go func() { - n, err := c1.Write(data) - if err != nil { - t.Error("async write", err) - } else if n != len(data) { - t.Error("io.Writer faulty implementation") + n, werr := c1.Write(data) + if werr == nil && n != len(data) { + werr = fmt.Errorf("async write %d of %d bytes", n, len(data)) } - err = c1.Close() - if err != nil { - t.Fatal("async close", err) + if werr == nil { + werr = c1.Close() } + tgoro.FinishWithErr(werr) }() } else { n, err := c1.Write(data) @@ -389,7 +383,20 @@ func testCloseTransmitsPending(tst *tester, s1, s2 *StackAsync, c1, c2 *tcp.Conn exchanging := 1 tcpData := 0 totalRead := 0 + writerDone := false for exchanging > 0 || c1.State().TxDataOpen() { + if async && !writerDone { + // Block until the writer parks on a full Tx buffer (or finishes). Servicing + // each park with exactly one pump round below keeps progress deterministic + // without sleeping or guessing whether the writer will park again. + done, werr := tsched.AwaitGoroYieldOrDone() + if werr != nil { + t.Error("async write/close:", werr) + } + if done { + writerDone = true + } + } exchanges++ exchanging = exchangeEthernetOnce(t, s1, s2, buf) frm, ok := getTCPFrame(buf[:exchanging]) @@ -401,18 +408,21 @@ func testCloseTransmitsPending(tst *tester, s1, s2 *StackAsync, c1, c2 *tcp.Conn if err != nil { t.Error(err) } else if ngot != n { - t.Errorf("want %d data read c1->c1, got %d", n, ngot) + t.Errorf("want %d data read c1->c2, got %d", n, ngot) } else if !internal.BytesEqual(buf[:n], data[totalRead:totalRead+n]) { t.Errorf("exch%d data rx mismatch, want:\n%q\ngot:\n%q\n", exchanges, data[totalRead:totalRead+n], buf[:n]) } totalRead += ngot - runtime.Gosched() // Yield to let c1 write via goroutine. - acks := exchangeEthernetOnce(t, s2, s1, buf) // Send ACK s1's way. + acks := exchangeEthernetOnce(t, s2, s1, buf) // Send ACK s1's way, freeing its Tx buffer. if acks == 0 { t.Error("no data sent back to s1") } } } + if async && !writerDone { + // The ACK above freed Tx buffer space; release the writer to fill it and re-park. + tsched.YieldToGoro() + } } if c1.BufferedUnsent() != 0 { t.Errorf("done %s: want no data left unsent got %d/%d", c1.State(), c1.BufferedUnsent(), len(data)) diff --git a/x/xnet/xnet_listener_test.go b/x/xnet/xnet_listener_test.go index 894bfbd..931ff41 100644 --- a/x/xnet/xnet_listener_test.go +++ b/x/xnet/xnet_listener_test.go @@ -3,7 +3,6 @@ package xnet import ( "net/netip" "testing" - "time" "github.com/soypat/lneto" "github.com/soypat/lneto/ethernet" @@ -172,7 +171,7 @@ func TestStackAsyncListener_MultiSequentialConn(t *testing.T) { chw := [6]byte{0xbe, 0xef, 0, 0, 0, 1} sv.SetGatewayHardwareAddr(chw) tst := testerFrom(t, MTU) - doRequest := func(caddrp netip.AddrPort, sleep time.Duration, data []byte) { + doRequest := func(caddrp netip.AddrPort, data []byte) { var client StackAsync err := client.Reset(StackConfig{ Hostname: "Client", @@ -222,15 +221,12 @@ func TestStackAsyncListener_MultiSequentialConn(t *testing.T) { if len(data) > 0 { tst.TestTCPEstablishedSingleData(&client, sv, &clConn, svconn, data) } - if sleep > 0 { - time.Sleep(sleep) - } tst.TestTCPClose(&client, sv, &clConn, svconn) } for range 1000 { caddr := caddr.Next() - doRequest(netip.AddrPortFrom(caddr, uint16(sv.Prand32())), 0, []byte("HTTP 1.0\r\n")) + doRequest(netip.AddrPortFrom(caddr, uint16(sv.Prand32())), []byte("HTTP 1.0\r\n")) } } diff --git a/x/xnet/xnet_test.go b/x/xnet/xnet_test.go index 75f1cef..fc83a25 100644 --- a/x/xnet/xnet_test.go +++ b/x/xnet/xnet_test.go @@ -30,57 +30,6 @@ const ( finack = tcp.FlagFIN | tcp.FlagACK ) -func newstackTestScheduler(t testing.TB) stackTestScheduler { - return stackTestScheduler{ - t: t, - stackBackoffSignal: make(chan struct{}), - stackContinueSignal: make(chan struct{}), - timeout: time.Second, - } -} - -type stackTestScheduler struct { - t testing.TB - // when stack backs off it signals here and waits until channel read or timeout. - stackBackoffSignal chan struct{} - // when main goroutine is ready for more information this channel is written to to signal waiting on stack activity. - stackContinueSignal chan struct{} - timeout time.Duration -} - -func (ss *stackTestScheduler) backoffStack(consecutiveBackoffs uint) time.Duration { - timeout := time.After(ss.timeout) - select { - case ss.stackBackoffSignal <- struct{}{}: - case <-timeout: - ss.t.Fatal("timeout backing off, possible race condition? Multiple stacks using same backoff is unexpected pattern") - } - select { - case <-ss.stackContinueSignal: - case <-timeout: - ss.t.Fatal("timeout waiting for continue") - } - return lneto.BackoffFlagNop // backoff yield implemented on our side. -} - -func (ss *stackTestScheduler) mainGoroutineWaitForStackYield() { - timeout := time.After(ss.timeout) - select { - case <-ss.stackBackoffSignal: - case <-timeout: - ss.t.Fatal("timeout waiting for stack to backoff") - } -} - -func (ss *stackTestScheduler) mainGoroutineYieldToStack() { - timeout := time.After(ss.timeout) - select { - case ss.stackContinueSignal <- struct{}{}: - case <-timeout: - ss.t.Fatal("timeout while trying to yield to stack") - } -} - func TestTCPConn_ReadBlocksUntilDataAvailable(t *testing.T) { const seed = 5678 const MTU = ethernet.MaxMTU @@ -88,6 +37,21 @@ func TestTCPConn_ReadBlocksUntilDataAvailable(t *testing.T) { client, sv, clconn, svconn := newTCPStacks(t, seed, MTU) tst := testerFrom(t, MTU) + // Drive svconn.Read from a single background goroutine whose backoff is + // controlled by the scheduler. This lets the test thread know deterministically + // when Read has parked waiting for data, instead of sleeping and hoping it blocked. + tsched := ltesto.NewSched(t) + tgoro := tsched.Goro() + err := svconn.Configure(tcp.ConnConfig{ + RxBuf: make([]byte, MTU), + TxBuf: make([]byte, MTU), + TxPacketQueueSize: 4, + RWBackoff: tgoro.Yield, + }) + if err != nil { + t.Fatal(err) + } + tst.TestTCPSetupAndEstablish(sv, client, svconn, clconn, svPort, 1337) // Verify no data buffered initially. @@ -96,27 +60,25 @@ func TestTCPConn_ReadBlocksUntilDataAvailable(t *testing.T) { } sendData := []byte("blocking test data") - readDone := make(chan struct{}) var readN int - var readErr error var readBuf [64]byte // Start a goroutine to read from svconn - this should block since no data available. go func() { - readN, readErr = svconn.Read(readBuf[:]) - close(readDone) + n, err := svconn.Read(readBuf[:]) + readN = n + tgoro.FinishWithErr(err) }() - // Give Read time to enter blocking state. - select { - case <-readDone: - t.Fatal("Read returned immediately without data - expected blocking") - case <-time.After(50 * time.Millisecond): - // Good - Read is blocking as expected. + // AwaitGoroYield blocks until Read parks in backoff: deterministic proof that + // Read found no data available and is waiting. + tsched.AwaitGoroYield() + if readN != 0 { + t.Fatal("Read returned before data was available") } // Write data on client side. - _, err := clconn.Write(sendData) + _, err = clconn.Write(sendData) if err != nil { t.Fatal(err) } @@ -139,14 +101,9 @@ func TestTCPConn_ReadBlocksUntilDataAvailable(t *testing.T) { t.Fatal(err) } - // Now Read should unblock and return data. - select { - case <-readDone: - // Good - Read unblocked. - case <-time.After(500 * time.Millisecond): - t.Fatal("Read did not unblock after data became available") - } - + // Release Read; the data is now available so it returns instead of backing off again. + tsched.YieldToGoro() + readErr := <-tsched.Done() if readErr != nil { t.Fatalf("Read returned error: %v", readErr) } @@ -164,9 +121,9 @@ func TestStackGoTCPDialRetriesPendingControl(t *testing.T) { const tcptimeout = time.Second const yield = 1 * time.Millisecond client, sv, _, _ := newTCPStacks(t, seed, MTU) - tbackoffer := newstackTestScheduler(t) - - sg := client.StackBlocking(tbackoffer.backoffStack).StackGo(StackGoConfig{ + tsched := ltesto.NewSched(t) + tgoro := tsched.Goro() + sg := client.StackBlocking(tgoro.Yield).StackGo(StackGoConfig{ ListenerPoolConfig: TCPPoolConfig{ QueueSize: 4, TxBufSize: MTU, @@ -185,16 +142,15 @@ func TestStackGoTCPDialRetriesPendingControl(t *testing.T) { laddr := netip.AddrPortFrom(netip.AddrFrom4(client.Addr4()), 1234) raddr := netip.AddrPortFrom(netip.AddrFrom4(sv.Addr4()), 22) - done := make(chan error, 1) go func() { _, err := sg.SocketNetip(context.Background(), "tcp", syscall.AF_INET, sockSTREAM, laddr, raddr) - done <- err + tgoro.FinishWithErr(err) }() npacket := 0 ntcppacket := 0 var buf [ethernet.MaxMTU + ethernet.MaxOverheadSize]byte for !t.Failed() { // Tinygo does not implement failnow. - tbackoffer.mainGoroutineWaitForStackYield() + tsched.AwaitGoroYield() n, err := client.EgressEthernet(buf[:]) now += tcptimeout / 100 if err != nil { @@ -205,7 +161,7 @@ func TestStackGoTCPDialRetriesPendingControl(t *testing.T) { npacket++ frm, ok := getTCPFrame(buf[:]) if !ok { - tbackoffer.mainGoroutineYieldToStack() + tsched.YieldToGoro() continue } ntcppacket++ @@ -216,14 +172,17 @@ func TestStackGoTCPDialRetriesPendingControl(t *testing.T) { switch ntcppacket { case 1: now += 2 * tcptimeout - tbackoffer.mainGoroutineYieldToStack() + tsched.YieldToGoro() case 2: now += 2 * tcptimeout - tbackoffer.mainGoroutineYieldToStack() + tsched.YieldToGoro() select { case <-time.After(time.Second): t.Fatal("SocketNetip hanging") - case <-done: + case err := <-tsched.Done(): + if err != errDeadlineExceed { + t.Fatal("expected deadline exceeded", err) + } return // Test success. } }