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
2 changes: 1 addition & 1 deletion .github/workflows/benchmark-radix-tree.yml
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ jobs:
timeout-minutes: 40
run: |
source "$HOME/.cargo/env"
cargo bench --bench radix_tree_benchmark -- benchmark_summary --exact 2>&1 | tee benchmark_output.txt
cargo bench --bench radix_tree_benchmark -- benchmark_summary 2>&1 | tee benchmark_output.txt

- name: Upload benchmark results
if: always()
Expand Down
325 changes: 321 additions & 4 deletions model_gateway/benches/radix_tree_benchmark.rs
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
//! Benchmarks for the radix tree implementations used in cache-aware routing.
//!
//! This benchmark tests both implementations:
//! This benchmark tests all three implementations:
//! - StringTree: Character-based tree for HTTP router (text input)
//! - TokenTree: Token-based tree for gRPC router (pre-tokenized input)
//! - PositionalIndexer: Event-driven indexer for gRPC router (KV cache events)
//!
//! Run with: cargo bench --bench radix_tree_benchmark
//!
Expand All @@ -25,7 +26,10 @@ use std::{
};

use criterion::{criterion_group, criterion_main, Criterion};
use kv_index::{StringTree, TokenTree};
use kv_index::{
compute_content_hash, compute_request_content_hashes, ContentHash, PositionalIndexer,
SequenceHash, StoredBlock, StringTree, TokenTree,
};
use rand::{
distr::{Alphanumeric, SampleString},
rng as thread_rng, Rng,
Expand Down Expand Up @@ -393,11 +397,256 @@ macro_rules! bench_token_concurrent {
}};
}

// ============================================================================
// PositionalIndexer Helpers
// ============================================================================

/// Generate token chunks of `block_size` tokens each.
fn generate_token_chunks(num_blocks: usize, block_size: usize) -> Vec<Vec<TokenId>> {
let mut rng = thread_rng();
(0..num_blocks)
.map(|_| {
(0..block_size)
.map(|_| rng.random_range(0..50000))
.collect()
})
.collect()
}

fn chunks_to_stored_blocks(chunks: &[Vec<TokenId>]) -> Vec<StoredBlock> {
chunks
.iter()
.enumerate()
.map(|(i, tokens)| StoredBlock {
seq_hash: SequenceHash(i as u64 + 1),
content_hash: compute_content_hash(tokens),
})
.collect()
}

fn flatten_tokens(chunks: &[Vec<TokenId>]) -> Vec<TokenId> {
chunks.iter().flat_map(|c| c.iter().copied()).collect()
}

/// Build a populated indexer. Workers share `shared_prefix_blocks` initial blocks
/// (simulating common system prompt) then diverge.
fn build_populated_indexer(
workers: &[String],
blocks_per_worker: usize,
block_size: usize,
shared_prefix_blocks: usize,
jump_size: usize,
) -> (Arc<PositionalIndexer>, Vec<Vec<Vec<TokenId>>>) {
let indexer = Arc::new(PositionalIndexer::new(jump_size));

let shared_chunks = generate_token_chunks(shared_prefix_blocks, block_size);
let shared_blocks = chunks_to_stored_blocks(&shared_chunks);

let mut all_worker_chunks = Vec::with_capacity(workers.len());

for worker in workers {
indexer.apply_stored(worker, &shared_blocks, None).unwrap();

let unique_count = blocks_per_worker.saturating_sub(shared_prefix_blocks);
let unique_chunks = generate_token_chunks(unique_count, block_size);
let unique_blocks: Vec<StoredBlock> = unique_chunks
.iter()
.enumerate()
.map(|(i, tokens)| StoredBlock {
seq_hash: SequenceHash((shared_prefix_blocks + i) as u64 + 1),
content_hash: compute_content_hash(tokens),
})
.collect();
Comment on lines +452 to +459

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The logic for creating unique_blocks is very similar to the chunks_to_stored_blocks helper function. To improve code reuse and maintainability, you could consider refactoring chunks_to_stored_blocks to accept a starting index for the seq_hash calculation. This would allow you to replace these lines with a single call to the modified helper, for example: let unique_blocks = chunks_to_stored_blocks(&unique_chunks, shared_prefix_blocks); (after updating the helper and other call sites).

References
  1. Extract duplicated logic into a shared helper function to improve maintainability and reduce redundancy.


if !unique_blocks.is_empty() {
let parent = SequenceHash(shared_prefix_blocks as u64);
indexer
.apply_stored(worker, &unique_blocks, Some(parent))
.unwrap();
}

let mut worker_chunks = shared_chunks.clone();
worker_chunks.extend(unique_chunks);
all_worker_chunks.push(worker_chunks);
}

(indexer, all_worker_chunks)
}

// ============================================================================
// PositionalIndexer Benchmark Macros
// ============================================================================

/// Macro for STORE benchmarks with PositionalIndexer
macro_rules! bench_indexer_store {
($group:expr, $num_workers:expr, $blocks_per_worker:expr, $block_size:expr, $workers:expr) => {{
let printed = Arc::new(AtomicBool::new(false));
let bench_name = format!(
"indexer_store_{}w_{}blk_{}bs",
$num_workers, $blocks_per_worker, $block_size
);
let workers_clone = $workers.clone();

$group.bench_function(&bench_name, |b| {
let workers = workers_clone.clone();
let printed = printed.clone();

b.iter_custom(|iters| {
let start = Instant::now();
for _ in 0..iters {
let indexer = PositionalIndexer::new(64);
for worker in &workers {
let chunks = generate_token_chunks($blocks_per_worker, $block_size);
let blocks = chunks_to_stored_blocks(&chunks);
Comment on lines +499 to +500

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

STORE benchmark timing includes input-generation overhead.

Line 499 and Line 500 run token generation + block hashing inside the measured loop, so this benchmark is not isolating apply_stored throughput.

Proposed fix (pre-generate blocks outside timed loop)
         $group.bench_function(&bench_name, |b| {
             let workers = workers_clone.clone();
+            let worker_blocks: Vec<Vec<StoredBlock>> = (0..workers.len())
+                .map(|_| {
+                    let chunks = generate_token_chunks($blocks_per_worker, $block_size);
+                    chunks_to_stored_blocks(&chunks)
+                })
+                .collect();
             let printed = printed.clone();

             b.iter_custom(|iters| {
                 let start = Instant::now();
                 for _ in 0..iters {
                     let indexer = PositionalIndexer::new(64);
-                    for worker in &workers {
-                        let chunks = generate_token_chunks($blocks_per_worker, $block_size);
-                        let blocks = chunks_to_stored_blocks(&chunks);
-                        let _ = indexer.apply_stored(black_box(worker), black_box(&blocks), None);
+                    for (i, worker) in workers.iter().enumerate() {
+                        let blocks = &worker_blocks[i];
+                        let _ = indexer.apply_stored(black_box(worker), black_box(blocks), None);
                     }
                 }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
let chunks = generate_token_chunks($blocks_per_worker, $block_size);
let blocks = chunks_to_stored_blocks(&chunks);
$group.bench_function(&bench_name, |b| {
let workers = workers_clone.clone();
let worker_blocks: Vec<Vec<StoredBlock>> = (0..workers.len())
.map(|_| {
let chunks = generate_token_chunks($blocks_per_worker, $block_size);
chunks_to_stored_blocks(&chunks)
})
.collect();
let printed = printed.clone();
b.iter_custom(|iters| {
let start = Instant::now();
for _ in 0..iters {
let indexer = PositionalIndexer::new(64);
for (i, worker) in workers.iter().enumerate() {
let blocks = &worker_blocks[i];
let _ = indexer.apply_stored(black_box(worker), black_box(blocks), None);
}
}
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@model_gateway/benches/radix_tree_benchmark.rs` around lines 499 - 500, The
benchmark currently measures token generation and hashing because
generate_token_chunks and chunks_to_stored_blocks are called inside the timed
loop; move the generation and conversion out of the measured section so only
apply_stored is timed: precompute chunks with
generate_token_chunks($blocks_per_worker, $block_size) and convert them to
blocks via chunks_to_stored_blocks(...) before entering the benchmark timing
loop (or store a Vec of blocks and reuse/clone as needed inside each iteration),
then call apply_stored on those precomputed blocks inside the timed loop to
isolate throughput measurement.

let _ = indexer.apply_stored(black_box(worker), black_box(&blocks), None);

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

apply_stored failures are silently ignored in measured paths.

Line 501 and Line 613-614 discard the Result, so failed writes can be counted as successful ops and skew throughput numbers.

Proposed fix (enforce success-path measurement)
-                        let _ = indexer.apply_stored(black_box(worker), black_box(&blocks), None);
+                        indexer
+                            .apply_stored(black_box(worker), black_box(&blocks), None)
+                            .expect("STORE benchmark input must be valid");

...

-                                        let _ =
-                                            indexer.apply_stored(&worker, &blocks, Some(parent));
+                                        indexer
+                                            .apply_stored(&worker, &blocks, Some(parent))
+                                            .expect("CONCURRENT benchmark write must be valid");

Also applies to: 613-614

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@model_gateway/benches/radix_tree_benchmark.rs` at line 501, The benchmark is
silently ignoring failures from indexer.apply_stored which can make failed
writes count as successful ops; change the measured calls to check the Result
instead of discarding it (e.g., use expect/unwrap or match and panic/log on Err)
for the apply_stored invocation used with black_box(worker) and
black_box(&blocks) so any failure aborts the benchmark and only true successes
are measured; apply the same fix to the other occurrences around the later calls
(the ones at the spots corresponding to lines 613–614).

}
}
let duration = start.elapsed();

if !printed.swap(true, Ordering::Relaxed) {
let total_blocks =
iters as f64 * $num_workers as f64 * $blocks_per_worker as f64;
let ops_per_sec = iters as f64 / duration.as_secs_f64();
let latency_us = duration.as_nanos() as f64 / iters as f64 / 1000.0;
let blocks_per_sec = total_blocks / duration.as_secs_f64();
add_result(
"indexer",
format!(
"{:>3}w | {:>4}blk x {:>2}tok | STORE | {:>8.0} ops/s | {:>7.1} µs | {:>8.0} blk/s",
$num_workers, $blocks_per_worker, $block_size, ops_per_sec, latency_us, blocks_per_sec
),
);
}

duration
});
});
}};
}

/// Macro for MATCH benchmarks with PositionalIndexer (find_matches — the hot path)
macro_rules! bench_indexer_match {
($group:expr, $num_workers:expr, $query_blocks:expr, $block_size:expr,
$indexer:expr, $query_hashes:expr) => {{
let printed = Arc::new(AtomicBool::new(false));
let bench_name = format!(
"indexer_match_{}w_{}qblk_{}bs",
$num_workers, $query_blocks, $block_size
);
let indexer_clone = $indexer.clone();
let hashes_clone = $query_hashes.clone();

$group.bench_function(&bench_name, |b| {
let indexer = indexer_clone.clone();
let hashes = hashes_clone.clone();
let mut idx = 0;
let printed = printed.clone();

b.iter_custom(|iters| {
let start = Instant::now();
for _ in 0..iters {
let result = indexer.find_matches(black_box(&hashes[idx % hashes.len()]));
black_box(result);
idx += 1;
}
let duration = start.elapsed();

if !printed.swap(true, Ordering::Relaxed) {
let ops_per_sec = iters as f64 / duration.as_secs_f64();
let latency_us = duration.as_nanos() as f64 / iters as f64 / 1000.0;
let throughput_mblk =
(ops_per_sec * $query_blocks as f64) / 1_000_000.0;
add_result(
"indexer",
format!(
"{:>3}w | {:>4}blk x {:>2}tok | MATCH | {:>8.0} ops/s | {:>7.1} µs | {:>5.2} Mblk/s",
$num_workers, $query_blocks, $block_size, ops_per_sec, latency_us, throughput_mblk
),
);
}

duration
});
});
}};
}

/// Macro for CONCURRENT benchmarks with PositionalIndexer
macro_rules! bench_indexer_concurrent {
($group:expr, $num_workers:expr, $block_size:expr, $num_threads:expr, $ops_per_thread:expr) => {{
let printed = Arc::new(AtomicBool::new(false));
let bench_name = format!("indexer_concurrent_{}w", $num_workers);

$group.bench_function(&bench_name, |b| {
let printed = printed.clone();

b.iter_custom(|iters| {
let start = Instant::now();
for _ in 0..iters {
let workers = generate_worker_endpoints($num_workers);
let (indexer, worker_chunks) =
build_populated_indexer(&workers, 64, $block_size, 8, 64);

Comment on lines +586 to +589

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Concurrent benchmark measures setup + workload together.

Line 586-589 rebuilds worker endpoints and repopulates the indexer inside the timed section. That inflates/obscures the mixed read/write throughput signal.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@model_gateway/benches/radix_tree_benchmark.rs` around lines 586 - 589, The
benchmark currently rebuilds worker endpoints and repopulates the indexer inside
the timed section (calls to generate_worker_endpoints and
build_populated_indexer), which mixes setup cost with measured workload; move
the calls to generate_worker_endpoints($num_workers) and
build_populated_indexer(&workers, 64, $block_size, 8, 64) out of the timed
benchmark block so that the timed section only runs the workload (reads/writes)
against the already-populated indexer and worker_chunks, and ensure any
per-iteration state that must be reset is done outside timing or via cloned
lightweight copies before starting the timer.

let handles: Vec<_> = (0..$num_threads)
.map(|t| {
let indexer = Arc::clone(&indexer);
let worker = workers[t % workers.len()].clone();
let chunks = worker_chunks[t % workers.len()].clone();
let block_size = $block_size;

thread::spawn(move || {
let mut rng = thread_rng();
for i in 0..$ops_per_thread {
if i % 3 == 0 {
// Read: find_matches
let query_tokens = flatten_tokens(&chunks);
let content_hashes = compute_request_content_hashes(
&query_tokens,
block_size,
);
black_box(indexer.find_matches(&content_hashes));
} else {
// Write: apply_stored
let new_chunks = generate_token_chunks(4, block_size);
let blocks = chunks_to_stored_blocks(&new_chunks);
let parent = SequenceHash(rng.random_range(1u64..65));
let _ =
indexer.apply_stored(&worker, &blocks, Some(parent));
}
}
Comment on lines +598 to +616

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

In the concurrent benchmark, the data preparation for the read path (flatten_tokens and compute_request_content_hashes) is performed inside the hot loop. Since the chunks data doesn't change within the thread's execution, these computations can be hoisted out of the for loop. This change will ensure the benchmark more accurately measures the performance of find_matches under contention, rather than including the data preparation overhead in every iteration.

                                let mut rng = thread_rng();
                                let query_tokens = flatten_tokens(&chunks);
                                let content_hashes =
                                    compute_request_content_hashes(&query_tokens, block_size);

                                for i in 0..$ops_per_thread {
                                    if i % 3 == 0 {
                                        // Read: find_matches
                                        black_box(indexer.find_matches(&content_hashes));
                                    } else {
                                        // Write: apply_stored
                                        let new_chunks = generate_token_chunks(4, block_size);
                                        let blocks = chunks_to_stored_blocks(&new_chunks);
                                        let parent = SequenceHash(rng.random_range(1u64..65));
                                        let _ =
                                            indexer.apply_stored(&worker, &blocks, Some(parent));
                                    }
                                }

})
})
.collect();

for h in handles {
h.join().unwrap();
}
}
let duration = start.elapsed();

if !printed.swap(true, Ordering::Relaxed) {
let total_ops = iters * $num_threads as u64 * $ops_per_thread as u64;
let ops_per_sec = total_ops as f64 / duration.as_secs_f64();
add_result(
"indexer",
format!(
"{:>3}w | CONCURRENT | {:>7.0} ops/s | {} threads x {} ops",
$num_workers, ops_per_sec, $num_threads, $ops_per_thread
),
);
}

duration
});
});
}};
}

// ============================================================================
// Main Benchmark
// ============================================================================

/// Main benchmark for StringTree and TokenTree
/// Main benchmark for StringTree, TokenTree, and PositionalIndexer
fn bench_summary(c: &mut Criterion) {
let mut group = c.benchmark_group("benchmark_summary");

Expand Down Expand Up @@ -493,6 +742,60 @@ fn bench_summary(c: &mut Criterion) {
bench_token_concurrent!(group, num_workers, workers, NUM_THREADS, OPS_PER_THREAD);
}

// ========================================================================
// PositionalIndexer Benchmark
// ========================================================================
const BLOCK_SIZES: [usize; 2] = [16, 64];
const BLOCKS_PER_WORKER: [usize; 3] = [64, 256, 1024];
const QUERY_BLOCK_COUNTS: [usize; 3] = [32, 128, 512];
const SHARED_PREFIX_BLOCKS: usize = 8;
const JUMP_SIZE: usize = 64;

for &num_workers in &WORKER_COUNTS {
let workers = generate_worker_endpoints(num_workers);

for &block_size in &BLOCK_SIZES {
// STORE benchmarks
for &blocks_per_worker in &BLOCKS_PER_WORKER {
bench_indexer_store!(group, num_workers, blocks_per_worker, block_size, workers);
}

// MATCH benchmarks: build a populated indexer, then query it
let max_blocks = *BLOCKS_PER_WORKER.last().unwrap();
let (indexer, worker_chunks) = build_populated_indexer(
&workers,
max_blocks,
block_size,
SHARED_PREFIX_BLOCKS,
JUMP_SIZE,
);

for &query_blocks in &QUERY_BLOCK_COUNTS {
// Generate query hashes: mix of cached (from worker 0) and novel tokens
let num_queries = 100;
let query_hashes: Vec<Vec<ContentHash>> = (0..num_queries)
.map(|i| {
let chunks = &worker_chunks[i % worker_chunks.len()];
let take = query_blocks.min(chunks.len());
let tokens = flatten_tokens(&chunks[..take]);
compute_request_content_hashes(&tokens, block_size)
})
Comment on lines +774 to +782

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Query corpus is not mixed despite the inline comment.

Line 774 says “mix of cached and novel tokens,” but Line 776-782 builds all queries from already-indexed worker_chunks.

Proposed fix (actually mix cached + novel queries)
                 let query_hashes: Vec<Vec<ContentHash>> = (0..num_queries)
                     .map(|i| {
-                        let chunks = &worker_chunks[i % worker_chunks.len()];
-                        let take = query_blocks.min(chunks.len());
-                        let tokens = flatten_tokens(&chunks[..take]);
-                        compute_request_content_hashes(&tokens, block_size)
+                        if i % 2 == 0 {
+                            let chunks = &worker_chunks[i % worker_chunks.len()];
+                            let take = query_blocks.min(chunks.len());
+                            let tokens = flatten_tokens(&chunks[..take]);
+                            compute_request_content_hashes(&tokens, block_size)
+                        } else {
+                            let novel_chunks = generate_token_chunks(query_blocks, block_size);
+                            let novel_tokens = flatten_tokens(&novel_chunks);
+                            compute_request_content_hashes(&novel_tokens, block_size)
+                        }
                     })
                     .collect();
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@model_gateway/benches/radix_tree_benchmark.rs` around lines 774 - 782, The
comment points out query_hashes is built only from indexed worker_chunks rather
than a mixture; modify how query_hashes is created to combine cached queries
taken from worker_chunks and novel queries generated on the fly. In the block
that builds query_hashes (variable names: num_queries, query_hashes,
worker_chunks, query_blocks, block_size), produce a mix (e.g., half cached, half
novel) by mapping over indices: for cached indices reuse
flatten_tokens(&chunks[..take]) + compute_request_content_hashes, and for novel
indices synthesize new token sequences (or call the existing token generator
used elsewhere) then call compute_request_content_hashes on those; ensure the
final query_hashes Vec preserves order/length and still uses block_size and
query_blocks variables.

.collect();

bench_indexer_match!(
group,
num_workers,
query_blocks,
block_size,
indexer,
query_hashes
);
}
}

bench_indexer_concurrent!(group, num_workers, 16, NUM_THREADS, OPS_PER_THREAD);
}

group.finish();
}

Expand All @@ -507,12 +810,14 @@ fn print_summary() {
// Collect results by category
let mut string_results = Vec::new();
let mut token_results = Vec::new();
let mut indexer_results = Vec::new();

for (key, value) in results.iter() {
let category = key.split('_').skip(1).collect::<Vec<_>>().join("_");
match category.as_str() {
"string" => string_results.push(value.clone()),
"token" => token_results.push(value.clone()),
"indexer" => indexer_results.push(value.clone()),
_ => {}
}
}
Expand Down Expand Up @@ -541,7 +846,19 @@ fn print_summary() {
eprintln!("{v}");
}

eprintln!("\n{}", "=".repeat(90));
eprintln!("\n{}", "=".repeat(95));
eprintln!("POSITIONALINDEXER (kv_index::PositionalIndexer) — event-driven KV cache routing");
eprintln!("{}", "=".repeat(95));
eprintln!(
"{:>4} | {:>15} | {:>6} | {:>10} | {:>9} | {:>12}",
"Work", "Size", "Op", "Throughput", "Latency", "Bandwidth"
);
eprintln!("{}", "-".repeat(95));
for v in &indexer_results {
eprintln!("{v}");
}

eprintln!("\n{}", "=".repeat(95));
}

fn run_benchmarks(c: &mut Criterion) {
Expand Down