Documentation
¶
Overview ¶
Package batching merges concurrent writes against a narrow key space into one write per process.
It holds two shapes, and the difference between them is who waits:
GroupCommit is the blocking one. Callers submit items and block until the batch carrying them has been written, so read-your-write holds: submit, then read, and the row is there. Whatever arrives while a write is in flight rides the next one together, so however many callers are writing, exactly one statement is ever in flight.
Buffer is the non-blocking one. Callers add keys and return immediately; the keys are deduped into a pending set and flushed on an interval or when the set fills. Buffer.Take pulls keys back out of the pending set, for a caller that has to write them itself first and needs the buffered write not to race it.
The failure this exists to prevent ¶
It is not slowness. A read path that upserts one row per request puts as many concurrent INSERT … ON CONFLICT DO UPDATE statements against the same handful of popular rows as it has in-flight requests. Those statements take row locks in whatever order each caller happened to build them in, deadlock against each other (Postgres 40P01), and hold a pool connection while they do. The pool empties, and endpoints with nothing to do with that table start failing.
Merging fixes it at the root: one statement in flight, one entry per key however many callers named it, and — with WithMerge or WithOrder — one lock order. The busier the process gets, the larger the batches become and the fewer connections the write path holds, which is the opposite of how one-statement-per-caller behaves under the same load.
Two things that were learned rather than reasoned ¶
The flush does not inherit a waiter's context. A batch outlives whichever caller happened to open it, so cancelling one request must not abandon a merged write that other callers are blocked on. Both types give the flush a context of their own, bounded by WithFlushTimeout; waiters keep their own deadlines and simply stop waiting, and their items land anyway — which is the right outcome, because the work was still worth doing.
Ordering belongs to the batcher, not the write function. Lock acquisition on one total order is what turns contention into a queue instead of a deadlock cycle, and a write function that receives a map-ordered slice cannot supply that order however carefully it is written. WithMerge emits in key order for exactly this reason, and WithOrder supplies an order for batches that are not keyed.
What this is not ¶
Not a queue, not a store, and not a retry policy. The write function is the caller's, and so is what happens when it fails beyond reporting it: GroupCommit hands the error to that batch's waiters and Buffer logs and counts it, and neither retries, re-queues, or holds the items back for a second attempt. A caller that needs durability under a failing write wants a queue — see the workqueue package, which is itself built on GroupCommit.
Index ¶
- Constants
- Variables
- type Buffer
- type GroupCommit
- type Option
- func WithClock(c clock.Clock) Option
- func WithFlushInterval(d time.Duration) Option
- func WithFlushTimeout(d time.Duration) Option
- func WithLogger(logger logging.Logger) Option
- func WithMaxPending(n int) Option
- func WithMerge[T any, K cmp.Ordered](key func(T) K, merge func(existing, incoming T) T) Option
- func WithMetricsProvider(metricsProvider metrics.Provider) Option
- func WithOrder[T any](compare func(a, b T) int) Option
- func WithTracerProvider(tracerProvider tracing.Provider) Option
Examples ¶
Constants ¶
const ( // DefaultFlushTimeout bounds one flush when WithFlushTimeout is not // supplied. It is generous on purpose: the batch's waiters have already // given up their own deadlines to it, and a write that lands late still // lands. DefaultFlushTimeout = 10 * time.Second // DefaultFlushInterval is how often a Buffer flushes when // WithFlushInterval is not supplied. GroupCommit has no interval at all — // see NewGroupCommit. DefaultFlushInterval = 5 * time.Second // DefaultMaxPending is how many distinct keys a Buffer accumulates before // flushing early, when WithMaxPending is not supplied. DefaultMaxPending = 1024 )
Variables ¶
var ( // ErrNilWriteFunc indicates a nil write function was passed to NewGroupCommit // or NewBuffer. It wraps errors.ErrNilInputParameter, so a caller may check // either. // // There is no default. A batcher with nowhere to write is a batcher that // accepts everything and loses it, and every caller of Submit would be told // its items had landed. ErrNilWriteFunc = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil batch write function") // ErrClosed indicates a Submit that arrived after Close. It is returned // rather than parking the caller on a batch nothing will ever flush. ErrClosed = platformerrors.New("batcher is closed") // ErrItemTypeMismatch indicates WithMerge or WithOrder was given functions // for a type other than the batcher's. Option carries no type parameter, so // the compiler cannot catch this; the constructors report it instead. // // It is reported rather than ignored because the option that fails to apply // is the one carrying the lock ordering: a batcher that silently dropped it // would write in map order and deadlock under exactly the load it was added // to survive. ErrItemTypeMismatch = platformerrors.New("function item type does not match batcher type") )
Functions ¶
This section is empty.
Types ¶
type Buffer ¶
type Buffer[K comparable] struct { // contains filtered or unexported fields }
Buffer coalesces keys in memory and writes them on an interval or when it fills. Callers never block and never learn whether the write succeeded.
It is GroupCommit with the guarantee removed, and that is the whole choice between them: use a Buffer where the write is a side effect of serving a request — a last-seen timestamp, an access counter, a freshness marker — and nothing reads it back in the same breath. Ten thousand requests naming the same hundred keys become one statement per interval, and no request waits for it.
A Buffer owns a goroutine and must be Closed.
func NewBuffer ¶
func NewBuffer[K comparable](write func(ctx context.Context, keys []K) error, opts ...Option) (*Buffer[K], error)
NewBuffer starts a buffer that flushes through write, and begins its flusher goroutine.
A flush failure reaches no caller, because by then there is no caller: the request that added the key was answered long ago. It is logged through WithLogger and counted on the batching_buffer_errors counter, and the keys are dropped rather than held for another attempt — retrying is a policy, and a policy that lives inside a buffer nobody is watching is how a write path grows an unbounded queue in memory. A caller that cannot afford to lose the write wants GroupCommit, or a queue.
func (*Buffer[K]) Add ¶
func (b *Buffer[K]) Add(keys ...K)
Add records keys for a later flush. It never blocks, and repeats of a key already pending cost nothing — that collapse is the point, not an optimization on top of it.
Keys added after Close are dropped and counted; see Dropped. That is the one way a key can go missing without a failing write, and it means a caller kept serving requests through a component it had already shut down.
func (*Buffer[K]) Close ¶
Close stops the flusher and writes whatever it still holds, on ctx rather than on a flush timeout of its own — shutdown has a deadline the caller owns. Safe to call more than once; Add afterwards drops.
func (*Buffer[K]) Dropped ¶
Dropped reports how many keys have been discarded because they were added after Close.
func (*Buffer[K]) Pending ¶
Pending reports how many distinct keys are waiting to be written. It is a sampled level and not a synchronization point: the flusher may take them the instant it returns.
func (*Buffer[K]) Take ¶
Take removes keys from the buffer and returns those it was holding, ordered by WithOrder when one was supplied.
It is how a caller expresses an ordering dependency: a foreground write that touches the same rows takes its keys back first, so the buffered write cannot land between the caller's read and its write. What comes back is what the buffer would have written and now will not — a key it never held is simply absent, which is not an error.
If a flush is already carrying any of those keys, Take waits for it rather than returning while that write is still in flight; the wait is bounded by WithFlushTimeout, and ctx cancels it. On cancellation nothing is taken, so a caller that gets an error here has lost neither its keys nor the buffer's.
Example ¶
A Buffer's callers never block, and a caller that has to write a key itself takes it back first.
package main
import (
"context"
"fmt"
"log"
"strings"
"time"
"github.com/primandproper/platform-go/v10/batching"
)
// exampleTimeout bounds every wait in this file.
//
// Examples are executable tests that run in the package's one test binary, and
// they run it to completion: an example that blocks blocks every test after it,
// where a unit test that blocks fails only its own case. Each wait below is a
// wait on work another goroutine has to finish, so each of them is a wait that
// something else in this package could stop happening — which is exactly what
// mutation testing does to it on purpose.
const exampleTimeout = 30 * time.Second
func main() {
flushed := make(chan []string, 1)
buffer, err := batching.NewBuffer(func(_ context.Context, keys []string) error {
flushed <- keys
return nil
},
batching.WithOrder(strings.Compare),
batching.WithFlushInterval(time.Hour), // this example flushes on Close
)
if err != nil {
log.Fatal(err)
}
// Take waits out a flush that is already carrying the key it wants, and this
// is the bound the doc comment promises that wait has.
ctx, cancel := context.WithTimeout(context.Background(), exampleTimeout)
defer cancel()
defer func() { _ = buffer.Close(ctx) }()
// A read path marking what it touched, on the way out of a request handler.
buffer.Add("session-1", "session-2", "session-3")
// Something else is about to rewrite session-2 itself, and needs the
// buffered write not to land in the middle of it.
taken, err := buffer.Take(ctx, "session-2")
if err != nil {
log.Print(err)
return
}
fmt.Println("taken:", strings.Join(taken, " "))
if err = buffer.Close(ctx); err != nil {
log.Print(err)
}
// Close returns only once the write function has run, so the batch is in the
// channel by now or there was never going to be one — nothing to wait for
// either way.
select {
case keys := <-flushed:
fmt.Println("flushed:", strings.Join(keys, " "))
default:
fmt.Println("flushed: nothing")
}
}
Output: taken: session-2 flushed: session-1 session-3
type GroupCommit ¶
type GroupCommit[T any] struct { // contains filtered or unexported fields }
GroupCommit merges concurrent Submit calls into one write — group commit, the same trick a write-ahead log plays on fsync, for the same reason.
It is safe for concurrent use and is meant to be shared: one per process per thing being written. A GroupCommit per caller would merge nothing, which is the only way to hold this type and get no benefit from it.
A GroupCommit owns a goroutine and must be Closed.
Example ¶
Concurrent callers block until their own rows have landed, and land together.
package main
import (
"context"
"fmt"
"log"
"sort"
"strings"
"sync"
"time"
"github.com/primandproper/platform-go/v10/batching"
)
// exampleTimeout bounds every wait in this file.
//
// Examples are executable tests that run in the package's one test binary, and
// they run it to completion: an example that blocks blocks every test after it,
// where a unit test that blocks fails only its own case. Each wait below is a
// wait on work another goroutine has to finish, so each of them is a wait that
// something else in this package could stop happening — which is exactly what
// mutation testing does to it on purpose.
const exampleTimeout = 30 * time.Second
// view is one row of a "times seen" table, the shape a hot read path writes.
type view struct {
page string
count int
}
func main() {
var (
mu sync.Mutex
totals = map[string]int{}
)
// The write function is the caller's, and receives one merged, key-ordered
// batch per flush — which is what makes it safe to write with a single
// multi-row statement. Standing in for the table here is a map, so that the
// example's output does not depend on how the three callers interleaved.
commit, err := batching.NewGroupCommit(func(_ context.Context, rows []view) error {
mu.Lock()
defer mu.Unlock()
for _, row := range rows {
totals[row.page] += row.count
}
return nil
},
batching.WithMerge(
func(v view) string { return v.page },
func(existing, incoming view) view {
return view{page: existing.page, count: existing.count + incoming.count}
},
),
)
if err != nil {
log.Fatal(err)
}
// Submit blocks until the caller's own rows have landed, so the context it
// is given is what decides how long it can be kept waiting.
ctx, cancel := context.WithTimeout(context.Background(), exampleTimeout)
defer cancel()
defer func() { _ = commit.Close(ctx) }()
var wg sync.WaitGroup
for _, page := range []string{"/pricing", "/home", "/pricing"} {
wg.Go(func() {
if submitErr := commit.Submit(ctx, view{page: page, count: 1}); submitErr != nil {
log.Print(submitErr)
}
})
}
wg.Wait()
if err = commit.Close(ctx); err != nil {
log.Print(err)
}
// Each page was written once per flush it appeared in, never once per
// caller, and the two /pricing views were summed by the merge rather than
// racing each other.
var written []string
mu.Lock()
for page, count := range totals {
written = append(written, fmt.Sprintf("%s=%d", page, count))
}
mu.Unlock()
sort.Strings(written)
fmt.Println(strings.Join(written, " "))
}
Output: /home=1 /pricing=2
func NewGroupCommit ¶
func NewGroupCommit[T any](write func(ctx context.Context, items []T) error, opts ...Option) (*GroupCommit[T], error)
NewGroupCommit starts a batcher that flushes through write, and begins its flusher goroutine.
The batcher is deliberately not timer-driven. A flush starts as soon as the previous one finishes, so an idle process pays no latency at all and a busy one merges more the busier it gets. There is no interval to tune and no configuration that can make it wrong — which is why WithFlushInterval is a Buffer option and not one of these.
Pass WithMerge when several submissions can name the same row. Without it every submitted item is written, in arrival order, and the batcher is buying one statement per flush rather than one row per key.
func (*GroupCommit[T]) Close ¶
func (g *GroupCommit[T]) Close(ctx context.Context) error
Close stops the flusher and writes whatever was still accumulating, so a caller blocked in Submit during shutdown gets a real answer instead of hanging until its own context expires. Safe to call more than once.
The final flush runs on ctx rather than on a timeout of its own: shutdown has a deadline the caller owns, and this is the one flush with nobody else's waiters to protect.
func (*GroupCommit[T]) Pending ¶
func (g *GroupCommit[T]) Pending() int
Pending reports how many items the batch now accepting them holds — distinct keys, when WithMerge is in play. It is a sampled level and not a synchronization point: the flusher may swap the batch out the instant it returns.
func (*GroupCommit[T]) Submit ¶
func (g *GroupCommit[T]) Submit(ctx context.Context, items ...T) error
Submit adds items to the batch currently accepting them and blocks until that batch has been written.
Read-your-write holds: when Submit returns nil the items are durably where the write function put them, so a read that follows sees them. That is what makes this usable from a request handler — the caller gets the guarantee it would have had from writing itself, and the process gets one statement instead of one per handler.
A caller whose context expires stops waiting but does not cancel the flush: the batch is shared, and its other waiters still need it. Those items are therefore likely to land anyway, which is the right outcome — the work was still worth doing. What the caller loses is the confirmation, not the write, so a context error here means "I do not know", not "it did not happen".
Submitting nothing is a no-op rather than an empty flush.
type Option ¶
type Option func(*options)
Option configures a GroupCommit or a Buffer.
One type serves both, because the two shapes share most of what there is to configure — a clock, a flush timeout, an ordering, and the three observability pillars — and splitting it would mean two spellings of WithLogger. Each option below says which constructor reads it; one that the constructor has no use for is accepted and ignored, so a wiring site that builds both from one slice of options does not have to sort them out.
Option carries no type parameter even though both shapes do. Go cannot infer a type argument from a call's result type, so an Option[T] would force every call site to spell the item type out by hand — WithFlushTimeout[MyRow](d) — forever. WithMerge and WithOrder are the two options that depend on the item type; they stay generic but still need no annotation, because the type is inferable from the functions each is handed.
func WithClock ¶
WithClock swaps the clock driving a Buffer's flush interval and both shapes' latency measurement. Tests generally do not need it: under testing/synctest the default clock already runs on bubble time.
func WithFlushInterval ¶
WithFlushInterval sets how often a Buffer flushes what it has accumulated. Ignored by NewGroupCommit, which is deliberately not timer-driven.
func WithFlushTimeout ¶
WithFlushTimeout bounds one flush. Read by both constructors.
It is a fixed timeout rather than an inherited deadline on purpose: the batch outlives whichever caller opened it, so it cannot borrow one waiter's cancellation without abandoning the rest. Set it generously — a waiter that gives up first has already stopped caring, and the ones still blocked would rather the write finished.
func WithLogger ¶
WithLogger attaches a logger. It is what reports a Buffer's flush failures, which reach no caller by design — see NewBuffer.
func WithMaxPending ¶
WithMaxPending caps how many distinct keys a Buffer holds before it flushes without waiting for the interval. Ignored by NewGroupCommit.
It is a flush trigger, not a bound: Add never blocks and never drops, so a process adding faster than the write function drains will exceed it between flushes. What it prevents is an unbounded batch, not an unbounded buffer.
func WithMerge ¶
WithMerge deduplicates a GroupCommit's batch by key and emits it in key order. Ignored by NewBuffer, which dedupes by the key type itself.
merge folds an item into whatever the batch already held for that key, and is only called when there is something to fold into — the first item for a key is taken as it stands. A nil merge keeps the last item to arrive for each key.
The item and key types are inferred from key, so this needs no type argument:
batching.WithMerge(func(r row) string { return r.id }, mergeRows)
The key type must be ordered, because emitting in key order is half of what this option is for: the write function receives the batch in one total order, which is the lock ordering that keeps concurrent writers of the same rows queueing instead of deadlocking.
It must match the GroupCommit it configures; NewGroupCommit returns ErrItemTypeMismatch otherwise, since Option cannot carry the item type for the compiler to check.
func WithMetricsProvider ¶
WithMetricsProvider attaches a metrics provider, enabling the batching_group_commit_* and batching_buffer_* instruments.
The batch size histogram is the one to watch. Merging is working when it climbs with load; a histogram pinned at one under load means callers are not arriving concurrently and the batcher is buying nothing.
func WithOrder ¶
WithOrder sorts every flushed batch by compare (slices.SortFunc semantics), for batches whose order is not already decided by a merge key. Read by both constructors; for a Buffer, it also orders what Take hands back, since a caller taking keys is about to write them and wants the same lock order.
Applied on top of WithMerge rather than instead of it: merge decides which items are written, order decides in what sequence. Supplying both is how a batch keyed by one field is written in another field's order.
The item type is inferred from compare, so this needs no type argument:
batching.WithOrder(strings.Compare)
It must match the batcher it configures; the constructors return ErrItemTypeMismatch otherwise.
func WithTracerProvider ¶
WithTracerProvider attaches a tracer provider. A flush is traced as a span of its own rather than under whichever caller happened to trigger it, because that is what it is: one write on behalf of every waiter, none of whom own it.
Source Files