Documentation
¶
Overview ¶
Package compute is v5's off-thread pool for CPU-bound work (plan item P3.12).
It exists for the jobs that are neither rendering nor domain state: image decoding, parsing, compression, search indexing. Left on the render thread each one is a long frame; handed to the domain worker they compete with the commands that own application state.
§11-Q4 is resolved as "reuse services.wasm; no third binary" — a third artifact adds a second Go runtime's memory for no isolation, since compute jobs are the app's own Go code either way. The pool scales by worker count.
It reuses internal/services.Dispatcher for WORKER HEALTH and nothing else, and that split is deliberate. The dispatcher assigns units stickily, because a unit's worker holds state for it and moving it throws that away. A compute job holds no state, so stickiness buys nothing and costs balance: hashing job ids across workers leaves one worker with three jobs while another idles. The pool therefore assigns to the least-loaded ready worker.
Index ¶
- Constants
- Variables
- type Assignment
- type Job
- type JobID
- type Options
- type Pool
- func (parsePool *Pool) Complete(parseJobID JobID) bool
- func (parsePool *Pool) Degraded() bool
- func (parsePool *Pool) Dispatch() []Assignment
- func (parsePool *Pool) InFlight() int
- func (parsePool *Pool) QueueDepth() int
- func (parsePool *Pool) ReadyWorkers() []WorkerID
- func (parsePool *Pool) RecordHeartbeat(parseWorkerID WorkerID, parseSequence uint64) error
- func (parsePool *Pool) RecordHeartbeatTimeout(parseWorkerID WorkerID) (WorkerHealth, error)
- func (parsePool *Pool) RestartWorker(parseDeadWorkerID WorkerID, parseReplacementWorkerID WorkerID) (int, error)
- func (parsePool *Pool) SetWorkerHealth(parseWorkerID WorkerID, parseHealth WorkerHealth) error
- func (parsePool *Pool) Submit(parseJob Job) error
- type WorkerHealth
- type WorkerID
Constants ¶
const ( HealthReady = services.HealthReady HealthDegraded = services.HealthDegraded HealthRestarting = services.HealthRestarting HealthDead = services.HealthDead )
The worker health states, re-exported so callers need not reach into an internal package to name them.
const DefaultQueueLimit = 256
DefaultQueueLimit is the queue bound applied when Options.QueueLimit is unset.
Variables ¶
var ErrNoReadyWorker = errors.New("compute: no ready worker is available")
ErrNoReadyWorker is returned when the pool has no worker able to take work.
var ErrQueueFull = errors.New("compute: the job queue is at its limit")
ErrQueueFull is returned when backpressure refuses a submission.
Functions ¶
This section is empty.
Types ¶
type Assignment ¶
Assignment is a job paired with the worker that should run it.
type Job ¶
Job is one unit of CPU-bound work.
Payload is opaque; the pool schedules bytes and never interprets them.
type Options ¶
type Options struct {
// QueueLimit bounds jobs waiting for a worker. Zero uses DefaultQueueLimit.
//
// A bound rather than unbounded growth is the backpressure P3.12 requires:
// an unbounded queue converts a producer that outruns the pool into a memory
// leak that fails much later and somewhere else.
QueueLimit int
// MaxInFlightPerWorker caps concurrent jobs on one worker. Zero means one,
// which is right for a single-threaded wasm worker.
MaxInFlightPerWorker int
}
Options configure a pool.
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool schedules CPU-bound jobs across a set of workers.
Not safe for concurrent use; it belongs to the thread that owns the pool.
func (*Pool) Dispatch ¶
func (parsePool *Pool) Dispatch() []Assignment
Dispatch assigns as many queued jobs as there is capacity for.
Returns the assignments a caller should actually deliver. Jobs beyond the available capacity stay queued in submission order; a pool that dispatched everything and let the workers queue internally would move the backpressure somewhere it cannot be observed.
func (*Pool) QueueDepth ¶
QueueDepth reports how many jobs are waiting.
func (*Pool) ReadyWorkers ¶
ReadyWorkers lists workers able to take jobs.
func (*Pool) RecordHeartbeat ¶
RecordHeartbeat records a worker heartbeat.
func (*Pool) RecordHeartbeatTimeout ¶
func (parsePool *Pool) RecordHeartbeatTimeout(parseWorkerID WorkerID) (WorkerHealth, error)
RecordHeartbeatTimeout records a missed heartbeat window.
func (*Pool) RestartWorker ¶
func (parsePool *Pool) RestartWorker(parseDeadWorkerID WorkerID, parseReplacementWorkerID WorkerID) (int, error)
RestartWorker replaces a worker and returns its in-flight jobs to the queue.
This is P3.12's criterion: a worker restarts WITHOUT LOSING QUEUED JOBS. Two distinct sets have to survive it — the jobs merely waiting, which were never tied to the dead worker, and the jobs already handed to it, which would otherwise vanish with it. The second set is the one an implementation loses by accident.
Requeued jobs go to the FRONT, ahead of jobs submitted later. They have already waited once, and putting them behind newer work would starve exactly the jobs a restart already delayed.
Semantics: AT-LEAST-ONCE. A job interrupted by a worker death is dispatched again, because the pool cannot know how far the dead worker got. Dropping it instead would be worse — silent data loss rather than repeated work — so compute jobs must be idempotent or have a duplicate result discarded. Pinned by TestWorkerRestartLosesNoJobs, which asserts an interrupted job is dispatched exactly twice and an untouched one exactly once.
func (*Pool) SetWorkerHealth ¶
func (parsePool *Pool) SetWorkerHealth(parseWorkerID WorkerID, parseHealth WorkerHealth) error
SetWorkerHealth updates a worker's health.
type WorkerHealth ¶
type WorkerHealth = services.WorkerHealth
WorkerHealth is a worker's pool-visible state.
Source Files