From bb5547f1b368b5360912f49b09ab0a8f91b33556 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Thu, 20 Aug 2026 07:59:38 +0200 Subject: [PATCH] perf(catalog): parallelize production manifest hashing Production catalog generation previously prepared every game serially and computed transfer-chunk hashes for extracted .eti files even though those hashes were discarded. Enable Rayon-backed BLAKE3 hashing, process selected games through the shared available-CPU pool, and hash large read blocks with update_rayon. Extracted outputs now use a whole-file-only hash path. Manifest collection remains deterministic and the independent second verification pass is preserved. Test Plan: - `just clippy` -- passed - `just test` -- passed --- Cargo.lock | 41 +++++++ Cargo.toml | 3 +- crates/lanspread-compat/Cargo.toml | 1 + .../src/catalog_publisher/mod.rs | 6 +- .../src/catalog_publisher/package.rs | 110 +++++++++++++++--- .../src/catalog_publisher/unrar.rs | 7 +- 6 files changed, 143 insertions(+), 25 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1f5f29e..d37272b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -247,6 +247,7 @@ dependencies = [ "cfg-if", "constant_time_eq", "cpufeatures 0.3.0", + "rayon-core", ] [[package]] @@ -614,6 +615,25 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-deque" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-queue" version = "0.3.13" @@ -2056,6 +2076,7 @@ dependencies = [ "blake3", "eyre", "lanspread-db", + "rayon", "serde", "sqlx", "tokio", @@ -3067,6 +3088,26 @@ version = "0.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20675572f6f24e9e76ef639bc5552774ed45f1c30e2951e1e99c59888861c539" +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "rcgen" version = "0.14.8" diff --git a/Cargo.toml b/Cargo.toml index b4c88ec..0f6b5f3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,7 +13,7 @@ members = [ [workspace.dependencies] base64 = "0.23" -blake3 = "1" +blake3 = { version = "1", features = ["rayon"] } bytes = { version = "1", features = ["serde"] } cap-fs-ext = { version = "4", default-features = false } cap-primitives = "4" @@ -25,6 +25,7 @@ if-addrs = "0.15" log = "0.4" mdns-sd = "0.20" mimalloc = { version = "0.1", features = ["secure"] } +rayon = "1" rcgen = { version = "=0.14.8", default-features = false, diff --git a/crates/lanspread-compat/Cargo.toml b/crates/lanspread-compat/Cargo.toml index 5947621..a83fcd7 100644 --- a/crates/lanspread-compat/Cargo.toml +++ b/crates/lanspread-compat/Cargo.toml @@ -12,6 +12,7 @@ lanspread-db = { path = "../lanspread-db" } blake3 = { workspace = true } eyre = { workspace = true } +rayon = { workspace = true } serde = { workspace = true } sqlx = { workspace = true } tokio = { workspace = true } diff --git a/crates/lanspread-compat/src/catalog_publisher/mod.rs b/crates/lanspread-compat/src/catalog_publisher/mod.rs index 9592d45..6aa9749 100644 --- a/crates/lanspread-compat/src/catalog_publisher/mod.rs +++ b/crates/lanspread-compat/src/catalog_publisher/mod.rs @@ -34,6 +34,7 @@ use lanspread_db::content_manifest::{ write_canonical_manifest_atomic, }; pub use package::{build_manifest_from_package, verify_manifest_against_package}; +use rayon::prelude::*; /// The explicit set of catalog rows operated on by the publisher. #[derive(Clone, Debug, Eq, PartialEq)] @@ -127,8 +128,11 @@ pub async fn generate_catalog_manifests( ) .wrap_err_with(|| format!("failed to preflight package version for {}", game.game_id))?; } + // Rayon defaults its shared pool to the host's available parallelism. The + // same pool is used by BLAKE3's `update_rayon`, so game-level work and + // per-file hashing share one bounded CPU budget instead of nesting pools. let prepared = selected - .into_iter() + .into_par_iter() .map(|game| prepare_manifest(game, options)) .collect::>>()?; diff --git a/crates/lanspread-compat/src/catalog_publisher/package.rs b/crates/lanspread-compat/src/catalog_publisher/package.rs index 1ea4608..85a4211 100644 --- a/crates/lanspread-compat/src/catalog_publisher/package.rs +++ b/crates/lanspread-compat/src/catalog_publisher/package.rs @@ -19,7 +19,8 @@ use lanspread_db::content_manifest::{ use super::{CatalogGame, catalog::validate_regular_directory, unrar::scan_extracted_files}; -const HASH_BUFFER_SIZE: usize = 1024 * 1024; +const HASH_BUFFER_SIZE: usize = 4 * 1024 * 1024; +const RAYON_HASH_MIN_INPUT: usize = 128 * 1024; const MAX_VERSION_INI_BYTES: u64 = 64 * 1024; #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -265,9 +266,28 @@ pub(super) fn hash_exact( if chunk_size == 0 { eyre::bail!("hash chunk size cannot be zero"); } + let (file, chunks) = hash_exact_inner(reader, size, Some(chunk_size))?; + Ok(FileHashes { file, chunks }) +} + +/// Hashes exactly `size` bytes and returns only the whole-file digest. +/// +/// Extracted Stream Install files do not need transfer-chunk digests, so this +/// avoids doing a second BLAKE3 computation for each extracted byte. +pub(super) fn hash_exact_file(reader: &mut impl Read, size: u64) -> eyre::Result { + let (file, chunks) = hash_exact_inner(reader, size, None)?; + debug_assert!(chunks.is_empty()); + Ok(file) +} + +fn hash_exact_inner( + reader: &mut impl Read, + size: u64, + chunk_size: Option, +) -> eyre::Result<(Blake3Digest, Vec)> { let mut remaining = size; let mut whole = blake3::Hasher::new(); - let mut chunk = blake3::Hasher::new(); + let mut chunk = chunk_size.map(|_| blake3::Hasher::new()); let mut chunk_bytes = 0_u64; let mut chunks = Vec::new(); let mut buffer = vec![0_u8; HASH_BUFFER_SIZE]; @@ -279,31 +299,54 @@ pub(super) fn hash_exact( eyre::bail!("input ended with {remaining} expected byte(s) missing"); } let bytes = &buffer[..read]; - whole.update(bytes); + update_hasher(&mut whole, bytes); - let mut offset = 0; - while offset < bytes.len() { - let available = chunk_size - chunk_bytes; - let take = usize::try_from(available.min(u64::try_from(bytes.len() - offset)?))?; - chunk.update(&bytes[offset..offset + take]); - offset += take; - chunk_bytes += u64::try_from(take)?; - if chunk_bytes == chunk_size { - chunks.push(Blake3Digest::from_bytes(*chunk.finalize().as_bytes())); - chunk = blake3::Hasher::new(); - chunk_bytes = 0; + if let Some(chunk_size) = chunk_size { + let mut offset = 0; + while offset < bytes.len() { + let available = chunk_size - chunk_bytes; + let take = usize::try_from(available.min(u64::try_from(bytes.len() - offset)?))?; + let Some(chunk_hasher) = chunk.as_mut() else { + eyre::bail!("chunk hasher disappeared while hashing input"); + }; + update_hasher(chunk_hasher, &bytes[offset..offset + take]); + offset += take; + chunk_bytes += u64::try_from(take)?; + if chunk_bytes == chunk_size { + let Some(chunk_hasher) = chunk.take() else { + eyre::bail!("chunk hasher disappeared while finalizing input"); + }; + chunks.push(Blake3Digest::from_bytes( + *chunk_hasher.finalize().as_bytes(), + )); + chunk = Some(blake3::Hasher::new()); + chunk_bytes = 0; + } } } remaining -= u64::try_from(read)?; } - if chunk_bytes != 0 { - chunks.push(Blake3Digest::from_bytes(*chunk.finalize().as_bytes())); + if chunk_size.is_some() && chunk_bytes != 0 { + let Some(chunk_hasher) = chunk else { + eyre::bail!("chunk hasher disappeared while finalizing input"); + }; + chunks.push(Blake3Digest::from_bytes( + *chunk_hasher.finalize().as_bytes(), + )); } - Ok(FileHashes { - file: Blake3Digest::from_bytes(*whole.finalize().as_bytes()), + Ok(( + Blake3Digest::from_bytes(*whole.finalize().as_bytes()), chunks, - }) + )) +} + +fn update_hasher(hasher: &mut blake3::Hasher, bytes: &[u8]) { + if bytes.len() >= RAYON_HASH_MIN_INPUT { + hasher.update_rayon(bytes); + } else { + hasher.update(bytes); + } } fn read_bounded_regular_file(path: &Path, limit: u64) -> eyre::Result> { @@ -403,6 +446,28 @@ mod tests { ); } + #[test] + fn large_inputs_use_the_parallel_hash_path_without_changing_digests() { + let bytes = (0..(4 * 1024 * 1024 + 17)) + .map(|index| u8::try_from(index % 251).expect("pattern fits in byte")) + .collect::>(); + let chunk_size = 256 * 1024; + let hashes = hash_exact( + &mut Cursor::new(&bytes), + u64::try_from(bytes.len()).expect("test input length fits in u64"), + chunk_size, + ) + .expect("hashing should succeed"); + assert_eq!(hashes.file, Blake3Digest::hash(&bytes)); + assert_eq!( + hashes.chunks, + bytes + .chunks(usize::try_from(chunk_size).expect("chunk size fits in usize")) + .map(Blake3Digest::hash) + .collect::>() + ); + } + #[test] fn empty_input_has_a_whole_hash_and_no_chunks() { let hashes = hash_exact(&mut Cursor::new([]), 0, 4).expect("hashing should succeed"); @@ -410,6 +475,13 @@ mod tests { assert!(hashes.chunks.is_empty()); } + #[test] + fn whole_file_hash_omits_transfer_chunks() { + let bytes = b"abcdefghij"; + let hash = hash_exact_file(&mut Cursor::new(bytes), 10).expect("hashing should succeed"); + assert_eq!(hash, Blake3Digest::hash(bytes)); + } + #[test] fn truncated_input_is_rejected() { let error = diff --git a/crates/lanspread-compat/src/catalog_publisher/unrar.rs b/crates/lanspread-compat/src/catalog_publisher/unrar.rs index f87a0c6..055162e 100644 --- a/crates/lanspread-compat/src/catalog_publisher/unrar.rs +++ b/crates/lanspread-compat/src/catalog_publisher/unrar.rs @@ -10,7 +10,6 @@ use std::{ use eyre::WrapErr; use lanspread_db::content_manifest::{ - CATALOG_CHUNK_SIZE, CanonicalCatalogPath, CatalogExtractedEntry, MAX_CATALOG_ENTRIES, @@ -18,7 +17,7 @@ use lanspread_db::content_manifest::{ MAX_CATALOG_TOTAL_BYTES, }; -use super::package::hash_exact; +use super::package::hash_exact_file; // Retain enough technical listing data for the maximum catalog shape without // permitting a subprocess to grow publisher memory without bound. @@ -203,14 +202,14 @@ fn hash_archive_outputs( insert_output(outputs, &entry.path, ExtractedValue::Directory)?; } RarEntryKind::File => { - let hashes = hash_exact(&mut stdout, entry.size, CATALOG_CHUNK_SIZE) + let hash = hash_exact_file(&mut stdout, entry.size) .wrap_err_with(|| format!("failed to hash extracted file {}", entry.path))?; insert_output( outputs, &entry.path, ExtractedValue::File { size: entry.size, - hash: hashes.file, + hash, }, )?; }