bench_test.go

package spool

import (
	"fmt"
	"testing"
)

// Every spool opened here passes an explicit SegmentBytes AND MaxBytes, and
// every benchmark that grows the backlog drains it inside b.StopTimer.
//
// This is not tidiness. Options{} means MaxBytes 0, which used to mean
// UNBOUNDED — a Pop benchmark written against it wrote 12 GB of segments in 90
// seconds and took a machine from 11 GB free to 7.5 GB. The zero value is now
// DefaultMaxBytes, so an uncapped benchmark fails with ErrFull instead of
// filling the disk, but relying on that would make these measure the cap rather
// than the code. State the bound.

const (
	benchSegBytes = 4 << 20
	benchMaxBytes = 32 << 20
)

func benchOpts() Options {
	return Options{SegmentBytes: benchSegBytes, MaxBytes: benchMaxBytes}
}

func benchPayload(n int) []byte {
	b := make([]byte, n)
	for i := range b {
		b[i] = byte(i)
	}
	return b
}

func sizeName(n int) string {
	switch {
	case n >= 1<<20:
		return "1MiB"
	case n >= 64<<10:
		return "64KiB"
	case n >= 4<<10:
		return "4KiB"
	default:
		return "256B"
	}
}

// drainAll empties the queue without timing it, so a benchmark's disk footprint
// stays bounded however large b.N grows.
func drainAll(b *testing.B, s *Spool) {
	b.Helper()
	for {
		_, commit, ok, err := s.Pop()
		if err != nil {
			b.Fatalf("drain: %v", err)
		}
		if !ok {
			return
		}
		commit()
	}
}

// BenchmarkFrameSum isolates the per-frame checksum: the work Append adds on
// top of its write+fsync. It must stay allocation-free — the digest is
// stack-held — and it is ~3 orders of magnitude cheaper than the fsync it rides
// along with (compare against BenchmarkAppend), so integrity here is close to
// free. Read the two together before trading the checksum away for throughput.
func BenchmarkFrameSum(b *testing.B) {
	for _, size := range []int{256, 4 << 10, 64 << 10, 1 << 20} {
		data := benchPayload(size)
		var hdr [4]byte
		hdr[0], hdr[1], hdr[2], hdr[3] = byte(size>>24), byte(size>>16), byte(size>>8), byte(size)
		b.Run(sizeName(size), func(b *testing.B) {
			b.SetBytes(int64(size))
			b.ReportAllocs()
			var sink uint64
			for i := 0; i < b.N; i++ {
				sink = frameSum(hdr[:], data)
			}
			_ = sink
		})
	}
}

// BenchmarkAppend measures the full durable append (frame + write + fsync). The
// fsync dominates by design: Append must not return until the record is on
// disk, since the producer advances its checkpoint on the strength of it.
func BenchmarkAppend(b *testing.B) {
	for _, size := range []int{256, 4 << 10, 64 << 10} {
		data := benchPayload(size)
		b.Run(sizeName(size), func(b *testing.B) {
			s, err := Open(b.TempDir(), benchOpts())
			if err != nil {
				b.Fatal(err)
			}
			defer func() { _ = s.Close() }()
			b.SetBytes(int64(size))
			b.ReportAllocs()
			b.ResetTimer()
			for i := 0; i < b.N; i++ {
				if err := s.Append(data); err != nil {
					b.StopTimer()
					drainAll(b, s)
					b.StartTimer()
					if err := s.Append(data); err != nil {
						b.Fatal(err)
					}
				}
			}
		})
	}
}

// BenchmarkAppendNoSync isolates the write path from the fsync. One write(2)
// per record rather than two (header then payload) is what the frame-staging
// buffer buys; the difference is invisible under a per-record fsync and worth
// roughly a quarter of the per-record cost under group commit.
func BenchmarkAppendNoSync(b *testing.B) {
	for _, size := range []int{256, 4 << 10} {
		data := benchPayload(size)
		b.Run(sizeName(size), func(b *testing.B) {
			s, err := Open(b.TempDir(), benchOpts())
			if err != nil {
				b.Fatal(err)
			}
			defer func() { _ = s.Close() }()
			b.SetBytes(int64(size))
			b.ReportAllocs()
			b.ResetTimer()
			for i := 0; i < b.N; i++ {
				if err := s.AppendNoSync(data); err != nil {
					b.StopTimer()
					if err := s.Sync(); err != nil {
						b.Fatal(err)
					}
					drainAll(b, s)
					b.StartTimer()
					if err := s.AppendNoSync(data); err != nil {
						b.Fatal(err)
					}
				}
			}
			b.StopTimer()
			if err := s.Sync(); err != nil {
				b.Fatal(err)
			}
		})
	}
}

// BenchmarkGroupCommit measures the group-commit amortization: N small
// records per fsync (AppendNoSync xN + one Sync). Compare the ns/record
// metric against BenchmarkAppend/256B — the whole point of the API is that
// the ~ms fsync is paid once per group instead of once per record.
func BenchmarkGroupCommit(b *testing.B) {
	data := benchPayload(256)
	for _, group := range []int{10, 100, 1000} {
		b.Run(fmt.Sprintf("group-%d", group), func(b *testing.B) {
			s, err := Open(b.TempDir(), benchOpts())
			if err != nil {
				b.Fatal(err)
			}
			defer func() { _ = s.Close() }()
			b.SetBytes(int64(group * len(data)))
			b.ReportAllocs()
			b.ResetTimer()
			for i := 0; i < b.N; i++ {
				for j := 0; j < group; j++ {
					if err := s.AppendNoSync(data); err != nil {
						b.StopTimer()
						if err := s.Sync(); err != nil {
							b.Fatal(err)
						}
						drainAll(b, s)
						b.StartTimer()
						if err := s.AppendNoSync(data); err != nil {
							b.Fatal(err)
						}
					}
				}
				if err := s.Sync(); err != nil {
					b.Fatal(err)
				}
			}
			b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(b.N)/float64(group), "ns/record")
		})
	}
}

// BenchmarkPop measures the read path alone: two ReadAt calls plus the
// allocations Pop hands the caller. Popping without committing re-delivers the
// same frame, which is exactly what isolates the read cost from commit's
// cursor write — and bounds the disk to one record.
func BenchmarkPop(b *testing.B) {
	for _, size := range []int{256, 4 << 10, 64 << 10} {
		b.Run(sizeName(size), func(b *testing.B) {
			s, err := Open(b.TempDir(), benchOpts())
			if err != nil {
				b.Fatal(err)
			}
			defer func() { _ = s.Close() }()
			if err := s.Append(benchPayload(size)); err != nil {
				b.Fatal(err)
			}
			b.SetBytes(int64(size))
			b.ReportAllocs()
			b.ResetTimer()
			for i := 0; i < b.N; i++ {
				data, _, ok, err := s.Pop()
				if !ok || err != nil {
					b.Fatalf("Pop: ok=%v err=%v", ok, err)
				}
				_ = data
			}
		})
	}
}

// BenchmarkPopCommit measures the full consume step. The delta against
// BenchmarkPop IS the cursor write: with Options.CommitSync it is an fsync per
// record and lands on the same order as BenchmarkAppend, which is the ceiling
// on any consumer that commits per small record. Both modes are measured
// because the choice between them is the difference between a consumer that
// keeps up with group commit and one that cannot.
func BenchmarkPopCommit(b *testing.B) {
	const refill = 512
	for _, sync := range []bool{false, true} {
		b.Run(fmt.Sprintf("CommitSync=%v", sync), func(b *testing.B) {
			data := benchPayload(256)
			opts := benchOpts()
			opts.CommitSync = sync
			s, err := Open(b.TempDir(), opts)
			if err != nil {
				b.Fatal(err)
			}
			defer func() { _ = s.Close() }()
			b.SetBytes(int64(len(data)))
			b.ReportAllocs()
			b.ResetTimer()
			for done := 0; done < b.N; {
				n := refill
				if r := b.N - done; r < n {
					n = r
				}
				b.StopTimer()
				for j := 0; j < n; j++ {
					if err := s.AppendNoSync(data); err != nil {
						b.Fatal(err)
					}
				}
				if err := s.Sync(); err != nil {
					b.Fatal(err)
				}
				b.StartTimer()
				for j := 0; j < n; j++ {
					_, commit, ok, err := s.Pop()
					if !ok || err != nil {
						b.Fatalf("Pop: ok=%v err=%v", ok, err)
					}
					commit()
				}
				done += n
			}
		})
	}
}

// BenchmarkOpen measures restart cost. repairTail walks the tail segment frame
// by frame, reading AND checksumming every payload, so the cost tracks
// SegmentBytes and the record size — not the total backlog, since only the tail
// is walked. That is why reopening a large backlog can be cheaper than a small
// one: a bigger backlog means the tail is a fresh, mostly-empty segment.
func BenchmarkOpen(b *testing.B) {
	for _, records := range []int{1000, 10000} {
		b.Run(fmt.Sprintf("%d-records-256B", records), func(b *testing.B) {
			dir := b.TempDir()
			s, err := Open(dir, benchOpts())
			if err != nil {
				b.Fatal(err)
			}
			data := benchPayload(256)
			for i := 0; i < records; i++ {
				if err := s.AppendNoSync(data); err != nil {
					b.Fatal(err)
				}
			}
			if err := s.Sync(); err != nil {
				b.Fatal(err)
			}
			if err := s.Close(); err != nil {
				b.Fatal(err)
			}
			b.ReportAllocs()
			b.ResetTimer()
			for i := 0; i < b.N; i++ {
				q, err := Open(dir, benchOpts())
				if err != nil {
					b.Fatal(err)
				}
				if err := q.Close(); err != nil {
					b.Fatal(err)
				}
			}
		})
	}
}