Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
36 changes: 33 additions & 3 deletions crates/core/src/engine/common/sequence.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,22 @@ use crate::engine::common::hashing::{
BlockHash, SequenceHash, XXH3_SEED, compute_block_hash_for_tokens, compute_next_sequence_hash,
};
use rand::random;
use uuid::Uuid;
use xxhash_rust::xxh3::xxh3_64;

/// Synthetic token for output position `output_ordinal` of a request that has
/// no planned token there.
///
/// The token depends only on the request ID and position, so replaying a
/// request under the same ID reproduces its output tokens and KV block hashes,
/// while distinct requests receive independent streams. Hashing a fixed
/// little-endian encoding keeps values stable across platforms and toolchains.
fn synthetic_output_token(request_id: Uuid, output_ordinal: usize) -> u32 {
let mut bytes = [0; 24];
bytes[..16].copy_from_slice(request_id.as_bytes());
bytes[16..].copy_from_slice(&(output_ordinal as u64).to_le_bytes());
xxh3_64(&bytes) as u32
}

#[derive(Debug)]
struct FlatTokens {
Expand Down Expand Up @@ -103,6 +119,7 @@ impl BlockIdentity {
/// positional lineage, or allocation bookkeeping.
#[derive(Debug)]
pub(crate) struct RequestSequence {
request_id: Uuid,
tokens: FlatTokens,
block_size: usize,
max_output_tokens: usize,
Expand All @@ -117,6 +134,7 @@ pub(crate) struct RequestSequence {
impl RequestSequence {
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
request_id: Uuid,
tokens: Vec<u32>,
max_output_tokens: usize,
output_capacity_hint: usize,
Expand Down Expand Up @@ -159,6 +177,7 @@ impl RequestSequence {
}

let sequence = Self {
request_id,
tokens: FlatTokens::new(tokens, output_capacity_hint, block_size, emit_token_ids),
block_size,
max_output_tokens,
Expand Down Expand Up @@ -203,6 +222,7 @@ impl RequestSequence {

/// Append the next planned or synthetic token.
///
/// Positions without a planned token derive one from the request ID.
/// Returns the token and whether this append opened a new partial block.
pub(crate) fn generate_token(&mut self) -> (u32, bool) {
assert!(
Expand All @@ -213,7 +233,7 @@ impl RequestSequence {
.planned_output_ids
.as_ref()
.and_then(|ids| ids.get(self.generated_tokens).copied())
.unwrap_or_else(random::<u32>);
.unwrap_or_else(|| synthetic_output_token(self.request_id, self.generated_tokens));
let opened_partial = self.len().is_multiple_of(self.block_size);
self.tokens.push(token);
self.generated_tokens += 1;
Expand Down Expand Up @@ -355,6 +375,7 @@ mod tests {
planned_output_ids: Option<Vec<u32>>,
) -> (RequestSequence, Vec<BlockIdentity>) {
RequestSequence::new(
Uuid::nil(),
tokens,
max_output_tokens,
max_output_tokens,
Expand Down Expand Up @@ -452,8 +473,17 @@ mod tests {

#[test]
fn completion_footprint_is_bounded_by_requested_output() {
let (sequence, identities) =
RequestSequence::new(vec![0, 1, 2, 3, 4], 100, 3, 4, false, false, false, None);
let (sequence, identities) = RequestSequence::new(
Uuid::nil(),
vec![0, 1, 2, 3, 4],
100,
3,
4,
false,
false,
false,
None,
);
assert_eq!(sequence.current_known_blocks(), 2);
assert_eq!(sequence.to_completion_blocks(), 27);
assert_eq!(
Expand Down
1 change: 1 addition & 0 deletions crates/core/src/engine/kv_manager/g1_manager.rs
Original file line number Diff line number Diff line change
Expand Up @@ -394,6 +394,7 @@ mod tests {
let mut manager =
G1Manager::new_with_event_sink(8, 4, KvEventPublishers::new(Some(sink)), 0);
let (mut sequence, identities) = RequestSequence::new(
owner,
(0..8).collect(),
4,
4,
Expand Down
4 changes: 3 additions & 1 deletion crates/core/src/engine/kv_manager/state_cache_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ fn request(
output: usize,
) -> (RequestSequence, BlockRequestLease) {
let (sequence, identities) = RequestSequence::new(
owner,
(0..tokens).map(|value| value as u32).collect(),
output,
output,
Expand Down Expand Up @@ -895,8 +896,9 @@ fn kda_partial_page_can_match_when_tokens_after_the_hit_diverge() {
manager.finish_lease(owner, lease);
let mut tokens: Vec<u32> = (0..19).collect();
tokens[16] = 999;
let (mut query, ids) = RequestSequence::new(tokens.clone(), 0, 0, 6, true, true, false, None);
let owner = Uuid::from_u128(603);
let (mut query, ids) =
RequestSequence::new(owner, tokens.clone(), 0, 0, 6, true, true, false, None);
let mut reader = BlockRequestLease::new(owner, ids);
reader.configure_prefix_hashes(&tokens, 2);
assert_eq!(
Expand Down
18 changes: 14 additions & 4 deletions crates/core/src/engine/kv_manager/vllm_backend.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1709,7 +1709,8 @@ mod tests {
emit_token_ids: bool,
) -> (RequestSequence, BlockRequestLease) {
let tokens = (0..hashes.len() * 4).map(|token| token as u32).collect();
let (sequence, _) = RequestSequence::new(tokens, 0, 0, 4, true, true, emit_token_ids, None);
let (sequence, _) =
RequestSequence::new(owner, tokens, 0, 0, 4, true, true, emit_token_ids, None);
let identities = hashes
.iter()
.copied()
Expand Down Expand Up @@ -1948,8 +1949,17 @@ mod tests {
// blocks plus one partial tail occupy four physical blocks.
let first = Uuid::from_u128(1);
let first_tokens = (0..15).collect::<Vec<u32>>();
let (mut first_sequence, first_identities) =
RequestSequence::new(first_tokens, 2, 2, 4, true, true, false, Some(vec![15, 16]));
let (mut first_sequence, first_identities) = RequestSequence::new(
first,
first_tokens,
2,
2,
4,
true,
true,
false,
Some(vec![15, 16]),
);
let mut first_lease = BlockRequestLease::new(first, first_identities);
ready(manager.allocate_lease(first, &mut first_lease, 15, 0));
manager.finalize_lease_computed_prefix(first, &mut first_sequence, &mut first_lease, 0, 15);
Expand Down Expand Up @@ -1994,7 +2004,7 @@ mod tests {
let mut second_tokens = (0..10).collect::<Vec<u32>>();
second_tokens.extend([100, 101, 102, 103]);
let (mut second_sequence, second_identities) =
RequestSequence::new(second_tokens, 0, 0, 4, true, true, false, None);
RequestSequence::new(second, second_tokens, 0, 0, 4, true, true, false, None);
let mut second_lease = BlockRequestLease::new(second, second_identities);
let prefill = manager.get_lease_prefill_cost(&second_sequence, &second_lease);
assert_eq!(prefill.cached_tokens, 8);
Expand Down
2 changes: 1 addition & 1 deletion crates/core/src/engine/scheduler/vllm/host_offload.rs
Original file line number Diff line number Diff line change
Expand Up @@ -694,7 +694,7 @@ mod tests {
tokens: Vec<u32>,
) -> (RequestSequence, BlockRequestLease, VllmHostRequestState) {
let (sequence, identities) =
RequestSequence::new(tokens, 0, 0, 4, true, false, false, None);
RequestSequence::new(owner, tokens, 0, 0, 4, true, false, false, None);
let lease = BlockRequestLease::new(owner, identities);
let host = VllmHostRequestState::new(&sequence, &lease, 4);
(sequence, lease, host)
Expand Down
1 change: 1 addition & 0 deletions crates/core/src/engine/scheduler/vllm/request.rs
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ impl RequestKvState {
planned_output_ids: Option<Vec<u32>>,
) -> Self {
let (sequence, identities) = RequestSequence::new(
owner,
tokens,
max_output_tokens,
output_capacity_hint,
Expand Down
3 changes: 2 additions & 1 deletion crates/core/src/engine/trace.rs
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,9 @@ pub fn g1_parent_chain_events(block_size: usize) -> Vec<KvEvent> {
let prompt_len = computed_after - 2;
let prompt_len_u32 =
u32::try_from(prompt_len).expect("ordering regression prompt length must fit in u32");
let owner = Uuid::from_u128(1);
let (mut sequence, identities) = RequestSequence::new(
owner,
(0..prompt_len_u32).collect(),
2,
2,
Expand All @@ -41,7 +43,6 @@ pub fn g1_parent_chain_events(block_size: usize) -> Vec<KvEvent> {
false,
Some(vec![prompt_len_u32, prompt_len_u32 + 1]),
);
let owner = Uuid::from_u128(1);
let mut lease = BlockRequestLease::new(owner, identities);
let (events, sink) = capture_kv_event_sink();
let mut manager =
Expand Down
55 changes: 55 additions & 0 deletions crates/core/tests/artifacts.rs
Original file line number Diff line number Diff line change
Expand Up @@ -261,6 +261,61 @@ fn cache_events_advertise_reusable_prefixes_only_when_caching_is_enabled() {
}
}

#[test]
fn canonical_replay_reproduces_unplanned_output_tokens_and_kv_hashes() {
let output_tokens = |artifacts: &ReplayArtifacts, request_index: usize| -> Vec<u32> {
let request_id = artifacts.requests[request_index].request_id;
artifacts
.outputs
.iter()
.filter(|output| output.request_id == request_id)
.filter_map(|output| output.token_id)
.collect()
};
for backend in [Backend::Vllm, Backend::Trtllm, Backend::Sglang] {
let run = || {
let mut replay_spec = spec(backend, 1, 1);
// Identical length-only requests carry no output plan, so every
// generated token and output-block hash is engine-synthesized.
replay_spec.requests = (0..2)
.map(|index| {
let mut request =
replay_request(&format!("unplanned-{index}"), 0.0, (0..6).collect());
request.output_tokens = 9;
request
})
.collect();
Replayer::new(replay_spec, ReplayEngineFactory::new())
.unwrap()
.with_capture_options(ReplayCaptureOptions {
determinism: ReplayDeterminism::CanonicalV1,
..ReplayCaptureOptions::default()
})
.run_with_artifacts(ReplayArtifactKvEventVisibility::Native)
.unwrap()
.1
};
let first = run();
let second = run();
for request_index in 0..2 {
let tokens = output_tokens(&first, request_index);
assert_eq!(tokens.len(), 9, "{backend:?}");
assert_eq!(
tokens,
output_tokens(&second, request_index),
"{backend:?} request {request_index}"
);
}
assert_ne!(
output_tokens(&first, 0),
output_tokens(&first, 1),
"{backend:?}: distinct requests must not share synthetic output blocks"
);
assert!(!first.kv_events.is_empty(), "{backend:?}");
assert_eq!(kv_parts(&first), kv_parts(&second), "{backend:?}");
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
}
}

#[test]
fn capped_passes_respect_visibility_boundaries() {
let capped = |visibility| {
Expand Down
Loading