README
¶
In-Flight Load Producer Plugin
Type: inflight-load-producer
Tracks real-time in-flight request and token counts per endpoint by hooking into the request lifecycle. Writes an InFlightLoad attribute onto each endpoint in the Produce phase, consumed by the following plugins:
token-load-scorer: Scores endpoints based on in-flight tokens.active-request-scorer: Scores endpoints based on in-flight requests.concurrency-detector: Provides admission control based on in-flight requests/tokens.prefix-cache-affinity-filter: Uses in-flight tokens as a load gate to break stickiness.
Behavior
- Prefix Cache Discounting: Automatically detects if an endpoint has a prefix cache hit (via
PrefixCacheMatchInfo). Only the uncached portion of the prompt is added to the in-flight token counter, providing a more accurate estimate of the actual compute load. - Token Release Timing:
- If
addEstimatedOutputTokensisfalse(default): For streaming requests, all tokens are released as soon as the first chunk of the response is received (StartOfStream), as the prefill compute is complete. For non-streaming requests (or as a safety net), tokens are released when the response completes (EndOfStream). - If
addEstimatedOutputTokensistrue: The prompt portion is released atStartOfStream(for streaming) orEndOfStream, and the estimated output portion is released only when the response completes (EndOfStream).
- If
- Request Release: In-flight request counters are always released when the response completes (
EndOfStream).
The producer hooks three lifecycle phases:
- Produce: Writes current in-flight counts to each endpoint's attributes.
- PreRequest: Increments counters when a request is dispatched to an endpoint.
- ResponseBody: Decrements counters when a response completes or the request is aborted.
Endpoint departure events (pod removed from the pool) are handled via the EndpointExtractor interface to clean up stale counters.
Parameters
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
addEstimatedOutputTokens |
bool |
No | false |
If true, adds an estimate of the generated output tokens to the in-flight counter. |
Related Documentation
- Token Load Scorer
- Active Request Scorer
- Concurrency Detector
- Prefix Cache Affinity Filter
- Concurrency Attributes
Configuration Example:
plugins:
- type: inflight-load-producer
name: inflight-load
parameters:
addEstimatedOutputTokens: true
Documentation
¶
Index ¶
- Constants
- func InFlightLoadProducerFactory(name string, decoder *json.Decoder, handle fwkplugin.Handle) (fwkplugin.Plugin, error)
- type Config
- type InFlightLoadProducer
- func (p *InFlightLoadProducer) Consumes() fwkplugin.DataDependencies
- func (p *InFlightLoadProducer) DeleteEndpoint(endpointID string)
- func (p *InFlightLoadProducer) DumpState() (json.RawMessage, error)
- func (p *InFlightLoadProducer) Extract(ctx context.Context, event datalayer.EndpointEvent) error
- func (p *InFlightLoadProducer) GetRequests(eid string) int64
- func (p *InFlightLoadProducer) GetTokens(eid string) int64
- func (p *InFlightLoadProducer) PreRequest(ctx context.Context, request *fwksched.InferenceRequest, ...)
- func (p *InFlightLoadProducer) Produce(_ context.Context, request *fwksched.InferenceRequest, ...) error
- func (p *InFlightLoadProducer) Produces() map[fwkplugin.DataKey]any
- func (p *InFlightLoadProducer) RegisterDependencies(r datalayer.Registrar) error
- func (p *InFlightLoadProducer) ResponseBody(_ context.Context, request *fwksched.InferenceRequest, ...)
- func (p *InFlightLoadProducer) TypedName() fwkplugin.TypedName
- type SimpleTokenEstimator
- type TokenEstimator
Constants ¶
const (
InFlightLoadProducerType = inflightloadconstants.InFlightLoadProducerType
)
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Config ¶
type Config struct {
// AddEstimatedOutputTokens controls whether estimated output tokens are added to
// the in-flight token counter. Defaults to false.
AddEstimatedOutputTokens bool `json:"addEstimatedOutputTokens"`
// PrefixMatchInfoProducerName selects which prefix-cache producer's
// PrefixCacheMatchInfo to read for the cached-prefix discount. Empty defaults
// to the approximate-prefix producer; set it to a precise-prefix-cache
// producer's instance name to discount against precise cache state instead.
PrefixMatchInfoProducerName string `json:"prefixMatchInfoProducerName,omitempty"`
}
Config controls optional behaviors of InFlightLoadProducer.
type InFlightLoadProducer ¶
type InFlightLoadProducer struct {
PluginState *fwkplugin.PluginState
// contains filtered or unexported fields
}
func (*InFlightLoadProducer) Consumes ¶
func (p *InFlightLoadProducer) Consumes() fwkplugin.DataDependencies
Consumes declares TokenizedPrompt as required so the data-layer DAG orders a token-producer ahead of this producer and auto-creates one when none is configured; without it the input-token estimate silently reads zero. PrefixCacheMatchInfo is optional — used to discount the already-cached prompt prefix from the prefix producer selected by prefixMatchInfoProducerName (approximate by default, or a precise-prefix-cache producer).
func (*InFlightLoadProducer) DeleteEndpoint ¶
func (p *InFlightLoadProducer) DeleteEndpoint(endpointID string)
DeleteEndpoint removes an endpoint from the concurrency trackers to prevent memory leaks. This matches the design of the previous saturation detector and is called by the ExtractNotification hook to ensure deterministic cleanup of stateful data.
func (*InFlightLoadProducer) DumpState ¶
func (p *InFlightLoadProducer) DumpState() (json.RawMessage, error)
DumpState implements fwkplugin.StateDumper and exposes per-endpoint in-flight request and token counts for the /debug/plugins/state endpoint.
The request and token tracker maps are snapshotted under separate read locks, so the returned per-endpoint Requests and Tokens values are not guaranteed to correspond to the same instant in time and the endpoint set itself may change between the two snapshots. This is acceptable for a debug endpoint, where best-effort visibility is preferred over coordinating a single global lock that would contend with the hot path.
The endpoint list is capped to the busiest endpoints to keep the debug payload bounded when a deployment has a large endpoint set.
func (*InFlightLoadProducer) Extract ¶
func (p *InFlightLoadProducer) Extract(ctx context.Context, event datalayer.EndpointEvent) error
Extract handles endpoint lifecycle events to manage dynamic attributes.
func (*InFlightLoadProducer) GetRequests ¶
func (p *InFlightLoadProducer) GetRequests(eid string) int64
func (*InFlightLoadProducer) GetTokens ¶
func (p *InFlightLoadProducer) GetTokens(eid string) int64
func (*InFlightLoadProducer) PreRequest ¶
func (p *InFlightLoadProducer) PreRequest(ctx context.Context, request *fwksched.InferenceRequest, result *fwksched.SchedulingResult)
func (*InFlightLoadProducer) Produce ¶
func (p *InFlightLoadProducer) Produce(_ context.Context, request *fwksched.InferenceRequest, endpoints []fwksched.Endpoint) error
func (*InFlightLoadProducer) Produces ¶
func (p *InFlightLoadProducer) Produces() map[fwkplugin.DataKey]any
func (*InFlightLoadProducer) RegisterDependencies ¶
func (p *InFlightLoadProducer) RegisterDependencies(r datalayer.Registrar) error
RegisterDependencies declares that this plugin needs an endpoint-notification-source to track endpoint lifecycle events. The source is auto-created if not already in the config.
func (*InFlightLoadProducer) ResponseBody ¶
func (p *InFlightLoadProducer) ResponseBody( _ context.Context, request *fwksched.InferenceRequest, resp *requestcontrol.Response, _ *datalayer.EndpointMetadata, )
func (*InFlightLoadProducer) TypedName ¶
func (p *InFlightLoadProducer) TypedName() fwkplugin.TypedName
type SimpleTokenEstimator ¶
type SimpleTokenEstimator struct {
OutputRatio float64
}
SimpleTokenEstimator derives input tokens from the tokenized prompt and estimates output tokens as inputTokens * OutputRatio.
func (*SimpleTokenEstimator) Estimate ¶
func (e *SimpleTokenEstimator) Estimate(request *fwksched.InferenceRequest) int64
Estimate returns the total estimated token count (input + output) for the request. Output tokens are estimated as inputTokens * OutputRatio.
func (*SimpleTokenEstimator) EstimateInput ¶
func (e *SimpleTokenEstimator) EstimateInput(request *fwksched.InferenceRequest) int64
EstimateInput returns the input token count read from the tokenized prompt, or 0 when no tokenization is available.
func (*SimpleTokenEstimator) EstimateOutput ¶
func (e *SimpleTokenEstimator) EstimateOutput(inputTokens int64) int64
EstimateOutput returns the estimated output token count given the input token count.
type TokenEstimator ¶
type TokenEstimator interface {
// Estimate returns the total estimated token count (input + output) for the request.
Estimate(request *fwksched.InferenceRequest) int64
// EstimateInput returns only the estimated input token count for the request.
EstimateInput(request *fwksched.InferenceRequest) int64
// EstimateOutput returns the estimated output token count given the input token count.
EstimateOutput(inputTokens int64) int64
}
TokenEstimator estimates the number of tokens for an LLM request.
func NewSimpleTokenEstimator ¶
func NewSimpleTokenEstimator() TokenEstimator
NewSimpleTokenEstimator returns a SimpleTokenEstimator with default output ratio.
Source Files
Directories