Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 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
2 changes: 2 additions & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Original file line number Diff line number Diff line change
Expand Up @@ -584,6 +584,78 @@ Use `--router-prefill-load-model ais` when you want prompt-side load tracking to

Use `--router-queue-policy wspt` when your workload has a mix of short and long requests and you want to minimize average TTFT. Use the default `fcfs` when you want to minimize tail TTFT.

## Runtime Policy Updates

**Experimental.** The frontend admin API can update the built-in default policy's scoring
parameters without replacing the router's queue, KV index, or active request bookings.
The update replaces the scorer and picker between routing decisions. Requests that already
have a selected worker keep that worker; subsequent selections, including queued requests,
use the new parameters.

`GET /router_policy` lists frontend-local KV router instances. Use `?model=<name>` to filter
by the model name returned in the response. Each entry includes `router_id`, `model`,
`namespace`, `component`, `endpoint`, `worker_type`, and `configuration`. Configuration contains
a `revision` and the complete resolved `parameters`. A null configuration means that the
selected custom policy does not support runtime updates. Non-KV routing paths are not listed.
Routers that have stopped or do not respond within five seconds are omitted; healthy routers
remain available in the listing.

```bash
curl --get http://localhost:8000/router_policy --data-urlencode model=qwen
```

Use a returned `router_id` and revision in `PATCH /router_policy`:

```json
{
"router_id": "<router_id from GET>",
"expected_revision": 0,
"parameters": {
"overlap_score_credit": 1.5,
"prefill_load_scale": 2.0,
"router_temperature": 0.1
}
}
```

The default policy accepts the following finite, non-negative parameters:
`overlap_score_credit`, `overlap_score_credit_decay`, `prefill_load_scale`,
`decode_active_request_weight`, `host_cache_hit_weight`, `disk_cache_hit_weight`,
`shared_cache_multiplier`, and `router_temperature`. These change policy scoring only;
host-owned load accounting and queue settings remain unchanged.

Omitted parameters retain their current values. A parameter set to `null` restores that
parameter's resolved startup value, including explicit policy YAML overrides. To restore all
startup parameters, send `reset: true` instead of `parameters`. Every successful update or
reset advances the revision. The response returns the target metadata and applied configuration.

Updates must preserve the worker signals available when the pipeline was constructed. For
example, a policy initialized without cache inputs cannot enable cache-dependent scoring at
runtime. Rebuild that router with cache inputs enabled first. Temporarily disabling a scoring
term does not prevent restoring its startup value later.

| Status | Meaning |
| --- | --- |
| `200` | Query succeeded, or the complete update was applied. |
| `400` | Invalid JSON/query, empty patch, or conflicting reset/parameters. |
| `404` | The router instance ID is unknown or has expired. |
| `409` | `expected_revision` does not match. Query the current configuration before retrying. |
| `413` | Request body exceeds 16 KiB. |
| `415` | Request is not sent as `application/json`. |
| `422` | Unknown request fields, missing revision, unsupported policy, unknown parameter, invalid value, or a change requiring unavailable worker signals. No parameters are changed. |
| `503` | The scheduler targeted by a PATCH stopped or did not respond within five seconds. Query its state before retrying an update. |

The same frontend admin switch controls both this API and `/busy_threshold`:
`DYN_DISABLE_FRONTEND_ADMIN_API=1` disables them. Busy thresholds remain separately managed by
`/busy_threshold` and are not part of the policy revision.

Updates are process-local and not persisted. A rebuilt router receives a new ID and starts
from its configured startup values. A controller must address each frontend replica directly,
retain the desired settings, and reapply them to new router instances; a load-balanced request
does not update every replica. The standalone Python router does not expose this HTTP API.
Rust hosts can use `KvRouter::runtime_policy()` or `LocalScheduler::runtime_policy()` to access
the same actor-serialized control handle.

## Prometheus Metrics

The router exposes Prometheus metrics on the frontend's HTTP port (default 8000) at `/metrics`:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -608,3 +608,31 @@ worker_selection:
```

The provider also accepts `overlap_score_credit_decay`, `decode_active_request_weight`, `host_cache_hit_weight`, `disk_cache_hit_weight`, and `shared_cache_multiplier`. Omitted parameters inherit router configuration. If no shared-cache multiplier is set, the default policy uses `0.5` when shared cache is enabled. Values must be finite and non-negative. Per-request score and temperature overrides do not apply to this policy. The builtin skips cache inputs at zero overlap credit and for plain disaggregated decode.
Comment thread
tedzhouhk marked this conversation as resolved.
Resolved

## Runtime Configuration

Policies can opt in to process-local configuration updates with
`WorkerSelectionPolicy::with_runtime_config(initial_parameters, builder)`. The initial JSON
object defines the patchable parameter names and their resolved reset values. The builder
receives a complete parameter object and must validate it before returning a replacement policy.
It runs synchronously on the scheduler actor: do not perform I/O or mutate shared policy state
while constructing a replacement. Return `RuntimePolicyError::InvalidParameters` for invalid
values. Policies that do not opt in remain startup-only.

The scheduler control handle exposes `get()` and `update(RuntimePolicyUpdate)`. An update
requires the expected revision; use `RuntimePolicyUpdate::patch` for partial replacement or
`RuntimePolicyUpdate::reset` to restore startup parameters. Null values in a patch restore the
corresponding startup parameter. Unknown keys, empty patches, and stale revisions are rejected.
`RuntimePolicyConfig::revision()` and `parameters()` expose the applied state. Revisions start
at zero and are local to one policy instance; external controllers must also identify the router
instance to avoid applying a stale update after a rebuild.

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.

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
its scoring weights and temperature. See [Runtime Policy Updates](configuration-and-tuning.md#runtime-policy-updates)
for the frontend HTTP contract, scope, and persistence limits.
16 changes: 16 additions & 0 deletions examples/router/custom-policy-example/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -376,3 +376,19 @@ python3 -m dynamo.mocker \
```

Send the same `curl` request from a fourth terminal. The frontend log records separate prefill and decode selections. Each worker set runs its own filter, scorer, and picker.

## Runtime Parameters
Comment thread
tedzhouhk marked this conversation as resolved.
Resolved

The `simple-filter-score-pick` provider demonstrates optional runtime updates for
`min_device_overlap_blocks`. It supplies resolved startup parameters and a pure replacement
builder through `WorkerSelectionPolicy::with_runtime_config`. The builder reuses startup
validation and constructs new filter/scorer/picker state without changing the host's cache,
queue, or worker inputs.

Query `GET /router_policy` on the frontend to obtain a router instance ID and revision, then
use `PATCH /router_policy` with that ID, `expected_revision`, and
`parameters: {"min_device_overlap_blocks": 2.0}`. Other example policies remain startup-only.
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
A positive threshold requires usable device-overlap data. Cold or replay requests without
that data, including in the Mocker walkthrough, can have no eligible workers.
For reset semantics and response codes, see the
[runtime policy API](https://github.com/ai-dynamo/dynamo/blob/main/docs/fern/pages/developer-guide/knowledge-base/modular-components/router/configuration-and-tuning.md#runtime-policy-updates).
Original file line number Diff line number Diff line change
Expand Up @@ -14,3 +14,4 @@ description = "Filter, scorer, and picker custom worker-selection policy example
[dependencies]
dynamo-kv-router = { workspace = true, features = ["standalone-selection"] }
serde.workspace = true
serde_json.workspace = true
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ use std::sync::Arc;
use dynamo_kv_router::KvRouterConfig;
use dynamo_kv_router::plugins::RouterPluginRegistry;
use dynamo_kv_router::plugins::worker_selection::{
WorkerFilter, WorkerSelectionPolicy, WorkerSelectionPolicyFactory,
RuntimePolicyError, WorkerFilter, WorkerSelectionPolicy, WorkerSelectionPolicyFactory,
WorkerSelectionPolicyParameters, WorkerSelectionPolicyProviderError,
WorkerSelectionPolicyRegistryError,
};
Expand Down Expand Up @@ -49,20 +49,52 @@ fn provider(

Ok(Arc::new(
move |config: &KvRouterConfig, worker_type, _partition| {
let filters: Vec<Box<dyn WorkerFilter>> = vec![Box::new(MinimumDeviceOverlapFilter {
min_device_overlap_blocks,
})];
WorkerSelectionPolicy::new_with_filters(
config.clone(),
worker_type.as_str(),
filters,
vec![Box::new(ActiveRequestsScorer)],
Box::new(RequestAwarePicker),
let policy = build_policy(config, worker_type.as_str(), min_device_overlap_blocks);
let config = config.clone();
policy.with_runtime_config(
[(
"min_device_overlap_blocks".into(),
min_device_overlap_blocks.into(),
)]
.into_iter()
.collect(),
Arc::new(move |values| {
let parameters: Parameters = serde_json::from_value(serde_json::Value::Object(
values,
))
.map_err(|error| RuntimePolicyError::InvalidParameters(error.to_string()))?;
validate_min_device_overlap_blocks(parameters.min_device_overlap_blocks)
.map_err(|error| {
RuntimePolicyError::InvalidParameters(error.to_string())
})?;
Ok(build_policy(
&config,
worker_type.as_str(),
parameters.min_device_overlap_blocks,
))
}),
)
},
))
}

fn build_policy(
config: &KvRouterConfig,
worker_label: &'static str,
min_device_overlap_blocks: f64,
) -> WorkerSelectionPolicy {
let filters: Vec<Box<dyn WorkerFilter>> = vec![Box::new(MinimumDeviceOverlapFilter {
min_device_overlap_blocks,
})];
WorkerSelectionPolicy::new_with_filters(
config.clone(),
worker_label,
filters,
vec![Box::new(ActiveRequestsScorer)],
Box::new(RequestAwarePicker),
)
}

/// Register the `simple-filter-score-pick` policy type.
pub fn register(
registry: &mut RouterPluginRegistry,
Expand Down
1 change: 1 addition & 0 deletions lib/bindings/kvbm/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions lib/bindings/python/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

6 changes: 6 additions & 0 deletions lib/kv-router/src/plugins/worker_selection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,12 @@
mod config;
mod context;
mod inputs;
mod runtime;

pub(crate) use runtime::RuntimePolicyState;
pub use runtime::{
RuntimePolicyBuilder, RuntimePolicyConfig, RuntimePolicyError, RuntimePolicyUpdate,
};

pub use context::WorkerSelectionContext;
pub(crate) use inputs::{CacheSnapshot, CandidateData, WorkerCacheData};
Expand Down
Loading
Loading