diff --git a/notify_test.go b/notify_test.go new file mode 100644 index 0000000..41339a9 --- /dev/null +++ b/notify_test.go @@ -0,0 +1,203 @@ +package ringbuffer + +import ( + "io" + "testing" + "time" +) + +// waitNotify reports whether a signal arrives within d. +func waitNotify(rb *RingBuffer, d time.Duration) bool { + select { + case <-rb.Notify(): + return true + case <-time.After(d): + return false + } +} + +// TestNotifyOnWrite is the basic contract: a consumer parked on Notify wakes +// when data lands. +func TestNotifyOnWrite(t *testing.T) { + rb := New(64) + + if waitNotify(rb, 50*time.Millisecond) { + t.Fatal("signalled before anything was written") + } + + go func() { + time.Sleep(20 * time.Millisecond) + rb.Write([]byte("hello")) + }() + + if !waitNotify(rb, 2*time.Second) { + t.Fatal("no signal after a write") + } +} + +// TestNotifyWorksInNonBlockingMode — the existing wakeups are all guarded by +// r.block, so signalling only alongside them would leave Notify silent for a +// non-blocking buffer. It is signalled from the write path instead. +func TestNotifyWorksInNonBlockingMode(t *testing.T) { + rb := New(64) // default: non-blocking + rb.Write([]byte("x")) + if !waitNotify(rb, time.Second) { + t.Fatal("no signal in non-blocking mode") + } +} + +// TestNotifyCoalesces — capacity 1, so a burst of writes leaves exactly one +// pending signal. The contract is "something changed", not a count, and a woken +// consumer re-inspects the buffer. +func TestNotifyCoalesces(t *testing.T) { + rb := New(64) + for i := 0; i < 10; i++ { + rb.Write([]byte("x")) + } + + if !waitNotify(rb, time.Second) { + t.Fatal("no signal after writes") + } + if waitNotify(rb, 50*time.Millisecond) { + t.Error("a second signal was queued; signals must coalesce") + } +} + +// TestNotifyOnClose — a consumer waiting for data must also be woken when there +// will never be any, or it parks forever on a closed buffer. +func TestNotifyOnClose(t *testing.T) { + rb := New(64).SetBlocking(true) + drain(t, rb) + + go func() { + time.Sleep(20 * time.Millisecond) + rb.CloseWriter() + }() + + if !waitNotify(rb, 2*time.Second) { + t.Fatal("no signal on CloseWriter; a waiting consumer would hang") + } +} + +// TestNotifyOnCloseWithError — same, for the failure path. +func TestNotifyOnCloseWithError(t *testing.T) { + rb := New(64).SetBlocking(true) + drain(t, rb) + + go func() { + time.Sleep(20 * time.Millisecond) + rb.CloseWithError(io.ErrUnexpectedEOF) + }() + + if !waitNotify(rb, 2*time.Second) { + t.Fatal("no signal on CloseWithError; a waiting consumer would hang") + } +} + +// TestNotifyDoesNotConsume — Notify must not disturb the buffer. A consumer +// woken by it still has to Peek to see the data, and the data must be intact. +func TestNotifyDoesNotConsume(t *testing.T) { + rb := New(64) + rb.Write([]byte("payload")) + <-rb.Notify() + + if got := rb.Length(); got != 7 { + t.Errorf("Length after Notify = %d, want 7", got) + } + p := make([]byte, 7) + n, err := rb.Peek(p) + if err != nil || string(p[:n]) != "payload" { + t.Errorf("Peek after Notify = %q err=%v, want %q", p[:n], err, "payload") + } +} + +// TestPeekOnlyNeverSeesEOF documents a trap for consumers that only peek. +// +// readErr reports io.EOF only once the buffer is empty, so data always drains +// before end-of-stream is announced. A consumer that never consumes therefore +// never learns the writer is finished, and will wait forever. +// +// This is correct behaviour, not a defect — but a peek-driven consumer has to +// consume something eventually, and must not treat "no EOF yet" as "the writer +// is still alive". +func TestPeekOnlyNeverSeesEOF(t *testing.T) { + rb := New(64).SetBlocking(true) + rb.Write([]byte("data")) + rb.CloseWriter() + + scratch := make([]byte, 64) + for i := 0; i < 3; i++ { + n, err := rb.Peek(scratch) + if err == io.EOF { + t.Fatal("Peek reported EOF while data was still buffered") + } + if n != 4 { + t.Fatalf("Peek returned %d bytes, want 4", n) + } + } + + // Consuming is what surfaces it. + rb.Read(make([]byte, 4)) + if _, err := rb.Peek(scratch); err != io.EOF { + t.Errorf("after draining, Peek err = %v, want io.EOF", err) + } +} + +// TestNotifyDrivesPeekConsumeLoop is the pattern the API exists for: wait on a +// signal, look without consuming, send, and consume only what has been +// confirmed. Consuming on confirmation is also what eventually surfaces EOF. +func TestNotifyDrivesPeekConsumeLoop(t *testing.T) { + rb := New(1024).SetBlocking(true) + + go func() { + for _, s := range []string{"alpha", "beta", "gamma"} { + rb.Write([]byte(s)) + time.Sleep(10 * time.Millisecond) + } + rb.CloseWriter() + }() + + var seen string + sent := 0 + scratch := make([]byte, 1024) + discard := make([]byte, 1024) + + deadline := time.After(5 * time.Second) + for { + n, err := rb.Peek(scratch) + if n > sent { + seen += string(scratch[sent:n]) // only what has not gone out yet + sent = n + } + if err == io.EOF { + break + } + + // The client confirms everything sent, so consume it. Until this + // happens the buffer never empties and EOF never arrives. + if sent > 0 { + rb.Read(discard[:sent]) + sent = 0 + continue + } + + select { + case <-rb.Notify(): + case <-deadline: + t.Fatal("loop stalled waiting for a signal") + } + } + + if seen != "alphabetagamma" { + t.Errorf("got %q, want %q", seen, "alphabetagamma") + } + if got := rb.Length(); got != 0 { + t.Errorf("Length = %d, want 0 after everything was confirmed", got) + } +} + +// drain keeps a blocking buffer from wedging a test when nothing reads it. +func drain(t *testing.T, rb *RingBuffer) { + t.Helper() + t.Cleanup(func() { rb.CloseWithError(io.ErrClosedPipe) }) +} diff --git a/ring_buffer.go b/ring_buffer.go index 4048d0f..7b5eeeb 100644 --- a/ring_buffer.go +++ b/ring_buffer.go @@ -58,24 +58,27 @@ type RingBuffer struct { 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. - generation int64 // Incremented on Reset() to invalidate current waiters + readCond *sync.Cond // Signaled when data has been read. + writeCond *sync.Cond // Signaled when data has been written. + notify chan struct{} // Signaled when data is written or state changes; see Notify. + generation int64 // Incremented on Reset() to invalidate current waiters } // New returns a new RingBuffer whose buffer has the given size. func New(size int) *RingBuffer { return &RingBuffer{ - buf: make([]byte, size), - size: size, + buf: make([]byte, size), + size: size, + notify: make(chan struct{}, 1), } } // NewBuffer returns a new RingBuffer whose buffer is provided. func NewBuffer(b []byte) *RingBuffer { return &RingBuffer{ - buf: b, - size: len(b), + buf: b, + size: len(b), + notify: make(chan struct{}, 1), } } @@ -180,10 +183,15 @@ func (r *RingBuffer) setErr(err error, locked bool) error { return err default: r.err = err + // The conds only exist in blocking mode (see SetBlocking), so this guard + // is nil-safety rather than policy. notify is allocated by both + // constructors and must fire in either mode: a consumer selecting on + // Notify still has to learn that the buffer closed. if r.block { r.readCond.Broadcast() r.writeCond.Broadcast() } + r.signalNotify() } return err } @@ -564,6 +572,7 @@ func (r *RingBuffer) ReadFrom(rd io.Reader) (n int64, err error) { r.isFull = r.r == r.w && nr > 0 n += int64(nr) r.writeCond.Broadcast() + r.signalNotify() eof := rerr == io.EOF r.mu.Unlock() if eof { @@ -755,6 +764,10 @@ func (r *RingBuffer) write(p []byte) (n int, err error) { r.isFull = true } + if n > 0 { + r.signalNotify() + } + return n, err } @@ -837,6 +850,8 @@ func (r *RingBuffer) writeByte(c byte) error { r.isFull = true } + r.signalNotify() + return nil } @@ -1013,6 +1028,7 @@ func (r *RingBuffer) Reset() { r.generation++ r.readCond.Broadcast() r.writeCond.Broadcast() + r.signalNotify() r.r = 0 r.w = 0 r.err = nil @@ -1069,6 +1085,49 @@ func (rc *readCloser) Close() error { return err } +// Notify returns a channel that is signaled whenever data is written to the +// buffer, or when the buffer is closed or reset. +// +// It exists for consumers that cannot simply block in Read. A consumer that must +// wait on several things at once — new data, plus events from elsewhere — needs +// every source of wakeup to be selectable, and a blocking read is not. Peek is +// also non-blocking by design, so a consumer that wants to look at data without +// consuming it has nothing to wait on. This channel is that wait. +// +// The channel has capacity 1 and signals coalesce: it reports that something +// changed, never how much or how often. A woken consumer must re-inspect the +// buffer rather than assume anything about its state. A signal may also arrive +// spuriously, so treat it as a hint to look, not as a promise of data. +// +// Signals are delivered regardless of blocking mode. +func (r *RingBuffer) Notify() <-chan struct{} { + return r.notify +} + +// signalNotify performs a non-blocking send on the notify channel. Safe to call +// with r.mu held: the send never blocks, since a pending signal already conveys +// everything a later one would. +// +// Placement differs from writeCond.Broadcast on purpose. Broadcasts live in the +// exported entry points; notify is signalled from the internal write() and +// writeByte() funnels instead, for two reasons: +// +// - Coverage. writeByte has four callers and only two of them broadcast; the +// rest are covered transitively, which takes tracing to confirm. Signalling +// at the funnel means a new caller cannot be missed. +// - Blocking mode. Most broadcasts sit inside `if r.block` guards, because the +// conds are nil otherwise. Mirroring that placement would silence Notify for +// a non-blocking buffer, which consumers are entitled to use. +// +// Extra signals cost nothing: the channel has capacity 1 and coalesces, and the +// documented contract is that a wakeup means "look", not "there is data". +func (r *RingBuffer) signalNotify() { + select { + case r.notify <- struct{}{}: + default: + } +} + // Peek reads up to len(p) bytes into p without moving the read pointer. func (r *RingBuffer) Peek(p []byte) (n int, err error) { if len(p) == 0 {