workerpool

Documentation

Index

• Variables
• 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 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)

Variables

var DEFAULT_OPTIONS = &Options{
	WorkerCount:          2 * runtime.NumCPU(),
	QueueSize:            4 * runtime.NumCPU(),
	FlushTasksAtShutdown: false,
}

Types

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 task to this pool (it drops the task if not enough workers are available and the queue is full). It returns a channel to obtain the task result, and a boolean if the task 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 WorkerCount

func WorkerCount(workerCount int) Option

type Options

type Options struct {
	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)