spool

README

spool

spool is a simple worker-pool for Go. Each job is a func() which is sent to the worker-pool. A worker will pick up the job from the mailbox and execute it.

First, initialize the worker-poll with a mailbox of a certain size:

pool := New(n)
defer pool.Stop()

The mailbox is just a chan func(). In fact the worker-pool itself is defined as:

type WorkerPool chan func()

Jobs can be sent to the worker-pool in two different manners, blocking and nonblocking. To send a job to the worker-pool and block until it's completed:

pool.Blocking(ctx, func() {
    // ...
})

And to send a job to the worker-pool and then move on:

pool.SemiBlocking(ctx, func() {
    // ...
})

As long as there is an empty space in the mailbox, SemiBlocking will just queue the job, and moves on. When there are no more empty spaces in the mailbox, SemiBlocking becomes blocking.

A worker-pool by default has no workers and they should be added explicitly. To add workers to the worker-pool:

pool.Grow(ctx, 10)

Now, the worker-pool has 10 workers.

It's possible to add temporary workers to the worker-pool:

pool.Grow(ctx, 9, WithAbsoluteTimeout(time.Minute * 5))

Also, instead of using and absolute timeout, an idle timeout can be used. In this case, added workers will stop, if they are idle for a certain duration:

pool.Grow(ctx, 9, WithIdleTimeout(time.Minute * 5))

The Blocking and SemiBlocking methods will panic if the worker-pool is stopped - to enforce visibility on job execution.

spool serializes the jobs in single worker mode

• TODO

Documentation

Index

• type MailboxSize
• type T
• type WorkerPool
  • func New(mailboxSize MailboxSize) WorkerPool
  • func (pool WorkerPool) Blocking(ctx context.Context, callback func()) error
  • func (pool WorkerPool) Grow(ctx context.Context, growth int, options ...actor.Option)
  • func (pool WorkerPool) SemiBlocking(ctx context.Context, callback func()) error
  • func (pool WorkerPool) Stop()

Examples

• WorkerPool.Blocking
• WorkerPool.Grow
• WorkerPool.SemiBlocking

Types

type MailboxSize

type MailboxSize int

type T

type T = func()

type WorkerPool

type WorkerPool chan func()

func New

func New(mailboxSize MailboxSize) WorkerPool

New creates a new WorkerPool without any initial workers. To spawn workers, Grow must be called.

func (WorkerPool) Blocking

func (pool WorkerPool) Blocking(ctx context.Context, callback func()) error

Blocking will panic, if the workerpool is stopped.

Example

pool := New(1)
defer pool.Stop()
pool.Grow(context.Background(), 1)

var state int64
job := func() { atomic.AddInt64(&state, 19) }

_ = pool.Blocking(context.Background(), job)

fmt.Println(atomic.LoadInt64(&state))

Output:

19

func (WorkerPool) Grow

func (pool WorkerPool) Grow(ctx context.Context, growth int, options ...actor.Option)

Example

const n = 19
pool := New(1)
defer pool.Stop()

pool.Grow(context.Background(), 3) // spin up three new workers

var state int64
wg := &sync.WaitGroup{}
wg.Add(n)
for i := 0; i < n; i++ {
	_ = pool.SemiBlocking(context.Background(), func() { defer wg.Done(); atomic.AddInt64(&state, 1) })
}
wg.Wait()

fmt.Println(state)

Output:

19

func (WorkerPool) SemiBlocking

func (pool WorkerPool) SemiBlocking(ctx context.Context, callback func()) error

SemiBlocking sends the job to the worker in a non-blocking manner, as long as the mailbox is not full. After that, it becomes blocking until there is an empty space in the mailbox. If the workerpool is stopped, SemiBlocking will panic.

Example

pool := New(1)
defer pool.Stop()
pool.Grow(context.Background(), 1)

var state int64
jobDone := make(chan struct{})
job := func() {
	defer close(jobDone)
	atomic.AddInt64(&state, 19)
}

_ = pool.SemiBlocking(context.Background(), job)
<-jobDone

fmt.Println(state)

Output:

19

func (WorkerPool) Stop

func (pool WorkerPool) Stop()