type Interface … type TypedInterface … type Queue … // DefaultQueue is a slice based FIFO queue. func DefaultQueue[T comparable]() Queue[T] { … } type queue … func (q *queue[T]) Touch(item T) { … } func (q *queue[T]) Push(item T) { … } func (q *queue[T]) Len() int { … } func (q *queue[T]) Pop() (item T) { … } type QueueConfig … type TypedQueueConfig … // New constructs a new work queue (see the package comment). // // Deprecated: use NewTyped instead. func New() *Type { … } // NewTyped constructs a new work queue (see the package comment). func NewTyped[T comparable]() *Typed[T] { … } // NewWithConfig constructs a new workqueue with ability to // customize different properties. // // Deprecated: use NewTypedWithConfig instead. func NewWithConfig(config QueueConfig) *Type { … } // NewTypedWithConfig constructs a new workqueue with ability to // customize different properties. func NewTypedWithConfig[T comparable](config TypedQueueConfig[T]) *Typed[T] { … } // NewNamed creates a new named queue. // Deprecated: Use NewWithConfig instead. func NewNamed(name string) *Type { … } // newQueueWithConfig constructs a new named workqueue // with the ability to customize different properties for testing purposes func newQueueWithConfig[T comparable](config TypedQueueConfig[T], updatePeriod time.Duration) *Typed[T] { … } func newQueue[T comparable](c clock.WithTicker, queue Queue[T], metrics queueMetrics[T], updatePeriod time.Duration) *Typed[T] { … } const defaultUnfinishedWorkUpdatePeriod … type Type … type Typed … type empty … type set … func (s set[t]) has(item t) bool { … } func (s set[t]) insert(item t) { … } func (s set[t]) delete(item t) { … } func (s set[t]) len() int { … } // Add marks item as needing processing. func (q *Typed[T]) Add(item T) { … } // Len returns the current queue length, for informational purposes only. You // shouldn't e.g. gate a call to Add() or Get() on Len() being a particular // value, that can't be synchronized properly. func (q *Typed[T]) Len() int { … } // Get blocks until it can return an item to be processed. If shutdown = true, // the caller should end their goroutine. You must call Done with item when you // have finished processing it. func (q *Typed[T]) Get() (item T, shutdown bool) { … } // Done marks item as done processing, and if it has been marked as dirty again // while it was being processed, it will be re-added to the queue for // re-processing. func (q *Typed[T]) Done(item T) { … } // ShutDown will cause q to ignore all new items added to it and // immediately instruct the worker goroutines to exit. func (q *Typed[T]) ShutDown() { … } // ShutDownWithDrain will cause q to ignore all new items added to it. As soon // as the worker goroutines have "drained", i.e: finished processing and called // Done on all existing items in the queue; they will be instructed to exit and // ShutDownWithDrain will return. Hence: a strict requirement for using this is; // your workers must ensure that Done is called on all items in the queue once // the shut down has been initiated, if that is not the case: this will block // indefinitely. It is, however, safe to call ShutDown after having called // ShutDownWithDrain, as to force the queue shut down to terminate immediately // without waiting for the drainage. func (q *Typed[T]) ShutDownWithDrain() { … } func (q *Typed[T]) ShuttingDown() bool { … } func (q *Typed[T]) updateUnfinishedWorkLoop() { … }
Codebase Browser
kubernetes