fix tests not passing for RingTx

2026-08-14 20:03:45 +00:00 · 2025-01-24 21:46:39 -03:00
parent 5e54c8ef78
commit 83719ffe98
3 changed files with 155 additions and 82 deletions
@@ -1,11 +1,13 @@
-package lneto
+package lneto_test

 import (
 	"bytes"
 	"math/rand"
 	"testing"

+	"github.com/soypat/lneto"
 	"github.com/soypat/lneto/internal/ltesto"
+	"github.com/soypat/lneto/tcp"
 )

 func TestTCPMarshalUnmarshal(t *testing.T) {
@@ -16,7 +18,13 @@ func TestTCPMarshalUnmarshal(t *testing.T) {
 	src := make([]byte, maxSize)
 	dst := make([]byte, maxSize)
 	for i := 0; i < 512; i++ {
-		src = gen.AppendRandomIPv4TCPPacket(src[:0], rng)
+		src = gen.AppendRandomIPv4TCPPacket(src[:0], rng, tcp.Segment{
+			SEQ:     tcp.Value(rng.Int()),
+			ACK:     tcp.Value(rng.Int()),
+			DATALEN: tcp.Size(rng.Intn(256)),
+			WND:     tcp.Size(rng.Intn(1024)),
+			Flags:   tcp.FlagACK,
+		})
 		dst = dst[:len(src)]
 		testMoveTCPPacket(t, src, dst)
 		if !bytes.Equal(src, dst) {
@@ -29,31 +37,31 @@ func testMoveTCPPacket(t *testing.T, src, dst []byte) {
 	if len(src) != len(dst) {
 		panic("expect src and dst same length")
 	}
-	efrm, err := NewEthFrame(src)
+	efrm, err := lneto.NewEthFrame(src)
 	if err != nil {
 		t.Fatal(err)
 	}
 	epl := efrm.Payload()
-	ifrm, err := NewIPv4Frame(epl)
+	ifrm, err := lneto.NewIPv4Frame(epl)
 	if err != nil {
 		t.Fatal(err)
 	}
 	ipl := ifrm.Payload()
-	tfrm, err := NewTCPFrame(ipl)
+	tfrm, err := lneto.NewTCPFrame(ipl)
 	if err != nil {
 		t.Fatal(err)
 	}

-	efrm2, _ := NewEthFrame(dst)
+	efrm2, _ := lneto.NewEthFrame(dst)
 	*efrm2.DestinationHardwareAddr() = *efrm.DestinationHardwareAddr()
 	*efrm2.SourceHardwareAddr() = *efrm.SourceHardwareAddr()
 	efrm2.SetEtherType(efrm.EtherTypeOrSize())
-	if efrm.EtherTypeOrSize() == EtherTypeVLAN {
+	if efrm.EtherTypeOrSize() == lneto.EtherTypeVLAN {
 		efrm2.SetVLANTag(efrm.VLANTag())
 		efrm2.SetVLANEtherType(efrm.VLANEtherType())
 	}

-	ifrm2, _ := NewIPv4Frame(efrm2.Payload())
+	ifrm2, _ := lneto.NewIPv4Frame(efrm2.Payload())
 	ifrm2.SetVersionAndIHL(ifrm.VersionAndIHL())
 	ifrm2.SetToS(ifrm.ToS())
 	ifrm2.SetFlags(ifrm.Flags())
@@ -65,7 +73,7 @@ func testMoveTCPPacket(t *testing.T, src, dst []byte) {
 	*ifrm2.SourceAddr() = *ifrm.SourceAddr()
 	*ifrm2.DestinationAddr() = *ifrm.DestinationAddr()

-	tfrm2, _ := NewTCPFrame(ifrm2.Payload())
+	tfrm2, _ := lneto.NewTCPFrame(ifrm2.Payload())
 	tfrm2.SetSourcePort(tfrm.SourcePort())
 	tfrm2.SetDestinationPort(tfrm.DestinationPort())
 	tfrm2.SetSeq(tfrm.Seq())
@@ -7,15 +7,10 @@ import (
 	"github.com/soypat/lneto/internal"
 )

-func newRingTx(buf []byte, maxQueuedPackets int) *ringTx {
-	if maxQueuedPackets <= 0 || len(buf) < 2 || len(buf) < maxQueuedPackets {
-		panic("invalid argument to NewRingTx")
-	}
-	return &ringTx{
-		rawbuf:  buf,
-		packets: make([]ringidx, maxQueuedPackets),
-	}
-}
+const (
+	// this must be at least 2 for buffer to work.
+	minBufferSize = 2
+)

 // ringTx is a ring buffer with retransmission queue functionality added.
 type ringTx struct {
@@ -24,12 +19,15 @@ type ringTx struct {
 	// packets contains
 	packets []ringidx
 	// _firstPkt is the index of the oldest packet in the packets field.
-	_firstPkt int
-	_lastPkt  int
+	// _firstPkt int
+	// _lastPkt  int
 	// unsentOff is the offset of start of unsent data into rawbuf.
 	unsentoff int
 	// unsentend is the offset of end of unsent data in rawbuf.
 	unsentend int
+	seq       Value
+	// always empty ring.
+	emptyRing ringidx
 }

 // ringidx represents packet data inside RingTx
@@ -45,6 +43,40 @@ type ringidx struct {
 	// acked bool
 }

+// Reset resets the RingTx's internal state to use buf as the main ring buffer and creates or reuses
+// the packet ring buffer.
+func (rx *ringTx) Reset(buf []byte, maxqueuedPackets int, seq Value) error {
+	if maxqueuedPackets <= 0 {
+		return errors.New("queued packets <=0")
+	} else if len(buf) < minBufferSize || len(buf) < maxqueuedPackets {
+		return errors.New("invalid buffer size")
+	}
+	if cap(rx.packets) < maxqueuedPackets {
+		rx.packets = make([]ringidx, maxqueuedPackets)
+	}
+	*rx = ringTx{
+		rawbuf:  buf,
+		packets: rx.packets[:maxqueuedPackets],
+		seq:     seq,
+	}
+	for i := range rx.packets {
+		rx.packets[i].markRcvd()
+	}
+	return nil
+}
+
+// ResetOrReuse is identical to a call to [ringTx.Reset] with the additional detail that
+// the zero value of buf (nil) and maxQueuedPackets (0) will selectively reuse existing data buffer and/or packet index buffer.
+func (rx *ringTx) ResetOrReuse(buf []byte, maxQueuedPackets int, ack Value) error {
+	if buf == nil {
+		buf = rx.rawbuf
+	}
+	if maxQueuedPackets == 0 {
+		maxQueuedPackets = len(rx.packets)
+	}
+	return rx.Reset(buf, maxQueuedPackets, ack)
+}
+
 // Buffered returns the amount of unsent bytes.
 func (tx *ringTx) Buffered() int {
 	r := tx.unsentRing()
@@ -59,9 +91,9 @@ func (tx *ringTx) BufferedSent() int {

 // Write writes data to the underlying unsent data ring buffer.
 func (tx *ringTx) Write(b []byte) (n int, err error) {
-	first := tx.packets[tx._firstPkt]
+	first := tx.pkt(tx.firstPkt())
 	r := tx.unsentRing()
-	if first.off < 0 {
+	if !first.sent() {
 		// No packets in queue case.
 		n, err = r.Write(b)
 	} else {
@@ -76,32 +108,26 @@ func (tx *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 (tx *ringTx) MakePacket(b []byte) (int, error) {
-	nxtpkt := (tx._lastPkt + 1) % len(tx.packets)
-	if tx._firstPkt == nxtpkt {
-		return 0, errors.New("packet queue full")
+func (tx *ringTx) MakePacket(b []byte) (int, Value, error) {
+	nxtpkt := tx.nextPkt()
+	if tx.nextPkt() < 0 {
+		return 0, 0, errors.New("queue full")
 	}
-
 	r := tx.unsentRing()
 	start := r.Off
 	n, err := r.Read(b)
 	if err != nil {
-		return n, err
+		return n, 0, err
 	}
-	last := &tx.packets[tx._lastPkt]
-	rlast := tx.packetRing(tx._lastPkt)
+	plen := Value(n)
+	seq := tx.seq
 	tx.packets[nxtpkt].off = start
 	tx.packets[nxtpkt].end = tx.addOff(start, n)
-	tx.packets[nxtpkt].seq = last.seq + Value(rlast.Buffered())
-	tx._lastPkt = nxtpkt
-	tx.unsentoff = tx.addOff(tx.unsentoff, n)
-	return n, nil
-}
+	tx.packets[nxtpkt].seq = seq + plen

-// IsQueueFull returns true if the sent packet queue is full in which
-// case a call to ReadPacket is guaranteed to fail.
-func (tx *ringTx) IsQueueFull() bool {
-	return tx._firstPkt == (tx._lastPkt+1)%len(tx.packets)
+	tx.unsentoff = tx.addOff(tx.unsentoff, n)
+	tx.seq += plen
+	return n, seq, nil
 }

 func (tx *ringTx) packetRing(i int) internal.Ring {
@@ -114,41 +140,29 @@ func (tx *ringTx) packetRing(i int) internal.Ring {

 // RecvSegment processes an incoming segment and updates the sent packet queue
 func (tx *ringTx) RecvACK(ack Value) error {
-	i := tx._firstPkt
-	for {
+	for i := range tx.packets {
 		pkt := &tx.packets[i]
-		if ack >= pkt.seq {
-			// Packet was received by remote. Mark it as acked.
-			pkt.off = -1
-			tx._firstPkt++
-			continue
+		if pkt.sent() && pkt.seq.LessThanEq(ack) {
+			pkt.markRcvd()
 		}
-		if i == tx._lastPkt {
-			break
-		}
-		i = (i + 1) % len(tx.packets)
 	}
 	return nil
 }

 func (tx *ringTx) unsentRing() internal.Ring {
-	return tx.ring(tx.unsentoff, tx.unsentend)
-}
-
-func (tx *ringTx) freeRing() (internal.Ring, int) {
-	return tx.ring(tx.unsentoff, tx.unsentend), 0
-}
-
-func (tx *ringTx) a() {
-
+	off := tx.unsentoff
+	if off == tx.unsentend && off != 0 {
+		off--
+	}
+	return tx.ring(off, tx.unsentend)
 }

 func (tx *ringTx) sentRing() internal.Ring {
-	first := tx.packets[tx._firstPkt]
-	if first.off < 0 {
-		return tx.ring(0, 0)
+	first := tx.pkt(tx.firstPkt())
+	if !first.sent() {
+		return internal.Ring{}
 	}
-	last := tx.packets[tx._lastPkt]
+	last := tx.pkt(tx.lastPkt())
 	return tx.ring(first.off, last.end)
 }

@@ -165,12 +179,21 @@ func (tx *ringTx) addOff(a, b int) int {
 	return off
 }

+func (tx *ringTx) pkt(i int) *ringidx {
+	if i == -1 {
+		return &tx.emptyRing
+	} else if i < 0 || i >= len(tx.packets) {
+		panic("invalid packet index")
+	}
+	return &tx.packets[i]
+}
+
 func (tx *ringTx) firstPkt() int {
 	seq := tx.packets[0].seq
 	idx := -1
 	for i := 0; i < len(tx.packets); i++ {
 		pkt := &tx.packets[i]
-		if (pkt.end != 0 || pkt.off != 0) && seq.LessThanEq(pkt.seq) {
+		if pkt.sent() && seq.LessThanEq(pkt.seq) {
 			seq = pkt.seq
 			idx = i
 		}
@@ -183,10 +206,32 @@ func (tx *ringTx) lastPkt() int {
 	idx := -1
 	for i := 0; i < len(tx.packets); i++ {
 		pkt := &tx.packets[i]
-		if (pkt.end != 0 || pkt.off != 0) && pkt.seq.LessThanEq(seq) {
+		if pkt.sent() && pkt.seq.LessThanEq(seq) {
 			seq = pkt.seq
 			idx = i
 		}
 	}
 	return idx
 }
+
+func (tx *ringTx) nextPkt() int {
+	idx := -1
+	for i := 0; i < len(tx.packets); i++ {
+		pkt := &tx.packets[i]
+		if !pkt.sent() {
+			idx = i
+			break
+		}
+	}
+	return idx
+}
+
+func (pkt *ringidx) sent() bool {
+	return pkt.end != 0 || pkt.off != 0
+}
+
+func (pkt *ringidx) markRcvd() {
+	*pkt = ringidx{}
+	// pkt.end = 0
+	// pkt.off = 0
+}
@@ -5,32 +5,52 @@ import (
 	"testing"
 )

-func TestTxQueueWrite(t *testing.T) {
+func TestTxQueue_SequentialMessages(t *testing.T) {
 	const (
-		bufsize = 1024
-		maxPkt  = 3
-		msg     = "hello world"
+		bufsize  = 2
+		maxPkt   = 1
+		msg      = "hello world"
+		startAck = 0 // this is the initial sequence number.
 	)
 	buf := make([]byte, bufsize)
-	rtx := newRingTx(buf, maxPkt)
-
-	bufs := bytes.SplitAfter([]byte(msg), []byte("e"))
+	var rtx ringTx
+	err := rtx.Reset(buf, maxPkt, startAck)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// msgs := bytes.SplitAfter([]byte(msg), []byte("e"))
+	msgs := bytes.Split([]byte(msg), []byte(""))
 	var data [bufsize]byte
-	for i, buf := range bufs {
-		n, err := rtx.Write(buf)
+	prevSeq := Value(startAck)
+	for i, msg := range msgs {
+		n, err := rtx.Write(msg)
 		if err != nil {
 			t.Fatalf("writing packet %d: %s", i, err)
-		} else if n != len(buf) {
-			t.Fatalf("want %d written, got %d", len(buf), n)
+		} else if n != len(msg) {
+			t.Fatalf("want %d written, got %d", len(msg), n)
 		}
-
-		n, err = rtx.MakePacket(data[:])
+		unsent := rtx.Buffered()
+		if len(msg) != unsent {
+			t.Fatalf("want %d unsent buffered, got %d", unsent, len(msg))
+		}
+		sent := rtx.BufferedSent()
+		if sent > 0 {
+			t.Fatalf("want 0 bytes sent, got %d", sent)
+		}
+		n, seq, err := rtx.MakePacket(data[:])
 		if err != nil {
 			t.Fatalf("making packet %d: %s", i, err)
-		} else if n != len(buf) {
-			t.Fatalf("want %d packet read, got %d", len(buf), n)
-		} else if !bytes.Equal(buf, data[:n]) {
-			t.Fatalf("want data %q, got data read %q", buf, data[:n])
+		} else if n != len(msg) {
+			t.Fatalf("want %d packet read, got %d", len(msg), n)
+		} else if !bytes.Equal(msg, data[:n]) {
+			t.Fatalf("want data %q, got data read %q", msg, data[:n])
+		} else if seq != prevSeq {
+			t.Fatalf("want seq %d, got %d", prevSeq, seq)
+		}
+		prevSeq += Value(n)
+		err = rtx.RecvACK(prevSeq)
+		if err != nil {
+			t.Fatal(err)
 		}
 	}
 }