workerpool

Documentation

Index

• type BlockingQueueWorkerPoolTask
• type BlockingQueuedWorkerPool
  • func NewBlockingQueuedWorkerPool(optionalOptions ...Option) (result *BlockingQueuedWorkerPool)
  • func (b *BlockingQueuedWorkerPool) CreateTask(f func(), optionalStackTrace ...string) *BlockingQueueWorkerPoolTask
  • func (b *BlockingQueuedWorkerPool) DecreasePendingTasksCounter()
  • func (b *BlockingQueuedWorkerPool) GetPendingQueueSize() int
  • func (b *BlockingQueuedWorkerPool) GetWorkerCount() int
  • func (b *BlockingQueuedWorkerPool) IncreasePendingTasksCounter()
  • func (b *BlockingQueuedWorkerPool) IsRunning() (isRunning bool)
  • func (b *BlockingQueuedWorkerPool) Run()
  • func (b *BlockingQueuedWorkerPool) Start()
  • func (b *BlockingQueuedWorkerPool) Stop()
  • func (b *BlockingQueuedWorkerPool) StopAndWait()
  • func (b *BlockingQueuedWorkerPool) Submit(handler func())
  • func (b *BlockingQueuedWorkerPool) SubmitTask(task *BlockingQueueWorkerPoolTask)
  • func (b *BlockingQueuedWorkerPool) TrySubmit(f func()) (added bool)
  • func (b *BlockingQueuedWorkerPool) TrySubmitTask(task *BlockingQueueWorkerPoolTask) (added bool)
  • func (b *BlockingQueuedWorkerPool) WaitUntilAllTasksProcessed()
• type NonBlockingQueuedWorkerPool
  • func NewNonBlockingQueuedWorkerPool(workerFunc func(Task), optionalOptions ...Option) (result *NonBlockingQueuedWorkerPool)
  • func (wp *NonBlockingQueuedWorkerPool) GetPendingQueueSize() int
  • func (wp *NonBlockingQueuedWorkerPool) GetWorkerCount() int
  • func (wp *NonBlockingQueuedWorkerPool) Stop()
  • func (wp *NonBlockingQueuedWorkerPool) StopAndWait()
  • func (wp *NonBlockingQueuedWorkerPool) Submit(params ...interface{}) (chan interface{}, bool)
  • func (wp *NonBlockingQueuedWorkerPool) TrySubmit(params ...interface{}) (result chan interface{}, added bool)
• type Option
  • func FlushTasksAtShutdown(flush bool) Option
  • func QueueSize(queueSize int) Option
  • func WithAlias(alias string) Option
  • func WorkerCount(workerCount int) Option
• type Options
  • func (options Options) Override(optionalOptions ...Option) *Options
• type Task
  • func (task *Task) Param(index int) interface{}
  • func (task *Task) Return(result interface{})
• type WorkerPool
  • func New(workerFnc func(Task), optionalOptions ...Option) (result *WorkerPool)
  • func (wp *WorkerPool) GetPendingQueueSize() int
  • func (wp *WorkerPool) GetWorkerCount() int
  • func (wp *WorkerPool) Run()
  • func (wp *WorkerPool) Start()
  • func (wp *WorkerPool) Stop()
  • func (wp *WorkerPool) StopAndWait()
  • func (wp *WorkerPool) Submit(params ...interface{}) (result chan interface{}, added bool)
  • func (wp *WorkerPool) TrySubmit(params ...interface{}) (result chan interface{}, added bool)

Types

type BlockingQueueWorkerPoolTask

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

BlockingQueueWorkerPoolTask is a task that is executed by a BlockingQueuedWorkerPool.

type BlockingQueuedWorkerPool

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

BlockingQueuedWorkerPool represents a set of workers with a blocking queue of pending tasks.

func NewBlockingQueuedWorkerPool

func NewBlockingQueuedWorkerPool(optionalOptions ...Option) (result *BlockingQueuedWorkerPool)

NewBlockingQueuedWorkerPool returns a new stopped WorkerPool.

func (*BlockingQueuedWorkerPool) CreateTask

func (b *BlockingQueuedWorkerPool) CreateTask(f func(), optionalStackTrace ...string) *BlockingQueueWorkerPoolTask

CreateTask creates a new BlockingQueueWorkerPoolTask with the given handler and optional ClosureStackTrace.

func (*BlockingQueuedWorkerPool) DecreasePendingTasksCounter

func (b *BlockingQueuedWorkerPool) DecreasePendingTasksCounter()

DecreasePendingTasksCounter decreases the pending task counter.

func (*BlockingQueuedWorkerPool) GetPendingQueueSize

func (b *BlockingQueuedWorkerPool) GetPendingQueueSize() int

GetPendingQueueSize returns the amount of tasks pending to the processed.

func (*BlockingQueuedWorkerPool) GetWorkerCount

func (b *BlockingQueuedWorkerPool) GetWorkerCount() int

GetWorkerCount returns the worker count for the WorkerPool.

func (*BlockingQueuedWorkerPool) IncreasePendingTasksCounter

func (b *BlockingQueuedWorkerPool) IncreasePendingTasksCounter()

IncreasePendingTasksCounter increases the pending task counter.

func (*BlockingQueuedWorkerPool) IsRunning

func (b *BlockingQueuedWorkerPool) IsRunning() (isRunning bool)

IsRunning returns true if the WorkerPool is running.

func (*BlockingQueuedWorkerPool) Run

func (b *BlockingQueuedWorkerPool) Run()

Run starts the WorkerPool and waits for its shutdown.

func (*BlockingQueuedWorkerPool) Start

func (b *BlockingQueuedWorkerPool) Start()

Start starts the WorkerPool (non-blocking).

func (*BlockingQueuedWorkerPool) Stop

func (b *BlockingQueuedWorkerPool) Stop()

Stop stops the WorkerPool.

func (*BlockingQueuedWorkerPool) StopAndWait

func (b *BlockingQueuedWorkerPool) StopAndWait()

StopAndWait stops the WorkerPool and waits for its shutdown.

func (*BlockingQueuedWorkerPool) Submit

func (b *BlockingQueuedWorkerPool) Submit(handler func())

Submit submits a handler function to the queue and blocks if the queue is full.

func (*BlockingQueuedWorkerPool) SubmitTask

func (b *BlockingQueuedWorkerPool) SubmitTask(task *BlockingQueueWorkerPoolTask)

SubmitTask submits a task to the queue and blocks if the queue is full (it should only be used instead of Submit if manually handling the task is necessary to create better debug outputs).

func (*BlockingQueuedWorkerPool) TrySubmit

func (b *BlockingQueuedWorkerPool) TrySubmit(f func()) (added bool)

TrySubmit tries to queue the execution of the handler function and ignores the handler if there is no capacity for it to be added.

func (*BlockingQueuedWorkerPool) TrySubmitTask

func (b *BlockingQueuedWorkerPool) TrySubmitTask(task *BlockingQueueWorkerPoolTask) (added bool)

TrySubmitTask tries to queue the execution of the task and ignores the task if there is no capacity for it to be added (it should only be used instead of TrySubmit if manually handling the task is necessary to create better debug outputs).

func (*BlockingQueuedWorkerPool) WaitUntilAllTasksProcessed

func (b *BlockingQueuedWorkerPool) WaitUntilAllTasksProcessed()

WaitUntilAllTasksProcessed waits until all tasks are processed.

type NonBlockingQueuedWorkerPool

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

NonBlockingQueuedWorkerPool implements a non-blocking goroutine pool backed by a queue.

func NewNonBlockingQueuedWorkerPool

func NewNonBlockingQueuedWorkerPool(workerFunc func(Task), optionalOptions ...Option) (result *NonBlockingQueuedWorkerPool)

NewNonBlockingQueuedWorkerPool creates and starts a new worker pool for the supplied function, with the supplied options.

func (*NonBlockingQueuedWorkerPool) GetPendingQueueSize

func (wp *NonBlockingQueuedWorkerPool) GetPendingQueueSize() int

GetPendingQueueSize gets the current amount of pending tasks in the queue.

func (*NonBlockingQueuedWorkerPool) GetWorkerCount

func (wp *NonBlockingQueuedWorkerPool) GetWorkerCount() int

GetWorkerCount gets the configured worker count.

func (*NonBlockingQueuedWorkerPool) Stop

func (wp *NonBlockingQueuedWorkerPool) Stop()

Stop closes this pool. If FlushTasksAtShutdown was set, it allows currently running and pending tasks to complete.

func (*NonBlockingQueuedWorkerPool) StopAndWait

func (wp *NonBlockingQueuedWorkerPool) StopAndWait()

StopAndWait closes the pool and waits for tasks to complete.

func (*NonBlockingQueuedWorkerPool) Submit

func (wp *NonBlockingQueuedWorkerPool) Submit(params ...interface{}) (chan interface{}, bool)

Submit is an alias for TrySubmit.

func (*NonBlockingQueuedWorkerPool) TrySubmit

func (wp *NonBlockingQueuedWorkerPool) TrySubmit(params ...interface{}) (result chan interface{}, added bool)

TrySubmit submits a BlockingQueueWorkerPoolTask to this pool (it drops the BlockingQueueWorkerPoolTask if not enough workers are available and the queue is full). It returns a channel to obtain the BlockingQueueWorkerPoolTask result, and a boolean if the BlockingQueueWorkerPoolTask was successfully submitted to the queue.

type Option

type Option func(*Options)

func FlushTasksAtShutdown

func FlushTasksAtShutdown(flush bool) Option

func QueueSize

func QueueSize(queueSize int) Option

func WithAlias

func WithAlias(alias string) Option

func WorkerCount

func WorkerCount(workerCount int) Option

type Options

type Options struct {
	Alias                string
	WorkerCount          int
	QueueSize            int
	FlushTasksAtShutdown bool
}

func (Options) Override

func (options Options) Override(optionalOptions ...Option) *Options

type Task

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

func (*Task) Param

func (task *Task) Param(index int) interface{}

func (*Task) Return

func (task *Task) Return(result interface{})

type WorkerPool

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

func New

func New(workerFnc func(Task), optionalOptions ...Option) (result *WorkerPool)

func (*WorkerPool) GetPendingQueueSize

func (wp *WorkerPool) GetPendingQueueSize() int

func (*WorkerPool) GetWorkerCount

func (wp *WorkerPool) GetWorkerCount() int

func (*WorkerPool) Run

func (wp *WorkerPool) Run()

func (*WorkerPool) Start

func (wp *WorkerPool) Start()

func (*WorkerPool) Stop

func (wp *WorkerPool) Stop()

func (*WorkerPool) StopAndWait

func (wp *WorkerPool) StopAndWait()

func (*WorkerPool) Submit

func (wp *WorkerPool) Submit(params ...interface{}) (result chan interface{}, added bool)

func (*WorkerPool) TrySubmit

func (wp *WorkerPool) TrySubmit(params ...interface{}) (result chan interface{}, added bool)