Improve benchmark tests

README.md

@@ -11,6 +11,8 @@
 - [Design Philosophy](#design-philosophy)
 - [Performance Characteristics](#performance-characteristics)
 - [Benchmarks](#benchmarks)
+	- [Performance Tips](#performance-tips)
+	- [In-memory Throughput Benchmark](#in-memory-throughput-benchmark)
 - [Examples](#examples)
 	- [Batch Writes/Reads](#batch-writesreads)
 	- [Stream Historical Data](#stream-historical-data)
@@ -66,23 +68,22 @@ import (
 func main() {
 	stream := ringbuf.New[string](1000)
 
-	// Writer (producer)
+	// Single writer (producer)
 	go func() {
 		for i := range 10_000 {
 			stream.Write(fmt.Sprintf("event-%d", i))
 			time.Sleep(100 * time.Microsecond) // Simulate i/o latency.
 		}
 
-		stream.Close()
+		stream.Close() // Broadcast io.EOF (end of stream).
 	}()
 
 	// Subscriber (consumer)
 	sub := stream.Subscribe(context.TODO(), nil)
-
 	for event := range sub.Iter() {
 		fmt.Println(event)
 	}
-
+	
 	if err := sub.Err(); !errors.Is(err, io.EOF) {
 		log.Fatal(err)
 	}
@@ -114,39 +115,73 @@ If you need guaranteed delivery, persistence, replay, or backpressure, this is n
 - **Write path**: Lock-free using atomic operations (~5 ns/op)
 - **Read path**: Lock-free hot path with minimal synchronization when waiting for new data
 - **Memory**: No memory allocations during data reads (0 B/op)
-- **Scalability**: Optimized for thousands of concurrent readers (1-10,000 readers at ~5 ns/op)
+- **Scalability**: Optimized for thousands of concurrent subscribers (1-100,000 in-memory readers)
 - **Latency**: Sub-microsecond read/write operations in common scenarios
-- **Write-throughput**: 200M+ writes/sec on modern hardware (assuming readers can keep up)
-- **Read-throughput**: 5B+ reads/sec on modern hardware (e.g. 50k subscribers)
+- **Write-throughput**: 200M+ writes/sec on modern hardware
+- **Read-throughput**: 200B+ in-memory reads/sec on modern hardware
 
 ## Benchmarks
 
-Performance heavily depends on hardware and your ring buffer / subscriber configuration. Batched writes generally perform better because they wake readers less often. The bigger the ring buffer size, the more concurrent readers will be able to keep up with the writer's pace (e.g. survive burst writes). With a sufficiently large buffer, it's often OK to allow subscribers to lag behind the head by up to ~90% of the buffer size.
+Performance heavily depends on hardware, ring buffer size and subscriber configuration (e.g. MaxLag). In real-world use cases, subscribers will likely be limited by I/O, network, message encoding/decoding, and also by other CPU and Go scheduler overhead of your program.
+
+### Performance Tips
+
+- Batched writes generally perform better, as they wake subscribers less often.
+- The bigger the MaxLag and ring buffer size, the more concurrent readers will be able to keep up with the writer's pace (e.g. survive burst writes or allow less reliable/slow network connections).
+- With a sufficiently large buffer, it's often OK to allow subscribers to lag behind the head by up to ~90% of the buffer size.
+- We strongly advise users to tune their configuration based on testing.
+
+### In-memory Throughput Benchmark
+
+This repository comes with an in-memory throughput benchmark test. See the following results on MacBook M5.
+
+Here we rate-limit the writer to ~1,000 `Write()` calls/sec; and each write batches 100 messages, that is ~100,000 `uint64` messages/sec in total. We allow readers to read a batch of up to 100 messages at a time:
 
-In real-world scenarios, the subscribers will likely be limited by I/O, JSON marshaling, or by writing responses to slow HTTP clients. ringbuf is designed to allow slow readers and fail gracefully if they cannot keep up without putting any backpressure on the writer and other subscribers.
+```
+$ go test -bench=BenchmarkThroughput -run=^$ -buffer_size=200000 -subscribers=1,10,100,1_000,10_000,50_000,100_000,200_000,500_000,1_000_000 -write_rate=1000 -write_batch=100 -read_batch=100 .
+goos: darwin
+goarch: arm64
+pkg: github.com/golang-cz/ringbuf
+cpu: Apple M5
+BenchmarkThroughput/subscribers_1-10               100072 reads/s     100072 writes/s          0 errors
+BenchmarkThroughput/subscribers_10-10             1000722 reads/s     100072 writes/s          0 errors
+BenchmarkThroughput/subscribers_100-10           10007106 reads/s     100071 writes/s          0 errors
+BenchmarkThroughput/subscribers_1000-10         100075501 reads/s     100076 writes/s          0 errors
+BenchmarkThroughput/subscribers_10000-10       1000686559 reads/s     100069 writes/s          0 errors
+BenchmarkThroughput/subscribers_50000-10       5003855243 reads/s     100077 writes/s          0 errors
+BenchmarkThroughput/subscribers_100000-10     10004740196 reads/s     100047 writes/s          0 errors
+BenchmarkThroughput/subscribers_200000-10     20010092634 reads/s     100050 writes/s          0 errors
+BenchmarkThroughput/subscribers_500000-10       280261997 reads/s      560.5 writes/s          0 errors
+BenchmarkThroughput/subscribers_1000000-10      270855033 reads/s      270.9 writes/s          0 errors
+PASS
+ok  	github.com/golang-cz/ringbuf	74.387s
+```
 
-We strongly advise users to tune their configuration based on testing.
+We can see that up to 200,000 subscribers were able to keep up with the writer and read a total of ~20 billion messages/sec with no subscriber falling behind (`errors=0`). However, at 1,000,000 subscribers, we can see that the system was overloaded and the overall throughput degraded. The buffer size was very generous (200,000 items) and exceeded the number of writes, so we didn't see any errors.
 
-For example, see the following in-memory throughput benchmark on MacBook M5. Here we rate-limit the writer to ~1,000 `Write()` calls/sec; and we write batches of 100 messages, that is ~100,000 `uint64` messages/sec in total. We allow readers to read a batch of up to 100 messages at a time.
+However, when we decrease the buffer size from 200,000 to just 10,000 items:
 
 ```
-$ go test -bench=BenchmarkThroughput -run=^$ -buffer_size=200000 -subscribers=1,10,100,1_000,10_000,50_000,100_000 -write_rate=1000 -write_batch=100 -read_batch=100 .
+$ go test -bench=BenchmarkThroughput -run=^$ -buffer_size=10000 -subscribers=1,10,100,1_000,10_000,50_000,100_000,200_000,500_000,1_000_000 -write_rate=1000 -write_batch=100 -read_batch=100 .
 goos: darwin
 goarch: arm64
 pkg: github.com/golang-cz/ringbuf
 cpu: Apple M5
-BenchmarkThroughput/subscribers_1-10          100071 writes/s	      100071 reads/s	  0 errors	 1209	    999287 ns/op
-BenchmarkThroughput/subscribers_10-10         100072 writes/s	     1000719 reads/s	  0 errors	 1209	    999282 ns/op
-BenchmarkThroughput/subscribers_100-10        100072 writes/s	    10007191 reads/s	  0 errors	 1209	    999282 ns/op
-BenchmarkThroughput/subscribers_1000-10       100072 writes/s	   100072250 reads/s	  0 errors	 1209	    999279 ns/op
-BenchmarkThroughput/subscribers_10000-10      100066 writes/s	  1000656396 reads/s	  0 errors	 1207	    999344 ns/op
-BenchmarkThroughput/subscribers_50000-10      100035 writes/s	  5001744580 reads/s	  0 errors	 1210	    999651 ns/op
-BenchmarkThroughput/subscribers_100000-10       2729 writes/s	   272949625 reads/s	  0 errors	   79	  36636800 ns/op
+BenchmarkThroughput/subscribers_1-10      	       100072 reads/s     100072 writes/s          0 errors
+BenchmarkThroughput/subscribers_10-10     	      1000719 reads/s     100072 writes/s          0 errors
+BenchmarkThroughput/subscribers_100-10    	     10007222 reads/s     100072 writes/s          0 errors
+BenchmarkThroughput/subscribers_1000-10   	    100071101 reads/s     100071 writes/s          0 errors
+BenchmarkThroughput/subscribers_10000-10  	   1000782188 reads/s     100078 writes/s          0 errors
+BenchmarkThroughput/subscribers_50000-10  	   1923639298 reads/s     100077 writes/s      32403 errors
+BenchmarkThroughput/subscribers_100000-10 	    419123458 reads/s     100065 writes/s      97163 errors
+BenchmarkThroughput/subscribers_200000-10 	    138727297 reads/s     100072 writes/s     199996 errors
+BenchmarkThroughput/subscribers_500000-10 	    324328784 reads/s      648.7 writes/s          0 errors
+BenchmarkThroughput/subscribers_1000000-10	    164544105 reads/s      164.5 writes/s          0 errors
 PASS
-ok  	github.com/golang-cz/ringbuf	13.680s
+ok  	github.com/golang-cz/ringbuf	56.813s
 ```
 
-We can see that 50,000 subscribers were able to keep up with the writer and read a total of ~5,000,000,000 messages/sec with no subscriber falling behind (`errors=0`). However, at 100,000 subscribers, we can see that the system was overloaded and the overall throughput degraded. The buffer size was quite generous, and we still didn't see any errors. However, if we decreased the buffer size, we'd likely see subscribers falling behind.
+We can see that we were able to handle up to 10,000 subscribers with maximum throughput and no errors. However, at 50,000 subscribers, we start seeing errors (subscribers falling behind). We can see that our total read throughput peaked at ~19 billion messages/sec.
 
 ## Examples
 

ringbuf_throughput_bench_test.go

@@ -62,15 +62,11 @@ func BenchmarkThroughput(b *testing.B) {
 			// Shared stats + first error.
 			var deliveredReads atomic.Uint64
 			var fellBehind atomic.Uint64
-			var firstErrMu sync.Mutex
-			var firstErr error
+			var firstErr atomic.Pointer[error]
 			recordErr := func(err error) {
-				fellBehind.Store(1)
-				firstErrMu.Lock()
-				if firstErr == nil {
-					firstErr = err
-				}
-				firstErrMu.Unlock()
+				fellBehind.Add(1)
+				e := err
+				firstErr.CompareAndSwap(nil, &e)
 			}
 
 			// Start subscribers (wait until all are ready, then start together).
@@ -80,11 +76,10 @@ func BenchmarkThroughput(b *testing.B) {
 			wgReady.Add(subs)
 			wgReaders.Add(subs)
 
-			cancels := make([]context.CancelFunc, 0, subs)
-			for i := range subs {
-				ctx, cancel := context.WithCancel(context.Background())
-				cancels = append(cancels, cancel)
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
 
+			for i := range subs {
 				sub := stream.Subscribe(ctx, &ringbuf.SubscribeOpts{
 					Name:          fmt.Sprintf("sub-%d", i),
 					MaxLag:        *flagBufferSize * 9 / 10, // Allow readers to fall behind, but fail fast if they can't keep up.
@@ -126,7 +121,7 @@ func BenchmarkThroughput(b *testing.B) {
 				}
 			}
 
-			// Phase 1: write b.N items.
+			// Produce items: Write batch of items b.N times
 			b.ResetTimer()
 			t0 := time.Now()
 			for i := 0; i < b.N; i++ {
@@ -142,36 +137,26 @@ func BenchmarkThroughput(b *testing.B) {
 			elapsed := time.Since(t0)
 			b.StopTimer()
 
-			// Phase 2: stop readers (cancel + repeated broadcast-only flush).
-			for _, cancel := range cancels {
-				cancel()
-			}
-
 			done := make(chan struct{})
 			go func() {
 				wgReaders.Wait()
 				close(done)
 			}()
 
-			deadline := time.NewTimer(2 * time.Second)
+			deadline := time.NewTimer(1 * time.Second)
 			defer deadline.Stop()
 
-			shutdownDone := false
-			shutdownTimedOut := false
-			for {
-				select {
-				case <-done:
-					shutdownDone = true
-				case <-deadline.C:
-					recordErr(fmt.Errorf("benchmark shutdown timed out (readers did not exit)"))
-					shutdownTimedOut = true
-				default:
-					stream.Write() // broadcast-only "flush"
-					time.Sleep(100 * time.Microsecond)
-				}
-				if shutdownDone || shutdownTimedOut {
-					break
-				}
+			select {
+			case <-done:
+				// All readers exited.
+
+			case <-deadline.C:
+				// Wake up all readers to exit.
+				cancel()
+				stream.Close()
+
+				// Wait for all readers to exit.
+				<-done
 			}
 
 			// Report metrics.
@@ -184,11 +169,9 @@ func BenchmarkThroughput(b *testing.B) {
 			b.ReportMetric(float64(fellBehind.Load()), "errors")
 
 			// Shown with `-test.v` (keeps benchmark output readable by default).
-			firstErrMu.Lock()
-			if firstErr != nil {
-				b.Logf("benchmark overload: %v", firstErr)
+			if errPtr := firstErr.Load(); errPtr != nil {
+				b.Logf("overloaded: first subscriber error: %v", *errPtr)
 			}
-			firstErrMu.Unlock()
 		})
 	}
 }