micromegas_object_cache/
blocks.rs

1use bytes::Bytes;
2use std::ops::Range;
3
4pub fn blocks_for_range(start: u64, end: u64, block_size: u64) -> Range<u64> {
5    debug_assert!(start < end);
6    let first = start / block_size;
7    let last = (end - 1) / block_size;
8    first..last + 1
9}
10
11pub fn block_byte_range(block_idx: u64, block_size: u64, file_size: u64) -> Range<u64> {
12    let start = block_idx * block_size;
13    let end = (start + block_size).min(file_size);
14    start..end
15}
16
17/// Largest byte span one `coalesce_runs` run may cover: `max_coalesced_get_bytes`
18/// rounded down to a whole number of blocks (at least one). Shared by
19/// `coalesce_runs` below and the fetch-budget reservation/assertion math in
20/// `range_cache/mod.rs` and `range_cache/scheduler.rs`, so every one of them
21/// agrees on what the largest possible run is.
22pub fn max_run_bytes(block_size: u64, max_coalesced_get_bytes: u64) -> u64 {
23    (max_coalesced_get_bytes / block_size).max(1) * block_size
24}
25
26/// Group sorted, deduplicated, *owned* missing block indices into maximal
27/// contiguous runs, splitting any run whose byte span would exceed
28/// `max_coalesced_get_bytes` at block boundaries. Each returned block-index
29/// range becomes one `origin.get_range` call.
30pub fn coalesce_runs(
31    sorted_missing_owned: &[u64],
32    block_size: u64,
33    max_coalesced_get_bytes: u64,
34) -> Vec<Range<u64>> {
35    let max_blocks_per_run = max_run_bytes(block_size, max_coalesced_get_bytes) / block_size;
36    let mut runs = Vec::new();
37    let mut i = 0;
38    while i < sorted_missing_owned.len() {
39        let mut j = i;
40        while j + 1 < sorted_missing_owned.len()
41            && sorted_missing_owned[j + 1] == sorted_missing_owned[j] + 1
42            && sorted_missing_owned[j + 1] - sorted_missing_owned[i] < max_blocks_per_run
43        {
44            j += 1;
45        }
46        runs.push(sorted_missing_owned[i]..sorted_missing_owned[j] + 1);
47        i = j + 1;
48    }
49    runs
50}
51
52pub fn assemble_range(
53    blocks: &[(u64, Bytes)],
54    block_size: u64,
55    req_start: u64,
56    req_end: u64,
57) -> Bytes {
58    if req_start >= req_end || blocks.is_empty() {
59        return Bytes::new();
60    }
61    let mut result = Vec::with_capacity((req_end - req_start) as usize);
62    for (block_idx, block_data) in blocks {
63        let blk_start = block_idx * block_size;
64        let blk_end = blk_start + block_data.len() as u64;
65        let clip_start = req_start.max(blk_start);
66        let clip_end = req_end.min(blk_end);
67        if clip_start < clip_end {
68            let local_start = (clip_start - blk_start) as usize;
69            let local_end = (clip_end - blk_start) as usize;
70            result.extend_from_slice(&block_data[local_start..local_end]);
71        }
72    }
73    Bytes::from(result)
74}
micromegas_object_cache/blocks.rs

micromegas_object_cache/
blocks.rs
