From 0a75ef46701d3dd133a45570d2185d99ee4e5b94 Mon Sep 17 00:00:00 2001 From: soypat Date: Mon, 13 Jan 2025 19:36:17 -0300 Subject: [PATCH] txqueue work --- internal/ring.go | 12 +++++------- internal/ring_test.go | 14 ++++++++++++++ tcp/txqueue.go | 4 ++-- tcp/txqueue_test.go | 35 +++++++++++++++++++++++++++++++++++ 4 files changed, 56 insertions(+), 9 deletions(-) create mode 100644 tcp/txqueue_test.go diff --git a/internal/ring.go b/internal/ring.go index 52c0897..907d55d 100644 --- a/internal/ring.go +++ b/internal/ring.go @@ -10,15 +10,12 @@ import ( var errRingBufferFull = errors.New("lneto/ring: buffer full") -// NewRing returns a new ring buffer ready for use. -func NewRing(buf []byte) *Ring { - return &Ring{Buf: buf} -} - // Ring implements basic Ring buffer functionality. type Ring struct { // Buf is used to store data written into Ring // with Write methods and then read out with Read methods. + // The capacity of Buf is unused. + // There is no readable data when both Off and End are zero. Buf []byte // Start of readable data which indexes into Buf. Off int @@ -35,8 +32,9 @@ func (r *Ring) WriteLimited(b []byte, limitOffset int) (int, error) { if len(b) > len(r.Buf) { return 0, io.ErrShortBuffer } - writeEnd := r.Off + len(b) - if limitOffset >= r.Off && writeEnd > limitOffset { + + writeEnd := (r.Off + len(b)) % len(b) + if limitOffset > r.Off && writeEnd > limitOffset { return 0, errRingBufferFull } else if writeEnd > len(r.Buf) { writeEnd %= len(r.Buf) diff --git a/internal/ring_test.go b/internal/ring_test.go index bae7328..70bd798 100644 --- a/internal/ring_test.go +++ b/internal/ring_test.go @@ -172,6 +172,20 @@ func TestRing2(t *testing.T) { } } +func TestRingWriteLimited(t *testing.T) { + rng := rand.New(rand.NewSource(0)) + const bufSize = 10 + r := &Ring{ + Buf: make([]byte, bufSize), + } + var data [bufSize]byte + for i := 0; i < 32; i++ { + n, _ := rng.Read(data[:rng.Intn(10)+1]) + limoff := r.Off + r.WriteLimited() + } +} + func TestRing_findcrash(t *testing.T) { const maxsize = 33 const ntests = 800000 diff --git a/tcp/txqueue.go b/tcp/txqueue.go index 8268b02..4543345 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -68,9 +68,9 @@ func (tx *ringTx) Write(b []byte) (int, error) { return r.WriteLimited(b, first.off) } -// ReadPacket reads from the unsent data ring buffer and generates a new packet segment. +// 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) NewPacketAndRead(b []byte) (int, error) { +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") diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go new file mode 100644 index 0000000..00b3d72 --- /dev/null +++ b/tcp/txqueue_test.go @@ -0,0 +1,35 @@ +package tcp + +import ( + "bytes" + "testing" +) + +func TestTxQueueWrite(t *testing.T) { + const ( + bufsize = 1024 + maxPkt = 3 + msg = "hello world" + ) + buf := make([]byte, bufsize) + rtx := newRingTx(buf, maxPkt) + + bufs := bytes.SplitAfter([]byte(msg), []byte("e")) + var data [bufsize]byte + for i, buf := range bufs { + n, err := rtx.Write(buf) + 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) + } + n, 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]) + } + } +}