parallel

README

Parallel

Parallel is a go library containing helper functions for parallel processing.

Installation

go get github.com/pushwoosh/parallel

Examples

See examples directory for full examples.

Panic handling

A panic inside a callback happens in a worker goroutine spawned by this library. The caller can not recover it with its own defer/recover (recover only works inside the panicking goroutine), and an unrecovered panic in any goroutine kills the whole process. To prevent that, every worker recovers panics and the library propagates them to the caller:

• Blocking functions (ApplyChan, ApplySlice, Execute, ExecuteOpts, MapSlice, MapSliceOrdered) re-raise the first recovered panic in the calling goroutine as *parallel.PanicError once the workers finish. For the caller it looks exactly like a panic in synchronous code, so an existing recovery layer (e.g. a gRPC recovery interceptor) handles it. After the first panic no new items or callbacks are started; workers that are already running finish first. ApplyChan also stops reading from its input channel, so the panic is re-raised even if the channel is never closed.
• MapChan is non-blocking, so recovered panics are delivered to the returned errors channel as *parallel.PanicError values and processing continues.

PanicError keeps the original panic value (Value) and the stack trace of the worker goroutine (Stack). If the panic value is an error, PanicError unwraps to it, so errors.Is/errors.As work through it. If several workers panic, only the first panic is kept; the rest are discarded.

Apply

Apply executes given function on each element of the input slice or channel.

There are two versions of Apply: ApplySlice and ApplyChan:

func ApplyChan[T any](input <-chan T, fn func(in T), opts ...ApplyOption) {}
func ApplySlice[T any](input []T, fn func(in T), opts ...ApplyOption) {}

To stop processing you can close the input channel.

Options

• WithApplyConcurrency(int) - limits the number of parallel threads. Default: ApplyDefaultConcurrency.

Example

ch := make(chan int)
...
ApplyChan(ch, func(in int) {
    fmt.Println(in)
})

Map

Map executes given function on each element of the input slice or channel and returns the result.

There are two versions of Map: MapSlice and MapChan:

func MapChan[Input any, Output any](input <-chan Input, fn func(in Input) (Output, error), opts ...MapOption) (<-chan Output, <-chan error) {}
func MapSlice[Input any, Output any](input []Input, fn func(in Input) (Output, error), opts ...MapOption) ([]Output, []error) {}

To stop processing you can close the input channel.

Options

• WithMapConcurrency(int) - limits the number of parallel threads. Default: MapDefaultConcurrency.
• WithMapStopOnFirstError - forces executor to stop processing new items after the first error occurred.

Example

input := make(chan string, 100)
input <- "hello"
input <- "world"
close(input)

parallel.MapChan(input, func(s string) (string, error) { return strings.ToTitle(s), nil })

Flow control

Output channel is filled with results of the callback function by the following rules:

• If error is nil, result is sent to output channel.
• If error is ErrMapSkip, result is not sent to output channel.
• If error is not ErrMapSkip, result is not sent to output channel and error is sent to errors output channel.

Same rules apply to MapSlice function.

MapSliceOrdered

MapSliceOrdered is a special version of MapSlice that guarantees that output slice will contain results in the same order as input slice.

Output slice will always contain the same number of elements as input slice. If callback function returns an error, output slice will contain nil at the corresponding position.

Execute

Execute executes given functions in parallel. There is no limit on the number of parallel threads. All given functions will be started at the same time.

To limit concurrency use ExecuteOpts.

Options

• WithExecuteConcurrency(int) - limits the number of parallel threads. Default: ExecuteDefaultConcurrency.

Example

parallel.Execute(
    func() { fmt.Println("Hello") },
    func() { fmt.Println("World") },
)

ConcurrencyLimiter

ConcurrencyLimiter is a helper that can limit amount of concurrently processed requests.

Example

limiter := NewConcurrencyLimiter(ops.concurrency)
for i := 0; i < 10000; i++ {
    limiter.Acquire()
    go func(i int) {
        defer limiter.Release()
        longJob(i)
    }(i)
}

Documentation

Index

• Constants
• Variables
• func ApplyChan[T any](input <-chan T, fn func(in T), opts ...ApplyOption)
• func ApplySlice[T any](input []T, fn func(in T), opts ...ApplyOption)
• func Execute(cbs ...func() error) []error
• func ExecuteOpts(cbs []func() error, opts ...ExecuteOption) []error
• func MapChan[Input any, Output any](input <-chan Input, fn func(in Input) (Output, error), opts ...MapOption) (<-chan Output, <-chan error)
• func MapSlice[Input any, Output any](input []Input, fn func(in Input) (Output, error), opts ...MapOption) ([]Output, []error)
• func MapSliceOrdered[Input any, Output any](input []Input, fn func(in Input) (Output, error), opts ...MapOption) ([]Output, []error)
• type ApplyOption
  • func WithApplyConcurrency(concurrency int) ApplyOption
• type ConcurrencyLimiter
  • func NewConcurrencyLimiter(concurrency int) ConcurrencyLimiter
• type ExecuteOption
  • func WithExecuteConcurrency(concurrency int) ExecuteOption
• type MapOption
  • func WithMapConcurrency(concurrency int) MapOption
  • func WithMapStopOnFirstError() MapOption
• type PanicError
  • func (e *PanicError) Error() string
  • func (e *PanicError) Unwrap() error

Constants

const (
	ApplyDefaultConcurrency = 10
)

const (
	ExecuteDefaultConcurrency = 10
)

const (
	MapDefaultConcurrency = 10
)

Variables

var ErrMapSkip = errors.New("skip")

ErrMapSkip is a special error that can be returned from Map function to skip the item.

Functions

func ApplyChan

func ApplyChan[T any](input <-chan T, fn func(in T), opts ...ApplyOption)

ApplyChan executes `fn` on each element of `input` channel in multiple threads. Options:

WithApplyConcurrency(int) - limits the number of parallel threads. Default: ApplyDefaultConcurrency

To stop processing, close `input` channel.

A panic in `fn` does not kill the process: it is recovered in the worker, ApplyChan stops reading new items (so the panic surfaces even if `input` is never closed), waits for already-started workers, and re-raises the first panic in the calling goroutine as *PanicError.

func ApplySlice

func ApplySlice[T any](input []T, fn func(in T), opts ...ApplyOption)

ApplySlice does the same as ApplyChan, but works with slice instead of a channel.

A panic in `fn` does not kill the process: after the first recovered panic no new items are started, already-started workers finish, and the first panic is re-raised in the calling goroutine as *PanicError.

func Execute

func Execute(cbs ...func() error) []error

Execute executes multiple callback functions `cbs` in parallel.

A panic in a callback does not kill the process: it is recovered in the worker (all callbacks are started immediately, so the others still run), and the first panic is re-raised in the calling goroutine as *PanicError after all workers finish.

func ExecuteOpts

func ExecuteOpts(cbs []func() error, opts ...ExecuteOption) []error

ExecuteOpts executes slice of callback functions `cbs` with custom options.

A panic in a callback does not kill the process: after the first recovered panic no new callbacks are started, already-started ones finish, and the first panic is re-raised in the calling goroutine as *PanicError.

func MapChan

func MapChan[Input any, Output any](input <-chan Input, fn func(in Input) (Output, error), opts ...MapOption) (<-chan Output, <-chan error)

MapChan executes `fn` on each element of `input` channel in several threads.

A panic in `fn` does not kill the process: it is recovered in the worker and delivered to the returned errors channel as *PanicError. MapChan is non-blocking, so the panic can not be re-raised in the calling goroutine.

func MapSlice

func MapSlice[Input any, Output any](input []Input, fn func(in Input) (Output, error), opts ...MapOption) ([]Output, []error)

MapSlice does the same as MapChan, but works with slices instead of channels in input and output.

Unlike MapChan, MapSlice blocks until the workers finish, so the first panic recovered in a worker is re-raised in the calling goroutine as *PanicError. After the first panic no new items are started.

func MapSliceOrdered

func MapSliceOrdered[Input any, Output any](input []Input, fn func(in Input) (Output, error), opts ...MapOption) ([]Output, []error)

MapSliceOrdered does the same as MapSlice, but returns results in the same order as input.

Types

type ApplyOption

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

func WithApplyConcurrency

func WithApplyConcurrency(concurrency int) ApplyOption

type ConcurrencyLimiter

type ConcurrencyLimiter interface {
	// Acquire acquires a slot in the concurrency limiter.
	// Blocks until a slot is available.
	Acquire()

	// Release releases a slot in the concurrency limiter.
	Release()
}

ConcurrencyLimiter is a helper that can limit amount of concurrently processed requests. See concurrency_limiter_test.go for usage example.

func NewConcurrencyLimiter

func NewConcurrencyLimiter(concurrency int) ConcurrencyLimiter

type ExecuteOption

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

func WithExecuteConcurrency

func WithExecuteConcurrency(concurrency int) ExecuteOption

type MapOption

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

func WithMapConcurrency

func WithMapConcurrency(concurrency int) MapOption

func WithMapStopOnFirstError

func WithMapStopOnFirstError() MapOption

type PanicError

type PanicError struct {
	Value any
	Stack []byte
}

PanicError wraps a panic recovered in a worker goroutine.

A panic in a goroutine spawned by this package can not be recovered by the caller: recover() only works in the goroutine where the panic happened, and an unrecovered panic in any goroutine kills the whole process. To prevent that, every worker goroutine recovers panics itself and the package propagates them to the caller:

• blocking functions (ApplyChan, ApplySlice, Execute, ExecuteOpts, MapSlice, MapSliceOrdered) re-raise the first recovered panic in the calling goroutine after the workers finish, so the caller's own defer/recover (e.g. a gRPC recovery interceptor) can handle it just like a panic in synchronous code. After the first panic no new items or callbacks are started; workers that are already running finish first. ApplyChan also stops reading from its input channel, so the panic is re-raised even if the channel is never closed;
• MapChan is non-blocking, so recovered panics are delivered to the returned errors channel as *PanicError values and processing continues.

Value holds the original value passed to panic(), Stack holds the stack trace of the worker goroutine captured at the moment of recovery.

func (*PanicError) Error

func (e *PanicError) Error() string

func (*PanicError) Unwrap

func (e *PanicError) Unwrap() error

Unwrap returns the panic value if it is an error, so errors.Is and errors.As see through PanicError.