pipeline

README

pipeline

A Go package for building concurrent data processing pipelines using channels.

Overview

pipeline provides a set of composable stages for processing data streams concurrently. It handles context cancellation, error propagation, and goroutine lifecycle management automatically.

Stages are defined as structs in the stages package. Each stage has a Create method that integrates it into a pipeline by connecting input and output channels.

The package uses Go's runtime/trace to create trace regions for each stage, allowing for detailed performance analysis.

Installation

go get github.com/schraf/pipeline/v3

Usage

Basic Example

To create a pipeline, you define a Config and use NewPipeline. You then connect stages by calling their Create methods inside the Composer function.

package main

import (
	"context"
	"fmt"

	"github.com/schraf/pipeline/v3"
	"github.com/schraf/pipeline/v3/stages"
)

func main() {
	ctx := context.Background()

	// Transform stage: multiply by 2
	transformStage := stages.TransformStage[int, int]{
		Name:   "multiply",
		Buffer: 10,
		Transformer: func(ctx context.Context, x int) (*int, error) {
			result := x * 2
			return &result, nil
		},
	}

    // Configure the pipeline
	cfg := pipeline.Config[int, int]{
		Name:             "example",
		InputBufferSize:  10,
		OutputBufferSize: 10,
		Composer: func(composer pipeline.Composer[int, int]) {
			ctx := composer.Context()
			inputs := composer.Inputs()
			outputs := composer.Outputs()

			out := transformStage.Create(ctx, inputs.At(0))

			outputs.Link(ctx, 0, out)
		},
	}

	pipe, _ := pipeline.NewPipeline(ctx, cfg)

	pipe.Start()

	// Feed data into the pipeline
	go func() {
		defer pipe.CloseAllInputs()
		for i := 1; i <= 5; i++ {
			pipe.Inputs().Send(ctx, 0, i)
		}
	}()

	// Consume results from the last stage's output channel
	for v := range pipe.Outputs().SinkAtIter(ctx, 0) {
		fmt.Println(v)
	}

	// Wait for completion and check for errors
	if err := pipe.Wait(); err != nil {
		panic(err)
	}
}

Pipeline Configuration

The Config struct specifies the pipeline's basic parameters:

• Name: Used for tracing and identification.
• InputChannels: Number of input channels to create (defaults to 1).
• InputBufferSize: Buffer size for each input channel.
• OutputChannels: Number of output channels to create (defaults to 1).
• OutputBufferSize: Buffer size for each output channel.
• Composer: Function used to connect the pipeline inputs to outputs.

Pipeline Stages

Stages are located in the github.com/schraf/pipeline/v3/stages package.

Transform

Applies a transformation function to each value:

stage := stages.TransformStage[int, int]{
    Name:   "transform",
    Buffer: 10,
    Transformer: func(ctx context.Context, x int) (*int, error) {
        result := x * 2
        return &result, nil
    },
}
out := stage.Create(ctx, in)

Filter

Filters values based on a predicate:

stage := stages.FilterStage[int]{
    Name:   "filter",
    Buffer: 10,
    Filter: func(ctx context.Context, x int) (bool, error) {
        return x%2 == 0, nil
    },
}
out := stage.Create(ctx, in)

ParallelTransform

Applies transformation with multiple concurrent workers:

stage := stages.ParallelTransformStage[int, int]{
    Name:    "parallel",
    Buffer:  10,
    Workers: 5,
    Transformer: func(ctx context.Context, x int) (*int, error) {
        return &x, nil
    },
}
out := stage.Create(ctx, in)

Batch

Groups values into fixed-size batches:

stage := stages.BatchStage[int, []int]{
    Name:      "batch",
    Buffer:    10,
    BatchSize: 5,
}
out := stage.Create(ctx, in)

FanIn

Merges multiple input channels into one:

stage := stages.FanInStage[int]{
    Name:   "fan-in",
    Buffer: 10,
}
out := stage.Create(ctx, multiIn) // multiIn is a MultiChannelReceiver

FanOut

Distributes values to multiple output channels (broadcast):

stage := stages.FanOutStage[int]{
    Name:        "fan-out",
    OutputCount: 3,
    Buffer:      10,
}
outputs := stage.Create(ctx, in) // Returns MultiChannelReceiver

FanOutRoundRobin

Distributes values to multiple output channels in a round-robin fashion:

stage := stages.FanOutRoundRobinStage[int]{
    Name:        "fan-out-rr",
    OutputCount: 3,
    Buffer:      10,
}
outputs := stage.Create(ctx, in)

Split

Routes values to different channels based on a selector:

stage := stages.SplitStage[int]{
    Name:        "split",
    OutputCount: 3,
    Buffer:      10,
    Selector: func(ctx context.Context, x int) int {
        return x % 3
    },
}
outputs := stage.Create(ctx, in)

Reduce

Processes values incrementally using a reducer function:

stage := stages.ReduceStage[int, int]{
    Name:         "reduce",
    Buffer:       1,
    InitialValue: 0,
    Reducer: func(ctx context.Context, acc int, x int) (int, error) {
        return acc + x, nil
    },
}
out := stage.Create(ctx, in)

Flatten

Takes an input channel of slices and emits each element individually:

stage := stages.FlattenStage[int]{
    Name:   "flatten",
    Buffer: 10,
}
out := stage.Create(ctx, inSlices)

Limit

Limits the number of values passed through the stage:

stage := stages.LimitStage[int]{
    Name:   "limit",
    Buffer: 10,
    Limit:  5,
}
out := stage.Create(ctx, in)

Expand

Lazily expands single input items into multiple items using an iterator:

stage := stages.ExpandStage[int, int]{
    Name:   "expand",
    Buffer: 10,
    Expander: func(ctx context.Context, x int) iter.Seq2[int, error] {
        return func(yield func(int, error) bool) {
            yield(x, nil)
            yield(x*10, nil)
        }
    },
}
out := stage.Create(ctx, in)

Aggregate

Collects all values into a single slice:

stage := stages.AggregateStage[int]{
    Name:   "aggregate",
    Buffer: 1,
}
out := stage.Create(ctx, in) // out is <-chan []int

Error Handling

The pipeline automatically cancels all stages when an error occurs. The first error encountered is captured and returned by Wait():

if err := p.Wait(); err != nil {
    log.Fatal(err)
}

Requirements

• Go 1.24.0 or later

License

See LICENSE file for details.

Documentation

Overview

Package pipeline provides composable stages for building concurrent data processing pipelines using channels. It handles context cancellation, error propagation, and goroutine lifecycle management automatically.

Index

• type Composer
  • func (c Composer[In, Out]) Context() Context
  • func (c Composer[In, Out]) Inputs() MultiChannelReceiver[In]
  • func (c Composer[In, Out]) Outputs() MultiChannelSender[Out]
• type Config
• type Context
  • func (c Context) Go(name string, fn func(ctx context.Context) error)
• type InputChannel
• type MultiChannelReceiver
  • func NewMultiChannelReceiver[T any, C InputChannel[T]](in ...C) MultiChannelReceiver[T]
  • func (m MultiChannelReceiver[T]) At(index int) <-chan T
  • func (m MultiChannelReceiver[T]) Iter() iter.Seq[<-chan T]
  • func (m MultiChannelReceiver[T]) Len() int
  • func (m MultiChannelReceiver[T]) SinkAt(ctx context.Context, index int) []T
  • func (m MultiChannelReceiver[T]) SinkAtIter(ctx context.Context, index int) iter.Seq[T]
• type MultiChannelSender
  • func NewMultiChannelSender[T any, C OutputChannel[T]](out ...C) MultiChannelSender[T]
  • func (m MultiChannelSender[T]) At(index int) chan<- T
  • func (m MultiChannelSender[T]) Iter() iter.Seq[chan<- T]
  • func (m MultiChannelSender[T]) Len() int
  • func (m MultiChannelSender[T]) Link(ctx Context, index int, in <-chan T) error
  • func (m MultiChannelSender[T]) LinkAll(ctx Context, in MultiChannelReceiver[T]) error
  • func (m MultiChannelSender[T]) Send(ctx context.Context, index int, values ...T) error
  • func (m MultiChannelSender[T]) SendRoundRobin(ctx context.Context, values ...T) error
  • func (m MultiChannelSender[T]) SendToAll(ctx context.Context, values ...T) error
• type OutputChannel
• type Pipeline
  • func NewPipeline[In any, Out any](ctx context.Context, cfg Config[In, Out]) (*Pipeline[In, Out], context.Context)
  • func (p *Pipeline[In, Out]) CloseAllInputs()
  • func (p *Pipeline[In, Out]) Inputs() MultiChannelSender[In]
  • func (p *Pipeline[In, Out]) Outputs() MultiChannelReceiver[Out]
  • func (p *Pipeline[In, Out]) Start() error
  • func (p *Pipeline[In, Out]) Wait() error

Types

type Composer

type Composer[In any, Out any] struct {
	// contains filtered or unexported fields
}

func (Composer[In, Out]) Context

func (c Composer[In, Out]) Context() Context

func (Composer[In, Out]) Inputs

func (c Composer[In, Out]) Inputs() MultiChannelReceiver[In]

func (Composer[In, Out]) Outputs

func (c Composer[In, Out]) Outputs() MultiChannelSender[Out]

type Config

type Config[In any, Out any] struct {
	Name             string
	InputChannels    int
	InputBufferSize  int
	OutputChannels   int
	OutputBufferSize int
	Composer         func(Composer[In, Out])
}

Config defines the make up of a pipeline and is required for construction of it

type Context

type Context struct {
	context.Context
	// contains filtered or unexported fields
}

func (Context) Go

func (c Context) Go(name string, fn func(ctx context.Context) error)

type InputChannel

type InputChannel[T any] interface {
	~chan T | ~<-chan T
}

type MultiChannelReceiver

type MultiChannelReceiver[T any] []<-chan T

func NewMultiChannelReceiver

func NewMultiChannelReceiver[T any, C InputChannel[T]](in ...C) MultiChannelReceiver[T]

func (MultiChannelReceiver[T]) At

func (m MultiChannelReceiver[T]) At(index int) <-chan T

func (MultiChannelReceiver[T]) Iter

func (m MultiChannelReceiver[T]) Iter() iter.Seq[<-chan T]

func (MultiChannelReceiver[T]) Len

func (m MultiChannelReceiver[T]) Len() int

func (MultiChannelReceiver[T]) SinkAt

func (m MultiChannelReceiver[T]) SinkAt(ctx context.Context, index int) []T

func (MultiChannelReceiver[T]) SinkAtIter

func (m MultiChannelReceiver[T]) SinkAtIter(ctx context.Context, index int) iter.Seq[T]

type MultiChannelSender

type MultiChannelSender[T any] []chan<- T

func NewMultiChannelSender

func NewMultiChannelSender[T any, C OutputChannel[T]](out ...C) MultiChannelSender[T]

func (MultiChannelSender[T]) At

func (m MultiChannelSender[T]) At(index int) chan<- T

func (MultiChannelSender[T]) Iter

func (m MultiChannelSender[T]) Iter() iter.Seq[chan<- T]

func (MultiChannelSender[T]) Len

func (m MultiChannelSender[T]) Len() int

func (MultiChannelSender[T]) Link

func (m MultiChannelSender[T]) Link(ctx Context, index int, in <-chan T) error

func (MultiChannelSender[T]) LinkAll

func (m MultiChannelSender[T]) LinkAll(ctx Context, in MultiChannelReceiver[T]) error

func (MultiChannelSender[T]) Send

func (m MultiChannelSender[T]) Send(ctx context.Context, index int, values ...T) error

func (MultiChannelSender[T]) SendRoundRobin

func (m MultiChannelSender[T]) SendRoundRobin(ctx context.Context, values ...T) error

func (MultiChannelSender[T]) SendToAll

func (m MultiChannelSender[T]) SendToAll(ctx context.Context, values ...T) error

type OutputChannel

type OutputChannel[T any] interface {
	~chan T | ~chan<- T
}

type Pipeline

type Pipeline[In any, Out any] struct {
	// contains filtered or unexported fields
}

Pipeline coordinates concurrent processing stages, managing their lifecycle and propagating errors and cancellation signals across all stages.

func NewPipeline

func NewPipeline[In any, Out any](ctx context.Context, cfg Config[In, Out]) (*Pipeline[In, Out], context.Context)

NewPipeline creates a new Pipeline and a derived context for coordinating pipeline stages. The returned context is cancelled when any stage encounters an error. Use the returned Pipeline to register stages and wait for completion.

func (*Pipeline[In, Out]) CloseAllInputs

func (p *Pipeline[In, Out]) CloseAllInputs()

CloseAllInputs will close all of the input channels

func (*Pipeline[In, Out]) Inputs

func (p *Pipeline[In, Out]) Inputs() MultiChannelSender[In]

func (*Pipeline[In, Out]) Outputs

func (p *Pipeline[In, Out]) Outputs() MultiChannelReceiver[Out]

func (*Pipeline[In, Out]) Start

func (p *Pipeline[In, Out]) Start() error

func (*Pipeline[In, Out]) Wait

func (p *Pipeline[In, Out]) Wait() error

Wait blocks until all registered stages complete and returns the first error encountered by any stage, or nil if all stages completed successfully.