Documentation
¶
Overview ¶
Package internal provides the core worker implementation for the controller.FlowController.
The components in this package are the private, internal building blocks of the controller. This separation enforces a clean public API at the `controller` level and allows the internal mechanics of the engine to evolve independently.
Design Philosophy: The Single-Writer Actor Model ¶
The concurrency model for this package is built around a single-writer, channel-based actor pattern, as implemented in the ShardProcessor. All state-mutating operations for a given shard (primarily enqueuing new requests) are funneled through a single Run goroutine.
This design makes complex, multi-step transactions (like a hierarchical capacity check against both a shard's total limit and a priority band's limit) inherently atomic without locks. This avoids the performance bottleneck of a coarse, shard-wide lock and allows the top-level Controller to remain decoupled and highly concurrent.
Key Components ¶
- ShardProcessor: The implementation of the worker actor. Manages the lifecycle of requests for a single shard.
- FlowItem: The internal representation of a request, managing its state and synchronization across goroutines.
Index ¶
- Variables
- type FinalState
- type FlowItem
- func (fi *FlowItem) Done() <-chan *FinalState
- func (fi *FlowItem) EffectiveTTL() time.Duration
- func (fi *FlowItem) EnqueueTime() time.Time
- func (fi *FlowItem) FinalState() *FinalState
- func (fi *FlowItem) Finalize(cause error)
- func (fi *FlowItem) FinalizeWithOutcome(outcome types.QueueOutcome, err error)
- func (fi *FlowItem) Handle() flowcontrol.QueueItemHandle
- func (fi *FlowItem) OriginalRequest() flowcontrol.FlowControlRequest
- func (fi *FlowItem) SetHandle(handle flowcontrol.QueueItemHandle)
- type Processor
Constants ¶
This section is empty.
Variables ¶
var ErrProcessorBusy = errors.New("shard processor is busy")
ErrProcessorBusy is a sentinel error returned by the processor's Submit method indicating that the processor's. internal buffer is momentarily full and cannot accept new work.
Functions ¶
This section is empty.
Types ¶
type FinalState ¶
type FinalState struct {
Outcome types.QueueOutcome
Err error
}
FinalState encapsulates the terminal outcome of a FlowItem's lifecycle.
type FlowItem ¶
type FlowItem struct {
// contains filtered or unexported fields
}
FlowItem is the internal representation of a request managed by the Flow Controller.
Lifecycle Management ¶
Finalization (determining outcome) can be initiated by the Controller (e.g., Context expiry) or the Processor (e.g., Dispatch/Reject). It sets the outcome and signals the waiting goroutine.
Synchronization ¶
Atomic operations synchronize state across the Controller and Processor goroutines:
- finalState (atomic.Pointer): Safely publishes the outcome.
- handle (atomic.Pointer): Safely publishes the queue admission status.
func NewItem ¶
func NewItem(req flowcontrol.FlowControlRequest, effectiveTTL time.Duration, enqueueTime time.Time) *FlowItem
NewItem allocates and initializes a new FlowItem for a request lifecycle.
func (*FlowItem) Done ¶
func (fi *FlowItem) Done() <-chan *FinalState
Done returns a read-only channel that will receive the FinalState pointer exactly once.
func (*FlowItem) EffectiveTTL ¶
EffectiveTTL returns the actual time-to-live assigned to this item.
func (*FlowItem) EnqueueTime ¶
EnqueueTime returns the time the item was logically accepted by the FlowController.
func (*FlowItem) FinalState ¶
func (fi *FlowItem) FinalState() *FinalState
FinalState returns the FinalState if the item has been finalized, or nil otherwise. Safe for concurrent access.
func (*FlowItem) Finalize ¶
Finalize determines the item's terminal state based on the provided cause (e.g., Context error) and the item's current admission status (queued or not).
This method is intended for asynchronous finalization initiated by the Controller (e.g., TTL expiry). It is idempotent.
func (*FlowItem) FinalizeWithOutcome ¶
func (fi *FlowItem) FinalizeWithOutcome(outcome types.QueueOutcome, err error)
FinalizeWithOutcome sets the item's terminal state explicitly.
This method is intended for synchronous finalization by the Processor (Dispatch, Reject) or the Controller (Distribution failure). It is idempotent.
func (*FlowItem) Handle ¶
func (fi *FlowItem) Handle() flowcontrol.QueueItemHandle
Handle returns the QueueItemHandle for this item within a queue. Returns nil if the item is not in a queue. Safe for concurrent access.
func (*FlowItem) OriginalRequest ¶
func (fi *FlowItem) OriginalRequest() flowcontrol.FlowControlRequest
OriginalRequest returns the original FlowControlRequest object.
func (*FlowItem) SetHandle ¶
func (fi *FlowItem) SetHandle(handle flowcontrol.QueueItemHandle)
SetHandle associates a QueueItemHandle with this item. Called by the queue implementation (via Processor). Safe for concurrent access.
type Processor ¶
type Processor struct {
// contains filtered or unexported fields
}
Processor is the core worker of the FlowController.
It is paired one-to-one with a RegistryShard instance and is responsible for all request lifecycle operations on that shard, from the point an item is successfully submitted to it.
Request Lifecycle Management & Ownership ¶
The Processor takes ownership of a FlowItem only after it has been successfully sent to its internal enqueueChan via Submit or SubmitOrBlock (i.e., when these methods return nil). Once the Processor takes ownership, it is solely responsible for ensuring that item.Finalize() or item.FinalizeWithOutcome() is called exactly once for that item, under all circumstances (dispatch, rejection, sweep, or shutdown).
If Submit or SubmitOrBlock return an error, ownership remains with the caller (the Controller), which must then handle the finalization.
Concurrency Model ¶
To ensure correctness and high performance, the processor uses a single-goroutine, actor-based model. The main run loop is the sole writer for all state-mutating operations. This makes complex transactions (like capacity checks) inherently atomic without coarse-grained locks.
func NewProcessor ¶
func NewProcessor( ctx context.Context, poolName string, registry contracts.FlowRegistry, registryBackground contracts.FlowRegistryBackground, saturationDetector flowcontrol.SaturationDetector, endpointCandidates contracts.EndpointCandidates, usageLimitPolicy flowcontrol.UsageLimitPolicy, clock clock.WithTicker, cleanupSweepInterval time.Duration, enqueueChannelBufferSize int, logger logr.Logger, ) *Processor
NewProcessor creates a new Processor instance.
func (*Processor) Run ¶
Run is the main operational loop for the shard processor. It must be run as a goroutine. It uses a `select` statement to interleave accepting new requests with dispatching existing ones, balancing responsiveness with throughput.
func (*Processor) Submit ¶
Submit attempts a non-blocking handoff of an item to the processor's internal enqueue channel.
Ownership Contract:
- Returns nil: The item was successfully handed off. The ShardProcessor takes responsibility for calling Finalize on the item.
- Returns error: The item was not handed off. Ownership of the FlowItem remains with the caller, who is responsible for calling Finalize.
Possible errors:
- ErrProcessorBusy: The processor's input channel is full.
- types.ErrFlowControllerNotRunning: The processor is shutting down.
func (*Processor) SubmitOrBlock ¶
SubmitOrBlock performs a blocking handoff of an item to the processor's internal enqueue channel. It waits until the item is handed off, the caller's context is cancelled, or the processor shuts down.
Ownership Contract:
- Returns nil: The item was successfully handed off. The ShardProcessor takes responsibility for calling Finalize on the item.
- Returns error: The item was not handed off. Ownership of the FlowItem remains with the caller, who is responsible for calling Finalize.
Possible errors:
- ctx.Err(): The provided context was cancelled or its deadline exceeded.
- types.ErrFlowControllerNotRunning: The processor is shutting down.
Source Files