conveyor

package module
v0.17.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 3, 2026 License: MIT Imports: 8 Imported by: 0

README

Go Reference test list

The lib proposes a new way of building pipelines in Go orthogonal to classical goroutines + channels approach. You code the path of your data item through the steps instead of coding loops that execute the same steps for all items.

It's optimized and well suited for batch processing of messages from brokers (Kafka, RabbitMQ, etc.) but can be used for any pull-based processing.

go get github.com/cardinalby/go-conveyor

Concept

Basic example

Suppose we have a typical 3-step pipeline that:

  • read stage: consumes messages from Kafka / RabbitMQ and collects them in a batch
  • write stage: writes some data from the batch to a database
  • commit stage: acknowledges the message / commits offsets on the broker

1. A gopher who doesn't know about pipelines would

Spin 1 goroutine with fn in loop that executes steps one by one:

// (1) read messages
// (2) write to DB                    (nobody reads messages now - underutilized)
// (3) commit offsets / ack messages  (nobody writes to DB now - underutilized)

1-goroutine worker

  • fn is responsible for processing 1 batch of messages (1 "item")
  • Easy to read and write, but:
    • ⚠️ While one of the steps is busy, the other steps are idle and resources are underutilized.

2. A good gopher would:

  • spin 3 goroutines with 3 functions, each:
    • is responsible for one stage
    • loops over the input
  • connect goroutines with channels between stages
  • don't forget to provide the proper shutdown logic:
    • wait for all goroutines to finish
    • close the channels in the right order

The picture shows the saturated pipeline:

  • batch1 reached "commit" stage
  • batch2 reached "write" stage
  • batch3 entered "read" stage

simple 3-step classical pipeline

3. A "conveyor" user would:

  • create Conveyor, define stages (steps)
  • modify fn (from 1st example) by adding stage.MoveTo calls between stages
  • delegate the dispatching, goroutines and shutdown management to the conveyor using this single "ItemProcessor" fn
c := conveyor.New()
// "read" stage is implicitly created with items limit = 1
writeStage := c.AddStage()
commitStage := c.AddStage()

itemProcessor := func(ctx context.Context) error {
    // itemProcessor is started only once "read" stage is available 
    
    // (1) read messages:
    // batch := ...
    
    // Atomic:
    // - wait and enter "write" stage (once available) 
    // - exit "read" stage making it available for the next batch
    if err := writeStage.MoveTo(ctx); err != nil {
        // happens on ctx cancellation or conveyor shutdown
        return err 
    }
    
    // (2) write to DB or return an error
    // for item := range batch { ... }
    
    // - enter "commit", 
    // - release "write" stage
    if err := commitStage.MoveTo(ctx); err != nil {
        return err
    }
    
    // (3) commit offsets / ack messages or return an error:
    // broker.Ack(batch)
    
    return nil
    // Conveyor can reuse the goroutine for the next processor call
}

// - starts 1st itemProcessor immediately
// - starts new processors once "read" stage is released
shutdownErr := c.Run(ctx, itemProcessor)

simple 3-step conveyor

Stages are smart locks with ordered admission:

  • MoveTo is atomic: it waits for the stage to be available and releases the previous stage providing backpressure
  • Deadlock-free: you can move only forward to the next stage, not backward

You still have a single function that processes a single batch, but the Conveyor takes care of:

  • dispatching new items (batches) to the next available goroutine from the dynamic pool
  • graceful shutdown of the pipeline
  • observability, dynamic concurrency limits and queue sizes for stages, etc.

Pros

  1. Code the path of a single batch ("item") through the stages:
    • easier to write / read / debug / trace / reason about
    • less wiring (no channels, no goroutines, no loops in user code)
    • easy to upgrade from existing "1-goroutine" version (just add MoveTo calls)
  2. If "stage1" produces some data for "stage3", use it directly at "stage3":
    • you don't need to pass it through "stage2" like you would do in channels
    • you are inside a single function for a single item/batch and keep the batch processing state
  3. You can skip some stages with natural control flow (if/else, return):
    • with channels, you would need to send a "skip" signal through the channels
    • with channels, it's complicated to send data from "stage1" to "stage3" if "stage2" is a fan-out stage
  4. Supports deadlock-free scatter-gather stages whose work may grow while it runs: schedule in rounds, or let a task schedule its own follow-ups
    • SetBackpressure chooses when an item entering a fan-out releases the previous stage: on entry, at the first task start, or once every branch of its initial work has started a task (or ran out of tasks)
  5. Dynamic concurrency limits and queue sizes for stages (gracefully adjusted in runtime)
  6. Goroutines pool is dynamically adjusted (goroutines are reused for new items)
  7. Graceful shutdown happens naturally, just respect the context:
    • If item N fails:
      • the conveyor stops spawning new items
      • items with number > N receive a context cancellation error and can't enter any stage
      • items with number < N are allowed to finish (by default)
    • If the context passed to c.Run() is canceled:
      • all items that are already in the pipeline are allowed to finish (by default)
      • new items are not spawned
    • Drain timeout: cancel the remaining items after a timeout
    • UntilShutdown: drop items that have not started side effects yet
  8. Pull-based observability

Cons

  • New paradigm and API to learn
  • Not all topologies (possible with channels) can be expressed (windowing, feedback loops, splits without join, etc.)
  • Worse performance compared to channels for CPU-bound scenarios
    • The lib is optimized for IO-bound scenarios (especially if each item is a batch of messages)
    • Any significant difference is only observable if your stages take less than 100 microseconds to process an item / batch.

Features

See the docs for guides on these features:

★ Interactive demo

  • Build different conveyor topologies
  • Simulate them in browser (against lib's WebAssembly build)

Documentation

Overview

Package conveyor moves items through an ordered series of nodes, keeping their relative order while managing capacity, backpressure, and completion. The path of one item is written as a single function, the ItemProcessor.

Everyday vocabulary:

  • MoveTo advances the item into a node.
  • Schedule registers parallel work for a fan-out, before or after entering it.
  • FanOut.Wait joins the work scheduled so far without leaving the fan-out.
  • Retain (on a Stage or a FanOut) keeps the node occupied while the item moves on: a stage until the returned release is called, a fan-out until its work is done. Stage.RetainFor does the same for a background task.
  • TaskGroup.Wait waits for retained work later in the item's path.

Most processors need only MoveTo and Schedule. Adaptive rounds add FanOut.Wait; overlapping work across stages adds Retain or RetainFor.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrConveyorAlreadyRunning is returned by Run when the conveyor is already running.
	ErrConveyorAlreadyRunning = errors.New("conveyor is already running")

	// ErrInvalidContext classifies a context that cannot drive a node transition. Match it, or one of the two
	// variants below, with errors.Is.
	ErrInvalidContext = errors.New("invalid context")

	// ErrForeignContext indicates the context was never derived from an item's ctx.
	ErrForeignContext = fmt.Errorf("%w: not derived from a conveyor item's ctx", ErrInvalidContext)

	// ErrStaleContext indicates the context's owner has already finished: the item, a lane child whose callback
	// returned, or a pool task whose task group has finished.
	ErrStaleContext = fmt.Errorf("%w: the item or work it belongs to has finished", ErrInvalidContext)
)

Returned errors, inspectable with errors.Is on the value returned by Run and the node methods (MoveTo, TryMoveTo, FanOut.Schedule, FanOut.Wait, TaskGroup.Wait).

Functions

func ItemNoFromContext 

func ItemNoFromContext(ctx context.Context) (no int64, ok bool)

ItemNoFromContext returns the number of the item this context belongs to, for logging and tracing. A child item's work reports the number of the item that scheduled it. ok is false if ctx did not originate from a conveyor item.

func ShutdownCause added in v0.16.0 

func ShutdownCause(pre context.Context, err error) error

ShutdownCause maps err, returned by a call made with pre, to the reason pre stopped it: if pre is done and err is or wraps pre's own error (context.Canceled, context.DeadlineExceeded), it returns context.Cause(pre) — a ShutdownError once shutdown has begun, so returning it is an abort, not a failure. Any other err, nil included, is returned as is.

Use it in an ItemProcessor or a task for every error of a call that got an UntilShutdown context: 

pre := conveyor.UntilShutdown(ctx)
if err := db.Exec(pre, query); err != nil {
	return conveyor.ShutdownCause(pre, err)
}

func UntilShutdown added in v0.16.0 

func UntilShutdown(ctx context.Context) context.Context

UntilShutdown returns a context derived from ctx, an item's context, that is also done once shutdown of the run begins: the Run context is canceled, or an item fails. Its cause is then a ShutdownError. If shutdown has already begun, the context is born done. Its Done channel may close a moment after shutdown begins, but node methods called with it fail from the moment shutdown begins.

Use it for the part of the path that may stop at once, such as reading the input and moving into the first node with side effects; use ctx after that: 

pre := conveyor.UntilShutdown(ctx)
msg, err := reader.Fetch(pre)
if err != nil {
	return conveyor.ShutdownCause(pre, err) // an abort if pre stopped the fetch, not a failure
}
if err := write.MoveTo(pre); err != nil {
	return err // a ShutdownError once shutdown has begun
}
// from here on, use ctx

The error of a call canceled through it, e.g. one wrapping context.Canceled, is a failure if returned as is (see Conveyor.Run): pass it through ShutdownCause. An item may also go on with ctx after its UntilShutdown context is done, e.g. to finish a partial batch.

Calls with the item's own ctx return the same context. Any other ctx gets a new context that lives until ctx or the item is done, so call it once per item and derive from the result (context.WithValue(pre, ...)) instead of calling UntilShutdown(context.WithValue(ctx, ...)) in a loop.

A ctx that does not belong to a conveyor item is returned unchanged.

Types

type Branch 

type Branch interface {
	Unit

	// NewTask creates a task of one callback — one slot's worth of work on this branch. A nil fn panics.
	NewTask(fn TaskFunc) Task

	// NewTasks creates a task of count independent callbacks fn(ctx, 0)..fn(ctx, count-1), each one slot's worth
	// of work on this branch. A count of 0 yields a no-op task. A negative count panics, and so does a nil fn with
	// count > 0.
	NewTasks(count int, fn func(ctx context.Context, index int) error) Task
}

Branch is what a FanOut's work is scheduled on: a Pool or a Lane. It is the task-construction surface the two share, for code that builds work without caring which kind it got.

FanOut.Branches returns them in creation order. Type-assert to Pool or Lane for the kind's own methods.

type Conveyor 

type Conveyor interface {
	// AddStage adds a Stage to the end of the conveyor, admitting one item at a time by default. Chain SetLimit and
	// SetQueueSize to adjust its capacity, and pass WithName to name it.
	//
	// It panics if the conveyor is running or has already run.
	AddStage(opts ...NodeOption) Stage

	// AddFanOut adds a FanOut to the end of the conveyor: a node whose work runs in parallel on branches added with
	// AddPool or AddLane. It admits one item at a time by default; chain SetLimit and SetQueueSize to adjust its
	// capacity, and pass WithName to name it.
	//
	// It panics if the conveyor is running or has already run.
	AddFanOut(opts ...NodeOption) FanOut

	// Run starts the conveyor, creating one item per itemProcessor invocation until ctx is canceled or an item
	// fails. It blocks until every in-flight item has finished, then returns what began the shutdown, the first
	// event only:
	//   - an ItemError, if an item failed first. Its Unwrap is the item's own error.
	//   - a ShutdownError, if ctx was canceled first. Its Unwrap is ctx's cancellation cause.
	//   - nil, if neither happened.
	//
	// Tell them apart with a type switch on the returned error. errors.As searches the whole chain, and is for
	// inspecting causes (see RunError).
	//
	// Both are RunErrors. RunError.DrainError tells how the items in flight finished: the first failure among them,
	// or the drain timeout (WithDrainTimeout). To see every failure, log it where it happens (the
	// ItemProcessor, a task, a RetainFor task). Items aborted by the conveyor are not failures and are not
	// reported. ErrConveyorAlreadyRunning is returned as is.
	//
	// An item counts as aborted if it returns an error that is or wraps a ShutdownError (e.g. from a node method,
	// or context.Cause of an UntilShutdown context), or any error after the conveyor canceled it. Any other error
	// is a failure, also one that wraps context.Canceled: for a call canceled through UntilShutdown, return
	// ShutdownCause(pre, err) instead.
	//
	// An item that fails or is aborted cancels all younger items, so no younger item gets past it. Use UntilShutdown
	// to stop items at once when shutdown begins, before their side effects start.
	//
	// Item contexts carry ctx's values (trace spans, loggers), but not its cancellation.
	//
	// Build all nodes before calling Run; the topology is frozen from the first Run on. Run may be called again
	// after it returns, but a concurrent second call returns ErrConveyorAlreadyRunning.
	Run(ctx context.Context, itemProcessor ItemProcessor) error

	// SetItemLimit caps how many items may be in flight across the whole conveyor at once. One item is one
	// ItemProcessor call, from creation to completion, so this also bounds the number of worker goroutines. 0
	// means unlimited, the default; a negative n panics. It returns the conveyor for chaining.
	//
	// It is a global bound on top of the nodes' own limits and does not change any of them.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor. Lowering it never evicts an item
	// in flight; it only stops new items from being created until the count has fallen below the new limit.
	SetItemLimit(n int) Conveyor

	// ItemLimit returns the current cap on items in flight across the whole conveyor, or 0 if unlimited (the
	// default).
	ItemLimit() int

	// StartingStage returns the implicit stage every item starts in. The ItemProcessor code before the item's first
	// MoveTo runs in it, one item at a time, and the next item is created only when it is free. It has no MoveTo
	// and its limit is always 1.
	//
	// Use it to match the stage in Stats, or call Retain / RetainFor to keep its slot after the item moves on: the
	// next item is not created until the hold is released or that work returns.
	StartingStage() RetainableStage

	// Stats returns a snapshot of the active run's state and resets the gauge windows. Safe to call at any time,
	// from any goroutine; outside a run it reports the zero Stats.
	Stats() Stats

	// DebugUnitOccupants reports, for every unit, exactly which items occupy its body and its waiting room right
	// now. It exists for debugging and visualization; use Stats for production observability.
	//
	// Safe to call at any time, from any goroutine; outside a run it reports nil.
	DebugUnitOccupants() []UnitOccupants
}

Conveyor moves items through an ordered series of nodes, preserving their relative order. Create one with New, build the nodes, then call Run with an ItemProcessor. The conveyor runs one ItemProcessor per item, each on its own goroutine, and handles ordering, capacity, and backpressure between nodes.

A node is either a Stage (AddStage), whose code runs inline in the ItemProcessor, or a FanOut (AddFanOut), where the item schedules work onto branches (Pool or Lane) that run it in parallel. An item advances between nodes with MoveTo.

Background work started with Stage.RetainFor or FanOut.Retain is represented by a TaskGroup: wait for it with TaskGroup.Wait, or select on Finished. Wait returns the work's first error as a TaskError.

func New added in v0.17.0 

func New(options ...Option) Conveyor

New creates a conveyor with no nodes: add them with AddStage and AddFanOut, then call Run.

type DrainContextFunc added in v0.16.0 

type DrainContextFunc func(cause error) (context.Context, context.CancelFunc)

DrainContextFunc produces the context that bounds the drain of a shutdown. cause is the first item error, or the Run context's cancellation cause, whichever began the shutdown. See WithDrainContext.

type FanOut 

type FanOut interface {
	Unit

	// AddPool adds a Pool branch: a single step where a scheduled task runs to completion. Chain SetLimit to run
	// several tasks at once, and pass WithName to name it.
	//
	// It panics if the conveyor is running or has already run.
	AddPool(opts ...NodeOption) Pool

	// AddLane adds a Lane branch: a pipeline of its own interior nodes (AddStage, AddFanOut) through which each
	// piece of scheduled work travels as a child item. Pass WithName to name it.
	//
	// It panics if the conveyor is running or has already run.
	AddLane(opts ...NodeOption) Lane

	// MoveTo advances the item into this fan-out. It blocks until it is the item's turn and an item slot (SetLimit) is
	// free. Pool capacity does not gate entry.
	//
	// Work prepared with Schedule before entering becomes the item's initial batch and may start before MoveTo
	// returns. Otherwise the item enters with an empty body, and the first Schedule after entry is the initial batch.
	// SetBackpressure decides whether entering releases the previous node at once or keeps its slot until the initial
	// batch makes progress.
	//
	// The item behind cannot enter this fan-out until this item's initial batch is known (it schedules, leaves, or
	// retains); meanwhile it may wait in the fan-out's waiting room.
	//
	// Leaving a previous fan-out joins its body first (see FanOut): the call then returns the body's TaskError, if
	// any, without entering.
	//
	// It returns ErrForeignContext, ErrStaleContext, or the item's cancellation cause, whether the cancellation is
	// visible on ctx or only on the item's own context (see ItemProcessor). It panics on misuse: moving backward,
	// re-entering this node, or a node outside the item's own series.
	MoveTo(ctx context.Context) error

	// TryMoveTo is MoveTo without waiting: it enters the fan-out only if it can do so right now, and reports whether
	// it did. It bypasses the waiting room and never jumps an item already waiting there. Busy pools do not make it
	// decline; entering may keep the previous node's slot instead (see SetBackpressure).
	//
	// When entered is false nothing happened: the item stays where it is, and work prepared with Schedule stays
	// prepared for a later attempt or a MoveTo. A canceled item returns (false, its cancellation cause). Leaving a
	// previous fan-out whose tasks still run declines; if they have stopped, it joins them and returns
	// (false, the TaskError) on a task error. It panics on the same misuse as MoveTo.
	TryMoveTo(ctx context.Context) (entered bool, err error)

	// Schedule adds tasks to this item's work at this fan-out. It never blocks. Allowed callers:
	//   - the ItemProcessor, inside this fan-out with an open body (not yet left, not retained), or before entering it
	//     while the fan-out is still ahead of the item;
	//   - a task running on one of this fan-out's pools, before the task returns;
	//   - a child item of one of this fan-out's lanes, before its callback returns.
	//
	// Inside the fan-out, tasks are queued at once. Before entry, tasks are prepared: the Tasks are consumed, but
	// nothing runs until the item enters. All prepared calls together form the initial batch (see
	// FanOutBackpressure). Prepared work is discarded without running if the item passes the fan-out without entering
	// or returns, and once a canceled item calls a node method or returns; a declined TryMoveTo keeps it. Prepared
	// work does not appear in Stats.
	//
	// Preparing the work and then entering lets it start as part of the entry:
	//
	// 	if err := f.Schedule(ctx, db1.NewTask(writeDB1), db2.NewTask(writeDB2)); err != nil {
	// 		return err
	// 	}
	// 	if err := f.MoveTo(ctx); err != nil {
	// 		return err
	// 	}
	//
	// Work is queued per branch by item age, then by call order, then by argument order. Running work is never
	// preempted.
	//
	// Calling Schedule with no tasks is legal: it makes the initial batch known (and empty), which lets the item
	// behind enter.
	//
	// It returns ErrForeignContext; ErrStaleContext for the context of a finished item, a finished child, a finished
	// task group, or (before entry) a processor that has returned; the item's cancellation cause; or the TaskError of a
	// body whose task has failed. In these cases nothing is queued or prepared. It panics on misuse: a task of another
	// fan-out, a Task submitted twice, a fan-out the item has already passed, a body closed by leaving (or trying to
	// leave), Retain, or a task calling it for a fan-out it does not run under.
	Schedule(ctx context.Context, tasks ...Task) error

	// Wait blocks until every task scheduled so far in this fan-out's body has stopped, including work those tasks
	// scheduled themselves, and returns nil or their first error as a TaskError. The item keeps its slot and may
	// Schedule again afterwards, unless a task failed. Use it for rounds: wait, plan the next tasks from the results,
	// schedule again. Repeated calls return the same error again. If the conveyor canceled the item, it returns the
	// ShutdownError.
	//
	// If the item or the call context is canceled before the tasks stop, Wait returns that cause and the tasks are not
	// joined (see FanOut).
	//
	// Only the ItemProcessor (or a child item at a fan-out of its own lane) may call it, while the body is open. It
	// panics when called with a task's context (a task must never wait for other work), before entering, after leaving,
	// or after Retain.
	Wait(ctx context.Context) error

	// Retain hands the work scheduled here to the returned TaskGroup and lets the item move on without waiting for it.
	// This fan-out's slot stays held until the work is done and the item has moved on. The work may still grow from its
	// own running tasks. It is the fan-out counterpart of Stage.RetainFor: use it to overlap this fan-out's work with
	// later stages, then call TaskGroup.Wait before the step that needs the result. The work's first error is reported
	// by that Wait; if the item never waits, the error fails the item.
	//
	// After Retain the ItemProcessor may not Schedule or Wait here again. Retain does not release the previous node's
	// slot earlier than SetBackpressure allows.
	//
	// It panics on misuse: a fan-out the item does not currently occupy, or nothing to retain (never entered, already
	// retained, or the body was closed by leaving).
	Retain(ctx context.Context) TaskGroup

	// SetLimit sets how many items may be inside this fan-out at once (default 1), and returns the fan-out for
	// chaining. An item counts from entering until it moves into the next node or its waiting room; after Retain it
	// counts until the retained work is done and the item has moved on. The limit does not let the item behind enter
	// before this item's initial batch is known. A limit below 1 panics.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor.
	SetLimit(limit int) FanOut

	// SetQueueSize gives this fan-out a waiting room of size items in front of it (0, the default, means none), and
	// returns the fan-out for chaining. A negative size panics.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor.
	SetQueueSize(size int) FanOut

	// Limit returns how many items may be inside this fan-out at once.
	Limit() int

	// QueueSize returns the size of this fan-out's waiting room, or 0 if it has none.
	QueueSize() int

	// SetBackpressure selects when an item entering this fan-out releases the slot it held before: the previous node's
	// slot, or its place in this fan-out's waiting room. The default is BackpressureBalanced. See FanOutBackpressure.
	// A mode other than the three Backpressure constants panics.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor. It applies to items entering after
	// the call; items already inside keep the mode they entered with.
	SetBackpressure(mode FanOutBackpressure) FanOut

	// Backpressure returns the backpressure mode.
	Backpressure() FanOutBackpressure

	// Branches returns this fan-out's branches, pools and lanes alike, in creation order.
	Branches() []Branch
}

FanOut is a node whose work runs in parallel on branches. The item enters with MoveTo, adds tasks with Schedule (before or after entering, as often as needed), and the move that takes it out of the node waits for all of them to finish. Use it for concurrent work such as writing to several databases, or, with a Lane, to turn one item into several child journeys.

The tasks of one item at a fan-out (its body) share a context derived from the item's. The first task error cancels it: the other tasks see a canceled context, queued tasks are dropped, and Schedule returns that error. The item itself is not canceled. A join — FanOut.Wait, or the MoveTo / TryMoveTo of the next node that leaves the fan-out — waits until the tasks have stopped and returns nil, that first error as a TaskError, or the ShutdownError if the conveyor canceled the item. After a TaskError the processor decides: return it to fail the item, or handle it and go on. After a failed leave the body is closed and the item is still in the fan-out, so a second MoveTo moves on. If the processor returns without joining, the error fails the item.

A branch is either a Pool (AddPool), a single step where a task runs and is done, or a Lane (AddLane), a pipeline whose steps a task travels as a child item. Most fan-outs need only pools.

type FanOutBackpressure added in v0.12.0 

type FanOutBackpressure int

FanOutBackpressure selects when an item entering a fan-out releases the slot it held before: the previous node's slot, or its place in the fan-out's waiting room (see FanOut.SetBackpressure). Entry itself is the same in every mode: the item's turn and a free item slot (SetLimit). Pool capacity never gates entry.

The modes differ in how much of the item's initial batch must start before the previous slot is released. The initial batch is the work prepared with Schedule before entry or, if none, the first Schedule after entry. Later Schedule calls and work scheduled by running tasks never delay the release.

A task counts as started when its callback is dispatched; a lane task counts when its child item is created. A task without callbacks (NewTasks with count 0) counts for its branch as well. Work waiting behind a full pool has not started. 

const (
	// BackpressureBalanced is the default. The previous slot is released when the first task of the initial batch
	// starts on any branch, or when every branch of the batch ran out without a task. An item whose work is all
	// waiting for full pools keeps the previous node busy; an item with some work running does not. A good
	// general-purpose choice.
	BackpressureBalanced FanOutBackpressure = iota

	// BackpressureBuffered releases the previous slot on entering, like a plain stage. More items wait inside the
	// fan-out (within its limit) and upstream feels the pressure later. Use it when item N+1 should reach an idle pool
	// while item N waits for a busy one.
	BackpressureBuffered

	// BackpressureStrict releases the previous slot when every branch of the initial batch has started one task or ran
	// out. An item with any initial branch fully waiting keeps the previous slot; a younger item whose work can start
	// may still enter and run ahead of it on the branches. Fewer items in flight, at the cost of throughput when task
	// durations differ. Use it when per-item memory or upstream resources are costly.
	BackpressureStrict
)

func (FanOutBackpressure) String added in v0.12.0 

func (m FanOutBackpressure) String() string

String names the mode for messages and debugging.

type Gauge 

type Gauge struct {
	Min  int
	Max  int
	Last int
}

Gauge is a windowed view of an integer quantity: the minimum and maximum it reached since the previous Stats read, plus its value at that read.

type ItemError added in v0.15.0 

type ItemError interface {
	RunError
	// Unit is the node the item was in when it failed. For an item waiting in the queue in front of a node, it is
	// that node.
	Unit() Unit
	// contains filtered or unexported methods
}

ItemError is returned by Run when an item failed and the Run context was not canceled before. Its Unwrap gives the item's own error, so errors.Is and errors.As work with it. The item's error is the one the ItemProcessor returned, or a TaskError of tasks it never joined.

Its Error text names the node: "item failed at write: db timeout". When the item's error is a TaskError of that same node, the node is not repeated: "item failed: write task: db timeout".

type ItemProcessor 

type ItemProcessor func(ctx context.Context) error

ItemProcessor processes one item, moving it through the conveyor's nodes with Stage.MoveTo / FanOut.MoveTo using the context it receives. It need not enter every node and may return early.

Returning an error shuts the conveyor down: no new items are created, later items are canceled, and earlier ones are allowed to finish. An item aborted during the shutdown cancels all younger items too. Use UntilShutdown for the part of the path that should stop at once when shutdown begins.

A task error does not cancel the item: the join that reports it (TaskGroup.Wait, FanOut.Wait, or the MoveTo that leaves a fan-out) returns a TaskError, and the processor decides. Return it, or any error, to fail the item; return nil to go on, e.g. after sending the message to a dead-letter queue. An error of tasks the processor never joined fails the item when the processor returns.

Cancellation is judged by the item's own context. Once it is canceled (a shutdown, or for a lane child the failure of its fan-out body), every node method returns the cause, even when called with a context that hides the cancellation (context.WithoutCancel). A derived context with its own deadline still works. Code that must run after cancellation can still run in plain Go once the node method has returned; it just cannot run inside a node.

type Lane 

type Lane interface {
	Branch
	RetainableStage

	// AddStage adds an interior Stage to this lane. Chain SetLimit to let several children run its code at once.
	AddStage(opts ...NodeOption) Stage

	// AddFanOut adds an interior FanOut to this lane, so a child item may itself scatter work onto its own
	// branches.
	AddFanOut(opts ...NodeOption) FanOut

	// NewTask creates a task of one callback — one child journey through this lane. A nil fn panics.
	NewTask(fn TaskFunc) Task

	// NewTasks creates a task of count independent callbacks fn(ctx, 0)..fn(ctx, count-1), each one child journey
	// through this lane, admitted in index order. A count of 0 yields a no-op task. A negative count panics, and
	// so does a nil fn with count > 0.
	//
	// The ctx each callback receives is the child's own — pass it to the lane's stages, not the item's.
	NewTasks(count int, fn func(ctx context.Context, index int) error) Task
}

Lane is a branch of a FanOut's work that is a pipeline in its own right: it has interior nodes (AddStage, AddFanOut), and each piece of scheduled work runs as a child item travelling them with the same MoveTo an ItemProcessor uses on the conveyor.

A lane's entrance admits one child at a time (there is no SetLimit), so its parallelism comes from the interior stages' own limits. Children are created in queue order (item age, then Schedule order) and enter the interior nodes in that order. A child may Schedule more work at the fan-out its lane belongs to, before its callback returns. A child's context is canceled with the other tasks of its body, and a child that fails (its callback or its own tasks) fails the body, like a pool task error (see FanOut).

A lane with no interior nodes behaves as a Pool with limit 1.

The lane's entrance is the starting stage of each child: the callback code before the child's first MoveTo runs in it. Retain hands the entrance slot to background work, so the child moves on but the lane creates no next child until that work returns. Only a child of this lane may call it; a lane with no interior nodes creates no children, so its tasks cannot.

type NodeOption added in v0.17.0 

type NodeOption interface {
	// contains filtered or unexported methods
}

NodeOption configures a node at creation (AddStage, AddFanOut, AddPool, AddLane). WithName is the only one; capacity is set separately with SetLimit and SetQueueSize.

func WithName added in v0.17.0 

func WithName(name string) NodeOption

WithName names a node, so it shows up by that name in Stats and in error messages. An unnamed node falls back to a positional name that shifts when the topology changes around it.

type Option 

type Option func(c *conveyor)

Option configures a Conveyor at creation. See New, WithDrainTimeout and WithDrainContext.

func WithDrainContext added in v0.17.0 

func WithDrainContext(f DrainContextFunc) Option

WithDrainContext bounds how long items may keep running after a shutdown begins (the Run context is canceled, or an ItemProcessor fails). Once shutdown starts, f is asked for a context; when that context is done, every in-flight item's context is canceled. f is called once per shutdown, before Run returns, even if all items have finished by then. 

return context.WithTimeout(context.Background(), 30*time.Second) // a drain timeout, then cancel
return alreadyDoneCtx, nil                                       // cancel in-flight items at once
return nil, nil                                                  // no limit: leave them to finish

Without this option or WithDrainTimeout, items are left to finish on their own. Of the two options, the last one wins.

func WithDrainTimeout added in v0.17.0 

func WithDrainTimeout(d time.Duration) Option

WithDrainTimeout lets in-flight items run for d after a shutdown begins (the Run context is canceled, or an ItemProcessor fails), then cancels their contexts. d <= 0 cancels them at once.

Without this option or WithDrainContext, items are left to finish on their own. Of the two options, the last one wins.

type Pool 

type Pool interface {
	Branch

	// SetLimit sets how many of this pool's tasks may run at once (default 1), and returns the pool for chaining.
	// A limit below 1 panics.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor.
	SetLimit(limit int) Pool

	// Limit returns how many of this pool's tasks may run at once.
	Limit() int

	// NewTask creates a task of one callback — one slot's worth of work on this pool. A nil fn panics.
	NewTask(fn TaskFunc) Task

	// NewTasks creates a task of count independent callbacks fn(ctx, 0)..fn(ctx, count-1), each one slot's worth
	// of work on this pool, built lazily as slots free up. A count of 0 yields a no-op task. A negative count
	// panics, and so does a nil fn with count > 0.
	NewTasks(count int, fn func(ctx context.Context, index int) error) Task
}

Pool is a single-step branch of a FanOut's work: a task runs on the slot it was given and is done. SetLimit(n) bounds how many of its tasks run at once.

A pool task cannot move or wait: calling MoveTo or Wait with its context panics. Use a Lane for a branch whose work is a multi-step path. The one node method a pool task may call is Schedule of the pool's own fan-out, to add follow-up work before the task returns.

Work is queued per branch by item age: a free slot never goes to a younger item's work while an older item has work queued there. Work that has started is never taken back.

type RetainableStage added in v0.13.0 

type RetainableStage interface {
	Unit

	// Retain keeps this stage's slot held until the returned release is called, letting the item move on without
	// releasing the stage. The slot is freed once release is called and the item has moved on, so a release called
	// while the item is still in the stage changes nothing. Use it to keep the next item out of the stage while this
	// item's later work (in the ItemProcessor or in its fan-out tasks) still uses the same resource.
	//
	// release may be called from any goroutine, more than once. Each Retain call holds the slot until its own
	// release is called. A hold not released by then ends when the item completes, after all its tasks finished:
	// do not call release from a goroutine you started yourself and expect it to keep the stage after the item
	// returns — use RetainFor for background work.
	//
	// On a finished item the returned release does nothing.
	//
	// It panics on misuse: a context that does not belong to an item, a handle from another conveyor, or a stage the
	// item is not in now.
	Retain(ctx context.Context) (release func())

	// RetainFor runs the task fn in the background while keeping this stage's slot held, letting the item move on
	// without releasing the stage. The slot is freed once fn returns and the item has moved on. Join fn with the
	// returned TaskGroup (before or after a later MoveTo): Wait returns its error as a TaskError, and the processor
	// decides; an error of a task never joined fails the item. Several calls run their tasks in parallel and the
	// slot is held until all of them return.
	//
	// fn gets the item's context, canceled when the item is; node calls with it panic. On a canceled item — canceled
	// on ctx or on its own context — fn does not run and the returned task group is already finished, carrying the
	// cancellation cause.
	//
	// It panics on misuse: a nil fn, a handle from another conveyor, or a stage the item is not in now.
	RetainFor(ctx context.Context, fn TaskFunc) TaskGroup
}

RetainableStage is a stage whose slot the item can keep after moving on: a Stage, or the conveyor's starting stage (Conveyor.StartingStage).

type RunError added in v0.15.0 

type RunError interface {
	error
	// Unwrap returns the trigger of the shutdown, never nil: the Run context's cancellation cause for a
	// ShutdownError, the failed item's own error for an ItemError.
	Unwrap() error
	// DrainError is nil if the items in flight when the shutdown began all finished without failing. Else it is
	// the first problem of the drain: an ItemError, if another item failed first, or the cause of the drain
	// context (see WithDrainTimeout), e.g. context.DeadlineExceeded, if the drain timed out first and the remaining
	// items were canceled. It never repeats the trigger. Items aborted by the conveyor are not failures (see
	// Conveyor.Run).
	DrainError() error
	// contains filtered or unexported methods
}

RunError describes why a run shut down and how the shutdown went. Run returns one of its two kinds: a ShutdownError when the Run context was canceled first, or an ItemError when an item failed first. Only the event that came first decides the kind. DrainError tells how the items in flight at that moment finished.

Items see a ShutdownError too, as the cancellation cause of their context when the conveyor aborts them. There, DrainError is always nil: it describes the whole shutdown and is known only when Run returns.

To tell which event began the shutdown, use a type switch on the error Run returns: 

switch e := err.(type) {
case conveyor.ItemError: // an item failed first
	log.Printf("item failed at %s: %v", e.Unit(), e.Unwrap())
case conveyor.ShutdownError: // the Run context was canceled first
	log.Printf("stopped: %v", e.Unwrap())
}

errors.As and errors.Is search the whole chain: use them to inspect causes, not to tell the kind. For example, the context cause of an item aborted because an earlier item failed is a ShutdownError that wraps that ItemError, so errors.As(cause, &itemErr) is true although this item did not fail.

The interface is sealed: only the conveyor creates these errors.

type ShutdownError 

type ShutdownError interface {
	RunError
	// contains filtered or unexported methods
}

ShutdownError is returned by Run when the Run context was canceled and no item had failed before. Its Unwrap gives the context's cancellation cause, so errors.Is(err, ErrSignalReceived{}) works.

It is also the cancellation cause of an item's context when the conveyor aborts the item: because the Run context was canceled, an earlier item failed (Unwrap gives its ItemError) or was aborted, or the drain timed out. And it is the cause of an UntilShutdown context once shutdown begins. Recover it with errors.As. errors.Is(err, context.Canceled) reaches the wrapped trigger.

Its Error text is "conveyor is shutting down: <cause>".

type Stage 

type Stage interface {
	RetainableStage

	// MoveTo advances the item into this stage, releasing the previous node. It blocks until the stage (or its
	// waiting room) has room and it is the item's turn. Leaving a fan-out joins the item's tasks there first and
	// returns their TaskError, if any, without entering (see FanOut).
	//
	// It returns ErrForeignContext, ErrStaleContext, or the item's cancellation cause — whether the cancellation is
	// visible on ctx or only on the item's own context (see ItemProcessor). It panics on misuse: moving backward,
	// re-entering an already-entered stage, or a handle from another conveyor or series.
	MoveTo(ctx context.Context) error

	// TryMoveTo is MoveTo without waiting: it enters the stage only if a slot is free right now, and reports
	// whether it did. Use it to make a stage optional under load — skip it, or take a different path — instead of
	// queueing.
	//
	// entered == false means nothing happened: the item stays where it is and the stage remains unentered, so it
	// may be tried again or entered later with a blocking MoveTo. It bypasses the stage's waiting room
	// (SetQueueSize) and never jumps an item already waiting there.
	//
	// A canceled item returns (false, its cancellation cause), whether the cancellation is visible on ctx or only on
	// the item's own context. Leaving a fan-out whose tasks still run declines; if they have stopped, it joins them
	// and returns (false, the TaskError) on a task error. It panics on the same misuse as MoveTo.
	TryMoveTo(ctx context.Context) (entered bool, err error)

	// SetLimit sets how many items may run this stage's code at once (default 1), and returns the stage for
	// chaining. Item order through the stage is only guaranteed at limit 1. A limit below 1 panics.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor; it never evicts items already
	// admitted.
	SetLimit(limit int) Stage

	// SetQueueSize gives this stage a waiting room of size items in front of it (0, the default, means none), and
	// returns the stage for chaining. An item that cannot enter the stage directly waits here instead, freeing the
	// previous node. Only MoveTo uses the waiting room; TryMoveTo never does. A negative size panics.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor; it never evicts items already
	// waiting.
	SetQueueSize(size int) Stage

	// Limit returns how many items may run this stage's code at once.
	Limit() int

	// QueueSize returns the size of this stage's waiting room, or 0 if it has none.
	QueueSize() int
}

Stage is a node whose work runs inline: the item enters it with MoveTo, runs the stage's code in the ItemProcessor body while holding the slot, then moves on. A stage is entered at most once per item.

type Stats 

type Stats struct {
	// InFlight counts the conveyor's own items — one per live ItemProcessor call. Child items of a lane's work are
	// not counted; their work shows up as branch occupancy instead.
	InFlight    Gauge
	LiveWorkers Gauge      // worker goroutines alive for root items
	Units       []UnitStat // one entry per node and branch, in creation order (index 0 is the implicit start stage)
}

Stats is a snapshot of a conveyor's runtime state, for metrics and observability. Its gauge fields report the range observed since the previous Stats call, not just a point sample.

Reading resets the windows, so wire exactly one consumer to Stats.

type Task 

type Task struct {
	// contains filtered or unexported fields
}

Task is a bundle of work for one branch, produced by a branch's constructors (NewTask, NewTasks). Submit tasks with FanOut.Schedule. The tasks of one call may belong to different branches of the same fan-out; several tasks for the same branch start in the order they are listed.

A Task is single-use: submitting the same Task twice panics.

type TaskError added in v0.16.0 

type TaskError interface {
	error
	// Unwrap returns the error the task returned, unchanged.
	Unwrap() error
	// Unit is where the task ran: the Pool or Lane of a fan-out task, the Stage or Lane of RetainFor.
	Unit() Unit
	// contains filtered or unexported methods
}

TaskError is the first error of an item's tasks — fan-out tasks, lane children, a RetainFor task — reported by a join once those tasks have all stopped: TaskGroup.Wait, FanOut.Wait, or the MoveTo that leaves a fan-out. The processor decides what it means for the item: return it (or another error) to fail the item, or nil to go on without it.

It is never an abort: tasks stopped because the conveyor canceled the item are reported with the ShutdownError itself. Any other error from a join (a canceled call context, a ShutdownError, a stale context) means the tasks are not joined yet.

type TaskFunc 

type TaskFunc = func(ctx context.Context) error

TaskFunc is one task: a piece of a branch's work, or the background work of RetainFor. It receives a context — pass it to everything it calls, and to the lane's own nodes if the task runs on a Lane — and should return promptly when canceled.

A returned error does not cancel the item. In a fan-out it cancels the context of the item's other tasks there, as errgroup.WithContext does. The join (TaskGroup.Wait, FanOut.Wait, or the MoveTo that leaves the fan-out) returns it as a TaskError, and the processor decides: return it to fail the item, or handle it and go on.

On a Pool the context can be used with Schedule of the pool's fan-out, to add follow-up work before the task returns, and with nothing else: a pool's work has nowhere to go and must not wait for other work. A RetainFor task may call no node method with its context.

type TaskGroup added in v0.16.0 

type TaskGroup interface {
	// Finished is closed once every task has finished, including work those tasks scheduled themselves, or was
	// skipped because the item was canceled, and the item has stopped adding to it.
	Finished() <-chan struct{}

	// Wait blocks until every task has finished and returns their first error as a TaskError, or nil. From then on
	// the outcome is the processor's: what it returns decides the item's result. If the conveyor canceled the item,
	// the tasks it stopped are reported with that ShutdownError instead. Only the item that started the tasks may
	// call it, with its own context or one derived from it.
	//
	// If the item or the call context is canceled first, Wait returns that cause at once and the tasks stay
	// unjoined. A canceled item never gets nil.
	//
	// It returns ErrForeignContext for a context without an item and ErrStaleContext once the item has finished. It
	// panics on misuse: a task group of another item (a lane child may wait only on its own task groups), or a task's
	// context (a task must never wait for other work).
	Wait(ctx context.Context) error
}

TaskGroup is a handle to tasks an item started in the background: a RetainFor task, or the fan-out tasks handed over with FanOut.Retain. Join them with Wait, before or after the next MoveTo (the choice decides which node's slot the item holds meanwhile). Finished can be used in a select.

The tasks' first error is reported by Wait as a TaskError, and the processor decides what it means for the item. An error of tasks the item never joined fails the item when it completes.

type Unit 

type Unit interface {
	fmt.Stringer
	// contains filtered or unexported methods
}

Unit is a handle to something with capacity: a Stage, a FanOut, a Branch (Pool or Lane), or Conveyor.StartingStage. It names the node in Stats and in error messages.

type UnitOccupants 

type UnitOccupants struct {
	Unit Unit

	// InBody lists the item numbers holding a slot in the node itself, in arrival order; an item holding several slots
	// appears once per slot. A slot may belong to an item that has already moved on: one kept by Retain, RetainFor
	// or FanOut.Retain, or the previous node's slot kept under Balanced or Strict backpressure until the
	// item's initial batch makes progress (see FanOutBackpressure). Such an item appears in two units at once.
	InBody []int64

	// InQueue lists the item numbers waiting in front of the node, in arrival order. Under Balanced or Strict an entry
	// may belong to an item already admitted to the fan-out from its waiting room. For a branch it is the backlog in
	// queue order (item age, then Schedule order), one entry per submission whose tasks are not all handed out yet: a
	// Schedule call after entry, or all calls before entry together.
	InQueue []int64
}

UnitOccupants is a point-in-time snapshot of exactly which items occupy one unit right now. See Conveyor.DebugUnitOccupants.

type UnitStat 

type UnitStat struct {
	Unit     Unit
	Occupied Gauge // slots of the node in use since the previous Stats read
	Limit    int   // the node's capacity — the denominator for Occupied
	Queued   Gauge // work waiting in front of the node since the previous Stats read
}

UnitStat is the per-node portion of Stats. Unit is the handle (a Stage, a FanOut, a Pool, a Lane, or Conveyor.StartingStage) that produced it.

Occupied and Limit describe the node itself. A slot counts while an item runs the node's code or has work outstanding there, while Retain, RetainFor or FanOut.Retain keeps it after the item moved on, and while an item admitted to a Balanced or Strict fan-out still keeps the previous node's slot (see FanOutBackpressure). So a node's occupancy may include items that are already in the next node. Work prepared with Schedule before entering a fan-out is not counted anywhere.

Queued describes what waits in front of the node: items in a waiting room (under Balanced or Strict, possibly an item already admitted to the fan-out from that room), or, for a branch, the submissions whose tasks are not all handed out yet: one per Schedule call after entry, and one for all calls before entry together. Running work is never counted, and a branch's backlog is not bounded by its fan-out's limit.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.