tests/chunk_revolver_coalescing.rs
//! Phase 1 of the "Magazine → chunk_revolver consolidation" plan
//! (`.nornir/chunk_revolver-unsorted.md`): the coalescing / ref-counted mode.
//!
//! The fixed-slot mode (`try_get_chunk` / `return_chunk`, pinned by
//! `tests/chunk_revolver.rs`) is 1 write = 1 whole slot. znippy's Magazine fork
//! added the one capability it lacked: **many small payloads packed into ONE
//! slot** as variable-width slices, with a per-slot reference count so the slot
//! only returns to the free list after the LAST worker is done with it.
//!
//! These tests pin that capability in the canonical pool so the fork can die.
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use znippy_zoomies::chunk_revolver::ChunkRevolver;
/// N payloads → 1 slot, and the slot is freed only after the Nth release.
#[test]
fn coalesces_many_payloads_into_one_slot_and_frees_on_last_release() {
// One slot, 1 KiB, cut granularity 256 B.
let rev = ChunkRevolver::with_coalescing(1, 1024, 256);
assert_eq!(rev.num_slots(), 1);
assert_eq!(rev.slice_size(), 256);
let returner = rev.returner();
let mut filler = rev.claim_slot().expect("pool has a free slot");
assert_eq!(filler.remaining(), 1024);
// Pack four 100-byte payloads into the single slot.
for f in 0..4u64 {
let w = filler.writable(100);
assert_eq!(w.len(), 100);
w.fill(b'a' + f as u8);
filler.commit_slice(100, false, f, 0, 0);
}
assert_eq!(filler.remaining(), 1024 - 400);
let slot_id = filler.slot_id();
let slices = filler.publish();
assert_eq!(slices.len(), 4, "four commits must publish four slices");
// Every slice reads back exactly what was written, from inside ONE slot.
for (i, s) in slices.iter().enumerate() {
assert_eq!(s.slot_id, slot_id, "all four slices share the slot");
assert_eq!(s.file_index, i as u64);
let bytes = unsafe { s.as_slice() };
assert_eq!(bytes.len(), 100);
assert!(
bytes.iter().all(|&b| b == b'a' + i as u8),
"slice {i} corrupted"
);
}
// The slot is NOT free until the last slice is released.
for s in slices.iter().take(3) {
assert!(
rev.try_claim_slot().is_none(),
"slot handed back while slices were still outstanding"
);
returner.release_one(s.slot_id);
}
assert!(
rev.try_claim_slot().is_none(),
"slot handed back with one slice still outstanding"
);
returner.release_one(slices[3].slot_id);
assert!(
rev.try_claim_slot().is_some(),
"slot was not recycled after its last release"
);
}
/// A published slice must be readable from a worker thread while the reader
/// keeps packing the next slot — that is the whole point of the ref count.
#[test]
fn workers_read_published_slices_concurrently_with_the_reader() {
const SLOTS: usize = 4;
const SLOT: usize = 4096;
const PAYLOAD: usize = 64;
const ROUNDS: usize = 200;
let rev = Arc::new(ChunkRevolver::with_coalescing(SLOTS, SLOT, 512));
let returner = rev.returner();
let sum = Arc::new(AtomicU64::new(0));
let (tx, rx) = std::sync::mpsc::channel();
let mut workers = Vec::new();
let rx = Arc::new(std::sync::Mutex::new(rx));
for _ in 0..4 {
let rx = Arc::clone(&rx);
let ret = returner.clone();
let sum = Arc::clone(&sum);
workers.push(std::thread::spawn(move || {
loop {
let slice: znippy_zoomies::chunk_revolver::PackedSlice = {
let g = rx.lock().unwrap();
match g.recv() {
Ok(s) => s,
Err(_) => break,
}
};
let bytes = unsafe { slice.as_slice() };
let local: u64 = bytes.iter().map(|&b| b as u64).sum();
sum.fetch_add(local, Ordering::Relaxed);
ret.release_one(slice.slot_id);
}
}));
}
let mut expected: u64 = 0;
let mut published = Vec::new();
let mut filler = rev.claim_slot().unwrap();
for r in 0..ROUNDS {
if filler.remaining() < PAYLOAD {
filler.publish_into(&mut published);
for s in published.drain(..) {
tx.send(s).unwrap();
}
filler = rev.claim_slot().expect("a worker must free a slot");
}
let byte = (r % 251) as u8;
let w = filler.writable(PAYLOAD);
w.fill(byte);
expected += (byte as u64) * PAYLOAD as u64;
filler.commit_slice(PAYLOAD, false, r as u64, 0, r as u32);
}
filler.publish_into(&mut published);
for s in published.drain(..) {
tx.send(s).unwrap();
}
drop(tx);
for w in workers {
w.join().unwrap();
}
assert_eq!(
sum.load(Ordering::Relaxed),
expected,
"bytes read back != bytes written"
);
}
/// A Filler dropped without `publish` must NOT leak its slot.
#[test]
fn dropped_filler_returns_its_slot() {
let rev = ChunkRevolver::with_coalescing(1, 512, 128);
{
let mut filler = rev.claim_slot().unwrap();
filler.writable(10).fill(7);
filler.commit_slice(10, false, 0, 0, 0);
// dropped without publish
}
assert!(
rev.try_claim_slot().is_some(),
"a Filler dropped un-published leaked its slot"
);
}
/// An empty slot returns straight to the free list on publish.
#[test]
fn publishing_an_empty_slot_frees_it_immediately() {
let rev = ChunkRevolver::with_coalescing(1, 512, 128);
let filler = rev.claim_slot().unwrap();
let slices = filler.publish();
assert!(slices.is_empty());
assert!(
rev.try_claim_slot().is_some(),
"empty published slot was not freed"
);
}
/// Coalescing mode must not double-issue: the fixed-slot `try_get_chunk` path
/// cannot hand out memory that the coalescing free list also owns.
#[test]
fn coalescing_mode_does_not_double_issue_through_the_fixed_slot_path() {
let mut rev = ChunkRevolver::with_coalescing(2, 256, 64);
assert!(
rev.try_get_chunk().is_none(),
"fixed-slot path handed out a slot the coalescing free list owns"
);
}