README
¶
pool
A bounded worker pool (#804, ADR 0216). Pool runs tasks on a fixed set of
long-lived workers fed by a bounded queue: a refusal when there is no room, a
drain, and a task's panic kept on its worker.
p := pool.NewPool(ctx, pool.Config{
Workers: 8,
Queue: 64,
OnPanic: func(pv pool.PanicValue) { report(pv) },
OnError: func(err error) { report(err) },
})
if err := p.Submit(ctx, handle); errs.HasCode(err, pool.CodePoolFull) {
// shed the load
}
err := p.Drain(ctx) // runs what is queued, the workers exit
Queue0:Submithands the task to an idle worker or returnsPoolFullat once.Queue> 0: a full queue is waited on within the submitter's context.Drain(ctx)runs what is queued;Stop(ctx)abandons it and cancels the running tasks' context. Both wait for the workers withinctx.- A task receives the pool's context; what it needs from its submitter travels
in its closure, so a
Submitallocates nothing.
Published as pkg/v1/concur/pool.
Documentation
¶
Overview ¶
Package pool provides Pool, a bounded worker pool: a fixed set of long-lived workers consuming a bounded queue of tasks, with a refusal when there is no room, a drain, and a task's panic isolated on its worker.
It is the shape group and worker are not. group is a one-shot fan-out — a goroutine per task, one wait, used once —, worker is one daemon. A service that runs the same kind of work for as long as it lives — a probe per check, a fire per scheduled job, a sender per batch — needs workers that outlive a task, a queue that refuses when full instead of growing, and a fault in one task that does not end the others.
Submit hands a task over: without a queue (Config.Queue 0) an idle worker takes it or PoolFull answers at once — a refusal, never a park (ADR 0031) —; with one, a full queue is waited on within the submitter's context. Drain stops accepting and runs what is queued; Stop abandons the queue and cancels the context the running tasks received; both wait for the workers within their own context. A task runs under the pool's context — New's, cancelled by Stop —, not the submitter's: a per-task context carrying the submitter's values would cost an allocation per task, so the values a task needs travel in its closure.
A task's panic is recovered on its worker into a PanicValue with the worker's stack and handed to Config.OnPanic, and the worker serves the next task: unlike group, which re-raises because it is one-shot, the pool outlives a task's fault. With no OnPanic the panic is raised again — a fault nobody observes is not swallowed.
Stdlib-only and generic. Range 0.1.6.* (ADR 0005 kernel block, ADR 0216).
Package pool — the Config value type.
Package pool — declares the sentinels returned by Submit.
Index ¶
Constants ¶
const CodePoolClosed errs.Code = 0x00_01_06_02 // 0.1.6.2
CodePoolClosed identifies a Submit made after Drain or Stop.
const CodePoolFull errs.Code = 0x00_01_06_01 // 0.1.6.1
CodePoolFull identifies a Submit that found no room for its task.
Variables ¶
var ( // PoolFull is returned by Submit when there is no room for the task: every // worker busy and no queue, or a full queue waited on until the submitter's // context ended — then wrapping that context's error. PoolFull = errs.Static(CodePoolFull, "POOL_FULL", "The worker pool is full", "internal/kernel/concur/pool: Submit found every worker busy and the queue full", 0, 0) // PoolClosed is returned by Submit once Drain or Stop has begun: the pool // takes no more tasks. PoolClosed = errs.Static(CodePoolClosed, "POOL_CLOSED", "The worker pool is closed", "internal/kernel/concur/pool: Submit called after Drain or Stop", 0, 0) )
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// Workers is how many tasks run at once, each on a long-lived worker; a
// value below 1 is 1.
Workers int
// Queue is how many submitted tasks may wait for a worker. Zero is no
// queue: Submit refuses at once when every worker is busy. A positive
// queue makes a Submit that finds it full wait, within its context. A
// negative value is zero.
Queue int
// OnPanic receives a task's panic, recovered on its worker, which then
// serves the next task. Nil raises the panic again on the worker: a fault
// nobody observes is not swallowed, and the process ends as it would for
// a bare goroutine.
OnPanic func(p PanicValue)
// OnError receives a task's non-nil error, which has no caller to return
// to. Nil discards it.
OnError func(err error)
}
Config sizes a Pool and names the observers of what its tasks cannot return. The zero Config is usable: one worker, no queue, a task's error discarded and its panic raised again.
type PanicValue ¶
type PanicValue struct {
// Raised is the value the original panic carried, verbatim.
Raised any
// Stack is the stack of the worker goroutine, captured at recovery.
Stack []byte
}
PanicValue carries a panic a task raised to Config.OnPanic: Raised is the value the panic carried, verbatim, and Stack the stack of the worker that ran the task, captured at recovery. It is not an error and carries no code: a panic is a programming fault.
func (PanicValue) String ¶
func (p PanicValue) String() string
String renders the original panic followed by the stack of the worker that raised it, so a re-raise with no OnPanic prints the task's stack.
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool is a fixed set of long-lived workers consuming a bounded queue of tasks. Submit hands a task over — refused with PoolFull when there is no room —, Drain stops accepting and runs what is queued, Stop abandons the queue and cancels what runs. A task's panic is recovered on its worker and handed to Config.OnPanic, and the worker serves the next task. Safe for concurrent use; the zero value is NOT usable, build it with NewPool.
func NewPool ¶
NewPool starts a Pool of cfg.Workers workers — at least one — over a queue of cfg.Queue tasks, its tasks running under a context derived from ctx: the values ctx carries, its cancellation, and Stop's. Its goroutines — the workers and one that waits for them — live until Drain or Stop, which join them within their own context.
func (*Pool) Drain ¶
Drain stops accepting tasks, lets the workers run every task already queued, and waits for them to exit within ctx: nil once they have, ctx's error when it ends first — the workers then go on until the queue is empty; Stop abandons it. Idempotent.
func (*Pool) Stop ¶
Stop stops accepting tasks, abandons those still queued — they never run —, cancels the context the running ones received, and waits for them within ctx: nil once every worker has exited, ctx's error when it ends first. A task that ignores its context cannot be stopped from outside. Idempotent.
func (*Pool) Submit ¶
Submit hands task to the pool. Without a queue it is taken by an idle worker or refused with PoolFull at once — never parked. With one, a full queue is waited on until ctx ends, which returns PoolFull wrapping ctx's error. After Drain or Stop it returns PoolClosed. A task accepted runs exactly once, unless Stop abandons it first. A nil task panics.
Source Files