add tcp ringtx

internal/ring.go

@@ -12,14 +12,39 @@ var errRingBufferFull = errors.New("tseq/ring: buffer full")
 
 // NewRing returns a new ring buffer ready for use.
 func NewRing(buf []byte) *Ring {
-	return &Ring{buf: buf}
+	return &Ring{Buf: buf}
 }
 
 // Ring implements basic Ring buffer functionality.
 type Ring struct {
-	buf []byte
+	// Buf is used to store data written into Ring
-	off int
+	// with Write methods and then read out with Read methods.
-	end int
+	Buf []byte
+	// Start of readable data which indexes into Buf.
+	Off int
+	// End of readable data which indexes into Buf.
+	End int
+}
+
+// WriteLimited performs a write that does not write over the ring buffer's
+// limitOffset index, which points to a position to r.Buf.
+func (r *Ring) WriteLimited(b []byte, limitOffset int) (int, error) {
+	if limitOffset > len(r.Buf) {
+		panic("bad limit offset")
+	}
+	if len(b) > len(r.Buf) {
+		return 0, io.ErrShortBuffer
+	}
+	writeEnd := r.Off + len(b)
+	if limitOffset >= r.Off && writeEnd > limitOffset {
+		return 0, errRingBufferFull
+	} else if writeEnd > len(r.Buf) {
+		writeEnd %= len(r.Buf)
+		if writeEnd > limitOffset {
+			return 0, errRingBufferFull
+		}
+	}
+	return r.Write(b)
 }
 
 // WriteString is a wrapper around [Ring.Write] that avoids allocation of converting byte slice to string.
@@ -37,17 +62,17 @@ func (r *Ring) Write(b []byte) (int, error) {
 	if midFree > 0 {
 		// start     end       off    len(buf)
 		//   |  used  |  mfree  |  used  |
-		n := copy(r.buf[r.end:r.off], b)
+		n := copy(r.Buf[r.End:r.Off], b)
-		r.end += n
+		r.End += n
 		return n, nil
 	}
 	// start       off       end      len(buf)
 	//   |  sfree   |  used   |  efree   |
-	n := copy(r.buf[r.end:], b)
+	n := copy(r.Buf[r.End:], b)
-	r.end += n
+	r.End += n
 	if n < len(b) {
-		n2 := copy(r.buf, b[n:])
+		n2 := copy(r.Buf, b[n:])
-		r.end = n2
+		r.End = n2
 		n += n2
 	}
 	return n, nil
@@ -66,10 +91,10 @@ func (r *Ring) ReadDiscard(n int) {
 		panic("discard exceeds length")
 	case n == buffered:
 		r.Reset()
-	case n+r.off > len(r.buf):
+	case n+r.Off > len(r.Buf):
-		r.off = n - (len(r.buf) - r.off)
+		r.Off = n - (len(r.Buf) - r.Off)
 	default:
-		r.off += n
+		r.Off += n
 	}
 }
 
@@ -83,7 +108,7 @@ func (r *Ring) ReadAt(p []byte, off64 int64) (int, error) {
 		return 0, io.ErrUnexpectedEOF
 	}
 	r2 := *r
-	r2.off = (r2.off + off) % (r.Size())
+	r2.Off = (r2.Off + off) % (r.Size())
 	return r2.ReadPeek(p)
 }
 
@@ -99,29 +124,29 @@ func (r *Ring) Read(b []byte) (int, error) {
 	if err != nil {
 		return n, err
 	}
-	r.off = newOff
+	r.Off = newOff
 	r.onReadEnd()
 	return n, nil
 }
 
 func (r *Ring) read(b []byte) (n, newOff int, err error) {
-	newOff = r.off
+	newOff = r.Off
 	if r.Buffered() == 0 {
 		return 0, newOff, io.EOF
 	}
-	if r.end > r.off {
+	if r.End > r.Off {
 		// start       off       end      len(buf)
 		//   |  sfree   |  used   |  efree   |
-		n = copy(b, r.buf[r.off:r.end])
+		n = copy(b, r.Buf[r.Off:r.End])
 		newOff += n
 		return n, newOff, nil
 	}
 	// start     end       off     len(buf)
 	//   |  used  |  mfree  |  used  |
-	n = copy(b, r.buf[r.off:])
+	n = copy(b, r.Buf[r.Off:])
 	newOff += n
 	if n < len(b) {
-		n2 := copy(b[n:], r.buf[:r.end])
+		n2 := copy(b[n:], r.Buf[:r.End])
 		newOff = n2
 		n += n2
 	}
@@ -130,13 +155,13 @@ func (r *Ring) read(b []byte) (n, newOff int, err error) {
 
 // Reset flushes all data from ring buffer so that no data can be further read.
 func (r *Ring) Reset() {
-	r.off = 0
+	r.Off = 0
-	r.end = 0
+	r.End = 0
 }
 
 // Size returns the capacity of the ring buffer.
 func (r *Ring) Size() int {
-	return len(r.buf)
+	return len(r.Buf)
 }
 
 // Buffered returns amount of bytes ready to read from ring buffer. Always less than [ring.Size].
@@ -146,38 +171,38 @@ func (r *Ring) Buffered() int {
 
 // Free returns amount of bytes that can be read into ring buffer before reaching maximum capacity given by [ring.Size]. Always less than [ring.Size].
 func (r *Ring) Free() int {
-	if r.off == 0 {
+	if r.Off == 0 {
-		return len(r.buf) - r.end
+		return len(r.Buf) - r.End
 	}
 
-	if r.off < r.end {
+	if r.Off < r.End {
 		// start       off       end      len(buf)
 		//   |  sfree   |  used   |  efree   |
-		startFree := r.off
+		startFree := r.Off
-		endFree := len(r.buf) - r.end
+		endFree := len(r.Buf) - r.End
 		return startFree + endFree
 	}
 	// start     end       off     len(buf)
 	//   |  used  |  mfree  |  used  |
-	return r.off - r.end
+	return r.Off - r.End
 }
 
 func (r *Ring) midFree() int {
-	if r.end >= r.off {
+	if r.End >= r.Off {
 		return 0
 	}
-	return r.off - r.end
+	return r.Off - r.End
 }
 
 // onReadEnd does some cleanup of [ring.off] and [ring.end] fields if possible for contiguous read performance benefits.
 func (r *Ring) onReadEnd() {
-	if r.end == len(r.buf) {
+	if r.End == len(r.Buf) {
-		r.end = 0 // Wrap around.
+		r.End = 0 // Wrap around.
 	}
-	if r.off == len(r.buf) {
+	if r.Off == len(r.Buf) {
-		r.off = 0 // Wrap around.
+		r.Off = 0 // Wrap around.
 	}
-	if r.off == r.end {
+	if r.Off == r.End {
 		r.Reset() // We read everything, reset.
 	}
 }

internal/ring_test.go

@@ -11,7 +11,7 @@ func TestRing(t *testing.T) {
 	rng := rand.New(rand.NewSource(0))
 	const bufSize = 10
 	r := &Ring{
-		buf: make([]byte, bufSize),
+		Buf: make([]byte, bufSize),
 	}
 	const data = "hello"
 	_, err := r.WriteString(data)
@@ -138,7 +138,6 @@ func TestRing(t *testing.T) {
 		}
 
 		// ReadDiscard test.
-
 		discard := rng.Intn(nfirst+nsecond) + 1
 		r.ReadDiscard(discard)
 		n, err := r.Read(readback[:])
@@ -159,15 +158,12 @@ func TestRing(t *testing.T) {
 
 func TestRing2(t *testing.T) {
 	const maxsize = 6
-	const ntests = 800
+	const ntests = 80000
 	rng := rand.New(rand.NewSource(0))
 	data := make([]byte, maxsize)
 	ringbuf := make([]byte, maxsize)
 	auxbuf := make([]byte, maxsize)
 	rng.Read(data)
-	// TODO(soypat): This test fails for greater ntests.
-	// It was not fixed because of a compiler bug: https://github.com/golang/go/issues/64854
-	// and since the benefits of the changes in this PR are already much better than what we previously had.
 	for i := 0; i < ntests; i++ {
 		dsize := max(rng.Intn(len(data)), 1)
 		if !testRing1_loopback(t, rng, ringbuf, data[:dsize], auxbuf) {
@@ -180,7 +176,7 @@ func TestRing_findcrash(t *testing.T) {
 	const maxsize = 33
 	const ntests = 800000
 	r := Ring{
-		buf: make([]byte, maxsize*6),
+		Buf: make([]byte, maxsize*6),
 	}
 	rng := rand.New(rand.NewSource(0))
 	data := make([]byte, maxsize)
@@ -229,7 +225,7 @@ func testRing1_loopback(t *testing.T, rng *rand.Rand, ringbuf, data, auxbuf []by
 	}
 	dsize := len(data)
 	var r Ring
-	r.buf = ringbuf
+	r.Buf = ringbuf
 
 	nfirst := rng.Intn(dsize) / 2
 	nsecond := rng.Intn(dsize) / 2
@@ -283,21 +279,21 @@ func fragmentReadInto(r io.Reader, buf []byte) (n int, _ error) {
 
 func setRingData(t *testing.T, r *Ring, offset int, data []byte) {
 	t.Helper()
-	if len(data) > len(r.buf) {
+	if len(data) > len(r.Buf) {
 		panic("data too large")
 	}
-	n := copy(r.buf[offset:], data)
+	n := copy(r.Buf[offset:], data)
-	r.end = offset + n
+	r.End = offset + n
-	if len(data)+offset > len(r.buf) {
+	if len(data)+offset > len(r.Buf) {
 		// End of buffer not enough to hold data, wrap around.
-		n = copy(r.buf, data[n:])
+		n = copy(r.Buf, data[n:])
-		r.end = n
+		r.End = n
 	}
-	r.off = offset
+	r.Off = offset
 	r.onReadEnd()
 	// println("buf:", len(r.buf), "end:", r.end, "off:", r.off, offset, "data:", len(data))
 	free := r.Free()
-	wantFree := len(r.buf) - len(data)
+	wantFree := len(r.Buf) - len(data)
 	if free != wantFree {
 		t.Fatalf("free got %d; want %d", free, wantFree)
 	}
@@ -306,12 +302,12 @@ func setRingData(t *testing.T, r *Ring, offset int, data []byte) {
 	if buffered != wantBuffered {
 		t.Fatalf("buffered got %d; want %d", buffered, wantBuffered)
 	}
-	end := r.end
+	end := r.End
-	off := r.off
+	off := r.Off
 	sdata := r.string()
 	if sdata != string(data) {
 		t.Fatalf("data got %q; want %q", sdata, data)
 	}
-	r.end = end
+	r.End = end
-	r.off = off
+	r.Off = off
 }

lneto/tcp/txqueue.go

@@ -0,0 +1,143 @@
+package tcp
+
+import (
+	"errors"
+	"time"
+
+	"github.com/soypat/tseq/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),
+	}
+}
+
+// ringTx is a ring buffer with retransmission queue functionality added.
+type ringTx struct {
+	// rawbuf contains the ring buffer of ordered bytes. It should be the size of the window.
+	rawbuf []byte
+	// packets contains
+	packets []ringidx
+	// firstPkt is the index of the oldest packet in the packets field.
+	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
+}
+
+// ringidx represents packet data inside RingTx
+type ringidx struct {
+	// off is data start offset of packet data inside buf.
+	off int
+	// end is the ringed data end offset, non-inclusive.
+	end int
+	// seq is the sequence number of the packet.
+	seq Value
+	t   time.Time
+	// acked flags if this packet has been acknowledged. Useful for SACK (selective acknowledgement)
+	// acked bool
+}
+
+// Buffered returns the amount of unsent bytes.
+func (tx *ringTx) Buffered() int {
+	r := tx.unsentRing()
+	return r.Buffered()
+}
+
+// BufferedSent returns the total amount of bytes sent but not acked.
+func (tx *ringTx) BufferedSent() int {
+	r := tx.sentRing()
+	return r.Buffered()
+}
+
+// Write writes data to the underlying unsent data ring buffer.
+func (tx *ringTx) Write(b []byte) (int, error) {
+	first := tx.packets[tx.firstPkt]
+	r := tx.unsentRing()
+	if first.off < 0 {
+		// No packets in queue case.
+		return r.Write(b)
+	}
+	return r.WriteLimited(b, first.off)
+}
+
+// ReadPacket 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) {
+	nxtpkt := (tx.lastPkt + 1) % len(tx.packets)
+	if tx.firstPkt == nxtpkt {
+		return 0, errors.New("packet queue full")
+	}
+
+	r := tx.unsentRing()
+	start := r.Off
+	n, err := r.Read(b)
+	if err != nil {
+		return n, err
+	}
+	last := &tx.packets[tx.lastPkt]
+	rlast := tx.packetRing(tx.lastPkt)
+	tx.packets[nxtpkt].off = start
+	tx.packets[nxtpkt].end = r.Off
+	tx.packets[nxtpkt].seq = last.seq + Value(rlast.Buffered())
+	tx.lastPkt = nxtpkt
+	tx.unsentoff = r.Off
+	return n, nil
+}
+
+// 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)
+}
+
+func (tx *ringTx) packetRing(i int) internal.Ring {
+	pkt := tx.packets[i]
+	if pkt.off < 0 {
+		return internal.Ring{}
+	}
+	return tx.ring(pkt.off, pkt.end)
+}
+
+// RecvSegment processes an incoming segment and updates the sent packet queue
+func (tx *ringTx) RecvACK(ack Value) error {
+	i := tx.firstPkt
+	for {
+		pkt := &tx.packets[i]
+		if ack >= pkt.seq {
+			// Packet was received by remote. Mark it as acked.
+			pkt.off = -1
+			tx.firstPkt++
+			continue
+		}
+		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) sentRing() internal.Ring {
+	first := tx.packets[tx.firstPkt]
+	if first.off < 0 {
+		return tx.ring(0, 0)
+	}
+	last := tx.packets[tx.lastPkt]
+	return tx.ring(first.off, last.end)
+}
+
+func (tx *ringTx) ring(off, end int) internal.Ring {
+	return internal.Ring{Buf: tx.rawbuf, Off: off, End: end}
+}