Batching — Professional Level¶

Table of Contents¶

1. Introduction
2. Internals of Go Channels and Batchers
3. Ring Buffers and Lock-Free Accumulators
4. NUMA-Aware Batching
5. The Kafka Producer Internals
6. The Postgres COPY Protocol
7. The librdkafka Buffer Pool
8. Memory Allocation Patterns
9. CPU Cache Behavior
10. Profiling a Batcher
11. Persistent Buffers
12. Group Commit and WAL Batching
13. io_uring and Modern I/O Batching
14. Building a Lock-Free Batcher
15. Benchmarking Methodology
16. Frontier Research
17. Summary

Introduction¶

Senior batching is about architecture. Professional batching is about internals: how do channels work in Go's runtime, what does the Kafka producer actually do, why does Postgres COPY out-perform multi-row INSERT by 10x, when does NUMA awareness matter, what does a lock-free MPMC queue buy you?

This file dives deep. It assumes you have:

• Built and shipped a production batcher.
• Used pprof and traced GC behaviour.
• Read at least one of: Go runtime source, Linux kernel network stack, Postgres internals.
• Comfortable with assembler-level reasoning when needed.

If those do not describe you, the senior file is the right level. Come back here when you want to understand why the layers below behave the way they do.

After this file you will:

• Know what make(chan T, n) allocates, how send and receive interact with the runtime scheduler, and what bottlenecks emerge at extreme scale.
• Be able to implement a lock-free MPMC ring buffer batcher and reason about its correctness.
• Understand NUMA effects on accumulators and when CPU pinning helps.
• Know what Kafka's linger.ms and batch.size parameters do internally and what the producer does between user code and the broker.
• Understand Postgres' COPY protocol and why it bypasses parser overhead.
• Profile a batcher down to the function call and the allocation.
• Discuss persistent buffers and write-ahead logs.

Internals of Go Channels and Batchers¶

A Go channel is a small struct (runtime.hchan) with a buffer, a lock, a pair of wait queues (senders and receivers), and metadata. When you make(chan T, n):

1. The runtime allocates an hchan struct plus space for n elements.
2. The dataqsiz field is set to n.
3. The closed flag is 0.
4. The wait queues are empty.

A send (ch <- v) does:

1. Lock hchan.lock.
2. Check if a receiver is waiting. If yes, copy directly to the receiver's stack and unblock it.
3. Otherwise, if buffer has space, copy into the buffer and increment sendx.
4. Otherwise, block: enqueue this goroutine on the senders' wait queue, park.
5. Unlock.

A receive is the mirror image.

For a batcher, the send is on the hot path. The interesting cases:

• Buffer not full, no receiver waiting: send copies to buffer, ~50 ns. Common in steady state.
• Buffer full, receiver waiting: send copies directly to receiver, ~70 ns (direct hand-off).
• Buffer full, no receiver: send blocks. Park the goroutine, schedule another. ~1 µs of overhead plus wait time.

The runtime's chansend and chanrecv are in runtime/chan.go. Reading these (~500 lines) gives you the floor on channel performance.

Implications for Batchers¶

• Steady-state channel cost: ~100 ns per send-receive pair.
• Channel lock contention: under heavy load (many producers), the lock is held briefly but frequently. At 10M sends/s, the lock is taken 10M times.
• The lock is per-channel. Multiple channels are independent.

When Channels Bottleneck¶

Roughly 10M sends/s is the channel ceiling on a single channel on modern hardware. Beyond that, lock contention dominates. Solutions:

• Sharded channels: hash producer to one of N channels, each with its own consumer.
• Lock-free queue: replace chan T with a lock-free MPMC queue (CAS-based).
• Per-CPU channels: one channel per goroutine, scheduled to one CPU.

For 99.9% of batchers, channels are not the bottleneck. The sink is.

Ring Buffers and Lock-Free Accumulators¶

A ring buffer is a fixed-size circular array with head and tail pointers. The classic data structure for high-throughput queues.

Single-Producer, Single-Consumer (SPSC)¶

type SPSCRing[T any] struct {
    buf  []T
    head atomic.Uint64
    tail atomic.Uint64
}

func (r *SPSCRing[T]) Push(v T) bool {
    h := r.head.Load()
    t := r.tail.Load()
    if h-t >= uint64(len(r.buf)) {
        return false // full
    }
    r.buf[h%uint64(len(r.buf))] = v
    r.head.Store(h + 1)
    return true
}

func (r *SPSCRing[T]) Pop() (T, bool) {
    var zero T
    t := r.tail.Load()
    h := r.head.Load()
    if t == h {
        return zero, false // empty
    }
    v := r.buf[t%uint64(len(r.buf))]
    r.tail.Store(t + 1)
    return v, true
}

Lock-free, ~5 ns per op. Limit: one producer, one consumer.

MPMC¶

Multi-producer, multi-consumer ring buffers are harder. Vyukov's MPMC ring buffer is a standard. Each slot has a sequence number; producers CAS the sequence to claim a slot; consumers CAS to release.

Implementation is ~30 lines of subtle code. Performance: ~20-50 ns per op.

For Go batchers, MPMC rings are rarely needed. Channels handle 10M ops/s, which covers almost everything.

When to Reach for Lock-Free¶

• Channel contention measurable in profiles (>10% time in runtime.chansend).
• Trading systems or HFT (microseconds matter).
• Interfacing with C code that uses ring buffers.

For everything else: chan T.

The Disruptor Pattern¶

LMAX's Disruptor is a ring buffer with batched producer/consumer barriers. It batches naturally: a consumer reads [old_tail, new_head) as a slice, not one item at a time. Many Java implementations; Go versions exist (smallnest/queue).

A Disruptor-style batcher would have the consumer read N items at a time directly from the ring, avoiding the slice copy. ~2x faster than channel for high throughput.

NUMA-Aware Batching¶

On NUMA (Non-Uniform Memory Access) systems, accessing memory on a different socket is 2-3x slower than local. A multi-socket box has multiple NUMA nodes; goroutines that bounce between sockets pay this cost.

For batchers handling > 100K ops/s, NUMA effects can matter. Strategies:

• Pin goroutines to a node: runtime.LockOSThread + sched_setaffinity.
• Per-node batchers: one batcher per NUMA node; producers route to local.
• Per-node memory pools: sync.Pool is per-P, naturally somewhat NUMA-friendly.

Verify with numactl --membind and benchmarks. Often the speedup is 20-30%. Sometimes the code complexity isn't worth it.

For laptop or single-socket cloud VM, NUMA is irrelevant.

The Kafka Producer Internals¶

A Kafka producer's life is more complex than "send a message". Internals:

Per-Topic Per-Partition Buffer¶

The producer maintains a per-topic-partition buffer. Records are appended; the buffer is flushed when full (batch.size) or after linger.ms.

Sender Thread¶

A separate thread (in Java; goroutine in Go clients) polls the buffer, builds produce requests, sends them.

Compression¶

The buffer is compressed (gzip, snappy, zstd) before sending. Compression amortises the per-record header overhead.

Acks¶

acks=0: producer doesn't wait. Fastest, no durability. acks=1: leader acks after writing to local log. Default. Some durability. acks=all: leader waits for replicas to ack. Strongest durability.

Retries¶

retries and retry.backoff.ms control transient failures. Idempotence prevents duplicates via producer IDs.

Putting It Together¶

Application: client.Produce(record)
  |
  v
Producer: append to partition buffer
  |
  v
Sender thread: poll buffers, build request when ready
  |
  v
Compression: compress batch
  |
  v
Network: send to broker
  |
  v
Broker: append to log, replicate, ack
  |
  v
Sender thread: invoke callback with success/failure
  |
  v
Application: callback fires

Notice: 4+ layers of batching/coordination between user code and durable storage. Each is tuned. If you write your own batcher in front of this, understand which of these you are duplicating.

Reading librdkafka¶

librdkafka (the C library, also wrapped by confluent-kafka-go) has ~30K lines. The batching logic is in rdkafka_partition.c and rdkafka_request.c. Worth reading once.

The Postgres COPY Protocol¶

Postgres has three insert paths:

1. INSERT (single row).
2. Multi-row INSERT (rows in one query).
3. COPY FROM (streaming bulk insert).

COPY is the fastest by 5-10x. Why?

What INSERT Does¶

Each INSERT goes through:

• Parse SQL text.
• Plan the query.
• Execute: row by row, check constraints, write to heap, write to indexes.
• Generate WAL records.
• Commit (fsync WAL).

For 1000 rows, this is 1000 parse, 1000 plan, 1000 execute. With multi-row, it is 1 parse, 1 plan, 1000 execute.

What COPY Does¶

COPY streams binary or text data through a dedicated protocol:

• One COPY FROM ... command sets up the stream.
• Rows arrive over the protocol as binary data.
• Server inserts directly into heap, bypassing parser.
• Constraint checks deferred until end (for non-deferred, checked per row but in tight loop).
• WAL written in larger chunks.

Result: per-row cost drops dramatically. A bulk insert of 1M rows takes 5 seconds via INSERT, 1 second via COPY.

Binary vs Text Format¶

Binary format: rows encoded in Postgres' binary protocol. Smaller, faster to parse. Text format: rows as tab-separated values. Easier to debug, slower.

Use binary for production batchers.

pgx CopyFrom¶

pgx.CopyFromRows builds a binary stream from [][]any and sends it. Behind the scenes:

• Opens a COPY FROM session.
• Encodes each value using type-specific binary encoders.
• Streams chunks (default 65 KB) to the server.
• Closes the session.

The Go client is doing the protocol work; the server is doing the bulk insert work. Both sides have minimal per-row CPU compared to INSERT.

Caveats¶

• COPY does not support ON CONFLICT.
• COPY does not return identifiers (no RETURNING).
• COPY locks tables differently (some metadata access is blocked during long COPY).

For idempotent inserts: temp table + INSERT SELECT ON CONFLICT. The COPY into the temp table is fast; the INSERT SELECT picks up.

The librdkafka Buffer Pool¶

librdkafka uses a custom buffer pool to avoid allocation churn. Each record is wrapped in an rd_kafka_msg_t; the pool reuses these.

Reading rdkafka_msg.c: the pool is per-thread, with a fast path that avoids malloc. Allocation is amortised across millions of records.

For Go batchers: sync.Pool plays a similar role but is GC-aware (objects may be reclaimed). For raw speed, manual pools work too.

Memory Allocation Patterns¶

A high-throughput batcher's allocations come from:

1. The item value itself (heap or stack).
2. The buffer slice (one alloc on first append; grows as needed).
3. The copy at flush time (make + copy).
4. The flushed batch's escape to the sink (often heap).
5. Serialisation (gzipping, marshalling).

To minimise allocations:

• Pre-allocate the buffer with full capacity.
• Use sync.Pool for flush slices.
• Use *bytes.Buffer from a pool for serialisation.
• Avoid interface{} and reflection in hot paths.
• Pre-allocate item structs in pools if pointer-shaped.

Profile with -memprofile to find the actual sources. Common surprises:

• Channel sends of large structs copy on every send. Use pointers.
• Closures capture variables; capture only what is needed.
• Logging (e.g., log.Printf("%v", batch)) can be a huge allocator.

CPU Cache Behavior¶

A modern CPU has L1 (~32 KB), L2 (~256 KB), L3 (~8 MB) caches. Cache lines are 64 bytes.

For a batcher:

• The buffer slice header is 24 bytes (ptr, len, cap). Fits in one cache line.
• A batch of 1000 items at 100 bytes each is 100 KB — fits in L2.
• A batch of 1M items at 100 bytes each is 100 MB — exceeds L3, hits main memory.

Cache effects make small batches faster per item than huge batches. There is a "cache knee" at ~64 KB-ish batch size where staying in L2 is fast. Going beyond means main-memory bandwidth dominates.

For most batchers, the network and disk dominate, and cache effects are noise. For in-memory transforms (combiners), cache fit matters.

False Sharing¶

Two atomic variables in the same cache line cause "false sharing": cache invalidation between cores even though the data is logically independent.

Fix: pad fields to cache-line boundaries. Go's runtime.padding package or manual _ [56]byte arrays after small atomics.

For a batcher's counters (enqueued, flushed_ok, etc), padding helps when these are read from many cores simultaneously. Usually negligible; check with profile.

Profiling a Batcher¶

A real profiling session:

// In main:
import _ "net/http/pprof"
go http.ListenAndServe("localhost:6060", nil)

Then:

# CPU profile, 30 seconds
go tool pprof -seconds=30 -http=:8080 localhost:6060/debug/pprof/profile

# Heap profile
go tool pprof -http=:8080 localhost:6060/debug/pprof/heap

# Goroutines
go tool pprof -http=:8080 localhost:6060/debug/pprof/goroutine

Look for:

• Top CPU consumers: should be sink-related (network, serialisation), not batcher overhead.
• Allocation hotspots: should be at flush boundaries (the make + copy), not per-item.
• Goroutine count: small and stable.

If CPU is in runtime.chansend more than the sink: contention; consider sharding.

If heap is dominated by item structs: items live too long; reduce buffer size or flush faster.

If goroutine count is growing: leak; investigate.

Trace¶

go tool trace shows scheduling, GC, syscalls. Useful for understanding pause behaviour.

Capture:

curl localhost:6060/debug/pprof/trace?seconds=10 > trace.out
go tool trace trace.out

Look for: GC pauses correlated with flushes (item ref pressure), long syscalls (slow sink), scheduling gaps (blocked goroutines).

Persistent Buffers¶

For high-value data, in-memory buffers risk loss on crash. Persistent buffers store items on disk before flushing to the downstream sink.

Design¶

producer -> in-memory queue -> disk log (WAL) -> in-memory queue -> sink

The disk log is the durability boundary. Items committed to the log are durable; the sink flush is just a propagation step.

Implementations¶

• BadgerDB: embedded key-value store with WAL semantics.
• bbolt: B-tree-based KV store.
• Custom log file: append-only, fsync on commit.

Trade-Offs¶

• Pro: zero loss on crash (with synchronous fsync).
• Con: disk I/O cost; usually limits throughput to 10K-100K records/s.
• Con: state to manage (rotation, compaction, recovery).

For most application-level batchers, in-memory + upstream-replay is preferred. Persistent buffers are for edge cases (constrained networks, regulatory requirements).

Vector and Fluent Bit¶

These telemetry shippers support disk buffers natively. Configuration: disk.max_size, disk.directory. The application sends events to the shipper (often via Unix socket); the shipper buffers, ships to the downstream. Crash recovery from the disk buffer.

Group Commit and WAL Batching¶

Inside Postgres, MySQL, and other databases: group commit is a batching mechanism for WAL fsync.

Postgres Group Commit¶

Parameters:

• commit_delay: how long to wait for siblings (microseconds).
• commit_siblings: minimum siblings before waiting.

When a transaction commits, it waits commit_delay if commit_siblings other transactions are in progress. The WAL fsync amortises across them.

Default: off (commit_delay = 0). For very write-heavy workloads, setting commit_delay = 100-1000 µs and commit_siblings = 5 can improve throughput.

MySQL Group Commit¶

Similar but tunable via binlog_group_commit_sync_delay and innodb_flush_log_at_trx_commit.

The Pattern¶

Two layers of batching:

• App-level: batch INSERTs.
• DB-level: batch fsyncs.

Both contribute to throughput. Tuning the DB-level requires DBA knowledge.

io_uring and Modern I/O Batching¶

Linux's io_uring (since 5.1, mature in 5.10+) lets applications submit batches of I/O operations to the kernel via a ring buffer.

For batchers:

• One io_uring_submit can carry many writes.
• Completion is signalled via a completion ring.
• Reduces syscalls dramatically.

Go libraries: github.com/iceber/iouring-go. Use cases: high-throughput log writers, custom databases.

Most batchers do not need io_uring. The bottleneck is upstream (sink latency), not syscall count. If profiling shows syscalls dominating, look at io_uring.

Building a Lock-Free Batcher¶

For extreme throughput (>10M items/s), a lock-free batcher uses MPMC ring buffers.

type LFBatcher[T any] struct {
    ring    *vyukov.MPMC[T]
    sink    Sink[T]
    maxSize int
    quit    atomic.Bool
}

func (b *LFBatcher[T]) Add(v T) bool { return b.ring.Enqueue(v) }

func (b *LFBatcher[T]) run() {
    buf := make([]T, 0, b.maxSize)
    for !b.quit.Load() {
        for len(buf) < b.maxSize {
            v, ok := b.ring.Dequeue()
            if !ok {
                break
            }
            buf = append(buf, v)
        }
        if len(buf) > 0 {
            _ = b.sink.Write(context.Background(), buf)
            buf = buf[:0]
        } else {
            runtime.Gosched()
        }
    }
}

Caveats:

• No time trigger (would require atomic timestamps).
• Spins if empty (consumes CPU). Real implementations adaptive-spin then park.
• Drop on overflow (no backpressure).

For 99.99% of Go batchers, channel-based is the right choice. Lock-free is for HFT and database internals.

Benchmarking Methodology¶

Reliable benchmarks of batchers require:

Warm Up¶

for i := 0; i < 10000; i++ {
    b.Add(i)
}
// Discard timing of warm-up.
b.WaitForDrain()

Steady State¶

Measure steady-state throughput, not startup:

start := time.Now()
for i := 0; i < N; i++ {
    b.Add(i)
}
b.WaitForDrain()
elapsed := time.Since(start)
ops := float64(N) / elapsed.Seconds()

Multiple Producers¶

Real load is multi-producer:

var wg sync.WaitGroup
for w := 0; w < numWorkers; w++ {
    wg.Add(1)
    go func(id int) {
        defer wg.Done()
        for i := 0; i < N/numWorkers; i++ {
            b.Add(Item{ID: id*1e9 + i})
        }
    }(w)
}
wg.Wait()

GC Disabled¶

For peak measurements, disable GC:

GOGC=off go test -bench .

Note: this hides real-world GC pressure. Re-enable for production-realistic.

Histogram, not Average¶

Latency: emit a histogram (use hdrhistogram library). The average lies; p99 tells truth.

Multiple Runs¶

Run 10 times; report median. Single-run results are noise.

Compare Apples to Apples¶

Batched vs unbatched must use the same sink, same hardware, same inputs.

Frontier Research¶

Areas of active research:

• Adaptive batch sizing with ML: Bandit algorithms tune MaxBatchSize online.
• Coalescing in service meshes: Istio/Envoy can batch outbound calls; research on optimal policies.
• Hierarchical batching: Multi-level amortisation in disaggregated storage.
• Persistent batchers with NVMe: Trade RAM for SSD without losing throughput.
• DPDK batching: Userspace networking; bypasses kernel for ultra-low-latency.

For most engineers, today's tools are enough. Stay aware of frontier work for the next 5 years.

A Real Distributed Pipeline Walkthrough¶

Imagine a service handling 1M events/s, sharded across 10 instances, writing to Kafka, then consumed and written to Postgres.

Stage 1: HTTP Ingestion (10 instances)¶

Each instance: - Receives 100K req/s. - Validates payload. - Batches into Kafka producer. - Returns 202.

Per-instance: 10 cores, 1 batcher feeding Kafka.

Stage 2: Kafka Brokers (3 brokers, 30 partitions)¶

Each broker: - Accepts produce requests. - Appends to partition logs. - Replicates to 2 other brokers. - Acks.

Produce throughput: 1M records/s across 30 partitions = 33K/s per partition.

Stage 3: Consumer Service (5 instances)¶

Each instance: - Consumes from 6 partitions. - Decodes records. - Batches into Postgres CopyFrom.

Per-instance: 200K events/s. Postgres can sustain this with CopyFrom (1M rows/s capacity).

Stage 4: Postgres (1 primary, 2 replicas)¶

Primary handles all writes. Replicas serve reads.

Per-write: COPY of 1000 rows in ~20 ms. 50 batches/s = 50K rows/s. Below capacity.

End-to-End Latency¶

• API ingress to Kafka: ~5 ms (batcher's MaxBatchDelay).
• Kafka produce to consume: ~50 ms (consumer poll interval).
• Consume to Postgres: ~50 ms (consumer batcher's MaxBatchDelay).
• Postgres COPY: ~20 ms.

End-to-end p99: ~200 ms. SLA: 500 ms. Comfortable.

Failure Modes¶

• HTTP instance crash: Kafka has 7-day retention; events not lost. New instance picks up new traffic.
• Kafka broker crash: replicas continue; broker rejoins.
• Consumer instance crash: another consumer takes over the partitions.
• Postgres replica crash: failover via Patroni or similar.

Each layer's batching makes the system efficient; each layer's redundancy makes it reliable.

A Real Distributed Pipeline Walkthrough (cont): Capacity Planning¶

Estimated Resource Use¶

• HTTP ingestion: 10 instances * 10 cores = 100 cores; 100 GB RAM.
• Kafka: 3 brokers * 8 cores = 24 cores; 3 * 64 GB RAM.
• Consumer: 5 instances * 8 cores = 40 cores; 40 GB RAM.
• Postgres: 1 primary + 2 replicas * 32 cores = 96 cores; 3 * 256 GB RAM.

Total: 260 cores, 1.3 TB RAM. Cost: $2-5K/month on cloud.

Without batching, the same throughput would require 10-50x more compute. Batching is what makes this affordable.

Inside Memcached: Multi-Get and Batching¶

Memcached supports get k1 k2 k3 k4. The server returns each value separately. The client batches keys to reduce round-trips.

Internally, the server processes each key independently but in one TCP message. Per-call cost amortised.

Implications¶

If a batcher reads from Memcached, batch the get calls. ~10x speedup typical.

Pipelined Get¶

Some clients use pipelined gets: send multiple get k commands without waiting for responses. The server processes in order; client matches responses.

Similar throughput to multi-get; different shape.

Inside Redis Pipelining¶

Redis supports pipelining: send multiple commands in one packet; receive multiple responses.

pipe := client.Pipeline()
pipe.Set(ctx, "k1", "v1", 0)
pipe.Set(ctx, "k2", "v2", 0)
cmds, _ := pipe.Exec(ctx)

Behind the scenes: one TCP write, one TCP read. Server processes each command; client matches responses to commands.

Throughput: 10-100x single-command.

Use With Batcher¶

A Redis sink:

func (s *RedisSink) Write(ctx context.Context, batch []Item) error {
    pipe := s.client.Pipeline()
    for _, item := range batch {
        pipe.Set(ctx, item.Key, item.Value, item.TTL)
    }
    _, err := pipe.Exec(ctx)
    return err
}

Pipelining + batching = max throughput for Redis sinks.

Inside Cassandra Batching¶

Cassandra has BEGIN BATCH ... APPLY BATCH syntax. The driver sends multiple writes in one frame.

Unlike Postgres or Redis, Cassandra's batching is mostly latency-focused, not throughput. Large batches can stress the coordinator node.

Recommendation: small batches (5-100 statements) for related rows in the same partition. Larger across partitions is an anti-pattern.

Inside DynamoDB BatchWriteItem¶

DynamoDB caps at 25 items per BatchWriteItem call. On partial success, some items are unprocessed; client retries them.

Sink Wrapper¶

func (s *DynamoSink) Write(ctx context.Context, batch []Item) error {
    for i := 0; i < len(batch); i += 25 {
        end := i + 25
        if end > len(batch) {
            end = len(batch)
        }
        // Build BatchWriteItem request from batch[i:end].
        // Handle UnprocessedItems by retrying.
    }
    return nil
}

Our batcher's MaxBatchSize should be a multiple of 25 (or 25 itself) to align.

Inside BigQuery Streaming Inserts¶

BigQuery's insertAll accepts up to 500 rows per call (or 10 MB).

Throughput: 100K rows/s/table.

Idempotency: each row has an insertId; duplicates dropped.

Sink Wrapper¶

func (s *BQSink) Write(ctx context.Context, batch []Event) error {
    rows := make([]*bigquery.RowInserter, len(batch))
    // ... build rows ...
    return s.table.Inserter().Put(ctx, rows)
}

For high throughput: multiple inserters per table (BigQuery handles parallelism).

A Closing Thought: The Discipline of Batching¶

After 5000 lines, what is the lesson?

Batching is one of the few "free" performance wins in software engineering. It is also one of the easiest to get wrong.

The discipline: - Measure first. Profile. Find the per-call cost. - Design deliberately. Both triggers. Bounded queue. Graceful shutdown. - Observe rigorously. Four metrics, alerts, runbooks. - Operate carefully. Drain on SIGTERM. Chaos test. Document.

A batcher is not a "set and forget" component. It evolves with the workload. Tuning is an ongoing conversation with the system.

Engineers who internalise this become the people their team consults when "the system is slow but I don't know why". The answer is, often, in how data flows from one component to another, and the batchers along the way.

Reading Recommendations¶

For deeper study:

• "Designing Data-Intensive Applications" by Martin Kleppmann: data flow, batching, streaming.
• "Systems Performance" by Brendan Gregg: profiling, tracing, observability.
• The Go runtime source: runtime/chan.go, runtime/proc.go, runtime/time.go.
• LMAX Disruptor papers: lock-free queues.
• Postgres source: WAL, checkpoint, replication, query planner.
• Kafka source: producer, consumer, broker.
• The OpenTelemetry SDK source: production-grade batchers.
• Brendan Gregg's tools: flame graphs, profiling.

Read code more than blog posts. The truth is in the source.

Detailed Failure Mode Analysis¶

For the production batcher above, let us enumerate failure modes.

Failure: Sink returns transient error¶

Behavior: flushFail counter increments. The batch is logged but not retried.

Mitigation: wrap the Sink in a RetryingSink decorator.

Failure: Sink panics¶

Behavior: the flushWorker goroutine dies. No recover in flushWorker. Pending and future batches are not flushed.

Mitigation: add recover in flushWorker:

func (b *Batcher[T]) flushWorker(id int) {
    defer func() {
        if r := recover(); r != nil {
            b.cfg.Logger.Error("flush worker panic", "worker", id, "panic", r)
            // Restart self.
            go b.flushWorker(id)
        }
    }()
    for job := range b.flushReq { ... }
}

Failure: Multiple flushWorkers panic¶

Behavior: all die; no flushes possible. b.flushReq fills; run loop blocks on handoff.

Mitigation: supervisor goroutine that restarts flushWorkers.

Failure: b.flushReq fills¶

Behavior: run loop blocks on b.flushReq <- job. Producers' Add blocks on the input channel.

Mitigation: monitor len(b.flushReq). Alert at 80%. The right answer is to scale flushers or fix the sink.

Failure: Input channel b.in fills¶

Behavior: Add blocks (or returns ctx.Err() with timeout). Producers slowed.

Mitigation: alert on queue_depth > 80%. Investigate downstream.

Failure: Shutdown timeout¶

Behavior: items in b.in and buf are lost; flushReq may have unsent batches; droppedOnShutdown counter does not capture all losses.

Mitigation: longer shutdown context; or accept the trade-off and document.

Failure: Race in Add-during-shutdown¶

The double-check pattern in Add should handle it: closeCh check first, then send. But a sufficiently-quick race could still panic.

Mitigation: ensure orchestrator stops producers before calling Shutdown. The race exists but is bounded.

Failure: GC pause during flush¶

Behavior: flush latency spikes. Other goroutines waited too. Producers see brief stall.

Mitigation: tune GOMEMLIMIT; profile and reduce allocations.

Failure: Network partition¶

Behavior: sink calls time out. flushFail accumulates. eventually circuit breaker opens (if present).

Mitigation: circuit breaker + DLQ + retry strategy.

Failure: Hard process kill (SIGKILL)¶

Behavior: everything in memory is gone. b.in items, buf items, in-flight batches, all lost.

Mitigation: upstream replay (Kafka with retention) or persistent buffer (disk WAL).

A Detailed Walk-Through: Implementing the Complete Production Batcher¶

Putting everything from junior, middle, senior, and professional together. The final code:

package batcher

import (
    "context"
    "errors"
    "log/slog"
    "runtime"
    "sync"
    "sync/atomic"
    "time"
)

var ErrClosed = errors.New("batcher: closed")

type Sink[T any] interface {
    Write(ctx context.Context, batch []T) error
}

type Config[T any] struct {
    Name          string
    MaxBatchSize  int
    MaxBatchDelay time.Duration
    QueueDepth    int
    FlushTimeout  time.Duration
    Flushers      int
    Sink          Sink[T]
    Logger        *slog.Logger
}

func (c *Config[T]) validate() error {
    if c.Sink == nil {
        return errors.New("batcher: Sink is required")
    }
    if c.MaxBatchSize <= 0 {
        return errors.New("batcher: MaxBatchSize must be positive")
    }
    if c.MaxBatchDelay <= 0 {
        return errors.New("batcher: MaxBatchDelay must be positive")
    }
    if c.QueueDepth <= 0 {
        c.QueueDepth = 1024
    }
    if c.FlushTimeout <= 0 {
        c.FlushTimeout = 5 * time.Second
    }
    if c.Flushers <= 0 {
        c.Flushers = 1
    }
    if c.Logger == nil {
        c.Logger = slog.Default()
    }
    return nil
}

type flushJob[T any] struct {
    batch  []T
    reason string
}

type Batcher[T any] struct {
    cfg       Config[T]
    in        chan T
    flushReq  chan flushJob[T]
    done      chan struct{}
    closeOnce sync.Once
    closeCh   chan struct{}
    pool      sync.Pool

    enqueued          atomic.Int64
    flushedOK         atomic.Int64
    flushedFail       atomic.Int64
    droppedOnShutdown atomic.Int64
}

func New[T any](cfg Config[T]) (*Batcher[T], error) {
    if err := cfg.validate(); err != nil {
        return nil, err
    }
    b := &Batcher[T]{
        cfg:      cfg,
        in:       make(chan T, cfg.QueueDepth),
        flushReq: make(chan flushJob[T], cfg.Flushers*2),
        done:     make(chan struct{}),
        closeCh:  make(chan struct{}),
        pool: sync.Pool{
            New: func() interface{} {
                s := make([]T, 0, cfg.MaxBatchSize)
                return &s
            },
        },
    }
    go b.run()
    for i := 0; i < cfg.Flushers; i++ {
        go b.flushWorker(i)
    }
    return b, nil
}

func (b *Batcher[T]) Add(ctx context.Context, item T) error {
    select {
    case <-b.closeCh:
        return ErrClosed
    default:
    }
    select {
    case b.in <- item:
        b.enqueued.Add(1)
        return nil
    case <-ctx.Done():
        return ctx.Err()
    case <-b.closeCh:
        return ErrClosed
    }
}

func (b *Batcher[T]) TryAdd(item T) bool {
    select {
    case <-b.closeCh:
        return false
    default:
    }
    select {
    case b.in <- item:
        b.enqueued.Add(1)
        return true
    default:
        return false
    }
}

func (b *Batcher[T]) run() {
    defer func() {
        if r := recover(); r != nil {
            b.cfg.Logger.Error("batcher run loop panic", "panic", r)
        }
        close(b.flushReq)
    }()

    buf := make([]T, 0, b.cfg.MaxBatchSize)
    var timer *time.Timer
    var timerC <-chan time.Time

    handoff := func(reason string) {
        if len(buf) == 0 {
            return
        }
        ptr := b.pool.Get().(*[]T)
        batch := (*ptr)[:0]
        batch = append(batch, buf...)
        select {
        case b.flushReq <- flushJob[T]{batch: batch, reason: reason}:
        case <-b.closeCh:
            // Drop on close-during-handoff; should be rare.
        }
        buf = buf[:0]
        if timer != nil {
            timer.Stop()
            timer = nil
            timerC = nil
        }
    }

    armTimer := func() {
        if timer == nil {
            timer = time.NewTimer(b.cfg.MaxBatchDelay)
            timerC = timer.C
        }
    }

    for {
        select {
        case item, ok := <-b.in:
            if !ok {
                handoff("shutdown")
                b.droppedOnShutdown.Add(int64(len(buf)))
                return
            }
            buf = append(buf, item)
            if len(buf) == 1 {
                armTimer()
            }
            if len(buf) >= b.cfg.MaxBatchSize {
                handoff("size")
            }
        case <-timerC:
            handoff("time")
        }
    }
}

func (b *Batcher[T]) flushWorker(id int) {
    for job := range b.flushReq {
        ctx, cancel := context.WithTimeout(context.Background(), b.cfg.FlushTimeout)
        err := b.cfg.Sink.Write(ctx, job.batch)
        cancel()
        if err != nil {
            b.flushedFail.Add(int64(len(job.batch)))
            b.cfg.Logger.Error("flush failed", "worker", id, "size", len(job.batch), "reason", job.reason, "err", err)
        } else {
            b.flushedOK.Add(int64(len(job.batch)))
        }
        // Return slice to pool.
        job.batch = job.batch[:0]
        b.pool.Put(&job.batch)
    }
    close(b.done)
}

func (b *Batcher[T]) Shutdown(ctx context.Context) error {
    b.closeOnce.Do(func() {
        close(b.closeCh)
        close(b.in)
    })
    select {
    case <-b.done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

type Stats struct {
    Enqueued          int64
    FlushedOK         int64
    FlushedFail       int64
    DroppedOnShutdown int64
    QueueDepth        int
}

func (b *Batcher[T]) Stats() Stats {
    return Stats{
        Enqueued:          b.enqueued.Load(),
        FlushedOK:         b.flushedOK.Load(),
        FlushedFail:       b.flushedFail.Load(),
        DroppedOnShutdown: b.droppedOnShutdown.Load(),
        QueueDepth:        len(b.in),
    }
}

// Compile-time check.
var _ runtime.Func // keep import

This is approximately 200 lines and implements:

• Both triggers (size, time).
• Pipeline architecture (1+ flush workers).
• Per-flush context with timeout.
• Defensive copy via sync.Pool.
• Idempotent shutdown.
• Per-Add context.
• Try-add for non-blocking.
• Atomic stats counters.
• Panic recovery.
• Generic over T.

Use this as the starting point for any production batcher. Add retries, metrics, tracing as decorators around the Sink.

Batcher Patterns Outside Go¶

For context, how do other languages handle this?

Java¶

java.util.concurrent.LinkedBlockingQueue plus a worker pool. ScheduledExecutorService for time triggers.

Spring Batch is a heavy-weight framework for batch jobs (mostly nightly, not micro-batching).

Rust¶

tokio::sync::mpsc::channel, tokio::time::interval for time triggers. Very similar pattern to Go.

Python¶

asyncio.Queue, asyncio.create_task for periodic flush. The GIL limits throughput but the shape is the same.

Node.js¶

EventEmitters + timers. Bulk write to Redis or Kafka via libraries.

Erlang/OTP¶

gen_server with a state and timeout. Idiomatic; the BEAM scheduler is light.

The pattern is universal. Each language has idioms; the algorithm is the same.

Hot Reload of Batcher Configuration¶

Production batchers need config changes without restart. Common need: dial MaxBatchSize up after a sink upgrade.

Atomic Pointer Pattern¶

type Batcher[T any] struct {
    cfg atomic.Pointer[Config]
}

func (b *Batcher[T]) UpdateConfig(c *Config) {
    b.cfg.Store(c)
}

func (b *Batcher[T]) run() {
    for {
        c := b.cfg.Load()
        // ... use c.MaxBatchSize, c.MaxBatchDelay
    }
}

Run loop reads config on each iteration. The atomic store is lock-free; the load is lock-free.

For time.Ticker interval, you cannot directly swap. Track lastInterval; if config changes, stop old ticker and start new:

if c.MaxBatchDelay != lastInterval {
    ticker.Stop()
    ticker = time.NewTicker(c.MaxBatchDelay)
    lastInterval = c.MaxBatchDelay
}

Sources of Config¶

• HTTP endpoint (POST /config).
• Etcd / Consul watch.
• File reload on SIGHUP.
• Environment variable poll (less common).

For Kubernetes services, ConfigMap mount + SIGHUP is standard.

A Note on Versioning the Batcher Library¶

If you maintain a batcher library used across teams:

Public API Stability¶

Method signatures, error types, configuration struct fields: stable.

Removing a field is a breaking change. Adding a field (with sensible default) is not.

Internal Evolution¶

The run loop, the flush strategy, the metric internals: can change between minor versions.

Deprecation Policy¶

When deprecating an API, mark with // Deprecated: comment and keep it working for at least 6 months.

When removing, bump major version.

Backward Compatibility¶

A user upgrading from v1.5 to v1.6 should not have to change code. From v1.x to v2.0: expect changes; provide migration guide.

A Final Comparison: Three Real-World Batcher Architectures¶

Let us compare three batchers in production at scale.

Architecture A: Stripe-Style Payment Auditor¶

Stripe-style: every API call's audit event must be durable. Used in financial services.

• Per-tenant batchers (large fleet).
• Postgres COPY backend.
• Single flusher per batcher (strict ordering).
• MaxBatchSize 500, MaxBatchDelay 100 ms.
• DLQ to S3.
• Drain timeout 60 s on shutdown.
• 99.999% durability SLO.

Architecture B: Datadog-Style Metrics Pipeline¶

Aggregate metrics from millions of agents.

• Per-tenant batchers (high cardinality).
• HTTP bulk endpoint.
• Many flushers per batcher (idempotent metrics).
• MaxBatchSize 5000, MaxBatchDelay 5 s.
• Drop on overflow.
• 99% durability SLO (metrics are aggregate; some loss tolerated).

Architecture C: Game Server-Style State Replicator¶

Replicate player state to a central server.

• One batcher per game server.
• gRPC streaming.
• MaxBatchSize 100, MaxBatchDelay 50 ms.
• Retry on transient.
• Drain immediately on player disconnect.
• 99.9% durability.

Common Patterns¶

All three: - Both triggers. - Bounded queues. - Reason-tagged metrics. - Graceful shutdown. - Per-tenant or per-source isolation.

Differences¶

The trade-offs reflect domain requirements: - Stripe: durability > latency. - Datadog: throughput > per-item. - Game server: latency > throughput.

Same pattern, different tunings. The architectural choice is in the configuration, not the algorithm.

Detailed Investigation: Why Are My Goroutines Leaking?¶

A real diagnostic walk.

Symptom¶

runtime.NumGoroutine() shows 50000 and rising. Heap memory growing.

Capture¶

curl localhost:6060/debug/pprof/goroutine?debug=1 > goroutines.txt

Grep for repeated stacks: sort | uniq -c | sort -rn | head.

Common Pattern¶

50000 goroutines:
    runtime.gopark
    runtime.chanrecv
    main.consumer (line 42)

50000 goroutines blocked on the same <-ch. The channel never closes.

Root Cause¶

Likely: producer goroutines that spawn for each request and forget to terminate. Or: a goroutine waiting on a context that never cancels.

Fix¶

Audit goroutine lifecycles. Every go f() should have an end condition: context cancel, channel close, exit signal.

For batchers specifically: ensure the run loop's exit is reachable. Test that Shutdown truly terminates the goroutine.

Detailed Investigation: Why Is My CPU 100%?¶

Symptom¶

CPU is pinned at 100%; throughput is the same.

Capture¶

go tool pprof -seconds=30 localhost:6060/debug/pprof/profile

In pprof: top, web.

Common Causes¶

1. Tight spin loop: for { check() } without sleep.
2. GC overload: runtime.gcMark* in top. Reduce allocations.
3. Channel contention: many producers; channel lock bouncing.
4. Lock contention: mutex profile via pprof.Lookup("mutex").

Fix Each¶

1. Add runtime.Gosched() or use a blocking primitive.
2. sync.Pool, pre-allocation, fewer allocations.
3. Shard channels.
4. Reduce critical section, use atomic, or shard.

Detailed Investigation: Why Is My Memory High?¶

Symptom¶

Heap memory growing without bound.

Capture¶

go tool pprof -inuse_space localhost:6060/debug/pprof/heap

top, web again.

Common Causes¶

1. Unbounded buffer: input channel without cap.
2. Goroutine leak: each holds stack memory.
3. Slice escape: slices that should be on stack escaping to heap.
4. Pool not draining: items pooled but never reused.

Fix Each¶

1. Bound the channel.
2. Audit goroutines as above.
3. Profile with -alloc_space; check escape analysis output.
4. Reduce pool reservation; ensure Put is called.

A Final Look at the Run Loop With All Optimisations¶

The "professional" run loop with every optimisation we've discussed:

func (b *Batcher[T]) run() {
    defer close(b.done)
    defer func() {
        if r := recover(); r != nil {
            b.cfg.Logger.Error("panic", "panic", r)
        }
    }()

    runtime.LockOSThread() // pin to OS thread
    defer runtime.UnlockOSThread()

    buf := make([]T, 0, b.cfg.MaxBatchSize)
    var timer *time.Timer
    var timerC <-chan time.Time

    flush := func(reason string) {
        if len(buf) == 0 {
            return
        }
        batch := batchPool.Get().([]T)[:0]
        batch = append(batch, buf...)
        b.flushReq <- flushJob[T]{batch: batch, reason: reason}
        buf = buf[:0]
        if timer != nil {
            timer.Stop()
            timer = nil
            timerC = nil
        }
    }

    armTimer := func() {
        if timer == nil {
            timer = time.NewTimer(b.cfg.MaxBatchDelay)
            timerC = timer.C
        }
    }

    for {
        // Bulk-drain channel.
        select {
        case item, ok := <-b.in:
            if !ok {
                flush("shutdown")
                return
            }
            buf = append(buf, item)
            if len(buf) == 1 {
                armTimer()
            }
            // Try to grab more without blocking.
            for len(buf) < b.cfg.MaxBatchSize {
                select {
                case more, ok := <-b.in:
                    if !ok {
                        flush("shutdown")
                        return
                    }
                    buf = append(buf, more)
                default:
                    goto checkSize
                }
            }
        checkSize:
            if len(buf) >= b.cfg.MaxBatchSize {
                flush("size")
            }
        case <-timerC:
            flush("time")
        }
    }
}

Optimisations applied:

• runtime.LockOSThread for scheduler stability.
• sync.Pool for batch slices.
• time.Timer for precise time trigger (not time.Ticker).
• Bulk drain on each wake (multi-item per select).
• Reason-tagged flush via channel.
• Pipeline architecture (flushReq channel).
• Panic recovery.

Result: ~50% less CPU than the naive run loop at high throughput.

Reading Real Production Code Critically¶

When reading a production batcher (yours or someone else's), questions to ask:

1. What are the triggers? All present?
2. What is the backpressure policy?
3. What does shutdown do? Is there a deadline?
4. How are errors handled?
5. What metrics are emitted?
6. How is it tested?
7. What was the last bug? What did it teach the team?

If any answer is unclear, the batcher has technical debt. Add to the backlog.

Documentation for Batchers¶

Every batcher in production should have:

A README¶

• Purpose: what does it batch?
• Configuration: what knobs?
• Metrics: what to monitor?
• Alerts: what should page?
• Runbook: how to respond to common failures?

A Design Document¶

• The shape of the implementation (channel vs lock-free vs mutex).
• The trade-offs considered.
• The rejected alternatives.

A Test Suite¶

• Unit tests for each trigger.
• Integration tests with real sink.
• Chaos tests for failure modes.
• Load tests for capacity.

Without these artifacts, the batcher is a black box that breaks at 3 AM and no one knows how to fix it.

Production Best Practices Summary¶

After everything, here are the principles:

1. Both triggers: size and time.
2. Defensive copy: never share buffer with sink.
3. Bounded queue: surface backpressure.
4. Graceful shutdown: with deadline.
5. Reason-tagged metrics: size, time, shutdown.
6. Retry layer: classify errors; backoff with jitter.
7. Dead-letter queue: for permanent failures.
8. Per-flush timeout: don't hang forever.
9. Idempotent shutdown: safe to call twice.
10. Test under load: not just unit tests.

These are not optional. A production batcher missing any of these has a known failure mode in its future.

What Makes a Batcher Great¶

After 4000+ lines, what separates a good batcher from a great one?

• Operability: someone other than the author can run it.
• Observability: failures are visible before they cause outages.
• Predictability: behavior under load is documented and verified.
• Recoverability: failures lead to recovery, not data loss.
• Documentation: a 3 AM engineer can fix it.

Greatness is rarely in the code. It is in the artifacts around the code: tests, metrics, alerts, runbooks, design docs.

A Detailed Look at Distributed Tracing for Batchers¶

Tracing across a batcher's async boundary requires careful instrumentation.

Per-Item Tracing¶

Each item's Add is wrapped in a span:

ctx, span := tracer.Start(ctx, "batcher.Add")
defer span.End()
item.TraceCtx = span.SpanContext()
batcher.Add(ctx, item)

The item carries its trace context.

Per-Batch Tracing¶

The flush creates a new span. Link to all items' spans:

batchCtx, batchSpan := tracer.Start(context.Background(), "batcher.Flush")
defer batchSpan.End()
links := make([]trace.Link, len(batch))
for i, item := range batch {
    links[i] = trace.Link{SpanContext: item.TraceCtx}
}
// Set links via tracer API
sink.Write(batchCtx, batch)

The flush span has many links; the trace UI renders fan-in.

Sampling¶

Tracing every item is expensive. Sample 1% with tail sampling:

sampler := sdktrace.TraceIDRatioBased(0.01)
provider := sdktrace.NewTracerProvider(
    sdktrace.WithSampler(sampler),
    ...
)

Or use tail sampling at the collector (Tempo, Jaeger Agent).

Span Attributes¶

Useful attributes:

• batcher.name: which batcher.
• batch.size: how many items.
• batch.reason: trigger reason.
• flush.duration: flush time.
• sink.error: error if any.

These make traces filterable.

Inside Linux Memory Management¶

For batchers using large buffers, kernel memory management matters.

Page Allocation¶

Go's allocator gets memory from the OS via mmap. Pages are 4 KB; huge pages 2 MB.

For a batcher holding 100 MB: 25K small pages or 50 huge pages.

Memory Mapping¶

Anonymous mmap returns zero-filled pages on first access (copy-on-write from a global zero page).

For a pre-allocated buffer: first writes incur page faults (~1 µs each). After warmup, zero overhead.

For huge pages (madvise(MADV_HUGEPAGE)): one page fault per 2 MB instead of per 4 KB. Faster startup for large buffers.

Swap¶

If the system swaps, performance dies. Configure for no swap on production (vm.swappiness=0).

Memory Pressure¶

The kernel kills processes under OOM. The batcher's memory should be bounded; otherwise the OOM killer picks the largest process.

Set cgroup memory limits in Kubernetes. Prefer the batcher dying explicitly (via Go's GOMEMLIMIT) to surprise OOM kills.

Profile-Guided Optimisation Details¶

Go 1.21 PGO. The compiler uses profile to: - Inline more aggressively. - Devirtualise interface calls (when one implementation dominates). - Re-order branches based on frequency.

Typical gain: 2-7%.

Workflow¶

1. Build initial binary: go build -o app.
2. Run with profile capture: go run -cpuprofile=cpu.pprof main.go.
3. Copy cpu.pprof to source dir as default.pgo.
4. Rebuild: go build -o app.

The compiler picks up default.pgo automatically.

For long-running batchers, capture profile from production. Update profile periodically (every few months).

A Real Optimisation Story¶

A team optimised a batcher from 10K/s to 100K/s. The journey:

Baseline (10K/s)¶

• Channel-based, MaxBatchSize=100, MaxBatchDelay=100ms.
• Sink: HTTP POST per batch.
• json.Marshal serialisation.
• No retries.

Step 1: Profile (no change)¶

Top: json.Marshal 40%, http.Client.Do 30%, runtime stuff 30%.

Step 2: easyjson (15K/s, +50%)¶

Replace json.Marshal with easyjson. Top: now json gone from top 10.

Step 3: sync.Pool for batch slices (18K/s, +20%)¶

Eliminate per-flush slice allocation.

Step 4: Larger MaxBatchSize (40K/s, +120%)¶

MaxBatchSize 100 -> 1000. Larger batches; sink throughput up.

Step 5: gzip compression (50K/s, +25%)¶

Compress before send. Network bandwidth was the limit; gzip 5x reduced bytes.

Step 6: HTTP/2 (55K/s, +10%)¶

Reuse one connection across many requests.

Step 7: Pipeline architecture (75K/s, +35%)¶

3 flush workers. Run loop no longer blocks on flush.

Step 8: Sharded input (100K/s, +33%)¶

4 channels by hash(item.UserID % 4). Reduced channel contention.

Summary¶

10x improvement through 8 changes. Each measured. Each addressed a specific bottleneck shown by profiles.

This is the senior+professional engineer's journey for a single component.

A Detailed Comparison: Library vs Custom Batcher¶

When to use a library vs build your own.

Libraries¶

• github.com/sourcegraph/conc/iter: simple iterators with batching.
• github.com/grafana/regexp: not a batcher; mentioned because it shows how to write performance-critical Go.
• github.com/golang-batch/batcher: a basic batcher library.

When to Use a Library¶

• Standard use case (size + time triggers, channel-based).
• Want maintenance from the community.
• No special integration needs.

When to Build Your Own¶

• Special triggers (byte size, per-tenant).
• Tight coupling with downstream (custom retry, custom DLQ).
• Performance requirements that the library does not meet.

A Hybrid¶

Use a library as the core, wrap with custom decorators. Most production batchers are this shape.

Real-Time vs Best-Effort Batching¶

A spectrum of guarantees.

Real-Time¶

• Tight latency bounds (sub-ms).
• Strict scheduling (RT priority).
• No GC pauses (custom allocator or off-heap).
• No retries (fail fast).

Examples: trading, control systems, robotics.

Soft Real-Time¶

• Loose latency bounds (10-100 ms).
• Best-effort scheduling.
• GC tolerated.
• Bounded retries.

Examples: video conferencing, game servers, ad serving.

Best-Effort¶

• No specific latency requirements.
• Standard Go runtime.
• Liberal retries.

Examples: audit logs, ETL, metrics.

Your batcher's tuning differs by class. For best-effort: simplicity. For real-time: every detail matters.

A High-Performance Logger Walkthrough¶

Consider zerolog or zap. They are themselves batchers (kind of):

• Per-call: format the log line into a buffer.
• The buffer is shared (sync.Pool).
• Periodically: flush the buffer to disk.

The batching is implicit, but the principles are the same:

• Pool buffers.
• Defer allocation.
• Bulk writes.

For very high log rates (10K+ lines/s), zerolog can sustain it with minimal overhead.

A Real Failure: Buffer Bloat at Scale¶

A failure mode from a real incident.

Setup¶

Service had a batcher with MaxBatchSize = 10000 and MaxBatchDelay = 60s. Aggressive amortisation.

The Problem¶

Postgres slowed by 10x for 5 minutes (storage issue). Flush durations jumped from 100 ms to 1 s. Then 2 s.

The batcher's input channel filled. Producers blocked. HTTP timeouts cascaded.

Within minutes, every Go service in the fleet was queueing requests behind blocked batchers.

Root Cause¶

MaxBatchDelay = 60s was too long. When the sink slowed, the batcher waited 60s per batch instead of 1s. The buffer never drained.

Fix¶

MaxBatchDelay = 5s. With faster flushes, the batcher catches up sooner.

Also: tighter pool timeouts so producers fail fast instead of blocking on Add.

Lesson¶

MaxBatchDelay is not just a latency knob — it is a recovery knob. Long delays mean long recovery from sink slowness.

For most production batchers, MaxBatchDelay between 100 ms and 5 s. Beyond 5 s only for tolerated-latency workloads (overnight ETL).

Inside Postgres' Buffer Manager¶

Postgres' shared_buffers is a fixed-size cache of disk pages. Bulk INSERTs interact with it.

Read-Then-Write¶

When a row is inserted, Postgres reads the target page (heap), modifies it, writes back.

If the page is in shared_buffers: fast (~1 µs). If not: disk read (~100 µs on SSD).

For COPY-heavy workloads, target pages are usually warm.

Dirty Page Eviction¶

shared_buffers cannot grow. When full, dirty pages are evicted to disk.

If the bgwriter (background writer) cannot keep up, foreground queries block on eviction.

Tuning: increase shared_buffers (25% of RAM), tune bgwriter (bgwriter_lru_maxpages, bgwriter_delay).

Checkpoint Storms¶

A checkpoint flushes all dirty pages. Postgres does this every checkpoint_timeout (5 min default) or when WAL exceeds max_wal_size.

During a checkpoint, foreground queries slow due to disk contention.

For audit log workloads, longer checkpoints (checkpoint_timeout = 30 min) reduce frequency.

Implications for Batchers¶

A bursty batcher can fill the WAL faster than the checkpointer drains. Solutions: - Larger max_wal_size. - Tune checkpoint_completion_target to spread writes. - Add WAL space.

Inside Postgres' Vacuum¶

Updates and deletes leave "dead tuples" — old versions of rows. Vacuum reclaims space.

Autovacuum¶

Runs in the background. Triggers based on dead tuple count.

For INSERT-heavy workloads (audit logs), few dead tuples. Autovacuum is rare.

For UPDATE/DELETE-heavy workloads, autovacuum can be busy. Tune autovacuum_max_workers, autovacuum_vacuum_cost_limit.

Implications for Batchers¶

A batcher doing INSERT ON CONFLICT DO UPDATE creates dead tuples. Significant updates = significant vacuum work.

Alternatives: - Periodic table swap (write to staging, then RENAME). - Partitioned tables (drop old partitions instead of vacuuming).

Inside Postgres' MVCC¶

Multi-version concurrency control: each row has version metadata (xmin, xmax). Reads see the version visible to their snapshot.

COPY Path¶

COPY rows are inserted with xmin = current transaction. Other transactions see them only after commit.

Snapshot Isolation¶

A transaction starts with a snapshot. It sees rows committed before snapshot start.

For long-running queries vs concurrent COPY: the query sees the pre-COPY state.

Implications¶

For analytical queries reading audit logs being written: consistent snapshot. No "torn write" issues.

A Deep Dive on select Statement Cost¶

Go's select is more expensive than a simple channel op. Let us measure.

Single Case Select¶

select {
case <-ch:
}

Essentially equivalent to <-ch. ~30-50 ns.

Two-Case Select With One Ready¶

select {
case <-ch1:
case <-ch2:
}

The runtime locks both channels, checks both, picks one. ~70-100 ns.

Two-Case Select With None Ready (Blocking)¶

The goroutine parks on both. ~150-200 ns to park; microseconds to wake.

N-Case Select¶

Each additional case adds ~30-50 ns to the steady-state cost. For N=10 select, hundreds of ns.

Default Case (Non-Blocking)¶

select {
case <-ch:
default:
}

Same cost as no-default-blocking, but does not park. ~50-80 ns.

Implications for Batchers¶

The standard batcher select has 2 cases (input + ticker). ~100 ns per iteration.

For 10M iterations/s: 1 second of CPU. The select is the bottleneck at this throughput.

Mitigations: - Reduce select cases. - Process multiple items per iteration (one select, many appends). - Lock-free queue (avoids select).

For typical batchers (< 1M items/s), select cost is negligible.

Bulk Append in the Run Loop¶

Instead of one append per item, bulk receive and append:

for {
    select {
    case item := <-b.in:
        buf = append(buf, item)
        // Try to grab more without blocking.
    drain:
        for len(buf) < b.maxSize {
            select {
            case more := <-b.in:
                buf = append(buf, more)
            default:
                break drain
            }
        }
        if len(buf) >= b.maxSize {
            flush()
        }
    case <-ticker.C:
        if len(buf) > 0 { flush() }
    }
}

Each "wake" of the run loop drains as much as is available, then either flushes or waits.

Reduces select overhead by amortising across multiple items.

When To Use¶

For ultra-high-throughput. The drain loop adds complexity.

For typical: one item per select iteration is fine.

Performance Anti-Patterns¶

Things that look fast but aren't.

Anti-Pattern A: Slice Growing in Hot Loop¶

buf := []int{}
for _, v := range source {
    buf = append(buf, v)
}

Each append may realloc + copy. For 1M items, multiple reallocs.

Fix: pre-size buf := make([]int, 0, expectedSize).

Anti-Pattern B: Map Growth¶

m := map[string]int{}
for k, v := range source {
    m[k] = v
}

Map grows via resize-copy. For 1M entries, many resizes.

Fix: pre-size m := make(map[string]int, expectedSize).

Anti-Pattern C: Reflection in Hot Path¶

reflect.ValueOf(item).Field(0) is ~50 ns. Many fields, many calls = milliseconds.

Fix: code-gen accessors, or use generics.

Anti-Pattern D: Channel of Pointers to Stack Values¶

ch := make(chan *int)
x := 5
ch <- &x // &x escapes to heap because of channel

Channels force heap allocation for pointer receives.

Fix: use values where possible.

Anti-Pattern E: Defer in Hot Loop¶

for _, item := range source {
    func() {
        defer cleanup()
        process(item)
    }()
}

Defer has ~30-50 ns overhead. For 1M iterations, 30-50 ms.

Fix: hoist cleanup out of the loop.

Anti-Pattern F: Capturing Loop Variable¶

for i, item := range items {
    go func() { _ = items[i] }() // <-- in pre-1.22, i is shared
}

Pre-Go 1.22, i is shared. Go 1.22+ fixes per-iteration scoping.

For Add-style batchers, this is a non-issue (no per-iteration goroutine).

Reading Kafka's Sender Goroutine¶

In franz-go, the sender (internal/sticky.go and pkg/kgo/sink.go) runs per-broker:

1. Wait for produce records to be ready (sized or lingered).
2. Build a produce request: for each topic-partition, include records up to limits.
3. Compress.
4. Send to broker.
5. On response, fire callbacks per record.

Per-Broker Sticky Partitioning¶

If you don't specify a partition, franz-go uses sticky partitioning: all records to one partition until full, then rotate. This maximises batch size per partition.

Idempotent Producer¶

enable.idempotence=true (default in newer versions): - Producer ID assigned by broker. - Each record has a producer ID + sequence number. - Broker dedupes on (producer ID, sequence).

Allows retries without duplicates. Lower throughput due to coordination overhead.

Transactional Producer¶

Beyond idempotent, transactional producer guarantees atomicity across partitions and topics. Used for read-modify-write to Kafka.

Implementation: a coordinator broker manages transaction state. Pre-commit fence + commit message.

Application-level: client.BeginTransaction(), send records, CommitTransaction().

For ETL pipelines that consume from Kafka, process, write to Kafka with exactly-once: use transactional producer with idempotent consumer.

A Note on runtime.Gosched¶

runtime.Gosched() yields the current goroutine. Useful in spin loops:

for !done() {
    runtime.Gosched()
}

Without Gosched, a tight spin loop on one P starves other goroutines on that P.

For lock-free batchers (spinning until ring has data), Gosched is the spin-back.

Alternative: time.Sleep(0) is similar but goes through the runtime's sleep path. Gosched is preferred for tight loops.

Memory Barriers in Go¶

Go's memory model is described in go.dev/ref/mem. For batchers:

• Channel send/receive synchronises memory.
• sync.Mutex.Unlock happens-before sync.Mutex.Lock.
• atomic operations have specific semantics; atomic.Store followed by atomic.Load synchronises.

For lock-free batchers, careful use of atomic is required. Go's atomics are sequentially consistent by default; that is stronger than needed but easier to reason about.

For weaker memory ordering (acquire-release), use Go's sync/atomic package since 1.19, which has Load/Store with no fence options — but the standard ones include necessary fences.

Generics in Hot Paths¶

Go 1.18+ generics. For batchers, Batcher[T any] avoids the interface{} boxing.

Cost¶

Generic functions are specialised per type at compile time. No runtime overhead.

The compiler generates one copy per "GC shape" (size class + pointer pattern). For Batcher[int64] and Batcher[string], two copies.

Limitations¶

Cannot use methods on type parameters without constraints. For sinks, the constraint is Sink[T any] interface.

When to Use¶

For library-style batchers reused across types: generics. For one-off batchers: concrete types.

Reading Real Production Code: HashiCorp's Memberlist Batching¶

HashiCorp's memberlist (gossip protocol) batches network messages:

• Per-peer outbound queue.
• Coalesces multiple messages into one UDP packet.
• Acks batched.

Code: github.com/hashicorp/memberlist. Reading shows another shape of batching — for low-latency UDP.

Reading Real Production Code: Caddy's HTTP/3 Send Coalescing¶

Caddy's HTTP/3 implementation coalesces small writes:

• Connection has a write buffer.
• Writes accumulate until 1500 bytes (MTU) or 1 ms passes.
• Flush to UDP.

Similar to TCP Nagle but for QUIC.

Reading Real Production Code: Cassandra's Java Driver¶

Cassandra's Java driver batches: - Per-connection request batching (multiple queries in one frame). - Driver-side throttling (max in-flight). - Per-coordinator routing.

Not directly Go but conceptually similar.

Inside io_uring for Batch I/O¶

Linux 5.1+ provides io_uring for batched async I/O. From a Go batcher perspective:

Setup¶

import "github.com/iceber/iouring-go"

iour, _ := iouring.New(1024) // 1024 entry submission queue
defer iour.Close()

Submit Batched Writes¶

var requests []iouring.PrepRequest
for i, batch := range batches {
    requests = append(requests,
        iouring.Write(fds[i], batch, uint64(i)))
}
ch, _ := iour.SubmitRequests(requests, nil)

One submission carries many writes. The kernel processes them asynchronously.

Wait for Completions¶

for i := 0; i < len(requests); i++ {
    res := <-ch
    if res.Err != nil {
        // handle
    }
}

Throughput¶

For thousands of small writes: io_uring is 2-10x faster than per-write syscall.

For larger writes (network sinks): not much advantage; one write is fine.

Use Cases¶

• Custom databases (we batch writes to many files).
• Log shippers (many output destinations).
• Network proxies (many connections).

For typical app-level batchers writing to one downstream: io_uring is overkill.

Custom Codecs for Hot Paths¶

For the hot serialisation path:

Pre-Calculated Sizes¶

If item size is known, pre-allocate the buffer:

size := 0
for _, item := range batch {
    size += sizeOf(item)
}
buf := make([]byte, 0, size)
for _, item := range batch {
    buf = encode(buf, item)
}

Avoids reallocations.

Code-Gen¶

go:generate to produce type-specific encoders. Used by easyjson, vtprotobuf, gogoproto.

Hand-written codec for hot types: faster still. Trade-off: maintenance.

Zero-Copy¶

For binary protocols, pass unsafe.Pointer to fields:

binary.LittleEndian.PutUint64(buf, *(*uint64)(unsafe.Pointer(&item.ID)))

Skips type conversion. Microseconds saved per call; over 1M calls, milliseconds.

Use only when profile shows codec as hot.

Building a Lock-Free MPMC Ring Buffer in Go¶

A complete implementation based on Vyukov's design:

package lockfree

import (
    "runtime"
    "sync/atomic"
)

type slot[T any] struct {
    seq atomic.Uint64
    val T
}

type Ring[T any] struct {
    buf      []slot[T]
    mask     uint64
    head     atomic.Uint64 // producer
    _pad1    [56]byte
    tail     atomic.Uint64 // consumer
    _pad2    [56]byte
}

func NewRing[T any](size uint64) *Ring[T] {
    if size & (size - 1) != 0 {
        panic("size must be power of 2")
    }
    r := &Ring[T]{
        buf:  make([]slot[T], size),
        mask: size - 1,
    }
    for i := range r.buf {
        r.buf[i].seq.Store(uint64(i))
    }
    return r
}

func (r *Ring[T]) Enqueue(v T) bool {
    var s *slot[T]
    pos := r.head.Load()
    for {
        s = &r.buf[pos & r.mask]
        seq := s.seq.Load()
        diff := int64(seq) - int64(pos)
        if diff == 0 {
            if r.head.CompareAndSwap(pos, pos+1) {
                break
            }
        } else if diff < 0 {
            return false // full
        } else {
            pos = r.head.Load()
        }
    }
    s.val = v
    s.seq.Store(pos + 1)
    return true
}

func (r *Ring[T]) Dequeue() (T, bool) {
    var zero T
    var s *slot[T]
    pos := r.tail.Load()
    for {
        s = &r.buf[pos & r.mask]
        seq := s.seq.Load()
        diff := int64(seq) - int64(pos+1)
        if diff == 0 {
            if r.tail.CompareAndSwap(pos, pos+1) {
                break
            }
        } else if diff < 0 {
            return zero, false // empty
        } else {
            pos = r.tail.Load()
        }
    }
    val := s.val
    s.seq.Store(pos + r.mask + 1)
    return val, true
}

Properties¶

• Multi-producer, multi-consumer.
• Lock-free (CAS-based).
• O(1) Enqueue and Dequeue.
• ~20-50 ns per op.
• Cache-friendly (slot includes its sequence number).

Use With a Batcher¶

type LFBatcher[T any] struct {
    ring    *Ring[T]
    sink    Sink[T]
    maxSize int
    quit    chan struct{}
}

func (b *LFBatcher[T]) Add(v T) bool {
    return b.ring.Enqueue(v)
}

func (b *LFBatcher[T]) run() {
    buf := make([]T, 0, b.maxSize)
    backoff := 0
    for {
        select {
        case <-b.quit:
            return
        default:
        }
        for len(buf) < b.maxSize {
            v, ok := b.ring.Dequeue()
            if !ok {
                break
            }
            buf = append(buf, v)
        }
        if len(buf) > 0 {
            b.sink.Write(context.Background(), buf)
            buf = buf[:0]
            backoff = 0
        } else {
            backoff++
            if backoff > 100 {
                runtime.Gosched() // yield CPU
                backoff = 0
            }
        }
    }
}