From f5157e7e4a3fafa5c5dcb45b633692afc513029b Mon Sep 17 00:00:00 2001 From: Patricio Whittingslow Date: Wed, 5 Nov 2025 19:21:30 -0300 Subject: [PATCH] huge fixes to ring buffer implementations and mostly passing tests --- internal/ring.go | 78 ++++++++++--------- internal/ring_test.go | 3 +- tcp/txqueue.go | 33 +++++--- tcp/txqueue_test.go | 175 +++++++++++++++++++++--------------------- 4 files changed, 155 insertions(+), 134 deletions(-) diff --git a/internal/ring.go b/internal/ring.go index 9f50338..f7ec7cc 100644 --- a/internal/ring.go +++ b/internal/ring.go @@ -28,37 +28,6 @@ type Ring struct { End int } -// FreeLimited returns the amount of bytes that can be written up to the -// argument offset limitOffset. See [Ring.WriteLimited]. -// If buffer is empty (End=0) write will begin at Off as a special case. -// If limitOffset is equal to the write starting place then FreeLimited returns 0. -func (r *Ring) FreeLimited(limitOffset int) (free int) { - if r.isFull() { - return 0 - } - - // Write start position. - var writeAt = r.End - if writeAt == 0 { - // Write start is End except when empty, in which case we writeAt at Off. - writeAt = r.Off - if limitOffset >= writeAt { - return limitOffset - writeAt // Contiguous case. - } - return r.Size() - writeAt + limitOffset // Wrap case. - } - - // normal (non-empty): write at End up to limitOffset, or Off, whichever comes first. - if writeAt <= limitOffset && writeAt <= r.Off { - return min(r.Off, limitOffset) - writeAt - } else if writeAt <= limitOffset { - return limitOffset - writeAt - } else if writeAt <= r.Off { - return r.Off - writeAt - } - return r.Size() - writeAt + min(limitOffset, r.Off) -} - // WriteLimited performs a write that does not write over the ring buffer's // limitOffset index, which points to a position to r.Buf. Up to [Ring.FreeLimited] bytes can be written. func (r *Ring) WriteLimited(b []byte, limitOffset int) (int, error) { @@ -83,10 +52,10 @@ func (r *Ring) WriteString(s string) (int, error) { // Write appends data to the ring buffer that can then be read back in order with [Ring.Read] methods. // An error is returned if length of data too large for buffer. Write is guaranteed to start at buffer index [Ring.Off]. func (r *Ring) Write(b []byte) (int, error) { - if r.isFull() { - return 0, errRingBufferFull - } else if len(b) == 0 { + if len(b) == 0 { return 0, errRingNoData + } else if r.IsFull() || r.Free() < len(b) { + return 0, errRingBufferFull } midFree := r.midFree() if midFree > 0 { @@ -170,7 +139,7 @@ func (r *Ring) Read(b []byte) (int, error) { } func (r *Ring) read(b []byte) (n int, err error) { - if r.Buffered() == 0 { + if r.IsEmpty() { return 0, io.EOF } if r.End > r.Off { @@ -229,10 +198,47 @@ func (r *Ring) midFree() int { return r.Off - r.End } -func (r *Ring) isFull() bool { +// FreeLimited returns the amount of bytes that can be written up to the +// argument offset limitOffset. See [Ring.WriteLimited]. +// If buffer is empty (End=0) write will begin at Off as a special case. +// If limitOffset is equal to the write starting place then FreeLimited returns 0. +func (r *Ring) FreeLimited(limitOffset int) (free int) { + if r.IsFull() { + return 0 + } + + // Write start position. + var writeAt = r.End + if writeAt == 0 { + // Write start is End except when empty, in which case we writeAt at Off. + writeAt = r.Off + if limitOffset >= writeAt { + return limitOffset - writeAt // Contiguous case. + } + return r.Size() - writeAt + limitOffset // Wrap case. + } + + // normal (non-empty): write at End up to limitOffset, or Off, whichever comes first. + if writeAt <= limitOffset && writeAt <= r.Off { + return min(r.Off, limitOffset) - writeAt + } else if writeAt <= limitOffset { + return limitOffset - writeAt + } else if writeAt <= r.Off { + return r.Off - writeAt + } + return r.Size() - writeAt + min(limitOffset, r.Off) +} + +// IsFull checks if ring buffer is full and cannot accept more data. +func (r *Ring) IsFull() bool { return r.End != 0 && (r.End == r.Off || (r.End == len(r.Buf) && r.Off == 0)) } +// IsEmpty checks if ring buffer is empty of data to read. Calls to Read on an empty buffer will return [io.EOF]. +func (r *Ring) IsEmpty() bool { + return r.End == 0 +} + // onReadEnd does some cleanup of [ring.off] and [ring.end] fields if possible for contiguous read performance benefits. func (r *Ring) onReadEnd(totalRead int) { if totalRead <= 0 { diff --git a/internal/ring_test.go b/internal/ring_test.go index 53b0940..ddca39b 100644 --- a/internal/ring_test.go +++ b/internal/ring_test.go @@ -324,7 +324,8 @@ func TestRingOverwrite(t *testing.T) { setRingData(t, r, off, rawbuf[:buf]) // Select write size overwriting data. for osz := bufSize - buf + 1; osz < bufSize+1; osz++ { - if osz <= r.Free() { + free := r.Free() + if osz <= free { panic("invalid test") } ngot, err := r.Write(auxbuf[:osz]) diff --git a/tcp/txqueue.go b/tcp/txqueue.go index f466dd9..52dbf09 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -104,12 +104,16 @@ func (rtx *ringTx) BufferedSent() int { // Write writes data to the underlying unsent data ring buffer. func (rtx *ringTx) Write(b []byte) (n int, err error) { - r, lim := rtx.unsentRing() - n, err = r.WriteLimited(b, lim) + unsent, lim := rtx.unsentRing() + if rtx.sentend == 0 { + n, err = unsent.Write(b) // catches case where limit matches with end when both buffers empty + } else { + n, err = unsent.WriteLimited(b, lim) + } if err != nil { return 0, err } - rtx.unsentend = r.End + rtx.unsentend = unsent.End return n, err } @@ -126,7 +130,7 @@ func (rtx *ringTx) MakePacket(b []byte, currentSeq Value) (int, error) { } // Reading unsent ring consumes unsent and converts it to "sent". unsent, _ := rtx.unsentRing() - oldSentOff := unsent.Off + oldUnsentOff := unsent.Off n, err := unsent.Read(b) if err != nil { return 0, err @@ -134,13 +138,22 @@ func (rtx *ringTx) MakePacket(b []byte, currentSeq Value) (int, error) { // unsentOff increases, sentEnd matches this value. // Start of buffer will be SENT, end of buffer will be UNSENT(or empty). // Packet generated has offset at old unsentOff. - newUnsentOff := unsent.Off - pkt := rtx.slist.AddPacket(n, oldSentOff, rtx.Size()) - if pkt.off != oldSentOff || pkt.end != addEnd(pkt.off, n, rtx.Size()) { + size := rtx.Size() + pkt := rtx.slist.AddPacket(n, oldUnsentOff, size) + if pkt.off != oldUnsentOff || pkt.end != addEnd(pkt.off, n, size) { panic("invalid generated packet") } - rtx.unsentoff = newUnsentOff - rtx.sentend = newUnsentOff + if rtx.sentend == 0 { + // Sent was previously empty, offset is reset from start of this packet + rtx.sentoff = pkt.off + } + if unsent.End == 0 { + // Fully read unsent buffer so offset is reset, need to recalculate. + rtx.unsentoff = pkt.end + } else { + rtx.unsentoff = unsent.Off + } + rtx.sentend = pkt.end rtx.unsentend = unsent.End return n, nil } @@ -282,7 +295,7 @@ func (sl *sentlist) AddPacket(datalen, off, bufsize int) *ringidx { panic("pkt buffer full") } lastPkt := sl.Newest() - if lastPkt != nil && off != lastPkt.end { + if lastPkt != nil && ((off != 0 && off != lastPkt.end) || (off == 0 && lastPkt.end != bufsize)) { panic("new sent packet offset must match last sent packet end") } sl.pkts = append(sl.pkts, ringidx{ diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index ea93bf8..b82b05f 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -13,7 +13,8 @@ import ( func TestRingTx_op(t *testing.T) { const maxBuf = 32 const maxpkt = 3 - const Nops = 3 + const Nops = 33 + const log = false type op uint8 const ( opWrite op = iota @@ -21,100 +22,100 @@ func TestRingTx_op(t *testing.T) { opAck opmax ) - - rng := rand.New(rand.NewSource(666)) - randop := func() op { return op(rng.Intn(int(opmax))) } var buf, auxbuf [maxBuf]byte dataWritten := make([]byte, 0, maxBuf*10) dataSent := make([]byte, 0, maxBuf*10) var rtx ringTx - for itest := 0; itest < 3; itest++ { - bufsize := rng.Intn(maxBuf/2) + maxBuf/2 - iss := Value(0) - npackets := rng.Intn(maxpkt-1) + 1 - err := rtx.Reset(buf[:bufsize], npackets, iss) - if err != nil { - t.Fatal(err) - } - // Prepare state for keeping track of test. - currentAcked := iss - currentSeq := iss - nsent := 0 - nunsent := 0 - nacked := 0 - dataWritten = dataWritten[:0] - dataSent = dataSent[:0] - for iop := 0; iop < Nops; iop++ { - free := bufsize - nsent - nunsent - availPkt := rtx.slist.Free() - op := randop() - var oplen int - var opname string - var opWriteData []byte - switch op { - case opWrite: - opname, oplen = "write", rng.Intn(free+1)+1 - opWriteData = auxbuf[:oplen] - rng.Read(opWriteData) - clear(auxbuf[oplen:]) - case opSend: - opname, oplen = "send", rng.Intn(nunsent+1)+1 - case opAck: - opname, oplen = "ack", rng.Intn(nsent+1)+1 + rng := rand.New(rand.NewSource(0)) + for iseed := int64(0); iseed < 10000; iseed++ { + rng.Seed(iseed) + for itest := 0; itest < 32; itest++ { + bufsize := rng.Intn(maxBuf/2) + maxBuf/2 + iss := Value(0) + npackets := rng.Intn(maxpkt-1) + 1 + err := rtx.Reset(buf[:bufsize], npackets, iss) + if err != nil { + t.Fatal(err) } - if itest < 2 { - continue // Debugging. - } - _ = opname - t.Logf("\n%s\nitest=%d iop=%d op=%s len=%d", rtx.mustAppendString(nil), itest, iop, opname, oplen) - switch op { - case opWrite: - // oplen=number of - nwgot, err := rtx.Write(opWriteData) - wantErr := oplen > free - if err != nil && oplen <= free { - t.Fatal(itest, iop, err) - } else if err == nil { - if wantErr { - panic("wanted write error") + // Prepare state for keeping track of test. + currentAcked := iss + currentSeq := iss + nsent := 0 + nunsent := 0 + nacked := 0 + dataWritten = dataWritten[:0] + dataSent = dataSent[:0] + for iop := 0; iop < Nops; iop++ { + free := bufsize - nsent - nunsent + availPkt := rtx.slist.Free() + op := op(rng.Intn(int(opmax))) + var oplen int + var opname string + var opWriteData []byte + switch op { + case opWrite: + opname, oplen = "write", rng.Intn(free+1)+1 + opWriteData = auxbuf[:oplen] + rng.Read(opWriteData) + case opSend: + opname, oplen = "send", rng.Intn(nunsent+1)+1 + case opAck: + opname, oplen = "ack", rng.Intn(nsent+1)+1 + } + if log { + t.Logf("\n%s\nseed=%d itest=%d iop=%d op=%s len=%d npkt=%d", rtx.mustAppendString(nil), iseed, itest, iop, opname, oplen, len(rtx.slist.pkts)) + } + switch op { + case opWrite: + // oplen=number of bytes to write into unsent buffer. + nwgot, err := rtx.Write(opWriteData) + wantErr := oplen > free + if err != nil && oplen <= free { + t.Fatal(itest, iop, err) + } else if err == nil { + if wantErr { + panic("wanted write error") + } + nunsent += nwgot + dataWritten = append(dataWritten, opWriteData[:nwgot]...) + } else if log { + t.Logf("opwrite: %s", err) } - nunsent += nwgot - dataWritten = append(dataWritten, opWriteData[:nwgot]...) - } else { - t.Logf("opwrite: %s", err) - } - case opSend: - // oplen=num bytes sent. - nsgot, err := rtx.MakePacket(auxbuf[:oplen], currentSeq) - megafail := nsgot > nunsent - if err != nil && oplen <= nunsent && availPkt > 0 { - t.Fatal(itest, iop, err) - } else if err == nil { - if megafail { - panic("megafail") + clear(opWriteData) + case opSend: + // oplen=num bytes to send in this operation. + nsgot, err := rtx.MakePacket(auxbuf[:oplen], currentSeq) + megafail := nsgot > nunsent + if err != nil && oplen <= nunsent && availPkt > 0 { + t.Fatal(itest, iop, err) + } else if err == nil { + if megafail { + panic("megafail") + } + nunsent -= nsgot + nsent += nsgot + dataSent = append(dataSent, auxbuf[:nsgot]...) + currentSeq += Value(nsgot) + } else if log { + t.Logf("opsend: %s", err) } - nunsent -= nsgot - nsent += nsgot - dataSent = append(dataSent, auxbuf[:nsgot]...) - currentSeq += Value(nsgot) - } else { - t.Logf("opsend: %s", err) + clear(auxbuf[:oplen]) + case opAck: + // oplen=acklength. + tryAck := currentAcked + Value(oplen) + err = rtx.RecvACK(tryAck) + if err != nil && oplen <= nsent { + t.Fatal(itest, iop, err) + } else if err == nil { + nsent -= oplen + nacked += oplen + currentAcked = tryAck + } else if log { + t.Logf("opack: %s", err) + } + default: + panic("unknown op") } - case opAck: - // oplen=acklength. - tryAck := currentAcked + Value(oplen) - err = rtx.RecvACK(tryAck) - if err != nil && oplen <= nsent { - t.Fatal(itest, iop, err) - } else if err == nil { - nsent -= oplen - nacked += oplen - currentAcked = tryAck - } else { - t.Logf("opack: %s", err) - } - default: - panic("unknown op") } } }