mirror of
https://github.com/soypat/lneto.git
synced 2026-08-12 02:43:44 +00:00
add ring buffer implementation
This commit is contained in:
soypat
@@ -0,0 +1,209 @@
|
||||
package internal
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"io"
|
||||
"math"
|
||||
"unsafe"
|
||||
)
|
||||
|
||||
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}
|
||||
}
|
||||
|
||||
// Ring implements basic Ring buffer functionality.
|
||||
type Ring struct {
|
||||
buf []byte
|
||||
off int
|
||||
end int
|
||||
}
|
||||
|
||||
// WriteString is a wrapper around [Ring.Write] that avoids allocation of converting byte slice to string.
|
||||
func (r *Ring) WriteString(s string) (int, error) {
|
||||
return r.Write(unsafe.Slice(unsafe.StringData(s), len(s)))
|
||||
}
|
||||
|
||||
// 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.
|
||||
func (r *Ring) Write(b []byte) (int, error) {
|
||||
free := r.Free()
|
||||
if len(b) > free {
|
||||
return 0, errRingBufferFull
|
||||
}
|
||||
midFree := r.midFree()
|
||||
if midFree > 0 {
|
||||
// start end off len(buf)
|
||||
// | used | mfree | used |
|
||||
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
|
||||
if n < len(b) {
|
||||
n2 := copy(r.buf, b[n:])
|
||||
r.end = n2
|
||||
n += n2
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// ReadDiscard is a performance auxiliary method that performs a dummy read or no-op read
|
||||
// for advancing the read pointer n bytes without actually copying data.
|
||||
// This method panics if amount of bytes is more than buffered (see [Ring.Buffered]).
|
||||
func (r *Ring) ReadDiscard(n int) {
|
||||
if n < 0 {
|
||||
panic("negative discard amount")
|
||||
}
|
||||
buffered := r.Buffered()
|
||||
switch {
|
||||
case n > buffered:
|
||||
panic("discard exceeds length")
|
||||
case n == buffered:
|
||||
r.Reset()
|
||||
case n+r.off > len(r.buf):
|
||||
r.off = n - (len(r.buf) - r.off)
|
||||
default:
|
||||
r.off += n
|
||||
}
|
||||
}
|
||||
|
||||
// ReadAt reads data at an offset from start of readable data but does not advance read pointer. [io.EOF] returned when no data available.
|
||||
func (r *Ring) ReadAt(p []byte, off64 int64) (int, error) {
|
||||
if math.MaxInt != math.MaxInt64 && off64+int64(len(p)) > math.MaxInt32 {
|
||||
return 0, errors.New("offset too large (32 bit overflow)") // Check only compiles for 32-bit platforms.
|
||||
}
|
||||
off := int(off64)
|
||||
if off+len(p) > r.Buffered() {
|
||||
return 0, io.ErrUnexpectedEOF
|
||||
}
|
||||
r2 := *r
|
||||
r2.off = (r2.off + off) % (r.Size())
|
||||
return r2.ReadPeek(p)
|
||||
}
|
||||
|
||||
// ReadPeek reads up to len(b) bytes from the ring buffer but does not advance the read pointer. [io.EOF] returned when no data available.
|
||||
func (r *Ring) ReadPeek(b []byte) (int, error) {
|
||||
n, _, err := r.read(b)
|
||||
return n, err
|
||||
}
|
||||
|
||||
// Read reads up to len(b) bytes from the ring buffer and advances the read pointer. [io.EOF] returned when no data available.
|
||||
func (r *Ring) Read(b []byte) (int, error) {
|
||||
n, newOff, err := r.read(b)
|
||||
if err != nil {
|
||||
return n, err
|
||||
}
|
||||
r.off = newOff
|
||||
r.onReadEnd()
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func (r *Ring) read(b []byte) (n, newOff int, err error) {
|
||||
newOff = r.off
|
||||
if r.Buffered() == 0 {
|
||||
return 0, newOff, io.EOF
|
||||
}
|
||||
if r.end > r.off {
|
||||
// start off end len(buf)
|
||||
// | sfree | used | efree |
|
||||
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:])
|
||||
newOff += n
|
||||
if n < len(b) {
|
||||
n2 := copy(b[n:], r.buf[:r.end])
|
||||
newOff = n2
|
||||
n += n2
|
||||
}
|
||||
return n, newOff, nil
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
// Size returns the capacity of the ring buffer.
|
||||
func (r *Ring) Size() int {
|
||||
return len(r.buf)
|
||||
}
|
||||
|
||||
// Buffered returns amount of bytes ready to read from ring buffer. Always less than [ring.Size].
|
||||
func (r *Ring) Buffered() int {
|
||||
return r.Size() - r.Free()
|
||||
}
|
||||
|
||||
// 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 < r.end {
|
||||
// start off end len(buf)
|
||||
// | sfree | used | efree |
|
||||
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
|
||||
}
|
||||
|
||||
func (r *Ring) midFree() int {
|
||||
if r.end >= r.off {
|
||||
return 0
|
||||
}
|
||||
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.off == len(r.buf) {
|
||||
r.off = 0 // Wrap around.
|
||||
}
|
||||
if r.off == r.end {
|
||||
r.Reset() // We read everything, reset.
|
||||
}
|
||||
}
|
||||
|
||||
func max(a, b int) int {
|
||||
if a > b {
|
||||
return a
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func min(a, b int) int {
|
||||
if a < b {
|
||||
return a
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func (r *Ring) string() string {
|
||||
var b bytes.Buffer
|
||||
r2 := *r
|
||||
b.ReadFrom(&r2)
|
||||
return b.String()
|
||||
}
|
||||
|
||||
func (r *Ring) _string(off int64) string {
|
||||
s := r.string()
|
||||
return s[off:]
|
||||
}
|
||||
@@ -0,0 +1,317 @@
|
||||
package internal
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"io"
|
||||
"math/rand"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestRing(t *testing.T) {
|
||||
rng := rand.New(rand.NewSource(0))
|
||||
const bufSize = 10
|
||||
r := &Ring{
|
||||
buf: make([]byte, bufSize),
|
||||
}
|
||||
const data = "hello"
|
||||
_, err := r.WriteString(data)
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
// Case where data is contiguous and at start of buffer.
|
||||
var buf [bufSize]byte
|
||||
n, err := fragmentReadInto(r, buf[:])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if string(buf[:n]) != data {
|
||||
t.Fatalf("got %q; want %q", buf[:n], data)
|
||||
}
|
||||
|
||||
// Case where data overwrites end of buffer.
|
||||
const overdata = "hello world"
|
||||
n, err = r.Write([]byte(overdata))
|
||||
if err == nil || n > 0 {
|
||||
t.Fatal(err, n)
|
||||
}
|
||||
|
||||
// Set Random data in ring buffer and read it back.
|
||||
for i := 0; i < 32; i++ {
|
||||
n := rng.Intn(bufSize)
|
||||
copy(buf[:], overdata[:n])
|
||||
offset := rng.Intn(bufSize - 1)
|
||||
setRingData(t, r, offset, buf[:n])
|
||||
|
||||
// Case where data wraps around end of buffer.
|
||||
n, err = r.Read(buf[:])
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
if string(buf[:n]) != overdata[:n] {
|
||||
t.Error("got", buf[:n], "want", overdata[:n])
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
if nfirst+nsecond > bufSize {
|
||||
nfirst = bufSize - nsecond
|
||||
}
|
||||
offset := rng.Intn(bufSize - 1)
|
||||
|
||||
copy(buf[:], overdata[:nfirst])
|
||||
setRingData(t, r, offset, buf[:nfirst])
|
||||
// println("test", r.end, r.off, offset, r)
|
||||
ngot, err := r.WriteString(overdata[nfirst : nfirst+nsecond])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ngot != nsecond {
|
||||
t.Errorf("%d did not write data correctly: got %d; want %d", i, ngot, nsecond)
|
||||
}
|
||||
buf = [bufSize]byte{}
|
||||
// Case where data wraps around end of buffer.
|
||||
n, err = r.Read(buf[:])
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
|
||||
if n != nfirst+nsecond {
|
||||
t.Errorf("got %d; want %d (%d+%d)", n, nfirst+nsecond, nfirst, nsecond)
|
||||
}
|
||||
if string(buf[:n]) != overdata[:n] {
|
||||
t.Errorf("got %q; want %q", buf[:n], overdata[:n])
|
||||
}
|
||||
}
|
||||
|
||||
var readback [bufSize]byte
|
||||
var zeros [bufSize]byte
|
||||
|
||||
// Set random data and write some more and read it back with ReadAt and ReadPeek and ReadDiscard.
|
||||
for i := 0; i < 32; i++ {
|
||||
nfirst := rng.Intn(len(data))/2 + 1 // write garbage data first.
|
||||
nsecond := rng.Intn(len(data))/2 + 1
|
||||
if nfirst+nsecond > bufSize {
|
||||
nfirst = bufSize - nsecond
|
||||
}
|
||||
r.Reset()
|
||||
|
||||
randOff := rng.Intn(bufSize)
|
||||
content := append([]byte{}, zeros[:nfirst]...)
|
||||
content = append(content, data[:nsecond]...)
|
||||
setRingData(t, r, randOff, content)
|
||||
// Two-tap ReadPeek to make sure pointer not advanced.
|
||||
for i := 0; i < 2; i++ {
|
||||
n, err = r.ReadPeek(readback[:])
|
||||
if err != nil && err != io.EOF {
|
||||
t.Fatal("read failed", err)
|
||||
} else if n != nfirst+nsecond {
|
||||
t.Errorf("want!=got bytes read %d, %d", nfirst+nsecond, n)
|
||||
} else if !bytes.Equal(readback[:nfirst], zeros[:nfirst]) {
|
||||
t.Error("first section not match")
|
||||
} else if !bytes.Equal(readback[nfirst:nfirst+nsecond], []byte(data[:nsecond])) {
|
||||
t.Error("second section not match")
|
||||
}
|
||||
}
|
||||
|
||||
// Two-tap ReadAt to make sure pointer not advanced.
|
||||
for i := 0; i < 2; i++ {
|
||||
off := rng.Intn(nfirst + nsecond)
|
||||
n, err = r.ReadAt(readback[:nfirst+nsecond-off], int64(off))
|
||||
|
||||
readat := readback[:n]
|
||||
first := zeros[min(off, nfirst):nfirst]
|
||||
secondOff := max(0, off-nfirst)
|
||||
second := []byte(data[secondOff:nsecond])
|
||||
gotSecond := readat[len(first):]
|
||||
if err != nil && err != io.EOF {
|
||||
t.Fatal("read failed", off, err)
|
||||
} else if n != nfirst+nsecond-off {
|
||||
t.Errorf("want!=got bytes read %d, %d", nfirst+nsecond-off, n)
|
||||
} else if len(first) > 0 && !bytes.Equal(readat[:nfirst-off], first) {
|
||||
t.Error("first section not match")
|
||||
} else if len(second) > 0 && !bytes.Equal(gotSecond, second) {
|
||||
t.Errorf("second section not match got=%q want=%q", gotSecond, second)
|
||||
}
|
||||
}
|
||||
|
||||
// ReadDiscard test.
|
||||
|
||||
discard := rng.Intn(nfirst+nsecond) + 1
|
||||
r.ReadDiscard(discard)
|
||||
n, err := r.Read(readback[:])
|
||||
if err != nil && err != io.EOF {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wantN := nfirst + nsecond - discard
|
||||
if wantN != n {
|
||||
t.Errorf("Want %d bytes read, got %d", wantN, n)
|
||||
}
|
||||
if !bytes.Equal(readback[:n], content[discard:]) {
|
||||
t.Errorf("want data read %q, got %q", content[discard:], readback[:n])
|
||||
}
|
||||
}
|
||||
|
||||
_ = r._string(0)
|
||||
}
|
||||
|
||||
func TestRing2(t *testing.T) {
|
||||
const maxsize = 6
|
||||
const ntests = 800
|
||||
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) {
|
||||
t.Fatalf("failed test %d", i)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRing_findcrash(t *testing.T) {
|
||||
const maxsize = 33
|
||||
const ntests = 800000
|
||||
r := Ring{
|
||||
buf: make([]byte, maxsize*6),
|
||||
}
|
||||
rng := rand.New(rand.NewSource(0))
|
||||
data := make([]byte, maxsize)
|
||||
|
||||
for i := 0; i < ntests; i++ {
|
||||
free := r.Free()
|
||||
if free < 0 {
|
||||
t.Fatal("free < 0")
|
||||
}
|
||||
if rng.Intn(2) == 0 {
|
||||
l := max(rng.Intn(len(data)), 1)
|
||||
if l > free {
|
||||
continue // Buffer full.
|
||||
}
|
||||
n, err := r.Write(data[:l])
|
||||
expectFree := free - n
|
||||
free = r.Free()
|
||||
if n != l {
|
||||
t.Fatal(i, "write failed", n, l, err)
|
||||
} else if expectFree != free {
|
||||
t.Fatal(i, "free not updated correctly", expectFree, free)
|
||||
}
|
||||
}
|
||||
buffered := r.Buffered()
|
||||
if buffered < 0 {
|
||||
t.Fatal("buffered < 0")
|
||||
}
|
||||
if rng.Intn(2) == 0 {
|
||||
l := max(rng.Intn(len(data)), 1)
|
||||
n, err := r.Read(data[:l])
|
||||
expectRead := min(buffered, l)
|
||||
expectBuffered := buffered - n
|
||||
buffered = r.Buffered()
|
||||
if n != expectRead {
|
||||
t.Fatal(i, "read failed", n, l, expectRead, err)
|
||||
} else if buffered != expectBuffered {
|
||||
t.Fatal(i, "buffered not updated correctly", expectBuffered, buffered)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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")
|
||||
}
|
||||
dsize := len(data)
|
||||
var r Ring
|
||||
r.buf = ringbuf
|
||||
|
||||
nfirst := rng.Intn(dsize) / 2
|
||||
nsecond := rng.Intn(dsize) / 2
|
||||
if nfirst == 0 || nsecond == 0 {
|
||||
return true
|
||||
}
|
||||
offset := rng.Intn(dsize - 1)
|
||||
|
||||
setRingData(t, &r, offset, data[:nfirst])
|
||||
ngot, err := r.Write(data[nfirst : nfirst+nsecond])
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
return false
|
||||
}
|
||||
if ngot != nsecond {
|
||||
t.Errorf("did not write data correctly: got %d; want %d", ngot, nsecond)
|
||||
}
|
||||
// Case where data wraps around end of buffer.
|
||||
n, err := r.Read(auxbuf[:])
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
return false
|
||||
}
|
||||
|
||||
if n != nfirst+nsecond {
|
||||
t.Errorf("got %d; want %d (%d+%d)", n, nfirst+nsecond, nfirst, nsecond)
|
||||
}
|
||||
if !bytes.Equal(auxbuf[:n], data[:n]) {
|
||||
t.Errorf("got %q; want %q", auxbuf[:n], data[:n])
|
||||
}
|
||||
return !t.Failed()
|
||||
}
|
||||
|
||||
func fragmentReadInto(r io.Reader, buf []byte) (n int, _ error) {
|
||||
maxSize := len(buf) / 4
|
||||
for {
|
||||
ntop := min(n+rand.Intn(maxSize)+1, len(buf))
|
||||
ngot, err := r.Read(buf[n:ntop])
|
||||
n += ngot
|
||||
if err != nil {
|
||||
if err == io.EOF {
|
||||
return n, nil
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
if n == len(buf) {
|
||||
return n, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func setRingData(t *testing.T, r *Ring, offset int, data []byte) {
|
||||
t.Helper()
|
||||
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) {
|
||||
// End of buffer not enough to hold data, wrap around.
|
||||
n = copy(r.buf, data[n:])
|
||||
r.end = n
|
||||
}
|
||||
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)
|
||||
if free != wantFree {
|
||||
t.Fatalf("free got %d; want %d", free, wantFree)
|
||||
}
|
||||
buffered := r.Buffered()
|
||||
wantBuffered := len(data)
|
||||
if buffered != wantBuffered {
|
||||
t.Fatalf("buffered got %d; want %d", buffered, wantBuffered)
|
||||
}
|
||||
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
|
||||
}
|
||||
Reference in New Issue
Block a user