soypat/lneto
1
0
Fork 0
mirror of https://github.com/soypat/lneto.git synced 2026-08-12 19:03:42 +00:00
Code Issues Packages Projects Releases 1 Wiki Activity

huge fixes to ring buffer implementations and mostly passing tests

Browse Source
This commit is contained in:
Patricio Whittingslow
2025-11-05 19:21:30 -03:00
parent 5e5f21a2b8
commit f5157e7e4a
4 changed files with 155 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/ring.go
+42 -36
View File
@@ -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 {
internal/ring_test.go
+2 -1
View File
@@ -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])
tcp/txqueue.go
+23 -10
View File
@@ -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{
tcp/txqueue_test.go
+88 -87
View File
@@ -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")
}
}
}