Incorporate strict test timing for TestStackGoTCPDialRetriesPendingControl (#135)

* rewrite tinygo failing test to be more real-time

* use channels to scheduler stack

* fix flakiness by increasing timeout and move implementation to top of file

x/xnet/stack-blocking.go

@@ -31,8 +31,20 @@ func (s *StackAsync) StackBlocking(stackProtoBackoff lneto.BackoffStrategy) Stac
 }
 
 type StackBlocking struct {
-	async    *StackAsync
+	async     *StackAsync
-	_backoff lneto.BackoffStrategy
+	_backoff  lneto.BackoffStrategy
+	_nanotime func() int64
+}
+
+func (s StackBlocking) nanotime() int64 {
+	if s._nanotime != nil {
+		return s._nanotime()
+	}
+	return time.Now().UnixNano()
+}
+
+func (s StackBlocking) deadlineTO(timeout time.Duration) int64 {
+	return int64(timeout) + s.nanotime()
 }
 
 func (s StackBlocking) DoDHCPv4(reqAddr [4]byte, timeout time.Duration) (*DHCPResults, error) {
@@ -41,7 +53,7 @@ func (s StackBlocking) DoDHCPv4(reqAddr [4]byte, timeout time.Duration) (*DHCPRe
 		return nil, err
 	}
 	var backoffs uint
-	deadline := time.Now().Add(timeout)
+	deadline := s.deadlineTO(timeout)
 	requested := false
 	var lastState dhcpv4.ClientState
 	for range maxIter {
@@ -108,7 +120,7 @@ func (s StackBlocking) DoNTP(hostAddr netip.Addr, timeout time.Duration) (offset
 		return -1, err
 	}
 
-	deadline := time.Now().Add(timeout)
+	deadline := s.deadlineTO(timeout)
 	var done bool
 	var backoffs uint
 	for range maxIter {
@@ -130,7 +142,7 @@ func (s StackBlocking) DoResolveHardwareAddress6(addr netip.Addr, timeout time.D
 		return hw, err
 	}
 	var backoffs uint
-	deadline := time.Now().Add(timeout)
+	deadline := s.deadlineTO(timeout)
 	for range maxIter {
 		hw, err = s.async.ResultResolveHardwareAddress6(addr)
 		if err == nil {
@@ -159,7 +171,7 @@ func (s StackBlocking) DoLookupIPType(host string, timeout time.Duration, qtype
 		return nil, err
 	}
 
-	deadline := time.Now().Add(timeout)
+	deadline := s.deadlineTO(timeout)
 	var backoffs uint
 	for range maxIter {
 		addrs, completed, err := s.async.ResultLookupIP(host)
@@ -189,7 +201,7 @@ func (s StackBlocking) DoDialTCP(conn *tcp.Conn, localPort uint16, addrp netip.A
 }
 
 func (s StackBlocking) waitDialTCP(conn *tcp.Conn, timeout time.Duration) (err error) {
-	deadline := time.Now().Add(timeout)
+	deadline := s.deadlineTO(timeout)
 	var backoffs uint
 	for range maxIter {
 		state := conn.State()
@@ -209,8 +221,8 @@ func (s StackBlocking) waitDialTCP(conn *tcp.Conn, timeout time.Duration) (err e
 	return errDeadlineExceed
 }
 
-func (s StackBlocking) checkDeadline(deadline time.Time) error {
+func (s StackBlocking) checkDeadline(deadline int64) error {
-	if time.Since(deadline) > 0 {
+	if s.nanotime() > deadline {
 		return errDeadlineExceed
 	}
 	return nil

x/xnet/stack-go.go

@@ -37,21 +37,18 @@ func (s *StackAsync) StackGo(stackProtoBackoff lneto.BackoffStrategy, cfg StackG
 }
 
 func (s StackBlocking) StackGo(cfg StackGoConfig) StackGo {
-	tcpDialTimeout := cfg.TCPDialTimeout
+	if cfg.TCPDialRetries <= 0 {
-	tcpDialRetries := cfg.TCPDialRetries
+		cfg.TCPDialRetries = defaultTCPDialRetries
-	// Defaults
-	if tcpDialTimeout <= 0 {
-		tcpDialTimeout = defaultTCPDialTimeout
 	}
-	if tcpDialRetries <= 0 {
+	if cfg.TCPDialTimeout <= 0 {
-		tcpDialRetries = defaultTCPDialRetries
+		cfg.TCPDialTimeout = defaultTCPDialTimeout
 	}
 
 	sg := StackGo{
 		blk:            s,
 		plcfg:          cfg.ListenerPoolConfig,
-		tcpDialTimeout: tcpDialTimeout,
+		tcpDialTimeout: cfg.TCPDialTimeout,
-		tcpDialRetries: tcpDialRetries,
+		tcpDialRetries: cfg.TCPDialRetries,
 	}
 	return sg
 }

x/xnet/xnet_test.go

@@ -30,6 +30,57 @@ 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
@@ -110,9 +161,12 @@ func TestTCPConn_ReadBlocksUntilDataAvailable(t *testing.T) {
 func TestStackGoTCPDialRetriesPendingControl(t *testing.T) {
 	const seed = 5678
 	const MTU = ethernet.MaxMTU
-
+	const tcptimeout = time.Second
+	const yield = 1 * time.Millisecond
 	client, sv, _, _ := newTCPStacks(t, seed, MTU)
-	sg := client.StackBlocking(backoffYield).StackGo(StackGoConfig{
+	tbackoffer := newstackTestScheduler(t)
+
+	sg := client.StackBlocking(tbackoffer.backoffStack).StackGo(StackGoConfig{
 		ListenerPoolConfig: TCPPoolConfig{
 			QueueSize: 4,
 			TxBufSize: MTU,
@@ -121,9 +175,13 @@ func TestStackGoTCPDialRetriesPendingControl(t *testing.T) {
 				return backoffYield
 			},
 		},
-		TCPDialTimeout: 10 * time.Millisecond,
+		TCPDialTimeout: tcptimeout,
 		TCPDialRetries: 2,
 	})
+	t.Log("start")
+	// closure to simulate time.
+	var now time.Duration
+	sg.blk._nanotime = func() int64 { return int64(now) }
 
 	laddr := netip.AddrPortFrom(netip.AddrFrom4(client.Addr4()), 1234)
 	raddr := netip.AddrPortFrom(netip.AddrFrom4(sv.Addr4()), 22)
@@ -132,29 +190,43 @@ func TestStackGoTCPDialRetriesPendingControl(t *testing.T) {
 		_, err := sg.SocketNetip(context.Background(), "tcp", syscall.AF_INET, sockSTREAM, laddr, raddr)
 		done <- err
 	}()
-
+	npacket := 0
+	ntcppacket := 0
 	var buf [ethernet.MaxMTU + ethernet.MaxOverheadSize]byte
-	waitForEgress := func() {
+	for !t.Failed() { // Tinygo does not implement failnow.
-		t.Helper()
+		tbackoffer.mainGoroutineWaitForStackYield()
-		deadline := time.Now().Add(100 * time.Millisecond)
+		n, err := client.EgressEthernet(buf[:])
-		for time.Now().Before(deadline) {
+		now += tcptimeout / 100
-			n, err := client.EgressEthernet(buf[:])
+		if err != nil {
-			if err != nil {
+			t.Fatal(err)
-				t.Fatal(err)
+		} else if n == 0 {
-			}
+			t.Fatal("expected packet from socketnetip", npacket, ntcppacket)
-			if n > 0 {
+		}
-				return
+		npacket++
-			}
+		frm, ok := getTCPFrame(buf[:])
-			time.Sleep(time.Millisecond)
+		if !ok {
+			tbackoffer.mainGoroutineYieldToStack()
+			continue
+		}
+		ntcppacket++
+		_, flags := frm.OffsetAndFlags()
+		if flags != tcp.FlagSYN {
+			t.Fatal("expected SYN packet")
+		}
+		switch ntcppacket {
+		case 1:
+			now += 2 * tcptimeout
+			tbackoffer.mainGoroutineYieldToStack()
+		case 2:
+			now += 2 * tcptimeout
+			tbackoffer.mainGoroutineYieldToStack()
+			select {
+			case <-time.After(time.Second):
+				t.Fatal("SocketNetip hanging")
+			case <-done:
+				return // Test success.
+			}
 		}
-		t.Fatal("timed out waiting for TCP dial egress packet")
-	}
-	waitForEgress()
-	waitForEgress()
-
-	err := <-done
-	if err == nil {
-		t.Fatal("expected TCP dial to fail after retries without peer response")
 	}
 }