worker

Documentation

Index

• type Allocator
• type BaseProcess
• type FSM
• type Streamer
• type SyncWorker
• type Watcher
• type Worker

Types

type Allocator

type Allocator func() (BaseProcess, error)

Allocator is responsible for worker allocation in the pool

type BaseProcess

type BaseProcess interface {
	fmt.Stringer

	// Pid returns worker pid.
	Pid() int64

	// Created returns time, worker was created at.
	Created() time.Time

	// State return receive-only WorkerProcess state object, state can be used to safely access
	// WorkerProcess status, time when status changed and number of WorkerProcess executions.
	State() FSM

	// Start used to run Cmd and immediately return
	Start() error

	// Wait must be called once for each WorkerProcess, call will be released once WorkerProcess is
	// complete and will return process error (if any), if stderr is presented it is value
	// will be wrapped as WorkerError. Method will return error code if php process fails
	// to find or Start the script.
	Wait() error

	// Stop sends soft termination command to the WorkerProcess and waits for process completion.
	Stop() error

	// Kill kills underlying process, make sure to call Wait() func to gather
	// error log from the stderr. Does not wait for process completion!
	Kill() error

	// Relay returns attached to worker goridge relay
	Relay() relay.Relay

	// AttachRelay used to attach goridge relay to the worker process
	AttachRelay(rl relay.Relay)
}

type FSM

type FSM interface {
	fmt.Stringer
	// NumExecs shows how many times WorkerProcess was invoked
	NumExecs() uint64
	// IsActive returns true if WorkerProcess not Inactive or Stopped
	IsActive() bool
	// RegisterExec using to registering php executions
	RegisterExec()
	// SetLastUsed sets worker last used time
	SetLastUsed(lu uint64)
	// LastUsed return worker last used time
	LastUsed() uint64
	// CurrentState returns the current state of the FSM
	CurrentState() int64
	// Transition used to move from one state to another
	Transition(from int64)
	// Compare compares state to the actual state and return true if they equal
	// false otherwise
	Compare(state int64) bool
}

FSM represents endure finite state machine

type Streamer

type Streamer interface {
	// BaseProcess provides basic functionality for the SyncWorker
	BaseProcess
	// ExecStream used to execute payload on the SyncWorker, there is no TIMEOUTS
	ExecStream(rqs *payload.Payload, resp chan *payload.Payload, stopCh chan struct{}) error
	// ExecStreamWithTTL used to handle Exec with TTL
	ExecStreamWithTTL(ctx context.Context, p *payload.Payload, resp chan *payload.Payload, stopCh chan struct{}) error
}

Streamer interface adds streams capabilities to the pool [BETA] interface, might be changed later

type SyncWorker

type SyncWorker interface {
	// BaseProcess provides basic functionality for the SyncWorker
	BaseProcess
	// Exec used to execute payload on the SyncWorker, there is no TIMEOUTS
	Exec(rqs *payload.Payload) (*payload.Payload, error)
	// ExecWithTTL used to handle Exec with TTL
	ExecWithTTL(ctx context.Context, p *payload.Payload) (*payload.Payload, error)
}

SyncWorker is not a good name, since it's just a sync executor, but not all worker is sync

type Watcher

type Watcher interface {
	// Watch used to add workers to the container
	Watch(workers []BaseProcess) error

	// Take takes the first free worker
	Take(ctx context.Context) (BaseProcess, error)

	// Release releases the worker putting it back to the queue
	Release(w BaseProcess)

	// Allocate - allocates new worker and put it into the WorkerWatcher
	Allocate() error

	// Destroy destroys the underlying container
	Destroy(ctx context.Context)

	// Reset will replace container and workers array, kill all workers
	Reset(ctx context.Context)

	// List return all container w/o removing it from internal storage
	List() []BaseProcess

	// Remove will remove worker from the container
	Remove(wb BaseProcess)
}

Watcher is an interface for the Sync workers lifecycle

type Worker

type Worker = SyncWorker

Worker is a non-bc replacement for the SDK