diff --git a/go.mod b/go.mod index d9ad533..aa34580 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,3 @@ -module github.com/smallnest/ringbuffer +module github.com/Arrayscape/ringbuffer go 1.19 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/peekat_test.go b/peekat_test.go new file mode 100644 index 0000000..571fbbc --- /dev/null +++ b/peekat_test.go @@ -0,0 +1,146 @@ +package ringbuffer + +import ( + "bytes" + "testing" +) + +// PeekAt exists for a consumer that runs ahead of what it can release: bytes +// stay buffered until something confirms them, so the read pointer sits where +// confirmation reached and the consumer reads from somewhere past it. + +func TestPeekAtReadsFromTheOffset(t *testing.T) { + rb := New(64) + rb.Write([]byte("0123456789")) + + p := make([]byte, 4) + for _, tc := range []struct { + off int + want string + }{ + {0, "0123"}, + {3, "3456"}, + {6, "6789"}, + } { + n, err := rb.PeekAt(tc.off, p) + if err != nil { + t.Fatalf("PeekAt(%d): %v", tc.off, err) + } + if got := string(p[:n]); got != tc.want { + t.Errorf("PeekAt(%d) = %q, want %q", tc.off, got, tc.want) + } + } + + if rb.Length() != 10 { + t.Errorf("Length = %d, want 10 — PeekAt must not consume", rb.Length()) + } +} + +func TestPeekAtShortAtTheEnd(t *testing.T) { + rb := New(64) + rb.Write([]byte("0123456789")) + + p := make([]byte, 8) + n, err := rb.PeekAt(7, p) + if err != nil { + t.Fatalf("PeekAt: %v", err) + } + if got := string(p[:n]); got != "789" { + t.Errorf("got %q, want %q", got, "789") + } +} + +// An offset at or past the end is the ordinary "nothing new yet" case for a +// consumer that has taken everything, and must not look like an empty buffer. +func TestPeekAtPastTheEndIsNotAnError(t *testing.T) { + rb := New(64) + rb.Write([]byte("0123456789")) + + p := make([]byte, 8) + for _, off := range []int{10, 11, 1000} { + n, err := rb.PeekAt(off, p) + if n != 0 || err != nil { + t.Errorf("PeekAt(%d) = %d, %v; want 0, nil", off, n, err) + } + } + if rb.Length() != 10 { + t.Errorf("Length = %d, want 10", rb.Length()) + } +} + +// The offset has to be applied in ring coordinates, not slice coordinates: the +// window it names can begin before the wrap and end after it. +func TestPeekAtAcrossTheWrap(t *testing.T) { + rb := New(10) + rb.Write([]byte("0123456789")) // full + rb.Read(make([]byte, 6)) // read pointer now at 6 + rb.Write([]byte("abcdef")) // wraps: buffered is "6789abcdef" + + if rb.Length() != 10 { + t.Fatalf("Length = %d, want 10", rb.Length()) + } + + p := make([]byte, 10) + n, err := rb.PeekAt(0, p) + if err != nil || string(p[:n]) != "6789abcdef" { + t.Fatalf("PeekAt(0) = %q, %v", p[:n], err) + } + + // Starting before the wrap and running past it. + n, _ = rb.PeekAt(2, p) + if got := string(p[:n]); got != "89abcdef" { + t.Errorf("PeekAt(2) = %q, want %q", got, "89abcdef") + } + // Starting after the wrap. + n, _ = rb.PeekAt(6, p) + if got := string(p[:n]); got != "cdef" { + t.Errorf("PeekAt(6) = %q, want %q", got, "cdef") + } +} + +func TestPeekAtOnAFullBuffer(t *testing.T) { + rb := New(8) + rb.Write([]byte("abcdefgh")) // exactly full: w == r and isFull + + p := make([]byte, 8) + n, _ := rb.PeekAt(0, p) + if got := string(p[:n]); got != "abcdefgh" { + t.Errorf("PeekAt(0) on a full buffer = %q", got) + } + n, _ = rb.PeekAt(5, p) + if got := string(p[:n]); got != "fgh" { + t.Errorf("PeekAt(5) on a full buffer = %q, want %q", got, "fgh") + } +} + +func TestPeekAtEmptyAndDegenerate(t *testing.T) { + rb := New(8) + + p := make([]byte, 4) + if n, err := rb.PeekAt(0, p); n != 0 || err != ErrIsEmpty { + t.Errorf("PeekAt on empty = %d, %v; want 0, ErrIsEmpty", n, err) + } + + rb.Write([]byte("abcd")) + if n, _ := rb.PeekAt(0, nil); n != 0 { + t.Errorf("PeekAt with no destination returned %d", n) + } + if n, _ := rb.PeekAt(-1, p); n != 0 { + t.Errorf("PeekAt with a negative offset returned %d", n) + } +} + +// PeekAt must agree with Peek where they overlap, or a consumer switching from +// one to the other would silently shift the stream. +func TestPeekAtAgreesWithPeek(t *testing.T) { + rb := New(32) + rb.Write([]byte("the quick brown fox jumps")) + + a := make([]byte, 32) + b := make([]byte, 32) + na, _ := rb.Peek(a) + nb, _ := rb.PeekAt(0, b) + if na != nb || !bytes.Equal(a[:na], b[:nb]) { + t.Errorf("Peek = %q, PeekAt(0) = %q", a[:na], b[:nb]) + } +} diff --git a/ring_buffer.go b/ring_buffer.go index ce9af00..b805da7 100644 --- a/ring_buffer.go +++ b/ring_buffer.go @@ -57,24 +57,27 @@ type RingBuffer struct { noClOnTo bool // do not set buffer in error on a timeout mu sync.Mutex 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), } } @@ -179,10 +182,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 } @@ -527,6 +535,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() if rerr == io.EOF { // We do not close. break @@ -710,6 +719,10 @@ func (r *RingBuffer) write(p []byte) (n int, err error) { r.isFull = true } + if n > 0 { + r.signalNotify() + } + return n, err } @@ -786,6 +799,8 @@ func (r *RingBuffer) writeByte(c byte) error { r.isFull = true } + r.signalNotify() + return nil } @@ -793,19 +808,7 @@ func (r *RingBuffer) writeByte(c byte) error { func (r *RingBuffer) Length() int { r.mu.Lock() defer r.mu.Unlock() - - if r.w == r.r { - if r.isFull { - return r.size - } - return 0 - } - - if r.w > r.r { - return r.w - r.r - } - - return r.size - r.r + r.w + return r.length() } // Capacity returns the size of the underlying buffer. @@ -962,6 +965,7 @@ func (r *RingBuffer) Reset() { r.generation++ r.readCond.Broadcast() r.writeCond.Broadcast() + r.signalNotify() r.r = 0 r.w = 0 r.err = nil @@ -1018,6 +1022,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 { @@ -1040,3 +1087,73 @@ func (r *RingBuffer) peek(p []byte) (n int, err error) { } return n, r.readErr(true) } + +// PeekAt reads up to len(p) bytes into p, starting off bytes past the read +// pointer, without moving it. +// +// The offset is for a consumer that has to run ahead of what it can release. +// Where bytes stay buffered until something confirms them, the read pointer is +// wherever confirmation has reached — not where reading should resume — and the +// two can be far apart. Peek always starts at the read pointer, so such a +// consumer would otherwise have to ask for everything from there and discard +// the part it already has, copying the skipped prefix on every call. +// +// An offset at or past the end of the buffered data returns 0 and no error: it +// means the consumer has taken everything there is, which is an ordinary thing +// for it to ask. That is distinct from an empty buffer, which returns +// ErrIsEmpty exactly as Peek does. +func (r *RingBuffer) PeekAt(off int, p []byte) (n int, err error) { + if len(p) == 0 || off < 0 { + return 0, r.readErr(false) + } + + r.mu.Lock() + defer r.mu.Unlock() + if err := r.readErr(true); err != nil { + return 0, err + } + + held := r.length() + if held == 0 { + // Nothing buffered at all, which is what Peek reports here too. + return 0, ErrIsEmpty + } + if off >= held { + // Buffered, but the consumer has already taken all of it. Not the same + // condition as an empty buffer, and not an error: it is what a consumer + // that has caught up sees every time it asks. + return 0, nil + } + + n = held - off + if n > len(p) { + n = len(p) + } + + start := r.r + off + if start >= r.size { + start -= r.size + } + if start+n <= r.size { + copy(p, r.buf[start:start+n]) + } else { + c1 := r.size - start + copy(p, r.buf[start:r.size]) + copy(p[c1:], r.buf[0:n-c1]) + } + return n, r.readErr(true) +} + +// length is Length without the lock, for callers that already hold it. +func (r *RingBuffer) length() int { + if r.w == r.r { + if r.isFull { + return r.size + } + return 0 + } + if r.w > r.r { + return r.w - r.r + } + return r.size - r.r + r.w +}