pool

Documentation

Index

• Constants
• Variables
• func Schedule(p *Pool) func(func())
• func ScheduleContext(p *Pool) func(context.Context, func()) error
• func ScheduleCustom(p *Pool) func(chan struct{}, func()) error
• func ScheduleTimeout(p *Pool) func(time.Duration, func()) error
• type Config
• type Pool
  • func Must(p *Pool, err error) *Pool
  • func New(c *Config) (*Pool, error)
  • func (p *Pool) Barrier()
  • func (p *Pool) BarrierContext(ctx context.Context) error
  • func (p *Pool) BarrierCustom(cancel <-chan struct{}) error
  • func (p *Pool) BarrierTimeout(tm time.Duration) error
  • func (p *Pool) Close() error
  • func (p *Pool) Done() <-chan struct{}
  • func (p *Pool) Schedule(t Task) error
  • func (p *Pool) ScheduleContext(ctx context.Context, t Task) error
  • func (p *Pool) ScheduleCustom(cancel <-chan struct{}, t Task) error
  • func (p *Pool) ScheduleImmediate(t Task) error
  • func (p *Pool) ScheduleTimeout(tm time.Duration, t Task) error
  • func (p *Pool) SetNoCork(v bool)
  • func (p *Pool) Wait()
  • func (p *Pool) WaitContext(ctx context.Context) error
  • func (p *Pool) WaitCustom(cancel <-chan struct{}) error
  • func (p *Pool) WaitTimeout(tm time.Duration) error
• type Task
• type TaskFunc
  • func (fn TaskFunc) Run()

Constants

const (
	DefaultExtraWorkerTTL = time.Second
)

Variables

var (
	ErrConfigMalformed        = errors.New("malformed pool config: max number of workers is lower than number of unstoppable workers")
	ErrConfigQueueUnstoppable = errors.New("malformed pool config: queue is buffered, but unstoppable workers set to 0")
	ErrUnavailable            = errors.New("pool: temporarily unavailable")
	ErrPoolClosed             = errors.New("pool: closed")
)

Functions

func Schedule

func Schedule(p *Pool) func(func())

Schedule is a helper that returns function which purpose is to schedule execution of next given function over p.

func ScheduleContext

func ScheduleContext(p *Pool) func(context.Context, func()) error

ScheduleTimeout is a helper that returns function which purpose is to schedule execution of next given function over p with context.

func ScheduleCustom

func ScheduleCustom(p *Pool) func(chan struct{}, func()) error

ScheduleTimeout is a helper that returns function which purpose is to schedule execution of next given function over p with cancellation chan.

func ScheduleTimeout

func ScheduleTimeout(p *Pool) func(time.Duration, func()) error

ScheduleTimeout is a helper that returns function which purpose is to schedule execution of next given function over p with timeout.

Types

type Config

type Config struct {
	// Number of workers that are always running.
	UnstoppableWorkers int

	// Maximum number of workers that could be spawned.
	// It includes UnstoppableWorkers count.
	// If MaxWorkers is <= 0 then it interpreted as extra workers are unlimited.
	// If MaxWorkers is in range [1, UnstoppableWorkers) then config is malformed.
	MaxWorkers int

	// When work flow becomes too much for unstoppable workers,
	// Pool could spawn extra worker if it fits max workers limit.
	// Spawned worker will be alive for this amount of time.
	ExtraWorkerTTL time.Duration

	// This parameter manages the size of work queue.
	// The smaller this value, the greater the probability of
	// spawning the extra worker. And vice versa.
	WorkQueueSize int

	// OnTaskIn, OnTaskOut are used to signal whan task is scheduled/pulled from queue.
	OnTaskIn, OnTaskOut func()

	// OnWorkerStart, OnWorkerStop are called on extra worker spawned/stopped.
	OnWorkerStart, OnWorkerStop func()
}

type Pool

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

func Must

func Must(p *Pool, err error) *Pool

Must is a helper that wraps a call to a function returning (*Pool, error) and panics if the error is non-nil. It is intended for use in variable initializations with New() function.

func New

func New(c *Config) (*Pool, error)

func (*Pool) Barrier

func (p *Pool) Barrier()

Barrier blocks until workers complete their current task.

func (*Pool) BarrierContext

func (p *Pool) BarrierContext(ctx context.Context) error

BarrierContext blocks until workers complete their current task or until context is done. In case when err is non-nil err is always a result of ctx.Err() call.

func (*Pool) BarrierCustom

func (p *Pool) BarrierCustom(cancel <-chan struct{}) error

BarrierCustom blocks until workers complete their current task or until given cancelation channel is non-empty.

func (*Pool) BarrierTimeout

func (p *Pool) BarrierTimeout(tm time.Duration) error

BarrierTimeout blocks until workers complete their current task or until timeout is expired.

func (*Pool) Close

func (p *Pool) Close() error

Close terminates all spawned goroutines.

func (*Pool) Done

func (p *Pool) Done() <-chan struct{}

Done returns channel which closure means that pool is done with all its work.

func (*Pool) Schedule

func (p *Pool) Schedule(t Task) error

Schedule makes task to be scheduled over pool's workers. It returns non-nil only if pool become closed and task can not be executed over its workers.

func (*Pool) ScheduleContext

func (p *Pool) ScheduleContext(ctx context.Context, t Task) error

ScheduleContext makes task to be scheduled over pool's workers.

Non-nil error only available when ctx is done. That is, if no workers are available during lifetime of context, it returns ctx.Err() call result.

func (*Pool) ScheduleCustom

func (p *Pool) ScheduleCustom(cancel <-chan struct{}, t Task) error

ScheduleCustom makes task to be scheduled over pool's workers.

Non-nil error only available when cancel closed. That is, if no workers are available until closure of cancel chan, it returns error.

This method useful for implementing custom cancellation logic by a caller.

func (*Pool) ScheduleImmediate

func (p *Pool) ScheduleImmediate(t Task) error

ScheduleImmediate makes task to be scheduled without waiting for free workers. That is, if all workers are busy and pool is not rubber, then ErrUnavailable is returned immediately.

func (*Pool) ScheduleTimeout

func (p *Pool) ScheduleTimeout(tm time.Duration, t Task) error

ScheduleTimeout makes task to be scheduled over pool's workers.

Non-nil error only available when tm is not 0. That is, if no workers are available during timeout, it returns error.

Zero timeout means that there are no timeout for awaiting for available worker.

func (*Pool) SetNoCork

func (p *Pool) SetNoCork(v bool)

func (*Pool) Wait

func (p *Pool) Wait()

Wait blocks until all previously scheduled tasks are executed.

func (*Pool) WaitContext

func (p *Pool) WaitContext(ctx context.Context) error

WaitContext blocks until all previously scheduled tasks are executed or until context is done. In case when err is non-nil err is always a result of ctx.Err() call.

func (*Pool) WaitCustom

func (p *Pool) WaitCustom(cancel <-chan struct{}) error

WaitCustom blocks until all previously scheduled tasks are executed or until given cancelation channel is non-empty.

func (*Pool) WaitTimeout

func (p *Pool) WaitTimeout(tm time.Duration) error

WaitTimeout blocks until all previously scheduled tasks are executed or until timeout is expired.

type Task

type Task interface {
	Run()
}

Task represents task that need to be executed by some worker.

type TaskFunc

type TaskFunc func()

func (TaskFunc) Run

func (fn TaskFunc) Run()