conveyor

README

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)

• 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

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.NewConveyor()
// "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)

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
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
     • all items that are already in the pipeline are allowed to finish
   • If the context passed to c.Run() is canceled:
     • all items that are already in the pipeline are allowed to finish
     • new items are not spawned
   • Force shutdown support
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:

• Retain previous stage longer
• Queues
• Shared Stages
• Fan-out (scatter/gather)
• Lanes: when a branch is a pipeline
• Conditional MoveTo
• Observability

★ Interactive demo

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

Documentation

Index

• Variables
• func ItemNoFromContext(ctx context.Context) (no int64, ok bool)
• type AnyUnitOption
• type Branch
• type Conveyor
  • func NewConveyor(options ...Option) Conveyor
• type FanOut
• type Gauge
• type ItemProcessor
• type Lane
• type OptName
• type Option
  • func OptShutdownContext(factory ShutdownContextFactory) Option
• type Pool
• type ShutdownContextFactory
• type ShutdownError
• type Stage
• type Stats
• type Task
• type TaskFunc
• type Tasks
  • func (ts *Tasks) Add(tasks ...Task) *Tasks
• type Unit
• type UnitOccupants
• type UnitStat
• type Wave

Variables

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 item has already finished.
	ErrStaleContext = fmt.Errorf("%w: item is finished", ErrInvalidContext)
)

Returned errors, inspectable with errors.Is on the value returned by Run and the node methods (Stage.MoveTo / FanOut.MoveTo), or on Wave.Err.

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.

Types

type AnyUnitOption

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

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

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 <= 0 yields a no-op task; a nil fn with count > 0 panics.
	NewTasks(count int, fn func(ctx context.Context, index int) error) Task

	// NewTasksGen creates a task whose callbacks are produced by gen, pulled one by one as slots free up.
	NewTasksGen(gen iter.Seq[TaskFunc]) Task

	// NewTasksChan creates a task whose callbacks are received from ch, pulled one by one as slots free up.
	NewTasksChan(ch <-chan TaskFunc) 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 reach a kind's own methods (Pool.SetLimit, Lane.AddStage).

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 OptName to name it.
	//
	// It panics if the conveyor is running or has already run.
	AddStage(opts ...AnyUnitOption) 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 OptName to name it.
	//
	// It panics if the conveyor is running or has already run.
	AddFanOut(opts ...AnyUnitOption) 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 the first item error, or else ctx's
	// cancellation cause.
	//
	// 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

	// SetItemsLimit caps how many items may be in flight across the whole conveyor at once, which in effect bounds
	// how many workers run concurrently: one worker drives one root item for its whole journey, from creation to
	// completion (see Run). A limit <= 0 means unlimited, the default. It returns the conveyor for chaining.
	//
	// Unlike a node's SetLimit, this bounds the conveyor globally, on top of whatever capacity the nodes themselves
	// admit — it does not replace per-node limits, and setting it does not change any of them.
	//
	// Safe to call at any time, from any goroutine, including on a running conveyor: raising it wakes the standby
	// worker so waiting to create the next item is picked up at once; lowering it never evicts an item already in
	// flight — it only stops new items from being created until the in-flight count has fallen below the new limit.
	SetItemsLimit(n int) Conveyor

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

	// StartUnit returns the handle of the implicit start stage that paces item creation, for matching it in Stats.
	// It is not a Stage; there is nothing to move to.
	StartUnit() Unit

	// 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 NewConveyor, 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), which schedules work onto branches (Pool or Lane) that run it in parallel. An item advances between nodes with MoveTo.

Background work started with Stage.Retain or FanOut.Detach is represented by a Wave, joined with a later MoveTo.

func NewConveyor

func NewConveyor(options ...Option) Conveyor

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

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 OptName to name it.
	//
	// It panics if the conveyor is running or has already run.
	AddPool(opts ...AnyUnitOption) 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 OptName to name it.
	//
	// It panics if the conveyor is running or has already run.
	AddLane(opts ...AnyUnitOption) Lane

	// MoveTo advances the item into this fan-out, releasing the previous node, joins the listed waves, and
	// schedules tasks onto the fan-out's branches. It returns once the tasks are scheduled, not finished: the work
	// runs in the background, and the item's next move waits for it unless it is handed off with Detach.
	//
	// Build tasks with the branches' constructors (Pool.NewTask / Lane.NewTask and their siblings) and pass them
	// as Tasks. Passing no tasks is legal.
	//
	// It returns ErrForeignContext, ErrStaleContext, or the item's cancellation cause on shutdown. It panics on
	// misuse: a task from another fan-out, resubmitting a Task, moving backward, re-entering this node, a node
	// outside the item's own series, or a wave from another item.
	MoveTo(ctx context.Context, tasks Tasks, joins ...Wave) error

	// TryMoveTo is MoveTo without waiting: it enters the node and schedules the tasks only if it can do so right
	// now, and reports whether it did.
	//
	// When entered is false nothing happened: the tasks are left unclaimed, so the same Tasks value may be
	// submitted later. It panics on the same misuse as MoveTo.
	TryMoveTo(ctx context.Context, tasks Tasks, joins ...Wave) (entered bool, err error)

	// Detach hands this fan-out's slot to the work already scheduled here, letting the item move on without
	// waiting for it to finish. Join the returned Wave in a later MoveTo to wait for it.
	//
	// It panics on misuse: a node the item does not currently occupy, or no outstanding work here to detach.
	Detach(ctx context.Context) Wave

	// SetLimit sets how many items may have work outstanding in this fan-out at once (default 1; a limit <= 0
	// means 1), and returns the fan-out for chaining. An item counts against it from the moment its work is
	// scheduled until it enters the next node.
	//
	// 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 (a size <= 0 means none), and
	// returns the fan-out for chaining.
	//
	// 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

	// 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 schedules tasks with MoveTo, and the move that takes it out of the node waits for 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.

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 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 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.

type Lane

type Lane interface {
	Branch

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

	// AddFanOut adds an interior FanOut to this lane, so a child item may itself scatter work onto its own
	// branches.
	AddFanOut(opts ...AnyUnitOption) 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 <= 0 yields a no-op task; a nil fn with count > 0
	// panics.
	//
	// 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

	// NewTasksGen creates a task whose callbacks are produced by gen, pulled one by one as the lane's entrance
	// frees up. Use it to stream work without materializing it upfront; gen should respect the ItemProcessor's
	// ctx. State it reads must not be mutated until the wave's Started channel closes. A nil gen is a no-op.
	NewTasksGen(gen iter.Seq[TaskFunc]) Task

	// NewTasksChan creates a task whose callbacks are received from ch, pulled one by one as the lane's entrance
	// frees up. The producer must run in its own goroutine, respect the ItemProcessor's ctx, and close the channel
	// when done. A nil channel is a no-op.
	NewTasksChan(ch <-chan TaskFunc) 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, using the same MoveTo an ItemProcessor uses on the conveyor.

A lane's entrance always admits one child at a time — there is no SetLimit — so its parallelism comes from the interior stages' own limits. Children are admitted to the interior nodes in the order their work was scheduled.

A lane with no interior nodes is legal but behaves as a Pool pinned at concurrency 1.

type OptName

type OptName string

OptName 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 NewConveyor and OptShutdownContext.

func OptShutdownContext

func OptShutdownContext(factory ShutdownContextFactory) Option

OptShutdownContext bounds how long items may keep running after a shutdown begins — triggered by the Run context being canceled, or by an ItemProcessor error. Once shutdown starts, the factory is asked for a context; when that context is done, every in-flight item's context is canceled.

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

Without this option, items are left to finish on their own.

type Pool

type Pool interface {
	Branch

	// SetLimit sets how many of this pool's tasks may run at once (default 1; a limit <= 0 means 1), and returns
	// the pool for chaining.
	//
	// 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 <= 0 yields a no-op task; a nil fn with
	// count > 0 panics.
	NewTasks(count int, fn func(ctx context.Context, index int) error) Task

	// NewTasksGen creates a task whose callbacks are produced by gen, pulled one by one as pool slots free up. Use
	// it to stream work without materializing it upfront; gen should respect the ItemProcessor's ctx. State it
	// reads must not be mutated until the wave's Started channel closes. A nil gen is a no-op.
	NewTasksGen(gen iter.Seq[TaskFunc]) Task

	// NewTasksChan creates a task whose callbacks are received from ch, pulled one by one as pool slots free up.
	// The producer must run in its own goroutine, respect the ItemProcessor's ctx, and close the channel when
	// done. A nil channel is a no-op.
	NewTasksChan(ch <-chan TaskFunc) 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's work has nowhere to travel, so calling MoveTo with a pool task's context panics — use a Lane for a branch whose work is a multi-step path.

Work is queued per branch in item order: everything an older item scheduled here starts before anything a younger item scheduled.

type ShutdownContextFactory

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

ShutdownContextFactory produces the context that bounds a shutdown. cause is the first item error, or the Run context's cancellation cause. See OptShutdownContext.

type ShutdownError

type ShutdownError interface {
	error
	// Cause returns the shutdown reason: the first item error, or the Run context's cancellation cause.
	Cause() error
	// contains filtered or unexported methods
}

ShutdownError is the cancellation cause of an item's context when the conveyor cancels it during a shutdown. Recover it with errors.As, or reach the wrapped cause with errors.Is (e.g. against context.Canceled).

The interface is sealed: only the conveyor creates ShutdownErrors, so matching one reliably means the conveyor canceled the item.

type Stage

type Stage interface {
	Unit

	// MoveTo advances the item into this stage, releasing the previous node, and joins the listed waves before the
	// stage's code runs. It blocks until the stage (or its waiting room) has room and it is the item's turn.
	//
	// It returns ErrForeignContext, ErrStaleContext, or the item's cancellation cause on shutdown. It panics on
	// misuse: moving backward, re-entering an already-entered stage, a handle from another conveyor or series, or
	// a wave from another item.
	MoveTo(ctx context.Context, joins ...Wave) 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.
	//
	// The joins are awaited only if the item entered. A canceled item returns (false, its cancellation cause). It
	// panics on the same misuse as MoveTo.
	TryMoveTo(ctx context.Context, joins ...Wave) (entered bool, err error)

	// Retain runs bgOp in the background while keeping this stage's slot held, letting the item move on without
	// releasing the stage. The slot is freed once bgOp returns and the item has moved on. Join the returned Wave
	// in a later MoveTo to wait for bgOp; an error from bgOp cancels the item.
	//
	// It panics on misuse: a handle from another conveyor, or a stage the item does not currently occupy.
	Retain(ctx context.Context, bgOp func() error) Wave

	// SetLimit sets how many items may run this stage's code at once (default 1; a limit <= 0 means 1), and
	// returns the stage for chaining. Item order through the stage is only guaranteed at limit 1.
	//
	// 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 (a size <= 0 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.
	//
	// 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, NewTasksGen, NewTasksChan). Collect tasks into a Tasks value and submit them with FanOut.MoveTo.

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

type TaskFunc

type TaskFunc = func(ctx context.Context) error

TaskFunc is one piece of a branch's work. It receives a context — pass it to everything it calls, and to the lane's own nodes if the branch is a Lane — and should return promptly when canceled. A returned error cancels the item.

On a Pool the context cannot be used with MoveTo: a pool's work has nowhere to go.

type Tasks

type Tasks []Task

Tasks is the collection of work passed to FanOut.MoveTo. Build it as a literal — conveyor.Tasks{s3.NewTasks(n, up), db.NewTask(idx)} — or accumulate it with Add. Its tasks may belong to different branches of the same fan-out; several tasks for the same branch run in the order they appear here.

func (*Tasks) Add

func (ts *Tasks) Add(tasks ...Task) *Tasks

Add appends tasks in place and returns the collection, for optional chaining.

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.StartUnit. 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.
	InBody []int64

	// InQueue lists the item numbers waiting in front of the node, in arrival order.
	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.StartUnit) that produced it, for matching a stat back to what was built.

Occupied and Limit describe the node itself: items running its code, with work outstanding, or occupying a branch's entrance. Queued describes what is waiting in front of it — a stage's or fan-out's waiting room, or a branch's accepted-but-not-started work.

type Wave

type Wave interface {
	// Started is closed once all the wave's work has been handed out — every task started, or the item canceled.
	// Relevant mainly for streaming sources (NewTasksGen, NewTasksChan): state they read must not be mutated until
	// it closes.
	Started() <-chan struct{}

	// Finished is closed once every task of this wave has finished, or was skipped because the item was canceled.
	// After it is closed, Err reports the outcome.
	Finished() <-chan struct{}

	// Err returns the first error the wave's work produced, or nil. It is only final once Finished is closed.
	// Reading it counts as observing the outcome.
	Err() error
}

Wave is a handle to background work an item started: a Stage.Retain operation, or work at a fan-out taken over with FanOut.Detach. Join it by naming it in a later MoveTo — commit.MoveTo(ctx, wave) — which waits for it and returns its error.

A wave whose error nobody observes still fails the item when it completes.