The highest tagged major version is
Documentation
¶
Index ¶
- Variables
- func RegisterQueuePreload(v *vm.VM)
- type Channel
- type Dispatcher
- func (d *Dispatcher) Clear() int
- func (d *Dispatcher) Complete(j *Job, failed bool, wait, exec time.Duration) Outcome
- func (d *Dispatcher) Len() int
- func (d *Dispatcher) MarkExpired()
- func (d *Dispatcher) NextDue(now time.Time) (j *Job, wait time.Duration, ok bool)
- func (d *Dispatcher) Snapshot() Metrics
- func (d *Dispatcher) Stop()
- func (d *Dispatcher) Stopped() bool
- func (d *Dispatcher) Submit(j *Job) error
- func (d *Dispatcher) Wake() <-chan struct{}
- type Job
- type Metrics
- type Outcome
- type Result
Constants ¶
This section is empty.
Variables ¶
var ( ErrFull = errors.New("queue is full") ErrStopped = errors.New("queue is stopped") )
Errors returned by Submit.
Functions ¶
func RegisterQueuePreload ¶
RegisterQueuePreload installs the `queue` module under package.preload.
Types ¶
type Channel ¶
type Channel struct {
// contains filtered or unexported fields
}
Channel is a Go channel carrying Lua values, usable from any goroutine.
The underlying data channel is never closed. Closing it would make a Send racing a Close panic ("send on closed channel") with no way for the sender to defend itself — the check-then-send window is unclosable. Instead Close closes a separate `done` channel, and Send/Receive select on it. That keeps Go's observable close semantics (senders fail, receivers drain the buffer first and only then see the close) without the panic.
func NewChannel ¶
NewChannel builds a channel with the given buffer capacity. capacity 0 is a Go-style unbuffered channel: a send blocks until a receiver takes the value.
func (*Channel) Close ¶
func (c *Channel) Close()
Close closes the channel. Idempotent. Buffered values remain receivable.
func (*Channel) Receive ¶
Receive blocks until a value arrives. A negative timeout waits forever; zero makes it a non-blocking try.
func (*Channel) Send ¶
Send blocks until the value is taken (or buffered). A negative timeout waits forever; zero makes it a non-blocking try.
A send that genuinely races a concurrent Close may still be delivered — the check-then-send window cannot be closed without a lock on the hot path, and Go's own answer to that race (panic) is not one a script should have to survive. Every non-racing send on a closed channel reports Closed.
type Dispatcher ¶
type Dispatcher struct {
// contains filtered or unexported fields
}
Dispatcher is a priority + delay scheduler. It is safe to submit to from any goroutine; it never runs a job itself. Callers pull work with NextDue and report the result with Complete.
Jobs are ordered by priority (higher first), FIFO within a priority. A job with a delay is parked in a separate heap until it comes due.
func NewDispatcher ¶
func NewDispatcher(capacity int) *Dispatcher
NewDispatcher builds an empty dispatcher. capacity bounds the number of pending jobs (ready + delayed); 0 leaves it unbounded.
func (*Dispatcher) Clear ¶
func (d *Dispatcher) Clear() int
Clear drops every pending job and returns how many were dropped.
func (*Dispatcher) Complete ¶
Complete records the result of an attempt and decides whether to retry.
`wait` is how long the job sat before starting and `exec` how long it ran; both feed the metrics. A failed job with attempts left is pushed back with its backoff applied and reported as Retried.
func (*Dispatcher) Len ¶
func (d *Dispatcher) Len() int
Len returns the number of pending jobs (ready + delayed).
func (*Dispatcher) MarkExpired ¶
func (d *Dispatcher) MarkExpired()
MarkExpired records a job dropped for missing its start deadline.
func (*Dispatcher) NextDue ¶
NextDue returns the highest-priority job eligible to run at `now`.
When no job is eligible, ok is false and `wait` says what the caller should do: a positive duration is how long until the earliest delayed job comes due (park that long, or until Wake fires); zero means there is nothing pending at all — the queue is drained, or stopped.
func (*Dispatcher) Snapshot ¶
func (d *Dispatcher) Snapshot() Metrics
Snapshot returns a copy of the current metrics.
func (*Dispatcher) Stop ¶
func (d *Dispatcher) Stop()
Stop makes the dispatcher reject new work and tells any parked pump to return. Pending jobs are left in place (Clear drops them); Stop is idempotent and safe to call from a running job.
func (*Dispatcher) Stopped ¶
func (d *Dispatcher) Stopped() bool
Stopped reports whether Stop has been called.
func (*Dispatcher) Submit ¶
func (d *Dispatcher) Submit(j *Job) error
Submit queues a job. It returns ErrStopped if the dispatcher has been stopped, or ErrFull if capacity is exhausted.
Unlike the channel-fed design this replaces, Submit pushes straight onto the heap under the lock. There is no receiver goroutine and no channel to close, which removes both the send-on-a-closed-channel panic that a Submit racing a Shutdown used to hit and the unbounded backlog that could build up behind a nominally "full" buffer.
func (*Dispatcher) Wake ¶
func (d *Dispatcher) Wake() <-chan struct{}
Wake is the channel a pump parks on while waiting for the next job to come due. It fires on Submit and on Stop.
type Job ¶
type Job struct {
ID string
Priority int64
Fn vm.Value // callable: *vm.Closure or *vm.GoFunc
Args []vm.Value // passed to Fn on every attempt
Payload vm.Value // opaque; carried along for the script's benefit
Enqueued time.Time // when the job first entered the queue
ReadyAt time.Time // job is not eligible to run before this instant
Timeout time.Duration // 0 = none; see Expired
Attempts int // completed attempts so far
Retries int // how many *re*-attempts a failure may trigger
Backoff time.Duration // delay before a retry becomes ready
// contains filtered or unexported fields
}
Job is one unit of work held by a Dispatcher. It is pure data: nothing in this file or in dispatcher.go ever *runs* a job. Execution happens in queue.go's pump, on the VM goroutine — see the package doc for why.
func (*Job) Expired ¶
Expired reports whether the job's start deadline has passed.
Timeout is a deadline on *starting*, not on running: a job that has sat in the queue past its deadline is dropped unrun. It deliberately does NOT abort a job already in progress — the VM is single-threaded and has no interrupt hooks, so there is no way to preempt a Lua function mid-call. A "cancel" that let the job keep running (and keep mutating VM state) while the caller was told it had been cancelled would be a lie, so the module enforces only the half of the contract it can actually honour: shedding stale work before it starts.
type Metrics ¶
type Metrics struct {
Enqueued int64
Processed int64
Succeeded int64
Failed int64
Retried int64
Expired int64
Dropped int64
TotalWait time.Duration
TotalExec time.Duration
MaxWait time.Duration
MaxExec time.Duration
}
Metrics is a snapshot of a dispatcher's counters.
It carries no mutex: the dispatcher's own lock guards the live copy, and Snapshot returns this struct by value. (Embedding a sync.Mutex here and returning *d.metrics — the shape this file used to have — copies the lock, which `go vet`'s copylocks check rejects and which silently hands callers a mutex in whatever state it happened to be in.)
Source Files