soypat/lneto
1
0
Fork 0
mirror of https://github.com/soypat/lneto.git synced 2026-08-20 14:39:02 +00:00
Code Issues Packages Projects Releases 1 Wiki Activity

rethinking ring buffer semantics and breaking everything in the process

Browse Source
This commit is contained in:
Patricio Whittingslow
2025-11-05 02:35:53 -03:00
parent 82bd54a6da
commit 5e5f21a2b8
6 changed files with 432 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/ring.go
+44 -14
View File
@@ -8,7 +8,10 @@ import (
"unsafe"
)
var errRingBufferFull = errors.New("lneto/ring: buffer full")
var (
errRingBufferFull = errors.New("lneto/ring: buffer full")
errRingNoData = errors.New("lneto/ring: empty write")
)
// Ring implements basic Ring buffer functionality.
type Ring struct {
@@ -18,26 +21,42 @@ type Ring struct {
// There is no readable data when End==0.
Buf []byte
// Start of readable data which indexes into Buf.
// If Off==End and End!=0 the buffer is full and data begins at Off.
// If Off==End and End!=0 the buffer is full and data begins at Off. Off<len(Buf) is always true.
Off int
// End of readable data which indexes into Buf, not including byte at End index.
// If End==0 then the buffer is empty. If End==Off and End!=0 the buffer is full.
End int
}
// SizeLimited returns the amount of bytes that can be written up to the
// argument offset limitOffset. See [Ring.WriteLimited]
// 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) {
end := r.End
if r.End == 0 {
end = r.Off
if r.isFull() {
return 0
}
if limitOffset > end {
free = limitOffset - end
} else {
free = len(r.Buf) - end + limitOffset
// 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.
}
return free
// 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
@@ -64,9 +83,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) {
free := r.Free()
if len(b) > free {
if r.isFull() {
return 0, errRingBufferFull
} else if len(b) == 0 {
return 0, errRingNoData
}
midFree := r.midFree()
if midFree > 0 {
@@ -74,6 +94,9 @@ func (r *Ring) Write(b []byte) (int, error) {
// | used | mfree | used |
n := copy(r.Buf[r.End:r.Off], b)
r.End += n
if r.End <= 0 {
panic("zero end after write") // TODO: remove panics after validation.
}
return n, nil
} else if r.End == 0 {
// To ensure Write begins on r.Off.
@@ -89,6 +112,9 @@ func (r *Ring) Write(b []byte) (int, error) {
r.End = n2
n += n2
}
if r.End <= 0 {
panic("zero end after write")
}
return n, nil
}
@@ -203,6 +229,10 @@ func (r *Ring) midFree() int {
return r.Off - r.End
}
func (r *Ring) isFull() bool {
return r.End != 0 && (r.End == r.Off || (r.End == len(r.Buf) && r.Off == 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
+90 -54
View File
@@ -21,8 +21,8 @@ func TestRing(t *testing.T) {
const data = "hello"
// Set random data and write some more and read it back.
for i := 0; i < 32; i++ {
nfirst := rng.Intn(bufSize) / 2
nsecond := rng.Intn(bufSize) / 2
nfirst := max(1, rng.Intn(bufSize)/2)
nsecond := max(1, rng.Intn(bufSize)/2)
if nfirst+nsecond > bufSize {
nfirst = bufSize - nsecond
}
@@ -222,9 +222,11 @@ func TestRingNonEmpty(t *testing.T) {
if checkWrite {
testRingSanity(t, r)
free := r.Size() - buf
n, err := r.Write(data[:free])
if n != free || err != nil {
t.Errorf("want %d to fill buffer, got n=%d err=%v", free, n, err)
if free != 0 {
n, err := r.Write(data[:free])
if n != free || err != nil {
t.Errorf("want %d to fill buffer, got n=%d err=%v", free, n, err)
}
}
}
if checkRead {
@@ -336,57 +338,60 @@ func TestRingOverwrite(t *testing.T) {
}
}
func TestRingWriteLimited(t *testing.T) {
rng := rand.New(rand.NewSource(2))
const bufSize = 8
r := &Ring{
Buf: make([]byte, bufSize),
}
testRingSanity(t, r)
var data [bufSize]byte
var wdata [bufSize]byte
for i := 0; i < 10000; i++ {
for i := range r.Buf {
r.Buf[i] = 0
}
buffered, _ := rng.Read(data[:rng.Intn(bufSize-2)+1])
off := rng.Intn(bufSize)
setRingData(t, r, off, data[:buffered])
if r.Buffered() != buffered {
t.Fatalf("failed to set buffered amount of data")
} else if r.Off != off {
t.Fatal("bad offset")
}
free := r.Free()
// func TestRingWriteLimited(t *testing.T) {
// rng := rand.New(rand.NewSource(2))
// const bufSize = 8
// r := &Ring{
// Buf: make([]byte, bufSize),
// }
// testRingSanity(t, r)
// var data [bufSize]byte
// var wdata [bufSize]byte
// for i := 0; i < 10000; i++ {
// for i := range r.Buf {
// r.Buf[i] = 0
// }
// buffered, _ := rng.Read(data[:rng.Intn(bufSize-2)+1])
// off := rng.Intn(bufSize)
// setRingData(t, r, off, data[:buffered])
// if r.Buffered() != buffered {
// t.Fatalf("failed to set buffered amount of data")
// } else if r.Off != off {
// t.Fatal("bad offset")
// }
// limOff := rng.Intn(bufSize) + 1
// freeLim := r.FreeLimited(limOff)
// free := r.Free()
toWrite := rng.Intn(free-1) + 1
rng.Read(wdata[:toWrite])
limOff := rng.Intn(bufSize) + 1
var wantN int
isContiguous := limOff > r.End
if isContiguous {
wantN = min(toWrite, limOff-r.End)
} else {
wantN = min(toWrite, len(r.Buf)-r.End+limOff)
}
overwrite := toWrite > wantN
// toWrite := rng.Intn(freeLim+1) + 1
// rng.Read(wdata[:toWrite])
n, err := r.WriteLimited(wdata[:toWrite], limOff)
if !overwrite && err != nil {
t.Errorf("limited write: %s", err)
} else if !overwrite && n != wantN {
t.Errorf("nwant=%d ngot=%d off=%d lim=%d towrite=%d buffered=%d/%d wantremain=%d gotremain=%d endOff=%d", wantN, n, off, limOff, toWrite, buffered, r.Size(), free-wantN, free-n, r.Off)
} else if overwrite && (err == nil || n != 0) {
t.Errorf("expected full buffer error and no data written on limit overwrite, got %d", n)
}
for i := r.End % r.Size(); i != limOff && i != r.Off; i = (i + 1) % r.Size() {
if r.Buf[i] != 0 {
t.Fatalf("OVERWRITE pos=%d end=%d lim=%d", i, r.End, limOff)
}
}
testRingSanity(t, r)
}
}
// var wantN int
// isContiguous := limOff > r.End
// if isContiguous {
// wantN = min(toWrite, limOff-r.End)
// } else {
// wantN = min(toWrite, len(r.Buf)-r.End+limOff)
// }
// overwrite := toWrite > wantN
// n, err := r.WriteLimited(wdata[:toWrite], limOff)
// if !overwrite && err != nil {
// t.Errorf("limited write: %s", err)
// t.Errorf("size=%d off=%d end=%d lim=%d nwrite=%d", r.Size(), r.Off, r.End, limOff, toWrite)
// } else if !overwrite && n != wantN {
// t.Errorf("nwant=%d ngot=%d off=%d lim=%d towrite=%d buffered=%d/%d wantremain=%d gotremain=%d endOff=%d", wantN, n, off, limOff, toWrite, buffered, r.Size(), free-wantN, free-n, r.Off)
// } else if overwrite && (err == nil || n != 0) {
// t.Errorf("expected full buffer error and no data written on limit overwrite, got %d", n)
// }
// for i := r.End % r.Size(); i != limOff && i != r.Off; i = (i + 1) % r.Size() {
// if r.Buf[i] != 0 {
// t.Fatalf("OVERWRITE pos=%d end=%d lim=%d", i, r.End, limOff)
// }
// }
// testRingSanity(t, r)
// }
// }
func TestRing_findcrash(t *testing.T) {
const maxsize = 33
@@ -437,6 +442,37 @@ func TestRing_findcrash(t *testing.T) {
}
}
func TestRingFreeLimited(t *testing.T) {
const n = 16
rng := rand.New(rand.NewSource(1))
buffer := make([]byte, n)
r := Ring{Buf: buffer}
for itest := 0; itest < 100000; itest++ {
r.Off = rng.Intn(n)
r.End = rng.Intn(n + 1)
limit := rng.Intn(n)
wantFreeLimited := 0
writeStart := r.End
if writeStart == 0 {
// Empty buffer case.
writeStart = r.Off
for i := writeStart % n; i != limit; i = (i + 1) % n {
wantFreeLimited++
}
} else {
// Buffer with contents. Beware limit and own offset.
for i := writeStart % n; i != limit && i != r.Off; i = (i + 1) % n {
wantFreeLimited++
}
}
gotFreeLimited := r.FreeLimited(limit)
if gotFreeLimited != wantFreeLimited {
t.Fatalf("[%d] len=%d off=%d end=%d limit=%d GOT=%d, WANT=%d", itest, n, r.Off, r.End, limit, gotFreeLimited, wantFreeLimited)
}
}
}
func testRing1_loopback(t *testing.T, rng *rand.Rand, ringbuf, data, auxbuf []byte) bool {
if len(data) > len(ringbuf) || len(data) > len(auxbuf) {
panic("invalid ringbuf or data")
internal/zoneprinter.go
+123
View File
@@ -0,0 +1,123 @@
package internal
import (
"errors"
"fmt"
"strconv"
)
type ZonePrinter struct {
zbuf []BufferZone
aux []byte
}
type BufferZone struct {
Name string
Start, End int
}
func (bp *ZonePrinter) AppendPrintZones(dst []byte, bufSize int, zones ...BufferZone) ([]byte, error) {
n := bufSize
if n == 0 {
return dst, errors.New("empty buffer")
}
if len(bp.zbuf) != 0 {
panic("race condition detected: zbuf not zero lengthed- ZonePrinter not to be used concurrently")
}
// Zeroed zone buffer. Will be used to detect collisions.
// TODO(soypat): Really wasteful, there's an obvious way to just have len(zones)+1 zbuf, but I've already spent too much time here. This also makes collision detection really easy so whatever.
bp.zbuf = append(bp.zbuf[:0], make([]BufferZone, n)...)
labels := bp.zbuf
// helper to paint an interval on the ring
paint := func(z BufferZone) error {
if z.End == 0 { // 0 == empty
return nil
}
if z.Start < z.End {
for i := z.Start; i < z.End; i++ {
if labels[i].Name != "" {
return fmt.Errorf("paint collision: %q/%q @%d size=%d", labels[i].Name, z.Name, i, bufSize)
}
labels[i].Name = z.Name
}
} else { // wrap
for i := z.Start; i < n; i++ {
if labels[i].Name != "" {
return fmt.Errorf("wrap(start) paint collision: %q/%q @%d size=%d", labels[i].Name, z.Name, i, bufSize)
}
labels[i].Name = z.Name
}
for i := 0; i < z.End; i++ {
if labels[i].Name != "" {
return fmt.Errorf("wrap(end) paint collision: %q/%q @%d size=%d", labels[i].Name, z.Name, i, bufSize)
}
labels[i].Name = z.Name
}
}
return nil
}
// paint
for _, zone := range zones {
err := paint(zone)
if err != nil {
return dst, err
}
}
// unpainted sections are free.
for i := range labels {
if labels[i].Name == "" {
labels[i].Name = "free"
}
}
// Compress to segments.
icurrent := 0
for i := 1; i < n; i++ {
current := labels[icurrent]
next := labels[i]
if current.Name != next.Name {
labels[icurrent].End = i
icurrent++
labels[icurrent].Start = i
labels[icurrent].Name = next.Name
}
}
segs := labels[:icurrent+1]
segs[icurrent].End = n
var l1 []byte = dst
var l2 []byte = bp.aux[:0]
for _, s := range segs {
l2start := len(l2)
if s.Name == "free" {
l2 = append(l2, "| "...)
} else {
l2 = append(l2, "|--"...)
}
l2 = append(l2, s.Name...)
l2 = append(l2, '(')
l2 = strconv.AppendInt(l2, int64(s.End-s.Start), 10)
if s.Name == "free" {
l2 = append(l2, ") "...)
} else {
l2 = append(l2, ")--"...)
}
fraglen := len(l2) - l2start
l1start := len(l1)
l1 = strconv.AppendInt(l1, int64(s.Start), 10)
paddingNeeded := fraglen - (len(l1) - l1start)
for range paddingNeeded {
l1 = append(l1, ' ')
}
}
l1 = strconv.AppendInt(l1, int64(bufSize), 10)
l1 = append(l1, '\n')
l2 = append(l2, '|')
l1 = append(l1, l2...)
bp.zbuf = bp.zbuf[:0] // zero out, used to detect concurrent usage of ZonePrinter.
return l1, nil
}