add Notify(): a selectable channel for consumers that can't block in Read

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) })
+}

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 {