Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions deploy/inference-gateway/ext-proc/src/selector.rs
Original file line number Diff line number Diff line change
Expand Up @@ -288,7 +288,7 @@ mod tests {

impl WorkerPicker for ExpectedOutputRecorder {
fn pick(
&mut self,
&self,
context: &WorkerSelectionContext<'_>,
input: WorkerInputView<'_>,
) -> Result<usize, WorkerSelectionPolicyError> {
Expand Down Expand Up @@ -459,7 +459,7 @@ models:

impl WorkerPicker for FirstEligiblePicker {
fn pick(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
input: WorkerInputView<'_>,
) -> Result<usize, WorkerSelectionPolicyError> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -124,7 +124,7 @@ impl WorkerFilter for MinimumDeviceOverlapFilter {
}

fn keep(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
candidate: WorkerCandidate<'_>,
) -> Result<bool, WorkerSelectionPolicyError> {
Expand Down Expand Up @@ -159,7 +159,7 @@ impl WorkerScorer for ActiveRequestsScorer {
}

fn score(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
candidates: WorkerCandidates<'_>,
costs: &mut [f64],
Expand Down Expand Up @@ -201,7 +201,7 @@ struct LowestCostPicker;

impl WorkerPicker for LowestCostPicker {
fn pick(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
input: WorkerInputView<'_>,
) -> Result<usize, WorkerSelectionPolicyError> {
Expand Down Expand Up @@ -451,6 +451,7 @@ If a component needs no optional worker data, return `WorkerInputs::NONE`. Combi
| `LOAD` | `decode_cost_blocks()` | Projected active decode footprint, including this request's additional active blocks |
| `LOAD` | `active_requests()` | Requests currently active on the worker |
| `PREFERRED_TAINT` | `preferred_taint_multiplier()` | Optional cost multiplier from matching preferred routing taints |
| `ACTIVE_REQUESTS` | `WorkerInputView::active_requests(row)` | Picker only. Requests in flight on the row's worker; `None` only when the picker did not declare the group or the row is out of range |
| `KV_TRANSFER` | `kv_transfer()` | `None` if the host has no transfer data for this request: no worker in the partition consumes router hints, or the indexer cannot retain the matched block chain |
| `KV_TRANSFER` | `source_prefix_blocks()` | Prefix blocks held by the source this worker would fetch from, and so held by the worker after the fetch; `None` when no eligible source holds more than the worker's device-resident plus host-pinned prefix |

Expand All @@ -460,6 +461,8 @@ The GPU, CPU, and disk overlap accessors return raw tier counts. A missing tier

A filter or scorer reads groups through `WorkerCandidate::cache()` or `load()`. A picker reads index-aligned views through `WorkerInputView`; its cache view supports `get(row)` and `iter()`. Each component must declare every group it reads; another component requesting a group does not grant access.

Declare `ACTIVE_REQUESTS` instead of `LOAD` when a request count is all the picker needs. It is cheaper, and `LOAD` needs the KV scheduler, which tracks each worker's active tokens and blocks.

A scorer reads `KV_TRANSFER` through `WorkerCandidate::kv_transfer()`. Filters and pickers do not receive it, and declaring it there has no effect. The host picks each worker's source with the [KV fetch hint](topology-aware-kv-transfer.md#kv-fetch-hints) rules, including topology islands, and booking a worker attaches a fetch hint for that same source. Fetches are best effort: the source may no longer hold the blocks when the request runs.

This scorer charges each worker for the prompt blocks it can neither reuse nor fetch. A worker without transfer data still counts its own cache:
Expand All @@ -477,7 +480,7 @@ impl WorkerScorer for UncoveredPromptScorer {
}
fn score(
&mut self,
&self,
context: &WorkerSelectionContext<'_>,
candidates: WorkerCandidates<'_>,
costs: &mut [f64],
Expand Down Expand Up @@ -634,6 +637,7 @@ impl RequestClassifier for InspectPrefixes {
- Treat candidate order as unspecified.
- Request only the signal groups that the component reads.
- Keep blocking I/O and panics out of `keep`, `score`, and `pick`.
- Keep any state your policy shares across requests correct under concurrent updates. Dynamo reuses one policy for every request to a model and worker role and handles many requests at once, so two selections can read and change that state at the same time. Policies without such state need nothing extra.
- Keep policy state local to the factory-created policy unless cross-partition sharing is a deliberate requirement.
- Build the policy against the same Dynamo revision as the frontend or EPP.
- If a signal adds work, storage, allocation, or another scan, run the worker-selection benchmark.
Expand Down Expand Up @@ -691,8 +695,9 @@ patch, and reset semantics.

Updates run between selection calls. A replacement cannot add worker-input groups beyond those
declared at startup, change the worker label, or change exclusive-affinity behavior. Queues,
worker discovery, cache indexes, and host accounting are retained. Scorer/picker-local scratch
state is rebuilt. A cancelled control request still waiting in the actor queue is skipped.
worker discovery, cache indexes, and host accounting are retained. The replacement's filters,
scorers, and picker are new instances, so state the previous instances held, such as a cursor,
does not carry over. A cancelled control request still waiting in the actor queue is skipped.

The [simple-filter-score-pick example](https://github.com/ai-dynamo/dynamo/blob/main/examples/router/custom-policy-example/simple-filter-score-pick/src/lib.rs)
opts in for its `min_device_overlap_blocks` parameter. The built-in default policy opts in for
Expand Down
8 changes: 8 additions & 0 deletions docs/fern/pages/reference/general/releases/deprecations.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,14 @@ Upgrading? Pick the version you run today to see the dependency migration and th

> **Migrate:** Set `DYN_ENABLE_FRONTEND_ADMIN_API=1` on frontends that need these routes and call port `8002` (`DYN_FRONTEND_ADMIN_PORT`). The listener binds to `127.0.0.1` (`DYN_FRONTEND_ADMIN_HOST`), so on Kubernetes use `kubectl port-forward pod/<frontend-pod> 8002:8002`. Remove `DYN_DISABLE_FRONTEND_ADMIN_API`; it is ignored.
<span className="dynref-badge dynref-badge--blue">Behavioral</span> **Worker-selection plugin callbacks take `&self` and require `Send + Sync`** ([#17712](https://github.com/ai-dynamo/dynamo/pull/17712)). `WorkerFilter::keep`, `WorkerScorer::score` and `WorkerPicker::pick` no longer take `&mut self`, and one plugin instance may be called from several threads at once.

> **Migrate:** Change each callback's receiver to `&self`. Move state that changes between calls behind atomics, a lock or a thread-local, and make sure the plugin type is `Sync`.
<span className="dynref-badge dynref-badge--amber">Deprecated</span> **Deprecated the `WorkerInputs::OCCUPANCY` worker-selection input group** in favor of `WorkerInputs::ACTIVE_REQUESTS`, which custom pickers read through `WorkerInputView::active_requests(row)`. The alias still compiles with a deprecation warning and is slated for removal in Dynamo v1.9.0.

> **Migrate:** Replace `WorkerInputs::OCCUPANCY` with `WorkerInputs::ACTIVE_REQUESTS` in custom worker-selection policies.
Comment thread
karen-sy marked this conversation as resolved.
Resolved
<a id="v160"></a>

## v1.6.0 - Oct 22, 2026
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ impl WorkerFilter for MinimumDeviceOverlapFilter {
}

fn keep(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
candidate: WorkerCandidate<'_>,
) -> Result<bool, WorkerSelectionPolicyError> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ pub(crate) struct PrefillPicker;
impl WorkerPicker for PrefillPicker {
/// Returns the prefill candidate row with the lowest cost.
fn pick(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
input: WorkerInputView<'_>,
) -> Result<usize, WorkerSelectionPolicyError> {
Expand All @@ -38,7 +38,7 @@ pub(crate) struct DecodePicker;
impl WorkerPicker for DecodePicker {
/// Returns the decode candidate row with the lowest cost.
fn pick(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
input: WorkerInputView<'_>,
) -> Result<usize, WorkerSelectionPolicyError> {
Expand All @@ -59,7 +59,7 @@ impl UnsupportedWorkerTypePicker {

impl WorkerPicker for UnsupportedWorkerTypePicker {
fn pick(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
_input: WorkerInputView<'_>,
) -> Result<usize, WorkerSelectionPolicyError> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ impl WorkerScorer for PrefillLoadScorer {
}

fn score(
&mut self,
&self,
context: &WorkerSelectionContext<'_>,
candidates: WorkerCandidates<'_>,
costs: &mut [f64],
Expand Down Expand Up @@ -50,7 +50,7 @@ impl WorkerScorer for DecodeLoadScorer {

/// Returns the projected decode cost as a lower-is-better cost.
fn score(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
candidates: WorkerCandidates<'_>,
costs: &mut [f64],
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ impl WorkerFilter for MinimumDeviceOverlapFilter {

/// Keeps a candidate when its device overlap meets the minimum.
fn keep(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
candidate: WorkerCandidate<'_>,
) -> Result<bool, WorkerSelectionPolicyError> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ impl WorkerPicker for RequestAwarePicker {
}

fn pick(
&mut self,
&self,
context: &WorkerSelectionContext<'_>,
input: WorkerInputView<'_>,
) -> Result<usize, WorkerSelectionPolicyError> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ impl WorkerScorer for ActiveRequestsScorer {

/// Find the batch minimum and write one relative load cost per worker.
fn score(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
candidates: WorkerCandidates<'_>,
costs: &mut [f64],
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ pub(crate) struct LowestCostPicker;

impl WorkerPicker for LowestCostPicker {
fn pick(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
input: WorkerInputView<'_>,
) -> Result<usize, WorkerSelectionPolicyError> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ impl WorkerScorer for ActiveRequestsScorer {
}

fn score(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
candidates: WorkerCandidates<'_>,
costs: &mut [f64],
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ impl WorkerScorer for UncachedBlocksScorer {
}

fn score(
&mut self,
&self,
context: &WorkerSelectionContext<'_>,
candidates: WorkerCandidates<'_>,
costs: &mut [f64],
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ impl WorkerPicker for SoftPinRepinPicker {
}

fn pick(
&mut self,
&self,
context: &WorkerSelectionContext<'_>,
input: WorkerInputView<'_>,
) -> Result<usize, WorkerSelectionPolicyError> {
Expand Down
10 changes: 5 additions & 5 deletions lib/kv-router/benches/worker_selection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ struct KeepAllFilter;

impl WorkerFilter for KeepAllFilter {
fn keep(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
_candidate: WorkerCandidate<'_>,
) -> Result<bool, WorkerSelectionPolicyError> {
Expand All @@ -65,7 +65,7 @@ struct CapacityFilter;

impl WorkerFilter for CapacityFilter {
fn keep(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
candidate: WorkerCandidate<'_>,
) -> Result<bool, WorkerSelectionPolicyError> {
Expand All @@ -83,7 +83,7 @@ impl WorkerScorer for BenchScorer {
}

fn score(
&mut self,
&self,
context: &WorkerSelectionContext<'_>,
candidates: WorkerCandidates<'_>,
costs: &mut [f64],
Expand Down Expand Up @@ -117,7 +117,7 @@ impl WorkerScorer for IgnoringPreferenceScorer {
}

fn score(
&mut self,
&self,
context: &WorkerSelectionContext<'_>,
candidates: WorkerCandidates<'_>,
costs: &mut [f64],
Expand All @@ -144,7 +144,7 @@ struct LowestCostPicker;

impl WorkerPicker for LowestCostPicker {
fn pick(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
input: WorkerInputView<'_>,
) -> Result<usize, WorkerSelectionPolicyError> {
Expand Down
2 changes: 1 addition & 1 deletion lib/kv-router/src/plugins/registry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -363,7 +363,7 @@ mod tests {

impl WorkerPicker for NeverPicker {
fn pick(
&mut self,
&self,
_context: &WorkerSelectionContext<'_>,
_input: WorkerInputView<'_>,
) -> Result<usize, WorkerSelectionPolicyError> {
Expand Down
24 changes: 17 additions & 7 deletions lib/kv-router/src/plugins/worker_selection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,10 @@ pub type WorkerSelectionPolicyFactory = Arc<
>;

/// Adds one finite cost contribution to each eligible worker.
pub trait WorkerScorer: Send {
///
/// One instance may be called from several threads at once, and successive calls may run on
/// different threads. Keep any state that changes between calls thread-safe.
pub trait WorkerScorer: Send + Sync {
/// Declare the worker-signal groups needed by this scorer.
fn required_worker_inputs(&self) -> WorkerInputs {
WorkerInputs::NONE
Expand All @@ -58,7 +61,7 @@ pub trait WorkerScorer: Send {
/// Write one finite, lower-is-better contribution to every slot. The host validates and adds
/// each contribution to that row's total. The candidate view and cost slice may not be retained after this call.
fn score(
&mut self,
&self,
context: &WorkerSelectionContext<'_>,
candidates: WorkerCandidates<'_>,
costs: &mut [f64],
Expand All @@ -68,31 +71,38 @@ pub trait WorkerScorer: Send {
/// Filters run in declaration order for each candidate. Callback order across different
/// candidates and scorers is unspecified; implementations must not depend on filters and scorers
/// being interleaved.
pub trait WorkerFilter: Send {
///
/// One instance may be called from several threads at once, and successive calls may run on
/// different threads. Keep any state that changes between calls thread-safe.
pub trait WorkerFilter: Send + Sync {
/// Declare the worker-signal groups needed by this filter.
fn required_worker_inputs(&self) -> WorkerInputs {
WorkerInputs::NONE
}

/// Return `true` to keep an eligible worker in the policy candidate set.
fn keep(
&mut self,
&self,
context: &WorkerSelectionContext<'_>,
candidate: WorkerCandidate<'_>,
) -> Result<bool, WorkerSelectionPolicyError>;
}

/// Selects one row after all filters and scorers run.
pub trait WorkerPicker: Send {
///
/// One instance may be called from several threads at once, and successive calls may run on
/// different threads. Keep any state that changes between calls thread-safe.
pub trait WorkerPicker: Send + Sync {
/// Declare the optional worker-signal columns needed by this picker.
fn required_worker_inputs(&self) -> WorkerInputs {
WorkerInputs::NONE
}

/// Return one row index from the host-owned eligible candidate table. Row order is
/// unspecified; inspect candidate data instead of relying on a stable position.
/// unspecified unless [`WorkerInputView`] says otherwise; inspect candidate data instead
/// of relying on a stable position.
fn pick(
&mut self,
&self,
context: &WorkerSelectionContext<'_>,
input: WorkerInputView<'_>,
) -> Result<usize, WorkerSelectionPolicyError>;
Expand Down
Loading
Loading