1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
//! Parallel chunk decompression using rayon with lane partitioning.
//!
//! When reading a chunked+compressed dataset with many chunks, this module
//! uses lane-partitioned parallel decompression: each thread receives a
//! deterministic, disjoint subset of chunks — no overlap, no coordination.
//!
//! The lane assignment is seeded by dataset metadata so repeated reads of
//! the same region produce identical partitions (cache-friendly, reproducible).
use crate::chunked_read::ChunkInfo;
use crate::convert::slice_range;
use crate::error::FormatError;
use crate::filter_pipeline::FilterPipeline;
use crate::filters::{ChunkContext, decompress_chunk};
use crate::lane_partition::{self, LaneStats, PartitionStats};
/// Threshold: only use parallel decompression when chunk count exceeds this.
const PARALLEL_THRESHOLD: usize = 4;
/// Result of decompressing a single chunk, tagged with its index for ordering.
struct DecompressedChunk {
index: usize,
data: Vec<u8>,
}
/// Returns `true` if the parallel path should be used for the given chunk count.
pub fn should_use_parallel(chunk_count: usize) -> bool {
chunk_count > PARALLEL_THRESHOLD
}
/// Decompress chunks in parallel using lane-partitioned assignment.
///
/// Instead of naive `par_iter`, chunks are deterministically assigned to lanes
/// (threads) using a seeded pseudorandom permutation. Each lane processes
/// only its assigned chunks — no redundant work, no coordination.
///
/// # Arguments
///
/// * `seed` - Seed for the partition permutation (e.g. dataset address + chunk range hash).
/// * `num_lanes` - Number of parallel lanes. Pass `None` to auto-detect from available cores.
///
/// # Errors
///
/// Returns the first error encountered by any worker thread.
pub fn decompress_chunks_lane_partitioned(
file_data: &[u8],
chunks: &[ChunkInfo],
pipeline: &FilterPipeline,
ctx: ChunkContext<'_>,
seed: u64,
num_lanes: Option<usize>,
) -> Result<(Vec<Vec<u8>>, PartitionStats), FormatError> {
use rayon::prelude::*;
let lanes = num_lanes.unwrap_or_else(|| {
std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1)
});
let assignments = lane_partition::partition_chunks(chunks.len(), lanes, seed);
let num_lanes = assignments.len();
// Each lane processes its assigned chunks and returns results + stats.
let lane_results: Result<Vec<(Vec<DecompressedChunk>, LaneStats)>, FormatError> = assignments
.into_par_iter()
.map(|indices| {
let mut results = Vec::with_capacity(indices.len());
let mut stats = LaneStats::default();
for &index in &indices {
let chunk_info = &chunks[index];
let r = slice_range(chunk_info.address, u64::from(chunk_info.chunk_size))?;
if r.end > file_data.len() {
return Err(FormatError::UnexpectedEof {
expected: r.end,
available: file_data.len(),
});
}
let raw_chunk = &file_data[r];
let decompressed = if chunk_info.filter_mask == 0 {
decompress_chunk(raw_chunk, pipeline, ctx)?
} else {
raw_chunk.to_vec()
};
stats.chunks_processed += 1;
stats.compressed_bytes += u64::from(chunk_info.chunk_size);
stats.decompressed_bytes += decompressed.len() as u64;
results.push(DecompressedChunk {
index,
data: decompressed,
});
}
Ok((results, stats))
})
.collect();
let lane_results = lane_results?;
// Aggregate stats
let mut partition_stats = PartitionStats::new(num_lanes);
partition_stats.total_chunks = chunks.len();
for (lane_idx, (_, stats)) in lane_results.iter().enumerate() {
partition_stats.per_lane[lane_idx] = stats.clone();
}
// Flatten and sort by original index to restore order
let mut all_chunks: Vec<DecompressedChunk> = lane_results
.into_iter()
.flat_map(|(chunks, _)| chunks)
.collect();
all_chunks.sort_by_key(|dc| dc.index);
let ordered = all_chunks.into_iter().map(|dc| dc.data).collect();
Ok((ordered, partition_stats))
}