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

@@ -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)
+}

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))

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"))
 	}
 }
 

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.
 			}
 		}