cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Revert "perf(blob): one syscall per new chunk in write_blob_bytes"

Browse Source 
This reverts ee51b32. The create_new change showed no measurable throughput
benefit — three 9-rep interleaved runs on the same ext4 device swung −12%..+21%
at the 256 KiB CDC size (a negative stat on a warm dentry cache is ~µs, below
the shared-disk noise floor). Applying the same "no change without a measured
win" bar used for the pool-sizing revert: the idiom/TOCTOU angle is real but the
race is already prevented upstream by the PG pin-or-classify serialisation, so
it's defence-in-depth only — not enough to keep an unmeasured change. Reverts
the production change, the bench, and its doc together.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JG5yYZ9s868mJwqT2Qz7ez
This commit is contained in:
Claude
2026-06-22 09:42:37 +00:00
parent ee51b32ba9
commit 601fbdedf9
4 changed files with 6 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
-8
View File
@@ -199,14 +199,6 @@ name = "bench_pool_concurrency"
path = "examples/bench_pool_concurrency.rs"
required-features = ["bench"]
# Blob write syscall benchmark — per-chunk stat removal in write_blob_bytes
# (try_exists+create → create_new). Measures new-chunk write throughput, old
# vs new strategy, swept over chunk size. No Postgres needed.
[[example]]
name = "bench_blob_write"
path = "examples/bench_blob_write.rs"
required-features = ["bench"]
# ACL owner-cache benchmark — owner query vs moka hit (needs the dev Postgres up).
[[example]]
name = "bench_owner_cache"
benches/BLOB-WRITE.md
-66
View File
@@ -1,66 +0,0 @@
# Blob write syscall benchmark — per-chunk `stat` removal
Measures the change to `write_blob_bytes` (`local_blob_backend.rs`): the new-chunk
create path went from `try_exists` (stat) + `File::create` (open `O_CREAT|O_TRUNC`)
— two metadata syscalls, each a `spawn_blocking` round-trip — to a single
`OpenOptions::create_new` (open `O_CREAT|O_EXCL`), treating `AlreadyExists` as the
idempotent skip. The bench writes N distinct content-addressed chunk files
scattered across the 256 hash-prefix dirs at the production fan-out
(`CHUNK_UPLOAD_CONCURRENCY = 8`), once per strategy, reporting write throughput.
## Reproduce
```bash
cargo run --release --features bench --example bench_blob_write
# tunables: BENCH_CHUNKS (4000) BENCH_CHUNK_KB (8,64,256) BENCH_CONCURRENCY (8) BENCH_REPS (3)
```
## Results — INCONCLUSIVE (effect below the noise floor)
Three 9-rep interleaved runs on the **same** ext4 device (`/dev/vda`, shared
container disk), Δ = new vs old chunks/s:
| run | 8 KiB | 64 KiB | 256 KiB |
|-----|------:|-------:|--------:|
| A (3 reps, sequential — order-biased, discard) | +28.7% | +9.3% | −12.2% |
| B (9 reps, interleaved, `/tmp`) | −5.4% | −1.0% | +0.5% |
| C (9 reps, interleaved, repo dir) | +8.1% | +7.2% | +21.2% |
The 256 KiB delta swings **−12% → +0.5% → +21%** across runs of identical code on
identical storage. The run-to-run variance (each rep writes ~1 GiB to a shared
container disk) is larger than the effect, so **the throughput impact is not
measurable in this environment**. Run A is additionally invalid (all-old-then-
all-new lets system drift penalise whichever strategy runs second); B and C are
methodologically sound (interleaved, alternating order) but disagree — that
disagreement *is* the finding.
## What IS deterministic (from the code, not the timer)
- **One metadata syscall instead of two** per new chunk: `openat(O_CREAT|O_EXCL)`
vs `newfstatat` + `openat(O_CREAT|O_TRUNC)`. On a large upload of unique data
that is thousands of `stat`s — and `spawn_blocking` dispatches — removed. The
wall-clock value of that is below the noise floor here because a negative
`stat` on a warm ext4 dentry cache is ~µs, dwarfed by the chunk's
create+write+flush (and, downstream, the fsync sweep).
- **Closes a TOCTOU race.** The old `try_exists`==false → `File::create`
(`O_TRUNC`) pair could truncate a file a racing writer created in between;
`O_EXCL` makes the check-and-create atomic and skips instead. (In practice the
PG pin-or-classify serialises writes per content hash, so this race is already
unreachable on the dedup path — the change is defence-in-depth.)
## Conclusion
This is a **code-quality / correctness micro-change**, not a measured throughput
win: `create_new` is the canonical Rust idiom replacing a check-then-create
anti-pattern, it is strictly fewer syscalls, and it has zero downside — but its
throughput effect is below what this shared-disk environment can resolve, and at
the realistic 256 KiB CDC chunk size the one saved `stat` is a tiny fraction of
the per-chunk cost regardless. Kept on those grounds, not on a benchmark number.
(Companion change *not* made: reusing the written `File` handle for the fsync
sweep instead of re-opening by path. The sweep is a single end-of-stream
`sync_blobs` over **all** the upload's new hashes (`dedup_service.rs:929`), so
retaining handles would hold thousands of FDs open for the whole upload — a 1 GiB
upload is ~4000 chunks > the default `ulimit -n` of 1024 → `EMFILE`. The
re-open-with-16-way-concurrency sweep is a deliberate FD-frugal design; the saved
`open` is negligible before the `fsync` it precedes anyway.)
examples/bench_blob_write.rs
-161
View File
@@ -1,161 +0,0 @@
//! Blob write syscall benchmark — per-chunk `stat` removal in `write_blob_bytes`.
//!
//! Isolates the one change: the new-chunk create path went from
//! `try_exists` (stat) + `File::create` (open O_CREAT|O_TRUNC) — 2 metadata syscalls
//! to
//! `OpenOptions::create_new` (open O_CREAT|O_EXCL) — 1 metadata syscall
//! each one a `spawn_blocking` round-trip on Tokio's blocking pool. The bench
//! writes N distinct content-addressed chunk files (scattered across the 256
//! hash-prefix dirs, like production) at the production fan-out
//! (`CHUNK_UPLOAD_CONCURRENCY = 8`), once per strategy, and reports write
//! throughput. "old" is the previous behaviour, "new" is the change.
//!
//! The gain is per-chunk and metadata-bound, so it scales **inversely with chunk
//! size** — largest for tiny chunks, smaller at the 256 KiB CDC average where the
//! data write dominates. The sweep shows that range.
//!
//! Run (no Postgres needed):
//! cargo run --release --features bench --example bench_blob_write
//! Tunables (env): BENCH_CHUNKS (4000) BENCH_CHUNK_KB (8,64,256)
//! BENCH_CONCURRENCY (8) BENCH_REPS (3)
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant};
use futures::{StreamExt, stream};
use oxicloud::application::ports::blob_storage_ports::BlobStorageBackend;
use oxicloud::infrastructure::services::local_blob_backend::LocalBlobBackend;
use tokio::fs::OpenOptions;
use tokio::io::AsyncWriteExt;
fn env_or<T: std::str::FromStr>(key: &str, default: T) -> T {
std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
}
#[derive(Clone, Copy, PartialEq)]
enum Strategy {
/// Old: try_exists (stat) + File::create (open O_CREAT|O_TRUNC).
Old,
/// New: OpenOptions::create_new (open O_CREAT|O_EXCL).
New,
}
async fn write_one(strategy: Strategy, path: PathBuf, data: Arc<Vec<u8>>) {
match strategy {
Strategy::Old => {
if tokio::fs::try_exists(&path).await.unwrap_or(false) {
return;
}
let mut f = tokio::fs::File::create(&path).await.expect("create");
f.write_all(&data).await.expect("write");
f.flush().await.expect("flush");
}
Strategy::New => match OpenOptions::new()
.write(true)
.create_new(true)
.open(&path)
.await
{
Ok(mut f) => {
f.write_all(&data).await.expect("write");
f.flush().await.expect("flush");
}
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {}
Err(e) => panic!("open: {e}"),
},
}
}
async fn run_strategy(
strategy: Strategy,
paths: &[PathBuf],
data: Arc<Vec<u8>>,
concurrency: usize,
) -> Duration {
let t = Instant::now();
let mut s = stream::iter(paths.iter().cloned())
.map(|p| write_one(strategy, p, data.clone()))
.buffer_unordered(concurrency);
while s.next().await.is_some() {}
t.elapsed()
}
#[tokio::main(flavor = "multi_thread")]
async fn main() {
let n: usize = env_or("BENCH_CHUNKS", 4000);
let chunk_kbs: Vec<usize> = std::env::var("BENCH_CHUNK_KB")
.ok()
.map(|s| s.split(',').filter_map(|x| x.trim().parse().ok()).collect())
.filter(|v: &Vec<usize>| !v.is_empty())
.unwrap_or_else(|| vec![8, 64, 256]);
let concurrency: usize = env_or("BENCH_CONCURRENCY", 8);
let reps: usize = env_or("BENCH_REPS", 3);
// Distinct, well-distributed hashes → scattered across the 256 prefix dirs.
let hashes: Vec<String> = (0..n)
.map(|i| blake3::hash(&(i as u64).to_le_bytes()).to_hex().to_string())
.collect();
println!("\n############################################################");
println!("# Blob write syscall benchmark — per-chunk stat removal");
println!(
"# {n} chunks/run, {concurrency}-way (CHUNK_UPLOAD_CONCURRENCY), median of {reps}"
);
println!("# old = try_exists+create (2 metadata syscalls); new = create_new (1)");
println!("############################################################\n");
println!(
"| {:>8} | {:>12} | {:>12} | {:>12} | {:>10} |",
"chunk", "old chunks/s", "new chunks/s", "new MB/s", "Δ chunks/s"
);
println!(
"|{:-<10}|{:-<14}|{:-<14}|{:-<14}|{:-<12}|",
"", "", "", "", ""
);
for &kb in &chunk_kbs {
let data = Arc::new(vec![0xABu8; kb * 1024]);
let mut old_rates: Vec<f64> = Vec::with_capacity(reps);
let mut new_rates: Vec<f64> = Vec::with_capacity(reps);
for rep in 0..reps {
// Interleave the two strategies within each rep, alternating which
// runs first, so any system drift (cache/tmpfs fill, thermal) hits
// both equally instead of penalising whichever runs second.
let order = if rep % 2 == 0 {
[Strategy::Old, Strategy::New]
} else {
[Strategy::New, Strategy::Old]
};
for strategy in order {
// Fresh dir per run so every chunk is genuinely new (no skips).
let tmp = tempfile::tempdir().expect("tempdir");
let backend = LocalBlobBackend::new(tmp.path());
backend.initialize().await.expect("init");
let paths: Vec<PathBuf> = hashes.iter().map(|h| backend.blob_path(h)).collect();
let dur = run_strategy(strategy, &paths, data.clone(), concurrency).await;
let rate = n as f64 / dur.as_secs_f64();
match strategy {
Strategy::Old => old_rates.push(rate),
Strategy::New => new_rates.push(rate),
}
}
}
old_rates.sort_by(|a, b| a.partial_cmp(b).unwrap());
new_rates.sort_by(|a, b| a.partial_cmp(b).unwrap());
let old = old_rates[old_rates.len() / 2];
let new = new_rates[new_rates.len() / 2];
let mb_s = new * (kb as f64) / 1024.0;
let delta = (new / old - 1.0) * 100.0;
println!(
"| {:>6}K | {:>12.0} | {:>12.0} | {:>12.1} | {:>9.1}% |",
kb, old, new, mb_s, delta
);
}
println!(
"\nΔ is the new (create_new) write throughput vs old (stat+create). The gain\n\
is metadata-bound, so it shrinks as the chunk size (data-write cost) grows;\n\
CDC chunks average 256 KiB.\n"
);
}
src/infrastructure/services/local_blob_backend.rs
+6 -23
View File
@@ -128,29 +128,12 @@ async fn fsync_paths_parallel(paths: Vec<PathBuf>, strict: bool) -> Result<(), D
/// (fsync now vs. deferred batch sync), or `None` when the blob already
/// existed (idempotent skip — content-addressed, so identical by definition).
async fn write_blob_bytes(blob_path: &Path, data: &Bytes) -> Result<Option<File>, DomainError> {
// `create_new` = `open(O_CREAT | O_EXCL)`: ONE syscall that both tests for
// existence and creates, replacing the old `try_exists` (stat) + `File::create`
// (two metadata syscalls, each its own `spawn_blocking` round-trip on Tokio's
// blocking pool). On a large upload of unique data that's thousands of stats
// saved. The blob is content-addressed, so an already-present file is
// byte-identical by definition → `AlreadyExists` is exactly the idempotent
// skip the old `try_exists` branch performed — and O_EXCL closes the TOCTOU
// window the check-then-create pair left open (no truncate-over-a-racing-writer).
let mut file = match fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(blob_path)
.await
{
Ok(file) => file,
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => return Ok(None),
Err(e) => {
return Err(DomainError::internal_error(
"Blob",
format!("Failed to create blob file: {}", e),
));
}
};
if fs::try_exists(blob_path).await.unwrap_or(false) {
return Ok(None);
}
let mut file = fs::File::create(blob_path).await.map_err(|e| {
DomainError::internal_error("Blob", format!("Failed to create blob file: {}", e))
})?;
file.write_all(data).await.map_err(|e| {
DomainError::internal_error("Blob", format!("Failed to write blob from bytes: {}", e))
})?;