diff --git a/internal/ring.go b/internal/ring.go new file mode 100644 index 0000000..12c0f8c --- /dev/null +++ b/internal/ring.go @@ -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:] +} diff --git a/internal/ring_test.go b/internal/ring_test.go new file mode 100644 index 0000000..e468a18 --- /dev/null +++ b/internal/ring_test.go @@ -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 +}