README
¶
EVA
Package EVA implements a fixed goroutine pool for managing and recycling a massive number of goroutines with unlimited task queue, allowing developers to limit the number of goroutines that created by your concurrent programs. For example, when processing jobs from HTTP requests that are CPU heavy you can create a pool with a size that matches your CPU count.
Features:
- Friendly interfaces: submitting tasks, getting the number. of running goroutines, readjusting capacity of pool dynamically, closing pool.
- Automatically managing and recycling a massive number of goroutines.
- Availability of overriding the Task or Callable interface.
- Unlimited task queue.
- Recover panic in eva.CustomTask.
Installation
go get github.com/pshopper/eva
Or, using dep:
dep ensure -add github.com/pshopper/eva
Usage
Async
Simple
config := &eva.Config{Size: 10, UnstoppableWorkers: 5}
p := eva.NewPool(&config)
t := eva.NewCustomTask(func(args ...interface{}) (interface{}, error) {
// some work here...
return result, err
}, args)
p.Submit(t)
p.Wait()
fmt.Printf("Task result=%v; error=%v; panic=%v, t.Get(), t.Error(), t.Panic())
p.Close()
With completion
config := &eva.Config{Size: 10, UnstoppableWorkers: 5}
p := eva.NewPool(&config)
completion := make(chan eva.Task, 3)
for i := 0; i < 100; i++ {
t := eva.NewCustomTask(func(args ...interface{}) (interface{}, error) {
// some work here...
return result, err
}, args)
p.SubmitWithCompletion(completion, t)
}
for i := 0; i < 100; i++ {
select {
case t := <-completion:
fmt.Printf("Task result=%v; error=%v; panic=%v, t.Get(), t.Error(), t.Panic())
}
}
p.Close()
With context
config := &eva.Config{Size: 10, UnstoppableWorkers: 5}
p := eva.NewPool(&config)
context, cancel := context.WithCancel(context.Background()) // WithDeadline or WithTimeout
for i := 0; i < 100; i++ {
t := eva.NewCustomTask(func(args ...interface{}) (interface{}, error) {
// some work here...
return result, err
}, args)
p.SubmitWithContext(context, t)
}
// some work here...
cancel()
p.Close()
With custom
config := &eva.Config{Size: 10, UnstoppableWorkers: 5}
p := eva.NewPool(&config)
context, cancel := context.WithCancel(context.Background()) // WithDeadline or WithTimeout
completion := make(chan eva.Task, 3)
for i := 0; i < 100; i++ {
t := eva.NewCustomTask(func(args ...interface{}) (interface{}, error) {
// some work here...
return result, err
}, args)
p.SubmitCustom(context, completion, t)
}
for i := 0; i < 3; i++ {
select {
case t := <-completion:
fmt.Printf("Task result=%v; error=%v; panic=%v, t.Get(), t.Error(), t.Panic())
}
}
cancel()
p.Close()
Immediate
config := &eva.Config{Size: 10, UnstoppableWorkers: 5}
p := eva.NewPool(&config)
t := eva.NewCustomTask(func(args ...interface{}) (interface{}, error) {
// some work here...
return result, err
}, args)
err := p.SubmitImmediate(t)
if err == nil {
fmt.Printf("Task result=%v; error=%v; panic=%v, t.Get(), t.Error(), t.Panic())
}
p.Close()
Sync
config := &eva.Config{Size: 10, UnstoppableWorkers: 5}
p := eva.NewPool(&config)
t := eva.NewCustomTask(func(args ...interface{}) (interface{}, error) {
// some work here...
return result, err
}, args)
p.Submit(t)
fmt.Printf("Task result=%v; error=%v; panic=%v, t.Get(), t.Error(), t.Panic()) // like CompletableFuture
p.Close()
Resize
You can change pool size:
pool.SetSize(1000)
pool.SetSize(1000000)
Or you can change changes unstoppable workers count
pool.SetUnstoppableWorkers(1000)
pool.SetUnstoppableWorkers(1000000)
This is safe to perform from any goroutine even if others are still processing.
Interfaces
You can implement Task interface and submit it in pool:
// Task interface for goroutines pool.
type Task interface {
// Cancel perform cancel task.
Cancel()
// Error returns task Run error.
// Blocked until the task is done.
Error() error
// Get returns task Run result.
// Blocked until the task is done.
Get() interface{}
// IsCancelled returns true if this task has been canceled.
IsCancelled() bool
// IsDone returns true if this task has been done.
IsDone() bool
// Panic returns task Run panic.
// Blocked until the task is done.
Panic() interface{}
// Run perform task.
Run()
}
Ordering
All the tasks submitted to EVA pool will not be guaranteed to be processed in order, because those tasks distribute among a series of concurrent workers, thus those tasks are processed concurrently.
Documentation
¶
Index ¶
- Constants
- Variables
- type Callable
- type Config
- type CustomCallable
- type CustomTask
- type Pool
- func (p *Pool) ActiveCount() int
- func (p *Pool) Close()
- func (p *Pool) IsClosed() bool
- func (p *Pool) SetSize(size int) error
- func (p *Pool) SetUnstoppableWorkers(count int) error
- func (p *Pool) Size() int
- func (p *Pool) Submit(t Task) error
- func (p *Pool) SubmitCustom(ctx context.Context, completion chan<- Task, t Task) error
- func (p *Pool) SubmitImmediate(t Task) error
- func (p *Pool) SubmitWithCompletion(completion chan<- Task, t Task) error
- func (p *Pool) SubmitWithContext(ctx context.Context, t Task) error
- func (p *Pool) TaskQueue() TaskQueue
- func (p *Pool) UnstoppableWorkers() int
- func (p *Pool) Wait()
- type Task
- type TaskQueue
Constants ¶
const (
DefaultSize = 10
)
Constants that are used throughout the Eva API.
Variables ¶
var ( ErrPoolSize = errors.New("pool size less than or equal 0") ErrUnstoppableWorkers = errors.New("unstoppable workers count less than 0") ErrPoolClosed = errors.New("pool closed") )
Errors that are used throughout the Eva API.
var ( ErrTaskQueueClosed = errors.New("task queue closed") ErrTaskQueueEmpty = errors.New("task queue is empty") )
Errors that are used throughout the Eva API.
Functions ¶
This section is empty.
Types ¶
type Callable ¶
type Callable interface {
// Call perform function.
Call()
// Error returns function Call error.
Error() error
// Result returns function Call result.
Result() interface{}
}
Callable is interface, that Call function for task in goroutines pool.
type Config ¶
type Config struct {
//Maximum number of workers that could be spawned.
// It includes ActiveWorkers count.
// If size == 0, then size poll change set to default (10)
Size int
// Number of workers that are always running.
// If UnstoppableWorkers > Size, the Size will be overwritten
UnstoppableWorkers int
}
Config is pool configuration
type CustomCallable ¶
type CustomCallable struct {
// contains filtered or unexported fields
}
CustomCallable (implement Callable interface) for task in goroutines pool.
func NewCustomCallable ¶
func NewCustomCallable(f func(...interface{}) (interface{}, error), args ...interface{}) *CustomCallable
NewCustomCallable is constructor for CustomCallable
func (*CustomCallable) Error ¶
func (cc *CustomCallable) Error() error
Error returns function Call error.
func (*CustomCallable) Result ¶
func (cc *CustomCallable) Result() interface{}
Result returns function Call result.
type CustomTask ¶
type CustomTask struct {
// contains filtered or unexported fields
}
CustomTask (implement Task interface) for goroutines pool.
func NewCustomTask ¶
func NewCustomTask(f func(...interface{}) (interface{}, error), args ...interface{}) *CustomTask
NewCustomTask is constructor for CustomTask
func (*CustomTask) Error ¶
func (ct *CustomTask) Error() error
Error returns task Run error. Blocked until the task is done.
func (*CustomTask) Get ¶
func (ct *CustomTask) Get() interface{}
Get returns task Run result. Blocked until the task is done.
func (*CustomTask) IsCancelled ¶
func (ct *CustomTask) IsCancelled() bool
IsCancelled returns true if this task has been canceled.
func (*CustomTask) IsDone ¶
func (ct *CustomTask) IsDone() bool
IsDone returns true if this task has been done.
func (*CustomTask) Panic ¶
func (ct *CustomTask) Panic() interface{}
Panic returns task Run panic. Blocked until the task is done.
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool is a struct that manages a collection of workers, each with their own goroutine. The Pool can initialize, expand, compress and close the workers, as well as processing task.
func (*Pool) ActiveCount ¶
ActiveCount returns the approximate number of workers that are actively executing tasks or idle.
func (*Pool) SetSize ¶
SetSize changes pool size. It returns ErrPoolClosed only if pool become closed. It returns ErrPoolSize only if size < 0.
func (*Pool) SetUnstoppableWorkers ¶
SetUnstoppableWorkers changes unstoppable workers count. It returns ErrPoolClosed only if pool become closed. It returns ErrUnstoppableWorkers only if count < 0.
func (*Pool) Submit ¶
Submit makes task to be scheduled over pool's workers. It returns ErrPoolClosed only if pool become closed and task can not be executed over its workers.
func (*Pool) SubmitCustom ¶
SubmitCustom is a combination of SubmitWithCompletion and SubmitWithContext.
func (*Pool) SubmitImmediate ¶
SubmitImmediate makes task to be scheduled without waiting for free workers. That is, if all workers are busy and pool is not rubber, then ErrUnavailable is returned immediately. It returns ErrPoolClosed only if pool become closed and task can not be executed over its workers.
func (*Pool) SubmitWithCompletion ¶
SubmitWithCompletion makes task to be scheduled over pool's workers with returning completed task by completion channel. It returns ErrPoolClosed only if pool become closed and task can not be executed over its workers. Be careful, may deadlock! You need read-loop from completion channel.
func (*Pool) SubmitWithContext ¶
SubmitWithContext makes task to be scheduled over pool's workers with context support. It returns ErrPoolClosed only if pool become closed and task can not be executed over its workers. It returns ctx.Err() error only available when ctx is done.
func (*Pool) UnstoppableWorkers ¶
UnstoppableWorkers returns the current number of workers in the pool.
type Task ¶
type Task interface {
// Cancel perform cancel task.
Cancel()
// Error returns task Run error.
// Blocked until the task is done.
Error() error
// Get returns task Run result.
// Blocked until the task is done.
Get() interface{}
// IsCancelled returns true if this task has been canceled.
IsCancelled() bool
// IsDone returns true if this task has been done.
IsDone() bool
// Panic returns task Run panic.
// Blocked until the task is done.
Panic() interface{}
// Run perform task.
Run()
}
Task interface for goroutines pool.
type TaskQueue ¶
type TaskQueue interface {
// Put inserts the task into this queue, waiting if necessary for
// space to become available.
Put(t Task) error
// Size returns the current number of tasks in the queue.
Size() int
// Take retrieves and removes the head of this queue, waiting
// if necessary until an task becomes available.
Take() (Task, error)
}
TaskQueue is interface that additionally supports operations that wait for the queue to become non-empty when retrieving a task, and wait for space to become available in the queue when storing a task.
Source Files