diff --git a/internal/ring.go b/internal/ring.go index 12c0f8c..22dc322 100644 --- a/internal/ring.go +++ b/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 - off int - end int + // Buf is used to store data written into Ring + // with Write methods and then read out with Read methods. + 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) - r.end += n + n := copy(r.Buf[r.End:r.Off], b) + r.End += n return n, nil } // start off end len(buf) // | sfree | used | efree | - n := copy(r.buf[r.end:], b) - r.end += n + n := copy(r.Buf[r.End:], b) + r.End += n if n < len(b) { - n2 := copy(r.buf, b[n:]) - r.end = n2 + n2 := copy(r.Buf, b[n:]) + 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): - r.off = n - (len(r.buf) - r.off) + case n+r.Off > len(r.Buf): + 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.end = 0 + r.Off = 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 { - return len(r.buf) - r.end + if r.Off == 0 { + 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 - endFree := len(r.buf) - r.end + startFree := r.Off + 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) { - r.end = 0 // Wrap around. + if r.End == len(r.Buf) { + r.End = 0 // Wrap around. } - if r.off == len(r.buf) { - r.off = 0 // Wrap around. + if r.Off == len(r.Buf) { + r.Off = 0 // Wrap around. } - if r.off == r.end { + if r.Off == r.End { r.Reset() // We read everything, reset. } } diff --git a/internal/ring_test.go b/internal/ring_test.go index e468a18..bae7328 100644 --- a/internal/ring_test.go +++ b/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) - r.end = offset + n - if len(data)+offset > len(r.buf) { + n := copy(r.Buf[offset:], data) + r.End = offset + n + if len(data)+offset > len(r.Buf) { // End of buffer not enough to hold data, wrap around. - n = copy(r.buf, data[n:]) - r.end = n + n = copy(r.Buf, data[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 - off := r.off + end := r.End + off := r.Off sdata := r.string() if sdata != string(data) { t.Fatalf("data got %q; want %q", sdata, data) } - r.end = end - r.off = off + r.End = end + r.Off = off } diff --git a/lneto/tcp/txqueue.go b/lneto/tcp/txqueue.go new file mode 100644 index 0000000..68ace79 --- /dev/null +++ b/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} +}