parallel

README

Parallel

Execute trivial operations in parallel without the hassle of maintaining channels, wait groups, error accumulators and other boilerplate code.

How to use

import (
    "github.com/kushsharma/parallel"
)

runner := parallel.NewRunner()
for _, j := range toomanyjobs {
	currentJob := j
	// queue operation for execution
	runner.Add(func() (interface{}, error) {
		//..do some operation here with (currentJob)..

		// first return value is result produced from the job
		// second return value is error if caused for some reason
		return nil, nil
	})
}

// Run() function is a blocking call and will start executing operations
// in parallel
for runIdx, state := range runner.Run() {
	if state.Err != nil {
		// handle error happened with job of index (runIdx)
	} else {
		// .. state.Val
		// result from from job with index (runIdx)
	}
}

Options

Limit concurrancy with

runner := parallel.NewRunner(paralle.WithLimit(10))

Documentation

Index

• type Runner
  • func NewRunner(opts ...RunnerOption) *Runner
  • func (p *Runner) Add(fn func() (interface{}, error))
  • func (p *Runner) Run() []State
  • func (p *Runner) RunSerial() []State
• type RunnerOption
  • func WithLimit(l int) RunnerOption
  • func WithTicket(l int) RunnerOption
• type State

Types

type Runner

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

Runner is a helper to execute operations in parallel and collect result/error in the end

func NewRunner

func NewRunner(opts ...RunnerOption) *Runner

NewRunner creates a new instance for parallel runner

func (*Runner) Add

func (p *Runner) Add(fn func() (interface{}, error))

Add will queue provided function for execution and will not immediately execute it. The order in which add was called will be used to provide the (result, error) pair

func (*Runner) Run

func (p *Runner) Run() []State

Run is a blocking operation and wait for all jobs to finish. result will be an ordered array of State which maps one to one with the order in which routines are added via Add() fn

func (*Runner) RunSerial

func (p *Runner) RunSerial() []State

RunSerial is here just to make sure runner can execute in serial as well if needed to

type RunnerOption

type RunnerOption func(*Runner)

func WithLimit

func WithLimit(l int) RunnerOption

WithLimit restricts n number of jobs executing in parallel

func WithTicket

func WithTicket(l int) RunnerOption

WithTicket restricts n number of jobs per second

type State

type State struct {
	Val interface{}
	Err error
}

State is provided as a result of each operation