pool

Documentation

Index

• Constants
• func NewPool(ctx context.Context, cmd func() *exec.Cmd, factory worker.Factory, cfg Config, ...) (pool.Pool, error)
• type After
• type Before
• type Config
  • func (cfg *Config) InitDefaults()
• type ErrorEncoder
• type Options
  • func ExecAfter(after ...After) Options
  • func ExecBefore(before ...Before) Options
• type StaticPool
  • func (sp *StaticPool) AddListener(listener events.EventListener)
  • func (sp *StaticPool) Destroy(ctx context.Context)
  • func (sp *StaticPool) Exec(p internal.Payload) (internal.Payload, error)
  • func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs internal.Payload) (internal.Payload, error)
  • func (sp *StaticPool) GetConfig() interface{}
  • func (sp *StaticPool) RemoveWorker(wb worker.BaseProcess) error
  • func (sp *StaticPool) Workers() (workers []worker.BaseProcess)
• type Supervised
• type SupervisorConfig
  • func (cfg *SupervisorConfig) InitDefaults()

Constants

const MB = 1024 * 1024

const StopRequest = "{\"stop\":true}"

StopRequest can be sent by worker to indicate that restart is required.

Functions

func NewPool

func NewPool(ctx context.Context, cmd func() *exec.Cmd, factory worker.Factory, cfg Config, options ...Options) (pool.Pool, error)

NewPool creates new worker pool and task multiplexer. StaticPool will initiate with one worker.

Types

type After

type After func(req internal.Payload, resp internal.Payload) internal.Payload

After is set of functions that executes AFTER Exec

type Before

type Before func(req internal.Payload) internal.Payload

Before is set of functions that executes BEFORE Exec

type Config

type Config struct {
	// Debug flag creates new fresh worker before every request.
	Debug bool

	// NumWorkers defines how many sub-processes can be run at once. This value
	// might be doubled by Swapper while hot-swap. Defaults to number of CPU cores.
	NumWorkers int64

	// MaxJobs defines how many executions is allowed for the worker until
	// it's destruction. set 1 to create new process for each new task, 0 to let
	// worker handle as many tasks as it can.
	MaxJobs int64

	// AllocateTimeout defines for how long pool will be waiting for a worker to
	// be freed to handle the task. Defaults to 60s.
	AllocateTimeout time.Duration

	// DestroyTimeout defines for how long pool should be waiting for worker to
	// properly destroy, if timeout reached worker will be killed. Defaults to 60s.
	DestroyTimeout time.Duration

	// Supervision config to limit worker and pool memory usage.
	Supervisor *SupervisorConfig
}

Configures the pool behaviour.

func (*Config) InitDefaults

func (cfg *Config) InitDefaults()

InitDefaults enables default config values.

type ErrorEncoder

type ErrorEncoder func(err error, w worker.BaseProcess) (internal.Payload, error)

ErrorEncoder encode error or make a decision based on the error type

type Options

type Options func(p *StaticPool)

func ExecAfter

func ExecAfter(after ...After) Options

func ExecBefore

func ExecBefore(before ...Before) Options

type StaticPool

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

StaticPool controls worker creation, destruction and task routing. Pool uses fixed amount of stack.

func (*StaticPool) AddListener

func (sp *StaticPool) AddListener(listener events.EventListener)

AddListener connects event listener to the pool.

func (*StaticPool) Destroy

func (sp *StaticPool) Destroy(ctx context.Context)

Destroy all underlying stack (but let them to complete the task).

func (*StaticPool) Exec

func (sp *StaticPool) Exec(p internal.Payload) (internal.Payload, error)

func (*StaticPool) ExecWithContext

func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs internal.Payload) (internal.Payload, error)

func (*StaticPool) GetConfig

func (sp *StaticPool) GetConfig() interface{}

Config returns associated pool configuration. Immutable.

func (*StaticPool) RemoveWorker

func (sp *StaticPool) RemoveWorker(wb worker.BaseProcess) error

func (*StaticPool) Workers

func (sp *StaticPool) Workers() (workers []worker.BaseProcess)

Workers returns worker list associated with the pool.

type Supervised

type Supervised interface {
	pool.Pool
	// Start used to start watching process for all pool workers
	Start()
}

type SupervisorConfig

type SupervisorConfig struct {
	// WatchTick defines how often to check the state of worker.
	WatchTick uint64

	// TTL defines maximum time worker is allowed to live.
	TTL uint64

	// IdleTTL defines maximum duration worker can spend in idle mode. Disabled when 0.
	IdleTTL uint64

	// ExecTTL defines maximum lifetime per job.
	ExecTTL uint64

	// MaxWorkerMemory limits memory per worker.
	MaxWorkerMemory uint64
}

func (*SupervisorConfig) InitDefaults

func (cfg *SupervisorConfig) InitDefaults()

InitDefaults enables default config values.