pool

README

Dedicated Worker Pool

The pool package builds on top of the Pulse rmap and streaming packages to provide scalable and reliable dedicated worker pools.

Overview

A dedicated worker pool uses a consistent hashing algorithm to assign long running jobs to workers. Each job is associated with a key and each worker with a range of hashed values. The pool hashes the job key when the job is dispatched to route the job to the proper worker.

Workers can be added or removed from the pool dynamically. Jobs get automatically re-assigned to workers when the pool grows or shrinks. This makes it possible to implement auto-scaling solutions, for example based on queueing delays.

Pulse uses the Jump Consistent Hash algorithm to assign jobs to workers which provides a good balance between load balancing and worker assignment stability.

%%{init: {'themeVariables': { 'edgeLabelBackground': '#7A7A7A'}}}%%
flowchart LR
    A[Job Producer]
    subgraph Pool["<span style='margin: 0 10px;'>Routing Pool Node</span>"]
        Sink["Job Sink"]
    end
    subgraph Worker[Worker Pool Node]
        Reader
        B[Worker]
    end
    A-->|Job+Key|Sink
    Sink-.->|Job|Reader
    Reader-.->|Job|B

    classDef userCode fill:#9A6D1F, stroke:#D9B871, stroke-width:2px, color:#FFF2CC;
    classDef pulse fill:#25503C, stroke:#5E8E71, stroke-width:2px, color:#D6E9C6;

    class A,B userCode;
    class Pool,Sink,Reader,Worker pulse;

    linkStyle 0 stroke:#DDDDDD,color:#DDDDDD,stroke-width:3px;
    linkStyle 1 stroke:#DDDDDD,color:#DDDDDD,stroke-width:3px;
    linkStyle 2 stroke:#DDDDDD,color:#DDDDDD,stroke-width:3px;

Usage

Job producer:

	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
	node, err := pool.AddNode(ctx, "example", rdb, pool.WithClientOnly())
	if err != nil {
		panic(err)
	}
	if err := node.DispatchJob(ctx, "key", []byte("payload")); err != nil {
		panic(err)
	}

Worker:

	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
	node, err := pool.AddNode(ctx, "example", rdb)
	if err != nil {
		panic(err)
	}
	handler := &JobHandler{}
	_, err := node.AddWorker(context.Background(), handler)
  if err != nil {
		panic(err)
	}

Job handler:

type JobHandler struct {
	// ...
}

// Pulse calls this method to start a job that was assigned to this worker.
func (h *JobHandler) Start(ctx context.Context, key string, payload []byte) error {
	// ...
}

// Pulse calls this method to stop a job that was assigned to this worker.
func (h *JobHandler) Stop(ctx context.Context, key string) error {
	// ...
}

Creating A Pool

The function AddNode is used to create a new pool node. It takes as input a name, a Redis client and a set of options.

The AddNode function returns a new pool node and an error. The pool node should be closed when it is no longer needed (see below).

The options are used to configure the pool node. The following options are available:

• WithClientOnly - specifies that this node will only be used to dispatch jobs to workers in other nodes, and will not run any workers itself.
• WithLogger - sets the logger to be used by the pool node.
• WithWorkerTTL - sets the worker time-to-live (TTL). This is the maximum duration a worker can go without sending a health check before it's considered inactive and removed from the pool. If a worker doesn't report its status within this time frame, it will be removed, allowing the pool to reassign its jobs to other active workers. The default value is 30 seconds.
• WithWorkerShutdownTTL - specifies the maximum time to wait for a worker to shutdown gracefully. This is the duration the pool will wait for a worker to finish its current job and perform any cleanup operations before forcefully terminating it. If the worker doesn't shut down within this time, it will be forcefully stopped. The default value is 2 minutes.
• WithMaxQueuedJobs - sets the maximum number of jobs that can be queued before the pool starts rejecting new jobs. This limit applies to the entire pool across all nodes. When this limit is reached, any attempt to dispatch new jobs will result in an error. The default value is 1000 jobs.
• WithAckGracePeriod - sets the grace period for job acknowledgment. If a worker doesn't acknowledge starting a job within this duration, the job becomes available for other workers to claim. This prevents jobs from being stuck if a worker fails to start processing them. The default value is 20 seconds.

Closing A Node

The Close method closes the pool node and releases all resources associated with it. It should be called when the node is no longer needed.

Note that closing a pool node does not stop remote workers. It only stops the local pool node. Remote workers can be stopped by calling the Shutdown method described below.

Shutting Down A Pool

The Shutdown method shuts down the entire pool by stopping all its workers gracefully. It should be called when the pool is no longer needed.

See the Data Flows section below for more details on the shutdown process.

Creating A Worker

The function AddWorker is used to create a new worker. It takes as input a job handler object.

The job handler must implement the Start and Stop methods used to start and stop jobs. The handler may also optionally implement a HandleNotification method to receive notifications.

The AddWorker function returns a new worker and an error. Workers can be removed from pool nodes using the RemoveWorker method.

Dispatching A Job

The DispatchJob method is used to dispatch a new job to the pool. It takes as input a job key and a job payload.

The job key is used to route the job to the proper worker. The job payload is passed to the worker's Start method.

The DispatchJob method returns an error if the job could not be dispatched. This can happen if the pool is full or if the job key is invalid.

Notifications

Nodes can send notifications to workers using the NotifyWorker method. The method takes as input a job key and a notification payload. The notification payload is passed to the worker's HandleNotification method.

Stopping A Job

The StopJob method is used to stop a job. It takes a job key as input and returns an error if the job could not be stopped. This can happen if the job key is invalid, the node is closed or the pool shutdown.

Scheduling

The Schedule method of the Node struct can be used to schedule jobs to be dispatched or stopped on a recurring basis. The method takes as input a job producer and invokes it at the specified interval. The job producer returns a list of jobs to be started and stopped.

Schedule makes it possible to maintain a pool of jobs for example in a multi-tenant system. See the examples for more details.

Documentation

Index

• Variables
• type Job
• type JobHandler
• type JobParam
• type JobPlan
• type JobProducer
• type Node
  • func AddNode(ctx context.Context, poolName string, rdb *redis.Client, opts ...NodeOption) (*Node, error)
  • func (node *Node) AddWorker(ctx context.Context, handler JobHandler) (*Worker, error)
  • func (node *Node) Close(ctx context.Context) error
  • func (node *Node) DispatchJob(ctx context.Context, key string, payload []byte) error
  • func (node *Node) IsClosed() bool
  • func (node *Node) IsShutdown() bool
  • func (node *Node) JobKeys() []string
  • func (node *Node) JobPayload(key string) ([]byte, bool)
  • func (node *Node) NewTicker(ctx context.Context, name string, d time.Duration, opts ...TickerOption) (*Ticker, error)
  • func (node *Node) NotifyWorker(ctx context.Context, key string, payload []byte) error
  • func (node *Node) PoolWorkers() []*Worker
  • func (node *Node) RemoveWorker(ctx context.Context, w *Worker) error
  • func (node *Node) Schedule(ctx context.Context, producer JobProducer, interval time.Duration) error
  • func (node *Node) Shutdown(ctx context.Context) error
  • func (node *Node) StopJob(ctx context.Context, key string) error
  • func (node *Node) Workers() []*Worker
• type NodeOption
  • func WithAckGracePeriod(ttl time.Duration) NodeOption
  • func WithClientOnly() NodeOption
  • func WithJobSinkBlockDuration(d time.Duration) NodeOption
  • func WithLogger(logger pulse.Logger) NodeOption
  • func WithMaxQueuedJobs(max int) NodeOption
  • func WithWorkerShutdownTTL(ttl time.Duration) NodeOption
  • func WithWorkerTTL(ttl time.Duration) NodeOption
• type NotificationHandler
• type Ticker
  • func (t *Ticker) Stop()
• type TickerOption
  • func WithTickerLogger(logger pulse.Logger) TickerOption
• type Worker
  • func (w *Worker) IsStopped() bool
  • func (w *Worker) Jobs() []*Job

Variables

var ErrJobExists = errors.New("job already exists")

ErrJobExists is returned when attempting to dispatch a job with a key that already exists.

var ErrRequeue = errors.New("requeue")

ErrRequeue indicates that a worker failed to process a job's start or stop operation and requests the job to be requeued for another attempt.

var ErrScheduleStop = fmt.Errorf("stop")

ErrScheduleStop is returned by JobProducer.Plan to indicate that the corresponding schedule should be stopped.

Types

type Job

type Job struct {
	// Key is used to identify the worker that handles the job.
	Key string
	// Payload is the job payload.
	Payload []byte
	// CreatedAt is the time the job was created.
	CreatedAt time.Time
	// Worker is the worker that handles the job.
	Worker *Worker
	// NodeID is the ID of the node that created the job.
	NodeID string
}

Job is a job that can be added to a worker.

type JobHandler

type JobHandler interface {
	// Start starts a job.
	Start(job *Job) error
	// Stop stops a job with a given key.
	Stop(key string) error
}

JobHandler starts and stops jobs.

type JobParam

type JobParam struct {
	// Key is the job key.
	Key string
	// Payload is the job payload.
	Payload []byte
}

JobParam represents a job to start.

type JobPlan

type JobPlan struct {
	// Jobs to start.
	Start []*JobParam
	// Job keys to stop.
	Stop []string
	// StopAll indicates that all jobs not in Jobs should be
	// stopped.  Stop is ignored if StopAll is true.
	StopAll bool
}

JobPlan represents a list of jobs to start and job keys to stop.

type JobProducer

type JobProducer interface {
	// Name returns the name of the producer. Schedule calls Plan on
	// only one of the producers with identical names across all
	// nodes.
	Name() string
	// Plan computes the list of jobs to start and job keys to stop.
	// Returning ErrScheduleStop indicates that the recurring
	// schedule should be stopped.
	Plan() (*JobPlan, error)
}

JobComputeFunc is the function called by the scheduler to compute jobs. It returns the list of jobs to start and job keys to stop.

type Node

type Node struct {
	ID       string
	PoolName string
	// contains filtered or unexported fields
}

Node is a pool of workers.

func AddNode

func AddNode(ctx context.Context, poolName string, rdb *redis.Client, opts ...NodeOption) (*Node, error)

AddNode adds a new node to the pool with the given name and returns it. The node can be used to dispatch jobs and add new workers. A node also routes dispatched jobs to the proper worker and acks the corresponding events once the worker acks the job.

The options WithClientOnly can be used to create a node that can only be used to dispatch jobs. Such a node does not route or process jobs in the background.

func (*Node) AddWorker

func (node *Node) AddWorker(ctx context.Context, handler JobHandler) (*Worker, error)

AddWorker adds a new worker to the pool and returns it. The worker starts processing jobs immediately. handler can optionally implement the NotificationHandler interface to handle notifications.

func (*Node) Close

func (node *Node) Close(ctx context.Context) error

Close stops the node workers and closes the Redis connection but does not stop workers running in other nodes. It requeues all the jobs run by workers of the node. One of Shutdown or Close should be called before the node is garbage collected unless it is client-only.

func (*Node) DispatchJob

func (node *Node) DispatchJob(ctx context.Context, key string, payload []byte) error

DispatchJob dispatches a job to the worker in the pool that is assigned to the job key using consistent hashing. It returns: - nil if the job is successfully dispatched and started by a worker - ErrJobExists if a job with the same key already exists in the pool - an error returned by the worker's start handler if the job fails to start - an error if the pool is closed or if there's a failure in adding the job

The method blocks until one of the above conditions is met.

func (*Node) IsClosed

func (node *Node) IsClosed() bool

IsClosed returns true if the node is closed.

func (*Node) IsShutdown

func (node *Node) IsShutdown() bool

IsShutdown returns true if the pool is shutdown.

func (*Node) JobKeys

func (node *Node) JobKeys() []string

JobKeys returns the list of keys of the jobs running in the pool.

func (*Node) JobPayload

func (node *Node) JobPayload(key string) ([]byte, bool)

JobPayload returns the payload of the job with the given key. It returns: - (payload, true) if the job exists and has a payload - (nil, true) if the job exists but has an empty payload - (nil, false) if the job does not exist

func (*Node) NewTicker

func (node *Node) NewTicker(ctx context.Context, name string, d time.Duration, opts ...TickerOption) (*Ticker, error)

NewTicker returns a new Ticker that behaves similarly to time.Ticker, but instead delivers the current time on the channel to only one of the nodes that invoked NewTicker with the same name.

func (*Node) NotifyWorker

func (node *Node) NotifyWorker(ctx context.Context, key string, payload []byte) error

NotifyWorker notifies the worker that handles the job with the given key.

func (*Node) PoolWorkers

func (node *Node) PoolWorkers() []*Worker

PoolWorkers returns the list of workers running in the entire pool.

func (*Node) RemoveWorker

func (node *Node) RemoveWorker(ctx context.Context, w *Worker) error

RemoveWorker stops the worker, removes it from the pool and requeues all its jobs.

func (*Node) Schedule

func (node *Node) Schedule(ctx context.Context, producer JobProducer, interval time.Duration) error

Schedule calls the producer Plan method on the given interval and starts and stops jobs accordingly. The schedule stops when the producer Plan method returns ErrScheduleStop. Plan is called on only one of the nodes that scheduled the same producer.

func (*Node) Shutdown

func (node *Node) Shutdown(ctx context.Context) error

Shutdown stops the pool workers gracefully across all nodes. It notifies all workers and waits until they are completed. Shutdown prevents the pool nodes from creating new workers and the pool workers from accepting new jobs. After Shutdown returns, the node object cannot be used anymore and should be discarded. One of Shutdown or Close should be called before the node is garbage collected unless it is client-only.

func (*Node) StopJob

func (node *Node) StopJob(ctx context.Context, key string) error

StopJob stops the job with the given key.

func (*Node) Workers

func (node *Node) Workers() []*Worker

Workers returns the list of workers running in the local node.

type NodeOption

type NodeOption func(*nodeOptions)

NodeOption is a worker creation option.

func WithAckGracePeriod

func WithAckGracePeriod(ttl time.Duration) NodeOption

WithAckGracePeriod sets the duration after which a job is made available to other workers if it wasn't started. The default is 20s.

func WithClientOnly

func WithClientOnly() NodeOption

WithClientOnly sets the pool to be client only. A client-only pool only supports dispatching jobs to workers and does not start background goroutines to route jobs.

func WithJobSinkBlockDuration

func WithJobSinkBlockDuration(d time.Duration) NodeOption

WithJobSinkBlockDuration sets the duration to block when reading from the job stream. The default is 5s. This option is mostly useful for testing.

func WithLogger

func WithLogger(logger pulse.Logger) NodeOption

WithLogger sets the handler used to report temporary errors.

func WithMaxQueuedJobs

func WithMaxQueuedJobs(max int) NodeOption

WithMaxQueuedJobs sets the maximum number of jobs that can be queued in the pool. The default is 1000.

func WithWorkerShutdownTTL

func WithWorkerShutdownTTL(ttl time.Duration) NodeOption

WithWorkerShutdownTTL sets the maximum time to wait for workers to shutdown. The default is 2 minutes.

func WithWorkerTTL

func WithWorkerTTL(ttl time.Duration) NodeOption

WithWorkerTTL sets the duration after which the worker is removed from the pool in case of network partitioning. The default is 10s. A lower number causes more frequent keep-alive updates from all workers.

type NotificationHandler

type NotificationHandler interface {
	// HandleNotification handles a notification.
	HandleNotification(key string, payload []byte) error
}

NotificationHandler handle job notifications.

type Ticker

type Ticker struct {
	C <-chan time.Time
	// contains filtered or unexported fields
}

Ticker represents a clock that periodically sends ticks to one of the pool nodes which created a ticker with the same name.

func (*Ticker) Stop

func (t *Ticker) Stop()

Stop turns off a ticker. After Stop, no more ticks will be sent. Stop does not close the channel, to prevent a concurrent goroutine reading from the channel from seeing an erroneous "tick".

type TickerOption

type TickerOption func(*tickerOptions)

TickerOption is a worker creation option.

func WithTickerLogger

func WithTickerLogger(logger pulse.Logger) TickerOption

WithTickerLogger sets the handler used to report temporary errors.

type Worker

type Worker struct {
	// Unique worker ID
	ID string
	// Time worker was created.
	CreatedAt time.Time
	// contains filtered or unexported fields
}

Worker is a worker that handles jobs with a given payload type.

func (*Worker) IsStopped

func (w *Worker) IsStopped() bool

IsStopped returns true if the worker is stopped.

func (*Worker) Jobs

func (w *Worker) Jobs() []*Job

Jobs returns the jobs handled by the worker.