Documentation
¶
Overview ¶
Package queue provides a bounded generic data queue and a task queue.
A queue moves through three states: Opened, Hanged and Closed. Once the sentinel data has been received and Queue.Done is called with true, the queue becomes Hanged: it stops accepting new data and only the remaining data may be received. Queue.Close closes the data channel and releases every caller blocked in Queue.Wait.
Write and Close are not mutually exclusive internally, so callers must serialize them, for example by sharing the same read-write lock; otherwise a write to an already closed channel panics. Passing rw to NewQueue makes the queue reuse the caller's read-write lock to provide that mutual exclusion.
Index ¶
Constants ¶
const ( Opened = iota // Opened: both reads and writes are allowed Hanged // Hanged: the end signal has been received, only the remaining data may be received Closed // Closed: the data channel has been closed )
The states of a Queue.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Queue ¶
type Queue[T any] struct { // contains filtered or unexported fields }
Queue is a bounded generic queue.
Queue uses a state machine and an end wait group to support the graceful shutdown semantics of "write the end signal, then wait for the remaining data to drain".
func NewQueue ¶
NewQueue returns a new Queue with the given capacity and write timeout. The timeout bounds how long a non-blocking Queue.Write waits when the queue is full; a timeout of 0 disables that bound and makes writes block until they succeed.
The optional rw is a read-write lock supplied by the caller. When it is provided, writes hold its read lock so that they are serialized against Queue.Close.
func (*Queue[T]) Clean ¶
func (q *Queue[T]) Clean(f func(T))
Clean receives and discards every remaining item of a closed queue, invoking f for each item. It must only be called after the queue has been closed; otherwise it blocks waiting for new data.
func (*Queue[T]) Close ¶
func (q *Queue[T]) Close()
Close closes the queue. It closes the data channel and releases the end wait; repeated calls do not close the channel twice. The caller must serialize Close with Queue.Write, otherwise a write to the already closed channel panics.
func (*Queue[T]) Done ¶
Done acknowledges that one item of the queue's data has been processed. It decrements the backpressure counter and, when isCloseSig is true and the queue is in the Opened state, moves the queue to Hanged and releases the end wait. Done does nothing when the queue is already hanged or closed.
func (*Queue[T]) Read ¶
func (q *Queue[T]) Read() <-chan T
Read returns the channel from which the queue's data is received. After receiving an item, the consumer must call Queue.Done to release the backpressure counter.
func (*Queue[T]) Wait ¶
func (q *Queue[T]) Wait()
Wait blocks until the queue is closed or the end signal has been received.
func (*Queue[T]) Write ¶
Write writes t to the queue. It reports errors.ErrQueueHanged when the queue is hanged and errors.ErrQueueClosed when it is closed.
When the write timeout is greater than 0 and the amount of pending data exceeds the queue capacity, a non-blocking write waits for at most the timeout and then reports errors.ErrWriteTimeout. Pass block as true, or configure a timeout <= 0, to block until the write succeeds.
type Tasker ¶
type Tasker struct {
// contains filtered or unexported fields
}
Tasker is a task runner.
A Tasker owns the asynchronous task queue of a node.
func NewTasker ¶
NewTasker returns a new Tasker whose task queue has the given capacity and write timeout. A timeout of 0 disables timeout control. The optional rw is passed through to NewQueue.
func (*Tasker) Clean ¶
func (t *Tasker) Clean()
Clean cancels every unfinished task still in the queue and releases their wait groups. It must only be called after the queue has been closed.
func (*Tasker) Commit ¶
Commit commits fn as a task. It takes a task object from the pool, stores fn in it and writes it to the task queue; when enqueuing fails, the task object is released immediately.
When wait is true, Commit returns a *sync.WaitGroup that the caller can wait on; otherwise it returns a nil WaitGroup.
func (*Tasker) Done ¶
Done stops accepting tasks. It writes a sentinel task that tells the consumer the task queue has finished; once the consumer handles the sentinel, the queue becomes hanged.
func (*Tasker) Handle ¶
Handle handles a queued task. A nil tk is the sentinel that hangs the queue. Handle first acknowledges the queue signal, then, when isExecute is true, runs the task function through xcall.Call, and finally returns the task object to the pool.
Source Files