From 83719ffe98a55576fde1c44d552c9a51e5ed1f73 Mon Sep 17 00:00:00 2001 From: soypat Date: Fri, 24 Jan 2025 21:46:39 -0300 Subject: [PATCH] fix tests not passing for RingTx --- lneto_test.go | 26 +++++--- tcp/txqueue.go | 157 ++++++++++++++++++++++++++++---------------- tcp/txqueue_test.go | 54 ++++++++++----- 3 files changed, 155 insertions(+), 82 deletions(-) diff --git a/lneto_test.go b/lneto_test.go index 91f218a..7be2a51 100644 --- a/lneto_test.go +++ b/lneto_test.go @@ -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()) diff --git a/tcp/txqueue.go b/tcp/txqueue.go index 4e3b611..3d24111 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -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 +} diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index 4715d8a..f6c2546 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -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) } } }