worker

README

Worker

The worker package implements a simple load balancing algorithm based on the master-worker scheme. In this algorithm, the workload is distributed among multiple worker processes by a single manager process.

Overview

The diagram below shows the available components and how they relate to each other:

Components

Click the name of a component to see its description.

Pool

Distributes the workload among registered pool workers.

PoolWorker

Represents a worker on the manager side. Uses a client to communicate with the worker.

Client

Sends requests to a server and receives responses from it. Used on the manager side.

Server

Receives requests from a client and sends back responses. Used on the worker side.

Loop

Represents the processing loop of a worker. Uses a server to communicate with a manager.

Communication

Although the communication between a worker and a manager is protocol-independent, the package comes with a default implementation based on the WebSocket protocol. You can find it in the websocket.go file.

Backoff algorithm

The package includes an implementation of the exponential backoff algorithm. This mechanism can be used to decrease the rate of requests in case of repeated failures. You can find the implementation of the algorithm in the backoff.go file.

Documentation

Index

• type Backoff
  • func (b *Backoff) Attempt()
  • func (b Backoff) Delay() time.Duration
  • func (b *Backoff) Reset()
• type Client
• type Loop
  • func NewLoop(server Server) *Loop
  • func (l *Loop) Run(handler RequestHandler)
• type Pool
  • func (p *Pool) AddWorker(worker *PoolWorker)
  • func (p *Pool) Process(height int64)
  • func (p *Pool) Run(handler ResponseHandler)
  • func (p *Pool) Stop()
  • func (p *Pool) Wait()
• type PoolWorker
  • func NewPoolWorker(client Client) *PoolWorker
  • func (pw *PoolWorker) Run(handler ResponseHandler, wg *sync.WaitGroup)
  • func (pw *PoolWorker) Stop()
• type Request
• type RequestHandler
• type Response
• type ResponseHandler
• type Server
• type WebsocketClient
  • func NewWebsocketClient(endpoint string) (*WebsocketClient, error)
  • func (wc *WebsocketClient) Close() error
  • func (wc *WebsocketClient) Connect() error
  • func (wc *WebsocketClient) Receive(res *Response) error
  • func (wc *WebsocketClient) Reconnect() error
  • func (wc *WebsocketClient) Send(req Request) error
• type WebsocketServer
  • func NewWebsocketServer(conn *websocket.Conn) *WebsocketServer
  • func (ws *WebsocketServer) Close() error
  • func (ws *WebsocketServer) Receive(req *Request) error
  • func (ws *WebsocketServer) Send(res Response) error

Types

type Backoff

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

Backoff implements an exponential backoff algorithm

func (*Backoff) Attempt

func (b *Backoff) Attempt()

Attempt increments the number of attempts

func (Backoff) Delay

func (b Backoff) Delay() time.Duration

Delay calculates the backoff time

func (*Backoff) Reset

func (b *Backoff) Reset()

Reset resets the number of attempts

type Client

type Client interface {
	Send(Request) error
	Receive(*Response) error
	Close() error
	Reconnect() error
}

Client interacts with a worker

type Loop

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

Loop represents the processing loop of a worker

func NewLoop

func NewLoop(server Server) *Loop

NewLoop creates a worker loop

func (*Loop) Run

func (l *Loop) Run(handler RequestHandler)

Run starts the worker loop

type Pool

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

Pool represents a pool of workers

func (*Pool) AddWorker

func (p *Pool) AddWorker(worker *PoolWorker)

AddWorker adds a worker to the pool

func (*Pool) Process

func (p *Pool) Process(height int64)

Process schedules the processing of a given height

func (*Pool) Run

func (p *Pool) Run(handler ResponseHandler)

Run starts the worker pool

func (*Pool) Stop

func (p *Pool) Stop()

Stop stops the worker pool

func (*Pool) Wait

func (p *Pool) Wait()

Wait blocks until all workers are finished

type PoolWorker

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

PoolWorker represents a worker in a pool

func NewPoolWorker

func NewPoolWorker(client Client) *PoolWorker

NewPoolWorker creates a pool worker

func (*PoolWorker) Run

func (pw *PoolWorker) Run(handler ResponseHandler, wg *sync.WaitGroup)

Run starts the pool worker

func (*PoolWorker) Stop

func (pw *PoolWorker) Stop()

Stop stops the pool worker

type Request

type Request struct {
	Height int64
}

Request respresents a manager request

type RequestHandler

type RequestHandler func(Request) error

RequestHandler handles a request from a manager

type Response

type Response struct {
	Height  int64
	Success bool
	Error   string
}

Response represents a worker response

type ResponseHandler

type ResponseHandler func(Response)

ResponseHandler handles a response from a worker

type Server

type Server interface {
	Receive(*Request) error
	Send(Response) error
}

Server interacts with a manager

type WebsocketClient

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

WebsocketClient interacts with a worker using a websocket

func NewWebsocketClient

func NewWebsocketClient(endpoint string) (*WebsocketClient, error)

NewWebsocketClient creates a websocket client

func (*WebsocketClient) Close

func (wc *WebsocketClient) Close() error

Close closes the websocket connection

func (*WebsocketClient) Connect

func (wc *WebsocketClient) Connect() error

Connect establishes a websocket connection

func (*WebsocketClient) Receive

func (wc *WebsocketClient) Receive(res *Response) error

Receive receives a response from a worker

func (*WebsocketClient) Reconnect

func (wc *WebsocketClient) Reconnect() error

Reconnect reestablishes a websocket connection

func (*WebsocketClient) Send

func (wc *WebsocketClient) Send(req Request) error

Send sends a request to a worker

type WebsocketServer

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

WebsocketServer interacts with a manager using a websocket

func NewWebsocketServer

func NewWebsocketServer(conn *websocket.Conn) *WebsocketServer

NewWebsocketServer creates a websocket server

func (*WebsocketServer) Close

func (ws *WebsocketServer) Close() error

Close closes the websocket connection

func (*WebsocketServer) Receive

func (ws *WebsocketServer) Receive(req *Request) error

Receive receives a request from a manager

func (*WebsocketServer) Send

func (ws *WebsocketServer) Send(res Response) error

Send sens a response back to a manager