From 0d90239f76a200dd1ce9ea5a435252adb51b5540 Mon Sep 17 00:00:00 2001 From: smallnest Date: Tue, 8 Sep 2026 16:09:12 +0800 Subject: [PATCH] Fix #28: ReadFrom no longer holds the lock while blocked in rd.Read ReadFrom took r.mu once and held it via defer for the whole loop, calling rd.Read while the mutex was still held. Whenever the source was idle (e.g. an idle TCP connection), rd.Read blocked with the lock held and froze the entire buffer: Read, Peek, Length, Free and IsEmpty all blocked until bytes arrived. This is reachable without naming the method, since io.Copy(rb, rd) dispatches to ReadFrom via io.ReaderFrom. Fix, preserving the zero-copy contract (ReadFrom reads directly into r.buf): - Add writeMu to serialize the write side. ReadFrom holds it for its whole duration and reserves the contiguous free window beyond the write pointer, releasing r.mu during rd.Read. Because the reserved window is disjoint from the committed [r, w) region, readers and inspectors stay responsive and correct while a read is in flight. - Guard Write/WriteByte/TryWrite/TryWriteByte with writeMu (lock order is always writeMu -> r.mu) so no concurrent writer scribbles into the reserved window or moves r.w out from under the snapshot. - Snapshot the generation before releasing r.mu; if Reset() lands while blocked in rd.Read, the stale bytes are dropped and the loop recomputes against the fresh state. Add readfrom_idle_test.go: a regression test proving inspectors return promptly while ReadFrom is parked in an idle read (deadlocks before this fix), plus concurrent slow-source and Reset stress tests under -race. Also bump go.mod to 1.26 and modernize (WaitGroup.Go, testing.B.Loop, any). --- go.mod | 2 +- readfrom_idle_test.go | 228 ++++++++++++++++++++++++++++++++++ ring_buffer.go | 73 +++++++++-- ring_buffer_benchmark_test.go | 16 +-- ring_buffer_test.go | 6 +- 5 files changed, 300 insertions(+), 25 deletions(-) create mode 100644 readfrom_idle_test.go 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...) }