diff --git a/x/xnet/stack-blocking.go b/x/xnet/stack-blocking.go index 575b8eb..de55193 100644 --- a/x/xnet/stack-blocking.go +++ b/x/xnet/stack-blocking.go @@ -31,8 +31,20 @@ func (s *StackAsync) StackBlocking(stackProtoBackoff lneto.BackoffStrategy) Stac } type StackBlocking struct { - async *StackAsync - _backoff lneto.BackoffStrategy + async *StackAsync + _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 { - if time.Since(deadline) > 0 { +func (s StackBlocking) checkDeadline(deadline int64) error { + if s.nanotime() > deadline { return errDeadlineExceed } return nil diff --git a/x/xnet/stack-go.go b/x/xnet/stack-go.go index 5d8fda6..3d519f8 100644 --- a/x/xnet/stack-go.go +++ b/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 - tcpDialRetries := cfg.TCPDialRetries - // Defaults - if tcpDialTimeout <= 0 { - tcpDialTimeout = defaultTCPDialTimeout + if cfg.TCPDialRetries <= 0 { + cfg.TCPDialRetries = defaultTCPDialRetries } - if tcpDialRetries <= 0 { - tcpDialRetries = defaultTCPDialRetries + if cfg.TCPDialTimeout <= 0 { + cfg.TCPDialTimeout = defaultTCPDialTimeout } sg := StackGo{ blk: s, plcfg: cfg.ListenerPoolConfig, - tcpDialTimeout: tcpDialTimeout, - tcpDialRetries: tcpDialRetries, + tcpDialTimeout: cfg.TCPDialTimeout, + tcpDialRetries: cfg.TCPDialRetries, } return sg } diff --git a/x/xnet/xnet_test.go b/x/xnet/xnet_test.go index bbca978..75f1cef 100644 --- a/x/xnet/xnet_test.go +++ b/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() { - t.Helper() - deadline := time.Now().Add(100 * time.Millisecond) - for time.Now().Before(deadline) { - n, err := client.EgressEthernet(buf[:]) - if err != nil { - t.Fatal(err) - } - if n > 0 { - return - } - time.Sleep(time.Millisecond) + for !t.Failed() { // Tinygo does not implement failnow. + tbackoffer.mainGoroutineWaitForStackYield() + n, err := client.EgressEthernet(buf[:]) + now += tcptimeout / 100 + if err != nil { + t.Fatal(err) + } else if n == 0 { + t.Fatal("expected packet from socketnetip", npacket, ntcppacket) + } + npacket++ + frm, ok := getTCPFrame(buf[:]) + 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") } }