perf: enable BLAKE3 multithreaded hashing (update_rayon) for files >10MB
This commit is contained in:
Generated
+1
@@ -289,6 +289,7 @@ dependencies = [
|
|||||||
"cfg-if",
|
"cfg-if",
|
||||||
"constant_time_eq",
|
"constant_time_eq",
|
||||||
"cpufeatures",
|
"cpufeatures",
|
||||||
|
"rayon-core",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
|
|||||||
+1
-1
@@ -41,7 +41,7 @@ http-range-header = "0.4"
|
|||||||
image = { version = "0.25", default-features = false, features = ["jpeg", "png", "gif", "webp"] }
|
image = { version = "0.25", default-features = false, features = ["jpeg", "png", "gif", "webp"] }
|
||||||
md5 = "0.8.0"
|
md5 = "0.8.0"
|
||||||
sha2 = "0.10.9"
|
sha2 = "0.10.9"
|
||||||
blake3 = "1.8.3"
|
blake3 = { version = "1.8.3", features = ["rayon"] }
|
||||||
hex = "0.4.3"
|
hex = "0.4.3"
|
||||||
http-body-util = "0.1.3"
|
http-body-util = "0.1.3"
|
||||||
percent-encoding = "2.3"
|
percent-encoding = "2.3"
|
||||||
|
|||||||
@@ -661,6 +661,10 @@ impl ChunkedUploadService {
|
|||||||
let mut output = StdBufWriter::with_capacity(524_288, raw_output);
|
let mut output = StdBufWriter::with_capacity(524_288, raw_output);
|
||||||
let mut hasher = blake3::Hasher::new();
|
let mut hasher = blake3::Hasher::new();
|
||||||
|
|
||||||
|
// For files >10 MB, use multithreaded BLAKE3 hashing (all cores)
|
||||||
|
const RAYON_THRESHOLD: u64 = 10 * 1024 * 1024;
|
||||||
|
let use_rayon = total_size > RAYON_THRESHOLD;
|
||||||
|
|
||||||
// Single 512 KB read buffer reused across all chunks (avoids N allocations)
|
// Single 512 KB read buffer reused across all chunks (avoids N allocations)
|
||||||
let mut buf = vec![0u8; 524_288];
|
let mut buf = vec![0u8; 524_288];
|
||||||
for (index, chunk_path) in &chunks_meta {
|
for (index, chunk_path) in &chunks_meta {
|
||||||
@@ -673,7 +677,11 @@ impl ChunkedUploadService {
|
|||||||
if n == 0 {
|
if n == 0 {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
if use_rayon {
|
||||||
|
hasher.update_rayon(&buf[..n]);
|
||||||
|
} else {
|
||||||
hasher.update(&buf[..n]);
|
hasher.update(&buf[..n]);
|
||||||
|
}
|
||||||
output.write_all(&buf[..n]).map_err(|e| {
|
output.write_all(&buf[..n]).map_err(|e| {
|
||||||
format!("Failed to write chunk {index} to assembled file: {e}")
|
format!("Failed to write chunk {index} to assembled file: {e}")
|
||||||
})?;
|
})?;
|
||||||
|
|||||||
@@ -52,6 +52,10 @@ use crate::domain::errors::{DomainError, ErrorKind};
|
|||||||
/// Block size for BLAKE3 file hashing (1MB — optimal syscall/throughput ratio).
|
/// Block size for BLAKE3 file hashing (1MB — optimal syscall/throughput ratio).
|
||||||
const HASH_BLOCK_SIZE: usize = 1024 * 1024;
|
const HASH_BLOCK_SIZE: usize = 1024 * 1024;
|
||||||
|
|
||||||
|
/// Files larger than this threshold use multithreaded BLAKE3 hashing via
|
||||||
|
/// `update_rayon()`, which splits the work across all available cores.
|
||||||
|
const RAYON_HASH_THRESHOLD: u64 = 10 * 1024 * 1024; // 10 MB
|
||||||
|
|
||||||
/// Chunk size for streaming file reads (256 KB)
|
/// Chunk size for streaming file reads (256 KB)
|
||||||
const STREAM_CHUNK_SIZE: usize = 256 * 1024;
|
const STREAM_CHUNK_SIZE: usize = 256 * 1024;
|
||||||
|
|
||||||
@@ -156,24 +160,42 @@ impl DedupService {
|
|||||||
// ── Hash helpers ─────────────────────────────────────────────
|
// ── Hash helpers ─────────────────────────────────────────────
|
||||||
|
|
||||||
/// Calculate BLAKE3 hash of content (~5× faster than SHA-256).
|
/// Calculate BLAKE3 hash of content (~5× faster than SHA-256).
|
||||||
|
///
|
||||||
|
/// For buffers larger than 10 MB the computation is parallelised across
|
||||||
|
/// all available cores via `update_rayon()`.
|
||||||
pub fn hash_bytes(content: &[u8]) -> String {
|
pub fn hash_bytes(content: &[u8]) -> String {
|
||||||
|
if content.len() as u64 > RAYON_HASH_THRESHOLD {
|
||||||
|
let mut hasher = blake3::Hasher::new();
|
||||||
|
hasher.update_rayon(content);
|
||||||
|
hasher.finalize().to_hex().to_string()
|
||||||
|
} else {
|
||||||
blake3::hash(content).to_hex().to_string()
|
blake3::hash(content).to_hex().to_string()
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Calculate BLAKE3 hash of a file (~5× faster than SHA-256).
|
/// Calculate BLAKE3 hash of a file (~5× faster than SHA-256).
|
||||||
///
|
///
|
||||||
/// Runs entirely on `spawn_blocking` with synchronous I/O so the Tokio
|
/// Runs entirely on `spawn_blocking` with synchronous I/O so the Tokio
|
||||||
/// worker threads are never blocked by CPU-bound hashing. Uses 1 MB
|
/// worker threads are never blocked by CPU-bound hashing.
|
||||||
/// reads for optimal syscall-to-throughput ratio.
|
///
|
||||||
|
/// For files larger than 10 MB the hash is computed with `update_rayon()`,
|
||||||
|
/// which splits the work across all available cores. Smaller files use
|
||||||
|
/// sequential 1 MB reads for optimal syscall-to-throughput ratio.
|
||||||
pub async fn hash_file(path: &Path) -> std::io::Result<String> {
|
pub async fn hash_file(path: &Path) -> std::io::Result<String> {
|
||||||
let path = path.to_path_buf();
|
let path = path.to_path_buf();
|
||||||
tokio::task::spawn_blocking(move || {
|
tokio::task::spawn_blocking(move || {
|
||||||
use std::io::Read;
|
let file_size = std::fs::metadata(&path)?.len();
|
||||||
|
|
||||||
let mut file = std::fs::File::open(&path)?;
|
|
||||||
let mut hasher = blake3::Hasher::new();
|
let mut hasher = blake3::Hasher::new();
|
||||||
let mut buffer = vec![0u8; HASH_BLOCK_SIZE];
|
|
||||||
|
|
||||||
|
if file_size > RAYON_HASH_THRESHOLD {
|
||||||
|
// Large file: read into memory and hash with all cores
|
||||||
|
let content = std::fs::read(&path)?;
|
||||||
|
hasher.update_rayon(&content);
|
||||||
|
} else {
|
||||||
|
// Small file: sequential streaming with 1 MB reads
|
||||||
|
use std::io::Read;
|
||||||
|
let mut file = std::fs::File::open(&path)?;
|
||||||
|
let mut buffer = vec![0u8; HASH_BLOCK_SIZE];
|
||||||
loop {
|
loop {
|
||||||
let n = file.read(&mut buffer)?;
|
let n = file.read(&mut buffer)?;
|
||||||
if n == 0 {
|
if n == 0 {
|
||||||
@@ -181,6 +203,7 @@ impl DedupService {
|
|||||||
}
|
}
|
||||||
hasher.update(&buffer[..n]);
|
hasher.update(&buffer[..n]);
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
Ok(hasher.finalize().to_hex().to_string())
|
Ok(hasher.finalize().to_hex().to_string())
|
||||||
})
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user