batchworkerpool

Documentation

Index

• Variables
• type BatchWorkerPool
  • func New(workerFnc func([]Task), optionalOptions ...Option) (result *BatchWorkerPool)
  • func (wp *BatchWorkerPool) GetBatchSize() int
  • func (wp *BatchWorkerPool) GetPendingQueueSize() int
  • func (wp *BatchWorkerPool) GetWorkerCount() int
  • func (wp *BatchWorkerPool) Run()
  • func (wp *BatchWorkerPool) Start()
  • func (wp *BatchWorkerPool) Stop()
  • func (wp *BatchWorkerPool) StopAndWait()
  • func (wp *BatchWorkerPool) Submit(params ...interface{}) (result chan interface{}, added bool)
• type Option
  • func BatchCollectionTimeout(batchCollectionTimeout time.Duration) Option
  • func BatchSize(batchSize int) 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{})

Variables

var DEFAULT_OPTIONS = &Options{
	WorkerCount:            2 * runtime.NumCPU(),
	QueueSize:              2 * runtime.NumCPU() * 64,
	BatchSize:              64,
	BatchCollectionTimeout: 15 * time.Millisecond,
	FlushTasksAtShutdown:   false,
}

Types

type BatchWorkerPool

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

func New

func New(workerFnc func([]Task), optionalOptions ...Option) (result *BatchWorkerPool)

func (*BatchWorkerPool) GetBatchSize

func (wp *BatchWorkerPool) GetBatchSize() int

func (*BatchWorkerPool) GetPendingQueueSize

func (wp *BatchWorkerPool) GetPendingQueueSize() int

func (*BatchWorkerPool) GetWorkerCount

func (wp *BatchWorkerPool) GetWorkerCount() int

func (*BatchWorkerPool) Run

func (wp *BatchWorkerPool) Run()

func (*BatchWorkerPool) Start

func (wp *BatchWorkerPool) Start()

func (*BatchWorkerPool) Stop

func (wp *BatchWorkerPool) Stop()

func (*BatchWorkerPool) StopAndWait

func (wp *BatchWorkerPool) StopAndWait()

func (*BatchWorkerPool) Submit

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

type Option

type Option func(*Options)

func BatchCollectionTimeout

func BatchCollectionTimeout(batchCollectionTimeout time.Duration) Option

func BatchSize

func BatchSize(batchSize int) Option

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
	BatchSize              int
	BatchCollectionTimeout time.Duration
	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{})