GF
Go Foundation API
worker API

worker

package

API reference for the worker package.

Imports

(4)
STD context STD fmt STD sync STD sync/atomic
T
type

Task

Task is a work function executed by the pool.

core/worker/worker.go:11-11 
type Task func(context.Context) error
S
struct

Future

Future holds the result of an asynchronous task.

core/worker/worker.go:14-16 
type Future struct

Methods

Wait
Method

Returns

error 
func (*Future) Wait() error
{
	r := <-f.ch
	return r.Err
}

Fields

Name Type Description
ch chan Result
S
struct

Result

Result holds the outcome of a completed Task.

core/worker/worker.go:19-21 
type Result struct

Fields

Name Type Description
Err error
S
struct

Pool

Pool is a fixed-size worker pool for concurrent task execution.

core/worker/worker.go:29-37 
type Pool struct

Methods

worker
Method

worker pulls tasks from the tasks channel until done is closed. 

func (*Pool) worker()
{
	defer p.wg.Done()
	for {
		select {
		case item := <-p.tasks:
			if p.closed.Load() {
				item.accepted <- false
				continue
			}
			item.accepted <- true
			_ = runTask(p.ctx, item.task)
		case <-p.done:
			return
		}
	}
}
Submit
Method

Submit enqueues a task for execution. Returns false if the pool is shut down or shutting down; the task is not executed in that case.

Parameters

task Task

Returns

bool 
func (*Pool) Submit(task Task) bool
{
	if task == nil || p.closed.Load() {
		return false
	}
	item := work{
		task:     task,
		accepted: make(chan bool, 1),
	}
	select {
	case p.tasks <- item:
		return <-item.accepted
	case <-p.done:
		return false
	}
}
Shutdown
Method

Shutdown stops accepting new tasks and waits for in-progress work to finish. The task channel is unbuffered, so no tasks can be pending pickup when done is closed; any Submit in progress will observe done and return false. 

func (*Pool) Shutdown()
{
	p.once.Do(func() {
		p.closed.Store(true)
		p.cancel()
		close(p.done)
		p.wg.Wait()
	})
}

Fields

Name Type Description
wg sync.WaitGroup
tasks chan work
done chan struct{}
cancel context.CancelFunc
ctx context.Context
once sync.Once
closed atomic.Bool
S
struct

work

core/worker/worker.go:39-42 
type work struct

Fields

Name Type Description
task Task
accepted chan bool

Uses

Task (worker)
F
function

NewPool

NewPool creates a fixed-size worker pool of n goroutines.

Parameters

n
int

Returns

*Pool
core/worker/worker.go:53-70 
func NewPool(n int) *Pool

{
	if n <= 0 {
		n = 1
	}

	ctx, cancel := context.WithCancel(context.Background())
	p := &Pool{
		tasks:  make(chan work),
		done:   make(chan struct{}),
		cancel: cancel,
		ctx:    ctx,
	}
	for i := 0; i < n; i++ {
		p.wg.Add(1)
		go p.worker()
	}
	return p
}

Example 

pool := worker.NewPool(4)
defer pool.Shutdown()
pool.Submit(func(ctx context.Context) error {
	return doWork(ctx)
})
F
function

runTask

Parameters

ctx
context.Context
task
Task

Returns

err
error
core/worker/worker.go:90-97 
func runTask(ctx context.Context, task Task) (err error)

{
	defer func() {
		if recovered := recover(); recovered != nil {
			err = fmt.Errorf("worker: task panic: %v", recovered)
		}
	}()
	return task(ctx)
}

Uses

Task (worker)