diff --git a/go.mod b/go.mod index d9ad533..a89d6b7 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,3 @@ module github.com/smallnest/ringbuffer -go 1.19 +go 1.26 diff --git a/readfrom_idle_test.go b/readfrom_idle_test.go new file mode 100644 index 0000000..9be2d94 --- /dev/null +++ b/readfrom_idle_test.go @@ -0,0 +1,228 @@ +package ringbuffer + +import ( + "bytes" + "io" + "sync" + "testing" + "time" +) + +// idleReader delivers a fixed set of chunks, then blocks in Read until gate is +// closed — modelling a network source that has gone quiet with nothing to send. +type idleReader struct { + chunks [][]byte + i int + gate chan struct{} +} + +func (rd *idleReader) Read(p []byte) (int, error) { + if rd.i < len(rd.chunks) { + n := copy(p, rd.chunks[rd.i]) + rd.i++ + return n, nil + } + <-rd.gate // idle: block here without any lock held by ReadFrom + return 0, io.EOF +} + +// mustReturn runs fn in a goroutine and fails if it does not complete within d. +// Before the fix, an inspection call made while ReadFrom is blocked in rd.Read +// never returns because ReadFrom holds r.mu across the blocked Read. +func mustReturn(t *testing.T, d time.Duration, what string, fn func()) { + t.Helper() + done := make(chan struct{}) + go func() { fn(); close(done) }() + select { + case <-done: + case <-time.After(d): + t.Fatalf("%s blocked for more than %v while ReadFrom was idle in rd.Read", what, d) + } +} + +// TestReadFromIdleSourceDoesNotBlockInspectors is the regression test for #28: +// while ReadFrom is parked in a blocked rd.Read, the read/inspection side of the +// buffer must stay responsive. +func TestReadFromIdleSourceDoesNotBlockInspectors(t *testing.T) { + rb := New(64).SetBlocking(true) + rd := &idleReader{chunks: [][]byte{[]byte("hello")}, gate: make(chan struct{})} + + rfDone := make(chan struct{}) + go func() { rb.ReadFrom(rd); close(rfDone) }() + + // Wait until the first chunk has been committed. After committing it, + // ReadFrom loops, calls rd.Read again, and blocks on the gate — now with + // r.mu released. + deadline := time.After(2 * time.Second) + for rb.Length() != 5 { + select { + case <-deadline: + t.Fatalf("data never landed; Length = %d", rb.Length()) + default: + time.Sleep(time.Millisecond) + } + } + + mustReturn(t, 2*time.Second, "Length", func() { rb.Length() }) + mustReturn(t, 2*time.Second, "Free", func() { rb.Free() }) + mustReturn(t, 2*time.Second, "IsEmpty", func() { rb.IsEmpty() }) + mustReturn(t, 2*time.Second, "Peek", func() { + p := make([]byte, 5) + if n, _ := rb.Peek(p); string(p[:n]) != "hello" { + t.Errorf("Peek = %q, want %q", p[:n], "hello") + } + }) + mustReturn(t, 2*time.Second, "Read", func() { + p := make([]byte, 5) + if n, _ := rb.Read(p); string(p[:n]) != "hello" { + t.Errorf("Read = %q, want %q", p[:n], "hello") + } + }) + + close(rd.gate) // let the source finish + select { + case <-rfDone: + case <-time.After(2 * time.Second): + t.Fatal("ReadFrom did not return after the source closed") + } +} + +// slowReader hands back one byte at a time with a small pause, so ReadFrom +// spends most of its life blocked in Read with the lock released. +type slowReader struct { + data []byte + i int +} + +func (rd *slowReader) Read(p []byte) (int, error) { + if rd.i >= len(rd.data) { + return 0, io.EOF + } + time.Sleep(200 * time.Microsecond) + p[0] = rd.data[rd.i] + rd.i++ + return 1, nil +} + +// TestReadFromConcurrentWithInspectorsAndDrain exercises ReadFrom against a slow +// source while other goroutines inspect and drain, to catch races in the +// reserve/commit handoff under -race. +func TestReadFromConcurrentWithInspectorsAndDrain(t *testing.T) { + const payload = "the quick brown fox jumps over the lazy dog, twice over" + rb := New(8).SetBlocking(true) // small buffer forces many wrap/full cycles + + var wg sync.WaitGroup + + // Writer. + wg.Go(func() { + rb.ReadFrom(&slowReader{data: []byte(payload)}) + rb.CloseWriter() + }) + + // Inspectors: must never observe corruption and must always return. + stop := make(chan struct{}) + wg.Go(func() { + p := make([]byte, 8) + for { + select { + case <-stop: + return + default: + } + _ = rb.Length() + _ = rb.Free() + _, _ = rb.Peek(p) + } + }) + + // Drainer. + var got bytes.Buffer + wg.Go(func() { + defer close(stop) + p := make([]byte, 3) + for { + n, err := rb.Read(p) + got.Write(p[:n]) + if err == io.EOF { + return + } + if err != nil { + t.Errorf("Read error: %v", err) + return + } + } + }) + + wg.Wait() + if got.String() != payload { + t.Errorf("drained %q, want %q", got.String(), payload) + } +} + +// TestReadFromConcurrentWithReset hammers Reset while ReadFrom is running, to +// exercise the generation snapshot/compare path (a Reset landing while we are +// blocked in rd.Read must be detected and the read discarded, not committed +// against a stale write pointer). Safety/termination only — content after a +// mid-copy Reset is intentionally undefined. +func TestReadFromConcurrentWithReset(t *testing.T) { + rb := New(8).SetBlocking(true) + + stop := make(chan struct{}) + var rf sync.WaitGroup // the ReadFrom writer + var churn sync.WaitGroup + + rf.Go(func() { + // A reader that keeps producing until told to stop, then EOFs. + rb.ReadFrom(readerFunc(func(p []byte) (int, error) { + select { + case <-stop: + return 0, io.EOF + default: + time.Sleep(100 * time.Microsecond) + p[0] = 'x' + return 1, nil + } + })) + }) + + // Reset and drain concurrently for a bounded number of rounds. + churn.Go(func() { + p := make([]byte, 4) + for range 500 { + rb.Reset() + _, _ = rb.Peek(p) + if l := rb.Length(); l < 0 || l > 8 { + t.Errorf("Length out of range: %d", l) + return + } + time.Sleep(50 * time.Microsecond) + } + }) + + churn.Go(func() { + p := make([]byte, 4) + for range 500 { + _, _ = rb.Read(p) + time.Sleep(50 * time.Microsecond) + } + }) + + // Shut down only after the churn goroutines are done, so no late Reset can + // clear the EOF and strand the winddown. + churn.Wait() + close(stop) + rb.CloseWriter() // unblock a ReadFrom parked in waitRead on a full buffer + + done := make(chan struct{}) + go func() { rf.Wait(); close(done) }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("ReadFrom did not settle; possible deadlock in the reserve/commit path") + } +} + +// readerFunc adapts a function to io.Reader. +type readerFunc func(p []byte) (int, error) + +func (f readerFunc) Read(p []byte) (int, error) { return f(p) } diff --git a/ring_buffer.go b/ring_buffer.go index ce9af00..296ca03 100644 --- a/ring_buffer.go +++ b/ring_buffer.go @@ -56,6 +56,7 @@ type RingBuffer struct { wTimeout time.Duration // Applies to read (wait for the write condition) noClOnTo bool // do not set buffer in error on a timeout mu sync.Mutex + writeMu sync.Mutex // serializes the write side so ReadFrom can release mu while blocked in rd.Read wg sync.WaitGroup readCond *sync.Cond // Signaled when data has been read. writeCond *sync.Cond // Signaled when data has been written. @@ -401,6 +402,8 @@ func (r *RingBuffer) Write(p []byte) (n int, err error) { if len(p) == 0 { return 0, r.setErr(nil, false) } + r.writeMu.Lock() + defer r.writeMu.Unlock() r.mu.Lock() defer r.mu.Unlock() if err := r.checkWriteErr(); err != nil { @@ -475,42 +478,75 @@ func (r *RingBuffer) waitWrite() (ok bool) { // Any error except EOF encountered during the read is also returned, // and the error will cause the Read side to fail as well. // ReadFrom only available in blocking mode. +// +// While a ReadFrom is in progress the read pointer is released between reads, so +// Read, Peek, Length, Free and IsEmpty stay responsive even when the source is +// idle and blocked in Read. Concurrent Write/WriteByte calls are serialized +// against ReadFrom and wait while a single read into the buffer is in flight. func (r *RingBuffer) ReadFrom(rd io.Reader) (n int64, err error) { if !r.block { return 0, errors.New("RingBuffer: ReadFrom only available in blocking mode") } + // ReadFrom is the write side for its whole duration. Serialize against other + // writers so we can release r.mu while blocked in rd.Read without a + // concurrent Write scribbling into the window we reserved or moving r.w out + // from under our snapshot. The read/inspection side never takes writeMu, so + // it is unaffected. + r.writeMu.Lock() + defer r.writeMu.Unlock() + zeroReads := 0 - r.mu.Lock() - defer r.mu.Unlock() for { + r.mu.Lock() if err = r.readErr(true); err != nil { + r.mu.Unlock() return n, err } if r.isFull { - // Wait for a read + // Wait for a read. if !r.waitRead() { + r.mu.Unlock() return 0, context.DeadlineExceeded } + r.mu.Unlock() continue } - // Calculate available space to read into + // Reserve the contiguous free window and snapshot what we will commit + // against. The window is free space beyond the write pointer, disjoint + // from the committed [r, w) region that readers and Peek/Length touch. var toRead []byte if r.w >= r.r { - // After reader, read until end of buffer + // After reader, read until end of buffer. toRead = r.buf[r.w:] } else { // Before reader, read until reader. if r.w >= r.size || r.r > r.size { - // Pointers are corrupted, return error to prevent panic + // Pointers are corrupted, return error to prevent panic. + r.mu.Unlock() return n, errors.New("RingBuffer: internal state corrupted") } toRead = r.buf[r.w:r.r] } + startW := r.w + gen := r.generation + r.mu.Unlock() + // Read with no lock held: this is where an idle source blocks, and where + // every other method must stay responsive. nr, rerr := rd.Read(toRead) + + r.mu.Lock() + if r.generation != gen { + // Reset() happened while we were blocked in rd.Read. The snapshot is + // stale and any bytes read belong to a buffer that no longer exists; + // drop them and recompute against the fresh state. + r.mu.Unlock() + continue + } if rerr != nil && rerr != io.EOF { err = r.setErr(rerr, true) + r.mu.Unlock() break } if nr == 0 && rerr == nil { @@ -518,16 +554,19 @@ func (r *RingBuffer) ReadFrom(rd io.Reader) (n int64, err error) { if zeroReads >= 100 { err = r.setErr(io.ErrNoProgress, true) } + r.mu.Unlock() continue } zeroReads = 0 - // Update write pointer with proper wrap-around using modulo - r.w = (r.w + nr) % r.size + // Commit. writeMu keeps us the only writer, so startW == r.w still holds. + r.w = (startW + nr) % r.size r.isFull = r.r == r.w && nr > 0 n += int64(nr) r.writeCond.Broadcast() - if rerr == io.EOF { + eof := rerr == io.EOF + r.mu.Unlock() + if eof { // We do not close. break } @@ -615,12 +654,10 @@ func (r *RingBuffer) WriteTo(w io.Writer) (n int64, err error) { func (r *RingBuffer) Copy(dst io.Writer, src io.Reader) (written int64, err error) { r.SetBlocking(true) var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, _ = r.ReadFrom(src) //nolint errcheck r.CloseWriter() - }() + }) defer wg.Wait() return r.WriteTo(dst) } @@ -631,6 +668,10 @@ func (r *RingBuffer) TryWrite(p []byte) (n int, err error) { if len(p) == 0 { return 0, r.setErr(nil, false) } + if !r.writeMu.TryLock() { + return 0, ErrAcquireLock + } + defer r.writeMu.Unlock() ok := r.mu.TryLock() if !ok { return 0, ErrAcquireLock @@ -715,6 +756,8 @@ func (r *RingBuffer) write(p []byte) (n int, err error) { // WriteByte writes one byte into buffer, and returns ErrIsFull if the buffer is full. func (r *RingBuffer) WriteByte(c byte) error { + r.writeMu.Lock() + defer r.writeMu.Unlock() r.mu.Lock() defer r.mu.Unlock() if err := r.checkWriteErr(); err != nil { @@ -745,6 +788,10 @@ func (r *RingBuffer) WriteByte(c byte) error { // TryWriteByte writes one byte into buffer without blocking. // If it does not succeed to acquire the lock, it returns ErrAcquireLock. func (r *RingBuffer) TryWriteByte(c byte) error { + if !r.writeMu.TryLock() { + return ErrAcquireLock + } + defer r.writeMu.Unlock() ok := r.mu.TryLock() if !ok { return ErrAcquireLock diff --git a/ring_buffer_benchmark_test.go b/ring_buffer_benchmark_test.go index c03482a..884940a 100644 --- a/ring_buffer_benchmark_test.go +++ b/ring_buffer_benchmark_test.go @@ -13,7 +13,7 @@ func BenchmarkRingBuffer_Sync(b *testing.B) { buf := make([]byte, 512) b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { _, _ = rb.Write(data) //nolint errcheck _, _ = rb.Read(buf) //nolint errcheck } @@ -32,7 +32,7 @@ func BenchmarkRingBuffer_AsyncRead(b *testing.B) { }() b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { _, _ = rb.Write(data) //nolint errcheck } } @@ -52,7 +52,7 @@ func BenchmarkRingBuffer_AsyncReadBlocking(b *testing.B) { }() b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { _, _ = rb.Write(data) //nolint errcheck } } @@ -69,7 +69,7 @@ func BenchmarkRingBuffer_AsyncWrite(b *testing.B) { }() b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { _, _ = rb.Read(buf) //nolint errcheck } } @@ -89,7 +89,7 @@ func BenchmarkRingBuffer_AsyncWriteBlocking(b *testing.B) { }() b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { _, _ = rb.Read(buf) //nolint errcheck } } @@ -122,7 +122,7 @@ func BenchmarkRingBuffer_ReadFrom(b *testing.B) { b.ResetTimer() b.SetBytes(sz) - for i := 0; i < b.N; i++ { + for b.Loop() { _, _ = io.ReadFull(rb, buf) //nolint errcheck } rb.CloseWithError(context.Canceled) //nolint errcheck @@ -141,7 +141,7 @@ func BenchmarkRingBuffer_WriteTo(b *testing.B) { b.ResetTimer() b.SetBytes(sz) - for i := 0; i < b.N; i++ { + for b.Loop() { _, err := rb.Write(data) if err != nil { b.Fatal(err) @@ -163,7 +163,7 @@ func BenchmarkIoPipeReader(b *testing.B) { b.ResetTimer() b.SetBytes(int64(len(data))) - for i := 0; i < b.N; i++ { + for b.Loop() { _, _ = pr.Read(buf) //nolint errcheck } } diff --git a/ring_buffer_test.go b/ring_buffer_test.go index 15fdc1a..6233726 100644 --- a/ring_buffer_test.go +++ b/ring_buffer_test.go @@ -430,7 +430,7 @@ func TestRingBuffer_Blocking(t *testing.T) { read = io.MultiWriter(read, &readBuf) wrote = io.MultiWriter(wrote, &wroteBuf) } - debugln := func(args ...interface{}) { + debugln := func(args ...any) { if debug { fmt.Println(args...) } @@ -593,7 +593,7 @@ func TestRingBuffer_BlockingBig(t *testing.T) { read = io.MultiWriter(read, &readBuf) wrote = io.MultiWriter(wrote, &wroteBuf) } - debugln := func(args ...interface{}) { + debugln := func(args ...any) { if debug { fmt.Println(args...) } @@ -755,7 +755,7 @@ func TestRingBuffer_ReadFromBig(t *testing.T) { wrote := io.Writer(wroteHash) read = io.MultiWriter(read, &readBuf) wrote = io.MultiWriter(wrote, &wroteBuf) - debugln := func(args ...interface{}) { + debugln := func(args ...any) { if debug { fmt.Println(args...) }