Skip to content
Closed
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
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,19 @@ previous partition and moved to the new one, which can leave the previous
partition not ready. Assign a unique ID to every live worker across the entire
service.

A worker that can consume router hints registers `router_hint_worker_type`
(its backend role, used to match hint sources to targets) and, when it can also
serve as a source, `router_hint_source_control_endpoints` keyed by global DP
rank:

```json
{
"worker_id": 1,
"router_hint_worker_type": "decode",
"router_hint_source_control_endpoints": { "0": "tcp://worker:5600", "1": "tcp://worker:5601" }
}
```

`PATCH /workers/{worker_id}` updates supplied fields, `DELETE
/workers/{worker_id}` removes the worker, and `GET /workers` lists catalog
state. `model_name` and `routing_group` scope selection, indexer, and load state;
Expand Down Expand Up @@ -222,6 +235,24 @@ blocks. `decode_busy` compares it against the worker's `total_kv_blocks` at
`conditional_disagg_decode_busy_threshold` and is omitted when either the
threshold or the capacity is unknown.

A booking response may also carry a `router_hint` when another worker of the
same `router_hint_worker_type` holds a longer cached prefix than the chosen
worker and advertises a source control endpoint for the matching DP rank:

```json
{
"router_hint": {
"source_control_endpoint": "tcp://worker-1:5600",
"block_hashes": [8713492873, 1928374650]
}
}
```

`block_hashes` are root-aligned external sequence hashes; entry `i` is request
block `i`, and the target decides which suffix to fetch from the source. Hints
require a local event-driven primary indexer (no approximate or remote primary)
and are never attached to query-only `select` responses.

### Advisory selection

`POST /select` accepts `"advisory": true` to select from current scheduler
Expand Down
39 changes: 32 additions & 7 deletions lib/kv-router/src/services/indexer/backend.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,9 @@ use crate::config::{ApproximateCachePolicyKind, KvRouterConfig};
use crate::identity::RoutingPartitionId;
use crate::indexer::{
ApproximateRetentionConfig, KvIndexer, KvIndexerInterface, KvIndexerMetrics, KvRouterError,
LowerTierIndexers, MatchDetails, RoutingDecisionHashes, SyncIndexer, TieredMatchDetails,
TieredMatchProvider, WireTieredMatchDetails, query_lower_tiers,
LowerTierIndexers, LowerTierQueryOptions, MatchDetails, RoutingDecisionHashes, SyncIndexer,
TieredMatchDetails, TieredMatchProvider, WireTieredMatchDetails,
query_lower_tiers_with_options,
};
use crate::protocols::{
KvCacheEventData, LocalBlockHash, OverlapScores, RouterEvent, WorkerId, WorkerWithDpRank,
Expand Down Expand Up @@ -663,24 +664,48 @@ impl Indexer {
&self,
sequence: Vec<LocalBlockHash>,
) -> std::result::Result<TieredMatchDetails, KvRouterError> {
self.find_tiered_matches_with_options(sequence, LowerTierQueryOptions::default())
.await
}

/// [`Self::find_tiered_matches`] with lower-tier query options. Router-hint
/// chain retention is honored only by a local event-driven primary (see
/// [`Self::supports_router_hint_chain_retention`]); other shapes ignore it.
pub async fn find_tiered_matches_with_options(
&self,
sequence: Vec<LocalBlockHash>,
options: LowerTierQueryOptions,
) -> std::result::Result<TieredMatchDetails, KvRouterError> {
let options = LowerTierQueryOptions {
retain_router_hint_chain: options.retain_router_hint_chain
&& self.supports_router_hint_chain_retention(),
};
let (device, lower_tier) = match self {
Indexer::Single {
primary,
lower_tier,
..
} => {
let device = primary.find_match_details(sequence.clone()).await?;
let lt = query_lower_tiers(lower_tier, &sequence, &device);
let device = primary
.find_match_details_with_options(
sequence.clone(),
options.retain_router_hint_chain,
)
.await?;
let lt = query_lower_tiers_with_options(lower_tier, &sequence, &device, options);
(device, lt)
}
Indexer::Concurrent {
primary,
lower_tier,
..
} => {
let device: MatchDetails =
primary.backend().find_match_details_impl(&sequence, false);
let lt = query_lower_tiers(lower_tier, &sequence, &device);
let device: MatchDetails = primary.backend().find_match_details_impl_with_options(
&sequence,
false,
options.retain_router_hint_chain,
);
let lt = query_lower_tiers_with_options(lower_tier, &sequence, &device, options);
(device, lt)
}
Indexer::Remote { primary, .. } => {
Expand Down
14 changes: 14 additions & 0 deletions lib/kv-router/src/services/selection/catalog.rs
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,20 @@ impl WorkerCatalog {
.flatten()
}

/// Whether any schedulable worker in `key`'s partition can consume router
/// hints. Worker-level metadata, so one representative rank suffices.
pub(super) fn has_router_hint_capable_workers(&self, key: &RoutingPartitionId) -> bool {
self.workers.read().values().any(|record| {
record.lifecycle == WorkerLifecycle::Schedulable
&& record.model_name == key.model_name
&& record.routing_group == key.routing_group
&& record
.router_hint_worker_type
.as_deref()
.is_some_and(|worker_type| !worker_type.is_empty())
})
}

pub(super) fn scheduler_configs_for_key(
&self,
key: &RoutingPartitionId,
Expand Down
Loading
Loading