From 270b8df043e5c55ac0fba5c1cdddb7485698d6bb Mon Sep 17 00:00:00 2001 From: Pat Whittingslow Date: Thu, 5 Mar 2026 21:35:16 +0100 Subject: [PATCH] Implement basic retransmit queue reset method (#52) * implement basic retransmit queue reset method * tcp: rto reset on new ack; accept ack after retransmit bugfix --- tcp/conn.go | 21 +++++ tcp/control.go | 29 +++++-- tcp/handler.go | 78 ++++++++++++++++- tcp/handler_retransmit_test.go | 147 +++++++++++++++++++++++++++++++++ tcp/handler_test.go | 1 + tcp/txqueue.go | 51 +++++++++--- tcp/txqueue_test.go | 22 ++--- x/xnet/tcppool.go | 50 +++++------ 8 files changed, 345 insertions(+), 54 deletions(-) create mode 100644 tcp/handler_retransmit_test.go diff --git a/tcp/conn.go b/tcp/conn.go index 164ed72..06f0798 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -28,6 +28,7 @@ type Conn struct { mu sync.Mutex h Handler remoteAddr []byte + nanoTime func() int64 // monotonic clock source; set by Configure. rdead time.Time wdead time.Time @@ -55,6 +56,10 @@ type ConnConfig struct { TxBuf []byte TxPacketQueueSize int Logger *slog.Logger + // NanoTime returns the current monotonic time in nanoseconds. + // Used for retransmission timing (RFC 6298). + // If nil, defaults to a function that calls time.Now().UnixNano(). + NanoTime func() int64 } func (conn *Conn) Configure(config ConnConfig) (err error) { @@ -65,9 +70,19 @@ func (conn *Conn) Configure(config ConnConfig) (err error) { return err } conn.logger.log = config.Logger + conn.nanoTime = config.NanoTime // nil is fine; conn.now() falls back to time.Now(). return nil } +// now returns the current monotonic time in nanoseconds. +// Uses the configured NanoTime function or falls back to time.Now().UnixNano(). +func (conn *Conn) now() int64 { + if conn.nanoTime != nil { + return conn.nanoTime() + } + return time.Now().UnixNano() +} + // LocalPort returns the local port on which the socket is listening or connected to. func (conn *Conn) LocalPort() uint16 { conn.mu.Lock() @@ -339,6 +354,7 @@ func (conn *Conn) Demux(buf []byte, off int) (err error) { return lneto.ErrMismatch } conn.trace("tcpconn.Recv", slog.Uint64("lport", uint64(conn.h.LocalPort())), slog.Uint64("rport", uint64(conn.h.remotePort))) + conn.h.SetNow(uint32(conn.now() / 1e6)) // ns → ms for accurate ACK timestamps. err = conn.h.Recv(buf[off:]) if err != nil { return err @@ -366,6 +382,11 @@ func (conn *Conn) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) } else if len(raddr) != len(conn.remoteAddr) { return 0, lneto.ErrMismatchLen } + conn.h.SetNow(uint32(conn.now() / 1e6)) // ns → ms. + // RFC 6298 §5.1: check RTO before sending new data. + if conn.h.ShouldRetransmit() { + conn.h.triggerRetransmit() + } n, err = conn.h.Send(carrierData[offsetToFrame:]) if err != nil || n == 0 { return 0, err diff --git a/tcp/control.go b/tcp/control.go index 4cdb09c..1830660 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -493,11 +493,24 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) { } case established && acksUnsentData: - err = errDropSegment - tcb.pending[0] |= FlagACK // Send ACK for unsent data; |= preserves any pending FIN. - if isDebug { - tcb.debug("rcv:ACK-unsent", slog.String("state", tcb._state.String()), - slog.Uint64("seg.ack", uint64(seg.ACK)), slog.Uint64("snd.nxt", uint64(tcb.snd.NXT))) + // After Retransmit() rewinds snd.NXT to snd.UNA, the remote may ACK + // data it received pre-rewind — a valid cumulative ACK that exceeds + // the rewound snd.NXT. Detect this case (NXT==UNA means rewind active) + // and accept the ACK if within the send window. + retransmitActive := tcb.snd.NXT == tcb.snd.UNA + if retransmitActive && seg.ACK.InWindow(tcb.snd.UNA, tcb.snd.WND) { + tcb.snd.NXT = seg.ACK + if isDebug { + tcb.debug("rcv:ACK-advance-nxt", slog.String("state", tcb._state.String()), + slog.Uint64("seg.ack", uint64(seg.ACK)), slog.Uint64("snd.nxt", uint64(tcb.snd.NXT))) + } + } else { + err = errDropSegment + tcb.pending[0] |= FlagACK // Send ACK for unsent data; |= preserves any pending FIN. + if isDebug { + tcb.debug("rcv:ACK-unsent", slog.String("state", tcb._state.String()), + slog.Uint64("seg.ack", uint64(seg.ACK)), slog.Uint64("snd.nxt", uint64(tcb.snd.NXT))) + } } case preestablished && (acksOld || acksUnsentData): @@ -562,6 +575,12 @@ func (tcb *ControlBlock) rstJump() Value { return 100 } +// Retransmit resets snd.NXT back to snd.UNA, allowing the next PendingSegment +// and Send calls to retransmit unacknowledged data. Must be paired with +// ringTx.RetransmitFromUNA to rewind the transmit buffer. +// Implements RFC 9293 §3.10.8 (RETRANSMISSION TIMEOUT). +func (tcb *ControlBlock) Retransmit() { tcb.snd.NXT = tcb.snd.UNA } + // Abort sets ControlBlock state to Closed and resets all sequence numbers and pending flag. // No more data can be sent nor received after the connection is aborted until opened again. // An abort call prepares the connection for opening an active connection via a diff --git a/tcp/handler.go b/tcp/handler.go index 31966ef..2d9db61 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -13,7 +13,6 @@ import ( // Handler is a low level TCP handling data structure. It implements logic // related to data buffering, frame sequencing and connection state handling. // Does NOT implement IP related logic, so no CRC calculation/validation or pseudo header logic. -// Does NOT implement connection lifetime handling, so NO deadlines, keepalives, backoffs or anything that requires use of time package. // // See [Conn] for a higher level abstraction of a TCP connection, and see [ControlBlock] for the lower level bits of a TCP connection. type Handler struct { @@ -25,12 +24,24 @@ type Handler struct { validator lneto.Validator localPort uint16 remotePort uint16 - // connid is a conenction counter that is incremented each time a new - // connection is established via Open calls. This disambiguate's whether + // connid is a connection counter that is incremented each time a new + // connection is established via Open calls. This disambiguates whether // Read and Write calls belong to the current connection. optcodec OptionCodec closing bool + + // Retransmission timer state — all uint32 milliseconds, no time package needed. + // rto is the current retransmission timeout in ms; starts at 1000 per RFC 6298 §2.1. + rto uint32 + // now is the current time in ms, set by Conn before Send/Recv via SetNow. + now uint32 + // lastACK is the last ACK value seen, for duplicate ACK detection (RFC 5681 §3.2). + lastACK Value + // dupACKs counts consecutive duplicate ACKs for fast retransmit (RFC 5681 §3.2). + dupACKs uint8 + // nRetx counts consecutive retransmissions for exponential backoff (RFC 6298 §5.5). + nRetx uint8 } func (h *Handler) SetLoggers(handler, scb *slog.Logger) { @@ -127,11 +138,19 @@ func (h *Handler) reset(localPort, remotePort uint16, iss Value) { validator: h.validator, logger: h.logger, closing: false, + rto: rtoInitial, // RFC 6298 §2.1: initial RTO = 1s. } h.bufTx.ResetOrReuse(nil, 0, iss) h.bufRx.Reset() } +const ( + // rtoInitial is the initial RTO per RFC 6298 §2.1: "the sender SHOULD set RTO <- 1 second". + rtoInitial uint32 = 1000 + // rtoMax caps exponential backoff per RFC 6298 §2.5. + rtoMax uint32 = 60_000 +) + // Recv receives an incoming TCP packet frame with the first byte being the first octet of the TCP frame. // The [Handler]'s internal state is updated if the packet is admitted successfully. func (h *Handler) Recv(incomingPacket []byte) error { @@ -166,6 +185,7 @@ func (h *Handler) Recv(incomingPacket []byte) error { return nil } prevState := h.scb.State() + prevUNA := h.scb.snd.UNA // Capture before Recv updates snd.UNA (RFC 6298 §5.3). err = h.scb.Recv(segIncoming) if err != nil { if h.scb.State() == StateClosed { @@ -191,6 +211,26 @@ func (h *Handler) Recv(incomingPacket []byte) error { if segIncoming.Flags.HasAny(FlagACK) { // Update TX ring buffer to free up acked data. h.bufTx.RecvACK(segIncoming.ACK) + // Dup-ACK tracking per RFC 5681 §3.2 and RTO reset per RFC 6298 §5.3. + if segIncoming.ACK != prevUNA && prevUNA.LessThan(segIncoming.ACK) { + // New data acknowledged — reset RTO and dup-ACK counter. + h.rto = rtoInitial // RFC 6298 §5.3. + h.nRetx = 0 + h.dupACKs = 0 + h.lastACK = segIncoming.ACK + } else if segIncoming.ACK == h.lastACK && segIncoming.DATALEN == 0 && + !segIncoming.Flags.HasAny(FlagSYN|FlagFIN) && h.bufTx.BufferedSent() > 0 { + // Duplicate ACK per RFC 5681 §2: same ACK, no data, no SYN/FIN, + // and receiver has outstanding data. + h.dupACKs++ + if h.dupACKs == 3 { + // RFC 5681 §3.2: "After receiving 3 duplicate ACKs [...] + // TCP performs a retransmission of what appears to be the + // missing segment, without waiting for the retransmission + // timer to expire." + h.triggerRetransmit() + } + } } if segIncoming.Flags.HasAny(FlagSYN) { // Parse remote MSS from TCP options. @@ -285,7 +325,7 @@ func (h *Handler) Send(b []byte) (int, error) { return 0, nil } if segment.DATALEN > 0 { - n, err := h.bufTx.MakePacket(b[sizeHeaderTCP:sizeHeaderTCP+segment.DATALEN], segment.SEQ) + n, err := h.bufTx.MakePacket(b[sizeHeaderTCP:sizeHeaderTCP+segment.DATALEN], segment.SEQ, h.now) if err != nil { return 0, err } else if n != int(segment.DATALEN) { @@ -429,6 +469,36 @@ func min(a, b int) int { return b } +// SetNow sets the current time in milliseconds for retransmission timing. +// Must be called by Conn before Send/Recv operations. +func (h *Handler) SetNow(ms uint32) { h.now = ms } + +// ShouldRetransmit returns true if the retransmission timeout has expired +// on the oldest unacknowledged segment. Per RFC 6298 §5.1 and §5.4. +func (h *Handler) ShouldRetransmit() bool { + oldest := h.bufTx.slist.Oldest() + if oldest == nil { + return false + } + return h.now-oldest.sentAt >= h.rto +} + +// triggerRetransmit rewinds the transmit queue and control block so the next +// Send call retransmits from snd.UNA. Per RFC 9293 §3.10.8, RFC 6298 §5.4–5.5. +func (h *Handler) triggerRetransmit() { + h.scb.Retransmit() + h.bufTx.RetransmitFromUNA() + // RFC 6298 §5.5: "The host MUST set RTO <- RTO * 2 ('back off the timer')." + h.nRetx++ + h.rto *= 2 + if h.rto > rtoMax { + h.rto = rtoMax + } + h.dupACKs = 0 + h.debug("tcp.Handler:retransmit", slog.Uint64("port", uint64(h.localPort)), + slog.Uint64("rto", uint64(h.rto)), slog.Uint64("nRetx", uint64(h.nRetx))) +} + func errstr(err error) string { if err == nil { return "" diff --git a/tcp/handler_retransmit_test.go b/tcp/handler_retransmit_test.go new file mode 100644 index 0000000..9baea99 --- /dev/null +++ b/tcp/handler_retransmit_test.go @@ -0,0 +1,147 @@ +package tcp + +import ( + "math/rand" + "testing" +) + +// TestRTOResetsOnNewACK is a regression test for a bug where prevUNA was +// captured AFTER ControlBlock.Recv updated snd.UNA, making the "new ACK" +// condition (seg.ACK != prevUNA) always false. This caused the RTO to never +// reset per RFC 6298 §5.3, leading to exponential backoff escalation even +// when the network was healthy. +// +// The fix: capture prevUNA before calling scb.Recv in Handler.Recv. +func TestRTOResetsOnNewACK(t *testing.T) { + const mtu = 1500 + const maxpackets = 3 + rng := rand.New(rand.NewSource(100)) + client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets) + setupClientServer(t, rng, client, server) + var rawbuf [mtu]byte + establish(t, client, server, rawbuf[:]) + + // Write and send data from client. + data := []byte("hello retransmit") + n, err := client.Write(data) + if err != nil { + t.Fatal("client write:", err) + } else if n != len(data) { + t.Fatal("short write") + } + + clear(rawbuf[:]) + n, err = client.Send(rawbuf[:]) + if err != nil { + t.Fatal("client send:", err) + } + + // Simulate prior retransmissions: RTO has been backed off and nRetx > 0. + client.rto = rtoInitial * 4 + client.nRetx = 2 + + // Server receives data and sends ACK. + err = server.Recv(rawbuf[:n]) + if err != nil { + t.Fatal("server recv:", err) + } + clear(rawbuf[:]) + n, err = server.Send(rawbuf[:]) + if err != nil { + t.Fatal("server send ACK:", err) + } + if n == 0 { + t.Fatal("expected server to send ACK") + } + + // Client receives ACK — RTO and nRetx should reset. + err = client.Recv(rawbuf[:n]) + if err != nil { + t.Fatal("client recv ACK:", err) + } + + if client.rto != rtoInitial { + t.Fatalf("BUG: RTO not reset on new ACK: got %d, want %d (RFC 6298 §5.3)", client.rto, rtoInitial) + } + if client.nRetx != 0 { + t.Fatalf("BUG: nRetx not reset on new ACK: got %d, want 0", client.nRetx) + } + if client.dupACKs != 0 { + t.Fatalf("dupACKs not reset on new ACK: got %d, want 0", client.dupACKs) + } +} + +// TestPostRetransmitACKAccepted is a regression test for a bug where after +// Retransmit() rewound snd.NXT to snd.UNA, a valid cumulative ACK from the +// remote (acknowledging data sent pre-rewind) was rejected as "acks unsent +// data" because seg.ACK > snd.NXT. +// +// The fix: in validateIncomingSegment, when snd.NXT == snd.UNA (retransmit +// active) and seg.ACK is within the send window, accept the ACK and advance +// snd.NXT to seg.ACK. +func TestPostRetransmitACKAccepted(t *testing.T) { + const mtu = 1500 + const maxpackets = 3 + rng := rand.New(rand.NewSource(200)) + client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets) + setupClientServer(t, rng, client, server) + var rawbuf [mtu]byte + establish(t, client, server, rawbuf[:]) + + // Client writes and sends data. + data := []byte("data before rewind") + n, err := client.Write(data) + if err != nil { + t.Fatal("client write:", err) + } else if n != len(data) { + t.Fatal("short write") + } + clear(rawbuf[:]) + n, err = client.Send(rawbuf[:]) + if err != nil { + t.Fatal("client send:", err) + } + + // Server receives data — its next ACK will acknowledge up to the + // original snd.NXT. + err = server.Recv(rawbuf[:n]) + if err != nil { + t.Fatal("server recv:", err) + } + + // Client triggers retransmit: snd.NXT rewound to snd.UNA. + preRewindNXT := client.scb.snd.NXT + client.triggerRetransmit() + if client.scb.snd.NXT != client.scb.snd.UNA { + t.Fatal("retransmit did not rewind snd.NXT to snd.UNA") + } + + // Server sends ACK for the data it already received. seg.ACK = preRewindNXT, + // which is > client.snd.NXT (now rewound to snd.UNA). + clear(rawbuf[:]) + n, err = server.Send(rawbuf[:]) + if err != nil { + t.Fatal("server send ACK:", err) + } + if n == 0 { + t.Fatal("expected server to send ACK") + } + + // Client receives ACK — should NOT be rejected. + err = client.Recv(rawbuf[:n]) + if err != nil { + t.Fatalf("BUG: post-retransmit ACK rejected: %v\n"+ + "After Retransmit() rewound snd.NXT to snd.UNA, the remote's cumulative\n"+ + "ACK (for data sent pre-rewind) exceeds the rewound snd.NXT and was\n"+ + "incorrectly rejected as 'acks unsent data'.", err) + } + + // snd.NXT should have advanced back to where it was before the rewind. + if client.scb.snd.NXT != preRewindNXT { + t.Fatalf("snd.NXT not restored: got %d, want %d", client.scb.snd.NXT, preRewindNXT) + } + // snd.UNA should have advanced to acknowledge the data. + if client.scb.snd.UNA != preRewindNXT { + t.Fatalf("snd.UNA not advanced: got %d, want %d", client.scb.snd.UNA, preRewindNXT) + } +} diff --git a/tcp/handler_test.go b/tcp/handler_test.go index 2be514b..36c92f0 100644 --- a/tcp/handler_test.go +++ b/tcp/handler_test.go @@ -53,6 +53,7 @@ func newHandler(t *testing.T, mtu, mintaxpackets int) *Handler { if err != nil { t.Fatal(err) } + h.rto = rtoInitial // Fake time: now=0 and sentAt=0, so RTO never fires in tests. return h } diff --git a/tcp/txqueue.go b/tcp/txqueue.go index f97901e..019d56b 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -34,7 +34,8 @@ type ringTx struct { iss Value } -// ringidx represents packet data inside RingTx +// ringidx represents packet data inside RingTx. +// Part of the retransmission queue required by RFC 9293 §3.4, §3.10.8. type ringidx struct { // off is data start offset of packet data inside buf. Follows [internal.Ring] semantics. off int @@ -44,7 +45,9 @@ type ringidx struct { seq Value // size is the size of the packet in bytes. size Size - // time is a measure of the instant of time message was sent at. + // sentAt is the time in milliseconds when this packet was first sent. + // Used for RTO detection per RFC 6298 §5. + sentAt uint32 } // Reset resets the RingTx's internal state to use buf as the main ring buffer and creates or reuses @@ -115,8 +118,9 @@ func (rtx *ringTx) Write(b []byte) (n int, err error) { } // MakePacket reads from the unsent data ring buffer and generates a new packet segment. -// It fails if the sent packet queue is full. -func (rtx *ringTx) MakePacket(b []byte, currentSeq Value) (int, error) { +// It fails if the sent packet queue is full. sentAt is the current time in milliseconds, +// stamped on the packet for RTO detection per RFC 6298 §5.1. +func (rtx *ringTx) MakePacket(b []byte, currentSeq Value, sentAt uint32) (int, error) { free := rtx.slist.Free() if free == 0 { return 0, lneto.ErrBufferFull @@ -137,7 +141,7 @@ func (rtx *ringTx) MakePacket(b []byte, currentSeq Value) (int, error) { // Start of buffer will be SENT, end of buffer will be UNSENT(or empty). // Packet generated has offset at old unsentOff. size := rtx.Size() - pkt := rtx.slist.AddPacket(n, oldUnsentOff, size, currentSeq) + pkt := rtx.slist.AddPacket(n, oldUnsentOff, size, currentSeq, sentAt) if pkt.off != oldUnsentOff || pkt.end != addEnd(pkt.off, n, size) { panic("invalid generated packet") } @@ -203,6 +207,32 @@ func (rtx *ringTx) ring(off, end int) internal.Ring { // Result of addEnd will never be 0 unless arguments are (0,0). func (rtx *ringTx) addEnd(a, b int) int { return addEnd(a, b, len(rtx.rawbuf)) } +// RetransmitFromUNA rewinds the transmit queue so that all sent-but-unacked +// data becomes unsent again. The next MakePacket call will re-send starting +// from snd.UNA. This is the smoltcp-style pointer-rewind approach: no extra +// mode flag, Send() has a single code path. +// +// Implements "send the segment at the front of the retransmission queue" +// per RFC 9293 §3.10.8 (RETRANSMISSION TIMEOUT). +func (rtx *ringTx) RetransmitFromUNA() { + oldest := rtx.slist.Oldest() + if oldest == nil { + return // Nothing in the retransmission queue. + } + unaSeq := oldest.seq + if rtx.sentend != 0 { + // Merge sent region [sentoff, sentend) back into unsent. + rtx.unsentoff = rtx.sentoff + if rtx.unsentend == 0 { + rtx.unsentend = rtx.sentend + } + rtx.sentoff = 0 + rtx.sentend = 0 + } + // Clear packet metadata; sequence tracking restarts from UNA. + rtx.slist.Reset(cap(rtx.slist.pkts), unaSeq) +} + func (rtx *ringTx) consolidateBufs() { unsentEmpty := rtx.unsentend == 0 sentEmpty := rtx.sentend == 0 @@ -293,7 +323,7 @@ func (sl *sentlist) Free() int { return cap(sl.pkts) - len(sl.pkts) } -func (sl *sentlist) AddPacket(datalen, off, bufsize int, seq Value) *ringidx { +func (sl *sentlist) AddPacket(datalen, off, bufsize int, seq Value, sentAt uint32) *ringidx { free := sl.Free() if free == 0 { panic("pkt buffer full") @@ -303,10 +333,11 @@ func (sl *sentlist) AddPacket(datalen, off, bufsize int, seq Value) *ringidx { panic("new sent packet offset must match last sent packet end") } sl.pkts = append(sl.pkts, ringidx{ - off: off, - end: addEnd(off, datalen, bufsize), - seq: seq, - size: Size(datalen), + off: off, + end: addEnd(off, datalen, bufsize), + seq: seq, + size: Size(datalen), + sentAt: sentAt, }) return &sl.pkts[len(sl.pkts)-1] } diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index f10d02c..6c90c50 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -84,7 +84,7 @@ func TestRingTx_op(t *testing.T) { clear(opWriteData) case opSend: // oplen=num bytes to send in this operation. - nsgot, err := rtx.MakePacket(auxbuf[:oplen], currentSeq) + nsgot, err := rtx.MakePacket(auxbuf[:oplen], currentSeq, 0) megafail := nsgot > nunsent if err != nil && oplen <= nunsent && availPkt > 0 { t.Fatal(itest, iop, err) @@ -138,17 +138,17 @@ func TestSentlist_multi(t *testing.T) { sl.Reset(3, 0) // Test multi packet x2. - p1 := sl.AddPacket(5, 0, bufsize, 0) - p2 := sl.AddPacket(5, p1.end, bufsize, p1.endSeq()) + p1 := sl.AddPacket(5, 0, bufsize, 0, 0) + p2 := sl.AddPacket(5, p1.end, bufsize, p1.endSeq(), 0) sl.RecvAck(Value(p2.size+p1.size), bufsize) if sl.Oldest() != nil { t.Fatal("expected full ack") } // multi packet x3. sl.Reset(3, 0) - p1 = sl.AddPacket(3, 0, bufsize, 0) - p2 = sl.AddPacket(3, p1.end, bufsize, p1.endSeq()) - p3 := sl.AddPacket(4, p2.end, bufsize, p2.endSeq()) + p1 = sl.AddPacket(3, 0, bufsize, 0, 0) + p2 = sl.AddPacket(3, p1.end, bufsize, p1.endSeq(), 0) + p3 := sl.AddPacket(4, p2.end, bufsize, p2.endSeq(), 0) sl.RecvAck(2, bufsize) oldest := sl.Oldest() if oldest != p1 { @@ -167,7 +167,7 @@ func TestSentlist_simple(t *testing.T) { // Test full ack. const bufsize = 16 const pkt = 10 - sl.AddPacket(pkt, 0, bufsize, 0) + sl.AddPacket(pkt, 0, bufsize, 0, 0) if sl.Oldest() == nil || sl.Newest() != sl.Oldest() { t.Error("expected same oldest/newest non-nil packet") } @@ -179,7 +179,7 @@ func TestSentlist_simple(t *testing.T) { } // Test partial ack. - sl.AddPacket(pkt, 0, bufsize, sl.ssn) + sl.AddPacket(pkt, 0, bufsize, sl.ssn, 0) for i := Value(0); i < pkt-1; i++ { ack++ sl.RecvAck(ack, bufsize) @@ -239,7 +239,7 @@ func TestTxQueue_multipacket(t *testing.T) { pktlen := rng.Intn(maxToPacket) + 1 pkt := rbuf[roff : roff+pktlen] expectPkt := wbuf[roff : roff+pktlen] - ngot, err := rtx.MakePacket(pkt, seq) + ngot, err := rtx.MakePacket(pkt, seq, 0) testQueueSanity(t, &rtx) roff += ngot seq += Value(ngot) @@ -374,7 +374,7 @@ func TestTxQueue(t *testing.T) { datalens = datalens[:0] for rtx.BufferedUnsent() != 0 { nbytes := rng.Intn(maxPacketSize-minBufferSize) + minBufferSize - n, err := rtx.MakePacket(readBuf[:nbytes], currentSeq) + n, err := rtx.MakePacket(readBuf[:nbytes], currentSeq, 0) if err != nil { t.Fatal(err) } else if n == 0 { @@ -572,7 +572,7 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new if wantRead != len(wantBufRead) { t.Fatalf("miscalculated expect read %d != %d", wantRead, len(wantBufRead)) } - n, err := rtx.MakePacket(readPacket, newPacketSeq) + n, err := rtx.MakePacket(readPacket, newPacketSeq, 0) if err != nil && wantRead != 0 { t.Errorf("error reading: %s", err) } else if n != wantRead { diff --git a/x/xnet/tcppool.go b/x/xnet/tcppool.go index 9dead6f..d67ba43 100644 --- a/x/xnet/tcppool.go +++ b/x/xnet/tcppool.go @@ -16,11 +16,11 @@ type TCPPool struct { naqcuired int conns []tcp.Conn userData []any - acquiredAt []time.Time - closingAt []time.Time - abortedAt []time.Time + acquiredAt []int64 + closingAt []int64 + abortedAt []int64 nextISS tcp.Value - _now func() time.Time + _now func() int64 estbTimeout time.Duration closingTimeout time.Duration logger *slog.Logger @@ -39,7 +39,11 @@ type TCPPoolConfig struct { Logger *slog.Logger ConnLogger *slog.Logger - Now func() time.Time + + // NanoTime returns the current monotonic time in nanoseconds. + // Used for pool timeout tracking and passed to each [tcp.Conn] for + // retransmission timing (RFC 6298). If nil, defaults to time.Now().UnixNano(). + NanoTime func() int64 // EstablishedTimeout sets the timeout for a TCP connection since it is acquired until it is established. // If the connection does not establish in this time it will be closed by the pool. EstablishedTimeout time.Duration @@ -56,12 +60,12 @@ func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) { } n := cfg.PoolSize pool := &TCPPool{ - acquiredAt: make([]time.Time, n), - closingAt: make([]time.Time, n), - abortedAt: make([]time.Time, n), + acquiredAt: make([]int64, n), + closingAt: make([]int64, n), + abortedAt: make([]int64, n), conns: make([]tcp.Conn, n), userData: make([]any, n), - _now: cfg.Now, + _now: cfg.NanoTime, estbTimeout: cfg.EstablishedTimeout, closingTimeout: cfg.ClosingTimeout, logger: cfg.Logger, @@ -76,6 +80,7 @@ func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) { TxBuf: bufSpace[txOff : txOff+cfg.TxBufSize], TxPacketQueueSize: cfg.QueueSize, Logger: cfg.ConnLogger, + NanoTime: cfg.NanoTime, }) if err != nil { return nil, err @@ -98,7 +103,7 @@ func (p *TCPPool) GetTCP() (conn *tcp.Conn, userData any, SuggestedISS tcp.Value defer p.mu.Unlock() p.debug("TCPPool:get") for i := range p.conns { - if p.acquiredAt[i].IsZero() { + if p.acquiredAt[i] == 0 { p.acquiredAt[i] = p.now() p.nextISS += 1000 p.naqcuired++ @@ -116,9 +121,9 @@ func (p *TCPPool) PutTCP(conn *tcp.Conn) { if &p.conns[i] == conn { // p.mu.Lock() p.conns[i].Abort() - p.acquiredAt[i] = time.Time{} - p.abortedAt[i] = time.Time{} - p.closingAt[i] = time.Time{} + p.acquiredAt[i] = 0 + p.abortedAt[i] = 0 + p.closingAt[i] = 0 p.naqcuired-- // p.mu.Unlock() return @@ -140,7 +145,7 @@ func (p *TCPPool) CheckTimeouts() { // p.mu.Lock() acq := p.acquiredAt[i] // p.mu.Unlock() - if acq.IsZero() { + if acq == 0 { continue } else if st.IsPreestablished() && p.since(acq) > p.estbTimeout { // Was acquired and did not reach establishment state so we close. @@ -148,12 +153,12 @@ func (p *TCPPool) CheckTimeouts() { conn.Close() } else if st.IsClosed() || st.IsClosing() { // p.mu.Lock() - if p.closingAt[i].IsZero() { + if p.closingAt[i] == 0 { p.closingAt[i] = p.now() - } else if p.abortedAt[i].IsZero() && p.since(p.closingAt[i]) > p.closingTimeout { + } else if p.abortedAt[i] == 0 && p.since(p.closingAt[i]) > p.closingTimeout { p.abortedAt[i] = p.now() conn.Abort() - } else if !p.abortedAt[i].IsZero() && p.since(p.abortedAt[i]) > 10*time.Second { + } else if p.abortedAt[i] != 0 && p.since(p.abortedAt[i]) > 10*time.Second { println("connection aborted and still not returned to TCPPool") println("source", conn.LocalPort(), "remote", conn.RemotePort(), "state", conn.State().String()) } @@ -161,16 +166,13 @@ func (p *TCPPool) CheckTimeouts() { } } -func (p *TCPPool) since(t time.Time) time.Duration { - if p._now == nil { - return time.Since(t) - } - return p._now().Sub(t) +func (p *TCPPool) since(t int64) time.Duration { + return time.Duration(p.now() - t) } -func (p *TCPPool) now() time.Time { +func (p *TCPPool) now() int64 { if p._now == nil { - return time.Now() + return time.Now().UnixNano() } return p._now() }