pipe

README

pipe is a framework for floating point signal processing. It utilizes pipeline pattern to build fast, asynchronomous and easy-to-extend pipes for sound processing. Each pipe consists of one Source, zero or multiple Processors and one or more Sinks. Phono also offers a friendly API for new pipe components implementations.

Getting started

Find examples in example repository.

Contributing

For a complete guide to contributing to pipe, see the Contribution guide

Documentation

Overview

Package pipe allows to build and execute DSP pipelines.

Concept

This package offers an opinionated perspective to DSP. It's based on the idea that the signal processing can have up to three stages:

Source - the origin of signal;
Processor - the manipulator of the signal;
Sink - the destination of signal;

It implies the following constraints:

Source and Sink are mandatory;
There might be 0 to n Processors;
All stages are executed sequentially.

Current implementation supports two execution modes: sync and async. In async mode every stage of every line is executed in by its own goroutine and channels are used to communicate between them. Sync mode allows to run one or more lines in the same goroutine. In this case lines and stages within lines are executed sequentially, as-provided. Pipe allows to use different modes in the same run.

Components

Each stage in the pipeline is implemented by components. For example, wav.Source reads signal from wav file and vst2.Processor processes signal with vst2 plugin. Components are instantiated with allocator functions:

SourceAllocatorFunc
ProcessorAllocatorFunc
SinkAllocatorFunc

Allocator functions return component structures and pre-allocate all required resources and structures. It reduces number of allocations during pipeline execution and improves latency.

Component structures consist of mutability, run closure and flush hook. Run closure is the function which will be called during the pipeline run. Flush hook is triggered when pipe is done or interruped by error or timeout. It enables to execute proper clean up logic. For mutability, refer to mutability package documentation.

Line definition and pipe

To run the pipeline, one first need to build it. It starts with a line definition:

l1 := pipe.Line{
    Source: wav.Source(reader),
    Processors: pipe.Processors(
        vst.Processor(vst2.Host{}).Allocator(nil),
    ),
    Sink: wav.Sink(writer),
}

Line defines the order in which DSP components form the pipeline. Once line is defined, components can be bound together. It's done by creating a pipe:

p, err := pipe.New(bufferSize, l1)

New executes all allocators provided by lines and binds components together into the pipe.

Execution

Once pipe is built, it can be executed. To do that Start method should be called:

errc := p.Start()
err := pipe.Wait(errc)

Start will start and asynchronously run all DSP components until either any of the following things happen: the source is done; the context is done; an error in any of the components occured.

Index

• func Run(ctx context.Context, bufferSize int, lines ...Line) error
• func Wait(errc <-chan error) error
• type ErrorRun
  • func (e *ErrorRun) Error() string
  • func (e *ErrorRun) Is(err error) bool
• type FlushFunc
• type Line
• type Pipe
  • func New(bufferSize int, lines ...Line) (*Pipe, error)
  • func (p *Pipe) AddLine(l Line) <-chan struct{}
  • func (p *Pipe) InsertProcessor(line, pos int, procAlloc ProcessorAllocatorFunc) <-chan struct{}
  • func (p *Pipe) Push(mutations ...mutable.Mutation)
  • func (p *Pipe) Start(ctx context.Context, initializers ...mutable.Mutation) <-chan error
• type ProcessFunc
• type Processor
• type ProcessorAllocatorFunc
  • func Processors(processors ...ProcessorAllocatorFunc) []ProcessorAllocatorFunc
• type SignalProperties
• type Sink
• type SinkAllocatorFunc
• type SinkFunc
• type Source
• type SourceAllocatorFunc
• type SourceFunc
• type StartFunc

Functions

func Run

func Run(ctx context.Context, bufferSize int, lines ...Line) error

Run executes the pipe in a single goroutine, sequentially.

func Wait

func Wait(errc <-chan error) error

Wait for successful finish or first error to occur.

Types

type ErrorRun

type ErrorRun struct {
	ErrExec  error
	ErrFlush error
}

ErrorRun is returned if runner was successfully started, but execution and/or flush failed.

func (*ErrorRun) Error

func (e *ErrorRun) Error() string

func (*ErrorRun) Is

func (e *ErrorRun) Is(err error) bool

Is checks if any of errors match provided sentinel error.q

type FlushFunc

type FlushFunc func(ctx context.Context) error

FlushFunc provides a hook to flush all buffers for the component or execute any other form of finalization logic.

type Line

type Line struct {
	mutable.Context
	Source     SourceAllocatorFunc
	Processors []ProcessorAllocatorFunc
	Sink       SinkAllocatorFunc
}

Line defines sequence of DSP components allocators. It has a single source, zero or many processors and single sink.

type Pipe

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

Pipe is a graph formed with multiple lines of bound DSP components.

func New

func New(bufferSize int, lines ...Line) (*Pipe, error)

New returns a new Pipe that binds multiple lines using the provided buffer size.

func (*Pipe) AddLine

func (p *Pipe) AddLine(l Line) <-chan struct{}

AddLine creates the line for provied route and adds it to the pipe.

func (*Pipe) InsertProcessor

func (p *Pipe) InsertProcessor(line, pos int, procAlloc ProcessorAllocatorFunc) <-chan struct{}

func (*Pipe) Push

func (p *Pipe) Push(mutations ...mutable.Mutation)

Push new mutators into pipe. Calling this method after pipe is done will cause a panic.

func (*Pipe) Start

func (p *Pipe) Start(ctx context.Context, initializers ...mutable.Mutation) <-chan error

Start starts the pipe execution.

type ProcessFunc

type ProcessFunc func(in, out signal.Floating) (int, error)

ProcessFunc takes the input buffer, applies processing logic and writes the result into output buffer.

type Processor

type Processor struct {
	mutable.Context
	ProcessFunc
	StartFunc
	FlushFunc
	SignalProperties
	// contains filtered or unexported fields
}

Processor is a mutator of signal data. Optinaly, mutability can be provided to handle mutations and flush hook to handle resource clean up.

type ProcessorAllocatorFunc

type ProcessorAllocatorFunc func(mctx mutable.Context, bufferSize int, input SignalProperties) (Processor, error)

ProcessorAllocatorFunc returns processor for provided buffer size. It is responsible for pre-allocation of all necessary buffers and structures. Along with the processor, output signal properties are returned.

func Processors

func Processors(processors ...ProcessorAllocatorFunc) []ProcessorAllocatorFunc

Processors is a helper function to use in line constructors.

type SignalProperties

type SignalProperties struct {
	SampleRate signal.Frequency
	Channels   int
}

SignalProperties contains information about input/output signal.

type Sink

type Sink struct {
	mutable.Context
	SinkFunc
	StartFunc
	FlushFunc
	SignalProperties
	// contains filtered or unexported fields
}

Sink is a destination of signal data. Optinaly, mutability can be provided to handle mutations and flush hook to handle resource clean up.

type SinkAllocatorFunc

type SinkAllocatorFunc func(mctx mutable.Context, bufferSize int, input SignalProperties) (Sink, error)

SinkAllocatorFunc returns sink for provided buffer size. It is responsible for pre-allocation of all necessary buffers and structures.

type SinkFunc

type SinkFunc func(in signal.Floating) error

SinkFunc takes the input buffer and writes that to the underlying destination.

type Source

type Source struct {
	mutable.Context
	SourceFunc
	StartFunc
	FlushFunc
	SignalProperties
	// contains filtered or unexported fields
}

Source is a source of signal data. Optinaly, mutability can be provided to handle mutations and flush hook to handle resource clean up.

type SourceAllocatorFunc

type SourceAllocatorFunc func(mctx mutable.Context, bufferSize int) (Source, error)

SourceAllocatorFunc returns source for provided buffer size. It is responsible for pre-allocation of all necessary buffers and structures.

type SourceFunc

type SourceFunc func(out signal.Floating) (int, error)

SourceFunc takes the output buffer and fills it with a signal data. If no data is available, io.EOF should be returned.

type StartFunc

type StartFunc func(ctx context.Context) error

StartFunc provides a hook to flush all buffers for the component.