taskengine

README

taskengine

Package taskengine can be used to concurrently execute a set of tasks assigned to multiple different workers.

Each worker can works all or a subset of the tasks.
The workers tasks can be automatically reordered to have each task be handled by a worker as soon as possible.

The main types defined by the package are:

• Engine
• Task
• Worker
• WorkerTasks

Engine

The NewEngine function initialize a new Engine object given the list of workers and the tasks of each worker.

func NewEngine(ctx context.Context, ws []*Worker, wts WorkerTasks) (*Engine, error)

The Execute method of the engine object returns a chan in which are enqueued the workers results for the input tasks.

func (eng *Engine) Execute(mode Mode) (chan Result, error)

The Mode enum type represents the mode of execution:

• FirstSuccessOrLastError: for each task it returns only one result: the first success or the last error. If a task can be handled by two or more workers, only the first success result is returned. The remaining job for same task are cancelled.
• FirstSuccessThenCancel: for each task it returns the error results preceding the first success and the first success. The remaining job for the same task are cancelled.
• All: for each task returns the result of all the workers. Multiple success results can be returned.

Task

A Task represents a unit of work to be executed. Each task can be assigned to one or more workers. Two tasks are considered equivalent if they have the same TaskID.
NOTE: tasks with the same TaskID can be different object with different information; this allows a task object assigned to a worker to contain information specific to that worker.

type Task interface {
    TaskID() TaskID      // Unique ID of the task
}

Worker

Each Worker has a WorkFunc that performs the task. Multiple instances of the same worker can be used in order to execute concurrently different tasks assign to the worker.

type Worker struct {
    WorkerID  WorkerID   // Unique ID of the worker
    Instances int        // Number of worker instances
    Work      WorkFunc   // The work function
}

The WorkFunc receives in input a context, the instance number of the worker and the Task, and returns an object that meets the Result interface.

type WorkFunc func(context.Context, int, Task) Result

The Result interface has only the Success method that must returns true in case of success and false otherwise.

type Result interface {
    Success() bool
}

WorkerTasks

WorkerTasks type is a map that contains the tasks list of each WorkerID.

type WorkerTasks map[WorkerID]Tasks

The SortTasks method reorder each worker tasks list to have each task be handled by a worker as soon as possible. For example given the following configuration:

worker1: [task1, task2]
worker2: [task1, task2, task3]
worker3: [task1, task2, task3]

SortTask method would give:

worker1: [task1, task2]
worker2: [task2, task3, task1]
worker3: [task3, task1, task2]

Documentation

Overview

Package taskengine can be used to concurrently execute a set of tasks assigned to multiple different workers.

A Task represents a unit of work to be executed. Each task can be assigned to one or more workers. Two tasks are considered equivalent if they have the same TaskID. Note that tasks with the same TaskID can be different object with different information; this allows a task object assigned to a worker to contain information specific to that worker.

Each Worker has a WorkFunc that performs the task. Multiple instances of the same worker can be used to concurrently execute different tasks assign to the worker.

The execution mode of the task is managed by the engine.Mode parameters:

- FirstSuccessOrLastError: For each task it returns only one result: the first success or the last error. If a task can be handled by two or more workers, only the first success result is returned. The remaining job for same task are skipped.

- UntilFirstSuccess: For each task returns the (not successfull) result of all the workers: after the first success the other requests are cancelled.

- All: For each task returns the result of all the workers. Multiple success results can be returned.

Index

• func Execute(ctx context.Context, workers []*Worker, tasks WorkerTasks, mode Mode) (chan Result, error)
• type Mode
• type Result
• type Task
• type TaskID
• type Tasks
  • func (ts *Tasks) Remove(i int) Task
• type WorkFunc
• type Worker
• type WorkerID
• type WorkerTasks
  • func (wts WorkerTasks) Clone() WorkerTasks

Functions

func Execute

func Execute(ctx context.Context, workers []*Worker, tasks WorkerTasks, mode Mode) (chan Result, error)

Execute function returns a chan that receives the Results of the workers for the input Requests.

Types

type Mode

type Mode int

Mode of execution for each task.

const (
	// For each task returns only one result:
	// the first success or the last error.
	FirstSuccessOrLastError Mode = iota

	// For each task returns the results until the first success:
	// after the first success the other requests are cancelled and not returned.
	// At most one success is returned.
	UntilFirstSuccess

	// For each task returns the result of all the workers.
	// Multiple success results can be returned.
	// After the first success, the remaining requests are cancelled.
	All
)

Values of mode of execution for each task.

type Result

type Result interface {
	// Success return true in case of a success response.
	// In this case no other Request will be worked for the same Job.
	Success() bool
}

Result is the interface that must be matched by the output of the Work function.

type Task

type Task interface {
	TaskID() TaskID
}

Task is a unit of work that can be executed by a worker Two or more task with the same TaskID are equivalent and possibly only one will be executed. Two or more task with the same TaskID can contain different information usefull for a specific worker.

type TaskID

type TaskID string

TaskID type definition.

type Tasks

type Tasks []Task

Tasks is an array of tasks.

func (*Tasks) Remove

func (ts *Tasks) Remove(i int) Task

Remove removes the i-th task of the list. It returns the rask removed.

type WorkFunc

type WorkFunc func(context.Context, int, Task) Result

WorkFunc is the worker function. - context: the context - int: the instance id of the worker - Task: the task to be eecuted

type Worker

type Worker struct {

	// Unique ID of the worker
	WorkerID WorkerID

	// Number of worker instances. Must be greater or equal 1
	Instances int

	// The work function
	Work WorkFunc
}

Worker is the unit (identified by WorkerID) that receives the Requests and executes a specific WorkFunc function to return the Responses. The Instances parameters represents the number of instances of each worker

type WorkerID

type WorkerID string

WorkerID type definition.

type WorkerTasks

type WorkerTasks map[WorkerID]Tasks

WorkerTasks is a map representing the tasks list of each worker

func (WorkerTasks) Clone

func (wts WorkerTasks) Clone() WorkerTasks

Clone method returns a cloned copy of the WorkerTasks object.