mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 08:18:18 +00:00
test(object-data-cache): add concurrency stress tests and GET benchmarks (#4711)
* test(object-data-cache): add concurrency stress tests for the fill machinery The race coverage so far pins specific interleavings with an injected barrier. These add sustained, real-parallelism contention to catch what pinned tests cannot — a leaked singleflight leader, a stranded semaphore permit, an index/cache divergence that only surfaces under volume: - moka_backend_concurrency_storm_leaves_no_leaked_state: 48 tasks x 400 mixed fill/lookup/invalidate ops on a 6-key pool over 4 worker threads, then asserts inflight_fills == 0 (no leaked in-flight fill), a clean post-storm fill/hit/invalidate sequence still works (state not corrupted), and clear() drains the cache. - moka_backend_gate_bounds_admission_under_concurrent_burst: a low-memory snapshot must reject an entire 32-fill burst — admission is bounded, not a check-then-act race that lets the burst through. (This does not cover the separate 5s-stale-snapshot overshoot, which needs byte-based reservation.) Both are mutation-verified: neutering invalidate_object fails the storm's "invalidation must win" assertion, and making allows_fill always-true fails the burst's full-rejection assertion. The storm passed 20/20 runs. `rt-multi-thread` is added to dev-deps so the tasks run on real worker threads. Refs: backlog#1107 Co-Authored-By: heihutu <heihutu@gmail.com> * bench(object-data-cache): measure the GET-path cost the audit argued statically Adds a criterion benchmark driving the public ObjectDataCache facade, so the "hot path is clean / high throughput" claim rests on numbers, not analysis: - plan_get ~104-112 ns (per-GET planning + metric label-set hash) - lookup_hit ~143-147 ns (a resident-body hit: moka get + Bytes clone + metric) - fill_distinct ~2.2-2.4 us (cold fill: plan + singleflight + index + moka + inner spawn) A hit at ~143 ns against a multi-millisecond erasure read is the ~1000x saving the cache exists for; the per-GET metric cost is real but ~100 ns, not a bottleneck; the fill cost (incl. the cancellation-safety spawn kept on purpose) is negligible on a background task. Run with `cargo bench -p rustfs-object-data-cache`. Refs: backlog#1107 Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
@@ -38,8 +38,16 @@ tokio = { workspace = true, features = ["sync", "rt", "time"] }
|
||||
tracing.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
criterion = { workspace = true }
|
||||
metrics-util = { version = "0.20", features = ["debugging"] }
|
||||
tokio = { workspace = true, features = ["macros", "rt", "time"] }
|
||||
# `rt-multi-thread` lets the concurrency stress tests run tasks on real worker
|
||||
# threads, so they exercise true parallelism on the shared singleflight/index
|
||||
# state rather than only cooperative interleaving.
|
||||
tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread", "time"] }
|
||||
|
||||
[[bench]]
|
||||
name = "cache_bench"
|
||||
harness = false
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//! Throughput / latency benchmarks for the object data cache
|
||||
//! (backlog#1107 production-readiness follow-up).
|
||||
//!
|
||||
//! The concurrency correctness was covered by deterministic and stress tests;
|
||||
//! these measure the GET-path cost the audit only argued about statically. Run
|
||||
//! with `cargo bench -p rustfs-object-data-cache`.
|
||||
|
||||
use bytes::Bytes;
|
||||
use criterion::{Criterion, criterion_group, criterion_main};
|
||||
use rustfs_object_data_cache::{ObjectDataCache, ObjectDataCacheConfig, ObjectDataCacheGetRequest, ObjectDataCacheMode};
|
||||
use std::hint::black_box;
|
||||
|
||||
fn fill_cache() -> ObjectDataCache {
|
||||
let config = ObjectDataCacheConfig {
|
||||
mode: ObjectDataCacheMode::FillBufferedOnly,
|
||||
max_bytes: 256 * 1024 * 1024,
|
||||
max_entry_bytes: 1024 * 1024,
|
||||
// Opt out of the memory gate so fills are deterministic regardless of
|
||||
// the host's live memory (matches the test suite's convention).
|
||||
min_free_memory_percent: 0,
|
||||
..ObjectDataCacheConfig::default()
|
||||
};
|
||||
ObjectDataCache::new(config).expect("cache should build")
|
||||
}
|
||||
|
||||
fn request(object: &str) -> ObjectDataCacheGetRequest<'_> {
|
||||
ObjectDataCacheGetRequest {
|
||||
bucket: "bucket",
|
||||
object,
|
||||
etag: "etag",
|
||||
size: 4096,
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
/// The per-GET planning cost on the hot path: key construction plus the metric
|
||||
/// label-set hash and recorder dispatch. `plan_get` is synchronous, so this is
|
||||
/// measured directly with no runtime overhead.
|
||||
fn bench_plan_get(c: &mut Criterion) {
|
||||
let cache = fill_cache();
|
||||
c.bench_function("plan_get", |b| {
|
||||
b.iter(|| {
|
||||
black_box(cache.plan_get(black_box(request("object"))));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/// Hit-path latency: lookup of a resident body. Batches 100 lookups per sample
|
||||
/// so the single `block_on` is amortised out of the measurement.
|
||||
fn bench_lookup_hit(c: &mut Criterion) {
|
||||
let runtime = tokio::runtime::Builder::new_current_thread().build().expect("runtime");
|
||||
let cache = fill_cache();
|
||||
let plan = cache.plan_get(request("hot-object"));
|
||||
runtime.block_on(async {
|
||||
let _ = cache.fill_body(&plan, Bytes::from(vec![0u8; 4096])).await;
|
||||
});
|
||||
|
||||
c.bench_function("lookup_hit_x100", |b| {
|
||||
b.iter(|| {
|
||||
runtime.block_on(async {
|
||||
for _ in 0..100 {
|
||||
black_box(cache.lookup_body(black_box(&plan)).await);
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/// Fill throughput for distinct cold keys. Batches 100 fills per sample.
|
||||
fn bench_fill(c: &mut Criterion) {
|
||||
let runtime = tokio::runtime::Builder::new_current_thread().build().expect("runtime");
|
||||
let cache = fill_cache();
|
||||
let body = Bytes::from(vec![0u8; 4096]);
|
||||
|
||||
c.bench_function("fill_distinct_x100", |b| {
|
||||
let mut n: u64 = 0;
|
||||
b.iter(|| {
|
||||
runtime.block_on(async {
|
||||
for _ in 0..100 {
|
||||
n += 1;
|
||||
let object = format!("fill-{n}");
|
||||
let plan = cache.plan_get(request(&object));
|
||||
black_box(cache.fill_body(black_box(&plan), body.clone()).await);
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
criterion_group!(benches, bench_plan_get, bench_lookup_hit, bench_fill);
|
||||
criterion_main!(benches);
|
||||
@@ -875,4 +875,125 @@ mod tests {
|
||||
}
|
||||
assert!(!cached, "a cancelled fill must not leave an orphaned body after a racing invalidation");
|
||||
}
|
||||
|
||||
// ---- Concurrency stress (backlog#1107 production-readiness follow-up) ----
|
||||
//
|
||||
// The deterministic race tests above pin specific interleavings with an
|
||||
// injected barrier. These drive sustained, real-parallelism contention on a
|
||||
// small key space to catch leaks and state corruption the pinned tests
|
||||
// cannot: a leaked singleflight leader, a stranded semaphore permit, an
|
||||
// index/cache divergence that only shows up under volume.
|
||||
|
||||
/// Hammer fill / lookup / invalidate concurrently on a few hot identities,
|
||||
/// then assert the shared machinery is neither leaked nor corrupted.
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
||||
async fn moka_backend_concurrency_storm_leaves_no_leaked_state() {
|
||||
let stats = Arc::new(ObjectDataCacheStats::default());
|
||||
// Long TTL so entries survive the storm — we want fill-vs-invalidate
|
||||
// races, not just expiry churn.
|
||||
let mut config = enabled_config();
|
||||
config.ttl = Duration::from_secs(30);
|
||||
config.time_to_idle = Duration::from_secs(30);
|
||||
let backend = Arc::new(MokaBackend::new(&config, Arc::clone(&stats)).expect("moka backend should build"));
|
||||
|
||||
const TASKS: usize = 48;
|
||||
const ITERS: usize = 400;
|
||||
const KEYS: usize = 6; // small pool → heavy contention on shared identities
|
||||
|
||||
let mut handles = Vec::new();
|
||||
for t in 0..TASKS {
|
||||
let backend = Arc::clone(&backend);
|
||||
handles.push(tokio::spawn(async move {
|
||||
for i in 0..ITERS {
|
||||
let k = (t + i) % KEYS;
|
||||
let object = format!("object-{}", k % 3);
|
||||
let plan = versioned_plan(&object, &format!("v{k}"), &format!("etag-{k}"));
|
||||
// Deterministic per-(task,iter) mix: mostly fills, plus
|
||||
// lookups and invalidations racing against them.
|
||||
match (t + i) % 4 {
|
||||
0 | 1 => {
|
||||
let _ = backend.fill_body(&plan, Bytes::from_static(b"hello")).await;
|
||||
}
|
||||
2 => {
|
||||
let _ = backend.lookup_body(&plan).await;
|
||||
}
|
||||
_ => {
|
||||
let identity = ObjectDataCacheIdentity::new("bucket", object.as_str());
|
||||
let _ = backend.invalidate_object(&identity).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}));
|
||||
}
|
||||
for handle in handles {
|
||||
handle.await.expect("stress task must not panic or deadlock");
|
||||
}
|
||||
|
||||
// Every fill_body call returned, so no singleflight leader may still be
|
||||
// registered: a non-zero count means a fill leaked its in-flight slot.
|
||||
assert_eq!(
|
||||
stats.snapshot().inflight_fills,
|
||||
0,
|
||||
"the concurrency storm leaked an in-flight fill (singleflight/semaphore not released)"
|
||||
);
|
||||
|
||||
// The shared state is not corrupted: a fresh, uncontended sequence must
|
||||
// still fill, hit, and invalidate correctly.
|
||||
let clean = versioned_plan("fresh-object", "v1", "fresh-etag");
|
||||
assert_eq!(
|
||||
backend.fill_body(&clean, Bytes::from_static(b"hello")).await,
|
||||
ObjectDataCacheFillResult::Inserted,
|
||||
"a clean fill after the storm must still insert"
|
||||
);
|
||||
assert!(matches!(backend.lookup_body(&clean).await, ObjectDataCacheLookup::Hit(_)));
|
||||
let _ = backend
|
||||
.invalidate_object(&ObjectDataCacheIdentity::new("bucket", "fresh-object"))
|
||||
.await;
|
||||
assert!(
|
||||
matches!(backend.lookup_body(&clean).await, ObjectDataCacheLookup::Miss),
|
||||
"invalidation must still win after a concurrency storm"
|
||||
);
|
||||
|
||||
// clear() plus a moka drain empties both the cache and the identity index.
|
||||
let _ = backend.clear().await;
|
||||
backend.cache.run_pending_tasks().await;
|
||||
assert_eq!(backend.entry_count(), 0, "clear() must drop every entry");
|
||||
}
|
||||
|
||||
/// A memory-constrained container (low snapshot) must reject an entire
|
||||
/// concurrent fill burst — admission is bounded, not a check-then-act race
|
||||
/// that lets the burst through. This is the CI-runnable proxy for the
|
||||
/// OOM-under-burst concern; it does not cover the separate 5s-stale-snapshot
|
||||
/// overshoot, which needs byte-based reservation in the gate.
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
||||
async fn moka_backend_gate_bounds_admission_under_concurrent_burst() {
|
||||
let stats = Arc::new(ObjectDataCacheStats::default());
|
||||
let backend = Arc::new(MokaBackend::new(&memory_gated_config(), Arc::clone(&stats)).expect("moka backend should build"));
|
||||
// Simulate a pod whose available memory is far below the free floor.
|
||||
backend
|
||||
.memory_gate
|
||||
.set_test_snapshot(Some(crate::memory::ObjectDataCacheMemorySnapshot {
|
||||
total_bytes: 1_000,
|
||||
available_bytes: 50,
|
||||
}));
|
||||
|
||||
const TASKS: usize = 32;
|
||||
let mut handles = Vec::new();
|
||||
for t in 0..TASKS {
|
||||
let backend = Arc::clone(&backend);
|
||||
handles.push(tokio::spawn(async move {
|
||||
let plan = versioned_plan(&format!("object-{t}"), "v1", &format!("etag-{t}"));
|
||||
backend.fill_body(&plan, Bytes::from_static(b"hello")).await
|
||||
}));
|
||||
}
|
||||
let mut rejected = 0;
|
||||
for handle in handles {
|
||||
if handle.await.expect("burst task should complete") == ObjectDataCacheFillResult::SkippedMemoryPressure {
|
||||
rejected += 1;
|
||||
}
|
||||
}
|
||||
|
||||
assert_eq!(rejected, TASKS, "the gate must reject every fill in the burst when memory is constrained");
|
||||
assert_eq!(backend.entry_count(), 0, "no body may be admitted under memory pressure");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user