use core::cell::UnsafeCell;
use core::ptr;
use core::sync::atomic::{AtomicBool, AtomicPtr, AtomicU32, AtomicU64, AtomicUsize, Ordering};
use std::fs::File;
use std::io::{Read, Seek, SeekFrom};
use super::percpu_lock::PerCpuLocks;
use super::plat::{Percentiles, PlatStats};
fn monotonic_nanos() -> u64 {
let mut ts: libc::timespec = unsafe { core::mem::zeroed() };
let rc = unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts) };
assert_eq!(rc, 0, "clock_gettime(CLOCK_MONOTONIC) failed");
(ts.tv_sec as u64) * 1_000_000_000 + ts.tv_nsec as u64
}
trait Linked: Sized {
fn next_link(&self) -> &AtomicPtr<Self>;
}
struct TreiberStack<T: Linked> {
head: AtomicPtr<T>,
}
impl<T: Linked> TreiberStack<T> {
fn new() -> Self {
Self {
head: AtomicPtr::new(ptr::null_mut()),
}
}
fn add(&self, node: *mut T) {
let link = unsafe { (*node).next_link() };
loop {
let old = self.head.load(Ordering::Acquire);
link.store(old, Ordering::Relaxed);
if self
.head
.compare_exchange(old, node, Ordering::AcqRel, Ordering::Acquire)
.is_ok()
{
return;
}
}
}
fn splice(&self) -> *mut T {
self.head.swap(ptr::null_mut(), Ordering::AcqRel)
}
fn splice_reversed(&self) -> *mut T {
let mut cur = self.head.swap(ptr::null_mut(), Ordering::AcqRel);
let mut rev: *mut T = ptr::null_mut();
while !cur.is_null() {
let link = unsafe { (*cur).next_link() };
let next = link.load(Ordering::Acquire);
link.store(rev, Ordering::Relaxed);
rev = cur;
cur = next;
}
rev
}
}
struct Request {
next: AtomicPtr<Request>,
}
impl Request {
fn new() -> Self {
Self {
next: AtomicPtr::new(ptr::null_mut()),
}
}
}
impl Linked for Request {
fn next_link(&self) -> &AtomicPtr<Self> {
&self.next
}
}
#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct SchbenchConfig {
pub message_threads: usize,
pub worker_threads: usize,
pub cache_footprint_kib: usize,
pub operations: usize,
pub sleep_usec: u64,
pub skip_locking: bool,
pub requests_per_sec: usize,
pub auto_rps: usize,
pub split_percent: Option<usize>,
pub pipe_transfer_bytes: usize,
}
pub(crate) const PIPE_TRANSFER_BUFFER: usize = 1024 * 1024;
const STOP_POLL_QUANTUM_NS: u64 = 50_000_000;
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct PipeTransferReport {
pub ops_per_sec: f64,
pub scaled: f64,
pub unit: &'static str,
}
pub fn pipe_transfer_report(
achieved_rps: f64,
pipe_transfer_bytes: usize,
nr_workers: usize,
) -> PipeTransferReport {
let n = nr_workers.max(1) as f64;
let bytes = pipe_transfer_bytes.min(PIPE_TRANSFER_BUFFER);
let ops_per_sec = achieved_rps / n;
let bytes_per_sec = achieved_rps * bytes as f64 / n;
let (scaled, unit) = pretty_size(bytes_per_sec);
PipeTransferReport {
ops_per_sec,
scaled,
unit,
}
}
fn pretty_size(mut number: f64) -> (f64, &'static str) {
const UNITS: [&str; 7] = ["B", "KB", "MB", "GB", "TB", "PB", "EB"];
let mut divs = 0;
while number >= 1024.0 && divs + 1 < UNITS.len() {
divs += 1;
number /= 1024.0;
}
(number, UNITS[divs])
}
impl Default for SchbenchConfig {
fn default() -> Self {
Self {
message_threads: 1,
worker_threads: 0,
cache_footprint_kib: 256,
operations: 5,
sleep_usec: 100,
skip_locking: false,
requests_per_sec: 0,
auto_rps: 0,
split_percent: None,
pipe_transfer_bytes: 0,
}
}
}
impl SchbenchConfig {
#[must_use = "builder methods consume self; bind the result"]
pub fn message_threads(mut self, n: usize) -> Self {
self.message_threads = n;
self
}
#[must_use = "builder methods consume self; bind the result"]
pub fn worker_threads(mut self, n: usize) -> Self {
self.worker_threads = n;
self
}
#[must_use = "builder methods consume self; bind the result"]
pub fn cache_footprint_kib(mut self, kib: usize) -> Self {
self.cache_footprint_kib = kib;
self
}
#[must_use = "builder methods consume self; bind the result"]
pub fn operations(mut self, n: usize) -> Self {
self.operations = n;
self
}
#[must_use = "builder methods consume self; bind the result"]
pub fn sleep_usec(mut self, usec: u64) -> Self {
self.sleep_usec = usec;
self
}
#[must_use = "builder methods consume self; bind the result"]
pub fn skip_locking(mut self, skip: bool) -> Self {
self.skip_locking = skip;
self
}
#[must_use = "builder methods consume self; bind the result"]
pub fn requests_per_sec(mut self, rps: usize) -> Self {
self.requests_per_sec = rps;
self
}
#[must_use = "builder methods consume self; bind the result"]
pub fn auto_rps(mut self, target_pct: usize) -> Self {
self.auto_rps = target_pct;
self
}
#[must_use = "builder methods consume self; bind the result"]
pub fn split_percent(mut self, percent: Option<usize>) -> Self {
debug_assert!(
percent.is_none_or(|p| p <= 100),
"split_percent must be 0..=100"
);
self.split_percent = percent;
self
}
#[must_use = "builder methods consume self; bind the result"]
pub fn pipe_transfer_bytes(mut self, bytes: usize) -> Self {
self.pipe_transfer_bytes = bytes;
self
}
pub(crate) fn matrix_size(&self) -> usize {
if self.operations > 0 && self.cache_footprint_kib > 0 {
((self.cache_footprint_kib * 1024 / 3 / core::mem::size_of::<u64>()) as f64).sqrt()
as usize
} else {
0
}
}
pub(crate) fn shared_matrix_size(&self) -> usize {
match self.split_percent {
Some(p) if self.operations > 0 && self.cache_footprint_kib > 0 => {
let shared_kb = self.cache_footprint_kib * (100 - p) / 100;
((shared_kb * 1024 / 3 / core::mem::size_of::<u64>()) as f64).sqrt() as usize
}
_ => 0,
}
}
pub(crate) fn private_matrix_size(&self) -> usize {
match self.split_percent {
Some(p) if self.operations > 0 && self.cache_footprint_kib > 0 => {
let private_kb = self.cache_footprint_kib * p / 100;
((private_kb * 1024 / 3 / core::mem::size_of::<u64>()) as f64).sqrt() as usize
}
_ => self.matrix_size(),
}
}
pub(crate) fn normalized_total_rps(&self) -> usize {
if self.auto_rps != 0 && self.requests_per_sec == 0 {
10
} else {
self.requests_per_sec
}
}
pub(crate) fn rps_per_message_thread(&self) -> usize {
self.normalized_total_rps() / self.message_threads.max(1)
}
pub(crate) fn resolve_worker_count(&self, allowed_count: usize) -> usize {
if self.worker_threads != 0 {
return self.worker_threads;
}
allowed_count.div_ceil(self.message_threads.max(1))
}
}
pub(crate) struct ThreadData {
next: AtomicPtr<ThreadData>,
futex: super::handshake::Handshake,
wake_time: AtomicU64,
wakeup_stats: UnsafeCell<PlatStats>,
request_stats: UnsafeCell<PlatStats>,
sched_delay_ns: UnsafeCell<u64>,
phase_snapshots: UnsafeCell<Vec<PhaseSnapshot>>,
requests: TreiberStack<Request>,
pending: AtomicU64,
pipe_bytes: usize,
pipe_page: UnsafeCell<Box<[u8]>>,
}
struct PhaseSnapshot {
epoch: u32,
wakeup: PlatStats,
request: PlatStats,
run_delay_ns: u64,
pcount: u64,
loop_count: u64,
}
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub(crate) struct SchbenchPhaseStats {
pub(crate) wakeup: PlatStats,
pub(crate) request: PlatStats,
pub(crate) rps: PlatStats,
pub(crate) msg_run_delay_ns: u64,
pub(crate) msg_pcount: u64,
pub(crate) worker_run_delay_ns: u64,
pub(crate) worker_pcount: u64,
pub(crate) loop_count: u64,
}
impl SchbenchPhaseStats {
pub(crate) fn merge(&mut self, other: &SchbenchPhaseStats) {
self.wakeup.combine(&other.wakeup);
self.request.combine(&other.request);
self.rps.combine(&other.rps);
self.msg_run_delay_ns = self.msg_run_delay_ns.saturating_add(other.msg_run_delay_ns);
self.msg_pcount = self.msg_pcount.saturating_add(other.msg_pcount);
self.worker_run_delay_ns = self
.worker_run_delay_ns
.saturating_add(other.worker_run_delay_ns);
self.worker_pcount = self.worker_pcount.saturating_add(other.worker_pcount);
self.loop_count = self.loop_count.saturating_add(other.loop_count);
}
}
impl Linked for ThreadData {
fn next_link(&self) -> &AtomicPtr<Self> {
&self.next
}
}
unsafe impl Sync for ThreadData {}
impl ThreadData {
fn new(pipe_bytes: usize) -> Self {
Self {
next: AtomicPtr::new(ptr::null_mut()),
futex: super::handshake::Handshake::new(),
wake_time: AtomicU64::new(0),
wakeup_stats: UnsafeCell::new(PlatStats::default()),
request_stats: UnsafeCell::new(PlatStats::default()),
sched_delay_ns: UnsafeCell::new(0),
phase_snapshots: UnsafeCell::new(Vec::new()),
requests: TreiberStack::new(),
pending: AtomicU64::new(0),
pipe_bytes,
pipe_page: UnsafeCell::new(vec![0u8; pipe_bytes].into_boxed_slice()),
}
}
unsafe fn pipe_fill(&self, val: u8) {
let page = unsafe { &mut *self.pipe_page.get() };
page.fill(val);
std::hint::black_box(page.as_ptr());
}
}
fn think_sleep(usec: u64) {
std::thread::sleep(std::time::Duration::from_micros(usec));
}
fn ops_split(operations: usize, p: usize) -> (usize, usize) {
let ops_private = (operations * p) / 100;
let ops_shared = operations - ops_private;
(ops_shared, ops_private)
}
#[allow(clippy::too_many_arguments)]
fn do_work(
private_buf: &mut [u64],
private_matrix_size: usize,
shared_buf: &[core::sync::atomic::AtomicU64],
shared_matrix_size: usize,
split_percent: Option<usize>,
operations: usize,
locks: Option<&PerCpuLocks>,
work_units: &mut u64,
) {
let _guard = locks.map(|l| l.lock_this_cpu());
match split_percent {
Some(p) => {
let (ops_shared, ops_private) = ops_split(operations, p);
if shared_matrix_size > 0 && ops_shared > 0 {
for _ in 0..ops_shared {
crate::workload::worker::matrix_multiply_shared(
shared_buf,
shared_matrix_size,
work_units,
);
}
}
if private_matrix_size > 0 && ops_private > 0 {
for _ in 0..ops_private {
crate::workload::worker::matrix_multiply(
private_buf,
private_matrix_size,
work_units,
);
}
}
}
None => {
for _ in 0..operations {
if private_matrix_size > 0 {
crate::workload::worker::matrix_multiply(
private_buf,
private_matrix_size,
work_units,
);
}
}
}
}
}
fn msg_and_wait(
td: &ThreadData,
msg_td: &ThreadData,
wait_list: &TreiberStack<ThreadData>,
stop: &AtomicBool,
) {
if td.pipe_bytes > 0 {
unsafe { td.pipe_fill(2) };
}
td.wake_time.store(monotonic_nanos(), Ordering::Release);
wait_list.add(td as *const ThreadData as *mut ThreadData);
msg_td.futex.post();
if !stop.load(Ordering::Acquire) {
td.futex.wait_forever();
}
let now = monotonic_nanos();
let wake = td.wake_time.load(Ordering::Acquire);
let delta_us = now.saturating_sub(wake) / 1000;
if delta_us > 0 {
unsafe { (*td.wakeup_stats.get()).add_lat(delta_us.min(u32::MAX as u64) as u32) };
}
}
fn rps_wait(td: &ThreadData, stop: &AtomicBool) -> *mut Request {
td.wake_time.store(monotonic_nanos(), Ordering::Release);
td.pending.store(0, Ordering::Release);
let chain = td.requests.splice_reversed();
if !chain.is_null() {
return chain; }
if stop.load(Ordering::Acquire) {
return ptr::null_mut();
}
td.futex.wait_forever();
if stop.load(Ordering::Acquire) {
return td.requests.splice_reversed();
}
let now = monotonic_nanos();
let wake = td.wake_time.load(Ordering::Acquire);
let delta_us = now.saturating_sub(wake) / 1000;
if delta_us > 0 {
unsafe { (*td.wakeup_stats.get()).add_lat(delta_us.min(u32::MAX as u64) as u32) };
}
td.requests.splice_reversed()
}
fn free_request_chain(mut req: *mut Request) {
while !req.is_null() {
let next = unsafe { (*req).next.load(Ordering::Acquire) };
drop(unsafe { Box::from_raw(req) });
req = next;
}
}
fn wake_all(wait_list: &TreiberStack<ThreadData>) {
let mut cur = wait_list.splice();
let now = monotonic_nanos();
while !cur.is_null() {
let td = unsafe { &*cur };
let next = td.next.load(Ordering::Acquire);
td.next.store(ptr::null_mut(), Ordering::Relaxed);
if td.pipe_bytes > 0 {
unsafe { td.pipe_fill(1) };
td.wake_time.store(monotonic_nanos(), Ordering::Release);
} else {
td.wake_time.store(now, Ordering::Release);
}
td.futex.post();
cur = next;
}
}
fn run_msg_thread(
msg_td: &ThreadData,
wait_list: &TreiberStack<ThreadData>,
stop: &AtomicBool,
phase_epoch: Option<&AtomicU32>,
) {
let tid = gettid_self();
let mut cur_epoch = phase_epoch.map_or(0, |e| e.load(Ordering::Relaxed));
let mut phase_ss_start = read_schedstat_raw(tid);
loop {
wake_all(wait_list);
if stop.load(Ordering::Acquire) {
wake_all(wait_list);
break;
}
msg_td.futex.wait_forever();
if let Some(pe) = phase_epoch {
let new_epoch = pe.load(Ordering::Relaxed);
if new_epoch != cur_epoch {
let ss_end = read_schedstat_raw(tid);
unsafe { drain_phase(msg_td, cur_epoch, phase_ss_start, ss_end, 0) };
cur_epoch = new_epoch;
phase_ss_start = ss_end;
}
}
}
let ss_end = read_schedstat_raw(tid);
unsafe { drain_phase(msg_td, cur_epoch, phase_ss_start, ss_end, 0) };
unsafe { *msg_td.sched_delay_ns.get() = mean_sched_delay(ss_end) };
}
fn run_rps_thread(
workers: &[ThreadData],
msg_td: &ThreadData,
stop: &AtomicBool,
phase_epoch: Option<&AtomicU32>,
live_rate: &AtomicUsize,
) {
let tid = gettid_self();
let mut cur_epoch = phase_epoch.map_or(0, |e| e.load(Ordering::Relaxed));
let mut phase_ss_start = read_schedstat_raw(tid);
const BATCH: u64 = 128;
const ONE_SEC_NS: u64 = 1_000_000_000;
let worker_count = workers.len().max(1);
let mut cur_tid: usize = 0;
while !stop.load(Ordering::Acquire) {
let requests_per_sec = live_rate.load(Ordering::Relaxed);
let start = monotonic_nanos();
let interval_ns = if requests_per_sec > 0 {
ONE_SEC_NS / requests_per_sec as u64
} else {
ONE_SEC_NS
};
let mut due = start;
for _ in 0..requests_per_sec {
if stop.load(Ordering::Acquire) {
break;
}
due = due.saturating_add(interval_ns);
let mut stop_during_pace = false;
loop {
if stop.load(Ordering::Acquire) {
stop_during_pace = true;
break;
}
let now = monotonic_nanos();
if now >= due {
break;
}
let nap = (due - now).min(STOP_POLL_QUANTUM_NS);
std::thread::sleep(std::time::Duration::from_nanos(nap));
}
if stop_during_pace {
break;
}
let worker = &workers[cur_tid % worker_count];
cur_tid += 1;
if worker.pending.load(Ordering::Acquire) > BATCH {
std::thread::sleep(std::time::Duration::from_micros(100));
continue;
}
worker.pending.fetch_add(1, Ordering::AcqRel);
let req = Box::into_raw(Box::new(Request::new()));
worker.requests.add(req);
worker.wake_time.store(monotonic_nanos(), Ordering::Release);
worker.futex.post();
}
loop {
if stop.load(Ordering::Acquire) {
break;
}
let elapsed = monotonic_nanos().saturating_sub(start);
if elapsed >= ONE_SEC_NS {
break;
}
let nap = (ONE_SEC_NS - elapsed).min(STOP_POLL_QUANTUM_NS);
std::thread::sleep(std::time::Duration::from_nanos(nap));
}
if let Some(pe) = phase_epoch {
let new_epoch = pe.load(Ordering::Relaxed);
if new_epoch != cur_epoch {
let ss_end = read_schedstat_raw(tid);
unsafe { drain_phase(msg_td, cur_epoch, phase_ss_start, ss_end, 0) };
cur_epoch = new_epoch;
phase_ss_start = ss_end;
}
}
if stop.load(Ordering::Acquire) {
for w in workers {
w.futex.post();
}
break;
}
}
let ss_end = read_schedstat_raw(tid);
unsafe { drain_phase(msg_td, cur_epoch, phase_ss_start, ss_end, 0) };
unsafe { *msg_td.sched_delay_ns.get() = mean_sched_delay(ss_end) };
}
fn gettid_self() -> libc::pid_t {
unsafe { libc::syscall(libc::SYS_gettid) as libc::pid_t }
}
fn read_schedstat_raw(tid: libc::pid_t) -> (u64, u64) {
match std::fs::read_to_string(format!("/proc/{tid}/schedstat")) {
Ok(s) => parse_schedstat_raw(&s),
Err(_) => (0, 0),
}
}
fn parse_schedstat_raw(s: &str) -> (u64, u64) {
let mut fields = s.split_whitespace();
let _run = fields.next(); let run_delay: u64 = fields
.next()
.and_then(|f| f.parse().ok())
.expect("schedstat field 2 (run_delay) must be a present integer");
let pcount: u64 = fields
.next()
.and_then(|f| f.parse().ok())
.expect("schedstat field 3 (pcount) must be a present integer");
(run_delay, pcount)
}
fn mean_sched_delay((run_delay, pcount): (u64, u64)) -> u64 {
if pcount == 0 { 0 } else { run_delay / pcount }
}
unsafe fn drain_phase(
td: &ThreadData,
epoch: u32,
ss_start: (u64, u64),
ss_end: (u64, u64),
loop_count: u64,
) {
unsafe {
let wakeup = (*td.wakeup_stats.get()).take();
let request = (*td.request_stats.get()).take();
(*td.phase_snapshots.get()).push(PhaseSnapshot {
epoch,
wakeup,
request,
run_delay_ns: ss_end.0.saturating_sub(ss_start.0),
pcount: ss_end.1.saturating_sub(ss_start.1),
loop_count,
});
}
}
struct WorkerCtx<'a> {
msg_td: &'a ThreadData,
wait_list: &'a TreiberStack<ThreadData>,
locks: Option<&'a PerCpuLocks>,
config: &'a SchbenchConfig,
stop: &'a AtomicBool,
progress: &'a AtomicU64,
phase_epoch: Option<&'a AtomicU32>,
shared: &'a [core::sync::atomic::AtomicU64],
shared_matrix_size: usize,
}
fn worker_loop(td: &ThreadData, ctx: &WorkerCtx) {
let WorkerCtx {
msg_td,
wait_list,
locks,
config,
stop,
progress,
phase_epoch,
shared,
shared_matrix_size,
} = *ctx;
let tid = gettid_self();
let private_matrix_size = config.private_matrix_size();
let mut private_buf = if private_matrix_size > 0 {
vec![0u64; 3 * private_matrix_size * private_matrix_size]
} else {
Vec::new()
};
let mut work_units = 0u64;
let mut cur_epoch = phase_epoch.map_or(0, |e| e.load(Ordering::Relaxed));
let mut phase_ss_start = read_schedstat_raw(tid);
let mut phase_loop_count = 0u64;
let pipe_mode = config.pipe_transfer_bytes > 0;
let rps_mode = !pipe_mode && config.rps_per_message_thread() != 0;
while !stop.load(Ordering::Acquire) {
let mut req: *mut Request = if rps_mode {
let chain = rps_wait(td, stop);
if stop.load(Ordering::Acquire) {
free_request_chain(chain);
break;
}
if chain.is_null() {
continue; }
chain
} else {
msg_and_wait(td, msg_td, wait_list, stop);
if stop.load(Ordering::Acquire) {
break;
}
ptr::null_mut()
};
loop {
let work_start = monotonic_nanos();
if !pipe_mode {
if config.sleep_usec > 0 {
think_sleep(config.sleep_usec);
}
do_work(
&mut private_buf,
private_matrix_size,
shared,
shared_matrix_size,
config.split_percent,
config.operations,
locks,
&mut work_units,
);
}
let now = monotonic_nanos();
let delta_us = now.saturating_sub(work_start) / 1000;
if delta_us > 0 {
unsafe { (*td.request_stats.get()).add_lat(delta_us.min(u32::MAX as u64) as u32) };
}
progress.fetch_add(1, Ordering::Relaxed);
phase_loop_count += 1;
if req.is_null() {
break; }
let next = unsafe { (*req).next.load(Ordering::Acquire) };
drop(unsafe { Box::from_raw(req) });
req = next;
if req.is_null() {
break;
}
}
if let Some(pe) = phase_epoch {
let new_epoch = pe.load(Ordering::Relaxed);
if new_epoch != cur_epoch {
let ss_end = read_schedstat_raw(tid);
unsafe { drain_phase(td, cur_epoch, phase_ss_start, ss_end, phase_loop_count) };
cur_epoch = new_epoch;
phase_ss_start = ss_end;
phase_loop_count = 0;
}
}
}
msg_td.futex.post();
let ss_end = read_schedstat_raw(tid);
unsafe { drain_phase(td, cur_epoch, phase_ss_start, ss_end, phase_loop_count) };
unsafe { *td.sched_delay_ns.get() = mean_sched_delay(ss_end) };
}
fn resolve_cpu_topology() -> (usize, usize) {
unsafe {
let mut set: libc::cpu_set_t = core::mem::zeroed();
let rc = libc::sched_getaffinity(0, core::mem::size_of::<libc::cpu_set_t>(), &mut set);
if rc != 0 {
return (1, 1);
}
let mut count = 0usize;
let mut max_id = 0usize;
for cpu in 0..libc::CPU_SETSIZE as usize {
if libc::CPU_ISSET(cpu, &set) {
count += 1;
max_id = cpu;
}
}
(count.max(1), (max_id + 1).max(1))
}
}
pub(crate) struct SchbenchResult {
pub(crate) wakeup: Percentiles,
pub(crate) request: Percentiles,
pub(crate) rps: Percentiles,
pub(crate) loop_count: u64,
pub(crate) nr_workers: usize,
pub(crate) achieved_rps: f64,
pub(crate) final_rps_goal: usize,
pub(crate) sched_delay_msg_ns: u64,
pub(crate) sched_delay_worker_ns: u64,
}
pub(crate) struct SchbenchOutcome {
pub(crate) whole_run: SchbenchResult,
pub(crate) phases: Vec<(u32, SchbenchPhaseStats)>,
}
struct MessageThreadResult {
whole_wakeup: PlatStats,
whole_request: PlatStats,
msg_sched_delay_ns: u64,
workers_sched_delay_sum: u64,
phases: std::collections::BTreeMap<u32, SchbenchPhaseStats>,
}
#[allow(clippy::too_many_arguments)]
fn run_one_message_thread(
worker_threads: usize,
locks: Option<&PerCpuLocks>,
config: &SchbenchConfig,
stop: &AtomicBool,
progress: &AtomicU64,
phase_epoch: Option<&AtomicU32>,
live_rate: &AtomicUsize,
shared: &[core::sync::atomic::AtomicU64],
shared_matrix_size: usize,
) -> MessageThreadResult {
let pipe_bytes = config.pipe_transfer_bytes.min(PIPE_TRANSFER_BUFFER);
let workers: Vec<ThreadData> = (0..worker_threads)
.map(|_| ThreadData::new(pipe_bytes))
.collect();
let msg_td = ThreadData::new(0);
let wait_list = TreiberStack::new();
let ctx = WorkerCtx {
msg_td: &msg_td,
wait_list: &wait_list,
locks,
config,
stop,
progress,
phase_epoch,
shared,
shared_matrix_size,
};
std::thread::scope(|inner| {
for w in &workers {
inner.spawn(|| worker_loop(w, &ctx));
}
if config.pipe_transfer_bytes == 0 && config.rps_per_message_thread() != 0 {
run_rps_thread(&workers, &msg_td, stop, phase_epoch, live_rate);
} else {
run_msg_thread(&msg_td, &wait_list, stop, phase_epoch);
}
for w in &workers {
w.futex.post();
}
});
if config.rps_per_message_thread() != 0 {
for w in &workers {
free_request_chain(w.requests.splice_reversed());
}
}
let mut whole_wakeup = PlatStats::default();
let mut whole_request = PlatStats::default();
let mut workers_sched_delay_sum = 0u64;
let mut phases: std::collections::BTreeMap<u32, SchbenchPhaseStats> =
std::collections::BTreeMap::new();
for w in &workers {
unsafe {
workers_sched_delay_sum =
workers_sched_delay_sum.saturating_add(*w.sched_delay_ns.get());
for snap in (*w.phase_snapshots.get()).drain(..) {
whole_wakeup.combine(&snap.wakeup);
whole_request.combine(&snap.request);
let e = phases.entry(snap.epoch).or_default();
e.wakeup.combine(&snap.wakeup);
e.request.combine(&snap.request);
e.worker_run_delay_ns = e.worker_run_delay_ns.saturating_add(snap.run_delay_ns);
e.worker_pcount = e.worker_pcount.saturating_add(snap.pcount);
e.loop_count = e.loop_count.saturating_add(snap.loop_count);
}
}
}
let msg_sched_delay = unsafe { *msg_td.sched_delay_ns.get() };
unsafe {
for snap in (*msg_td.phase_snapshots.get()).drain(..) {
let e = phases.entry(snap.epoch).or_default();
e.msg_run_delay_ns = e.msg_run_delay_ns.saturating_add(snap.run_delay_ns);
e.msg_pcount = e.msg_pcount.saturating_add(snap.pcount);
}
}
MessageThreadResult {
whole_wakeup,
whole_request,
msg_sched_delay_ns: msg_sched_delay,
workers_sched_delay_sum,
phases,
}
}
#[derive(Default)]
struct ReadBusyState {
fd: Option<File>,
prev_total: u64,
prev_idle: u64,
}
fn read_busy(s: &mut ReadBusyState) -> Option<f32> {
if s.fd.is_none() {
s.fd = File::open("/proc/stat").ok();
}
let f = s.fd.as_mut()?;
f.seek(SeekFrom::Start(0)).ok()?;
let mut buf = [0u8; 512];
let n = f.read(&mut buf).ok()?;
let text = core::str::from_utf8(&buf[..n]).ok()?;
let line = text.lines().next().unwrap_or("");
let mut fields = line.split_whitespace();
if fields.next() != Some("cpu") {
return None;
}
let mut total = 0u64;
let mut idle = 0u64;
for (i, tok) in fields.enumerate() {
let v: u64 = tok.parse().unwrap_or(0);
if i == 3 {
idle = v;
}
total += v;
}
if s.prev_total == 0 {
s.prev_total = total;
s.prev_idle = idle;
return None;
}
let dt = total.saturating_sub(s.prev_total);
let di = idle.saturating_sub(s.prev_idle);
s.prev_total = total;
s.prev_idle = idle;
if dt == 0 {
return None;
}
Some(100.0 - (di as f32 / dt as f32) * 100.0)
}
fn scale_rps_for_busy(
busy: f32,
live_rate: &AtomicUsize,
target_hit: &AtomicBool,
target_pct: usize,
) -> bool {
let target = target_pct as f32;
let rps = live_rate.load(Ordering::Relaxed) as f32;
let already_hit = target_hit.load(Ordering::Relaxed);
let mut just_hit = false;
let new_rate: usize = if busy < target {
let mut delta = target / busy;
if delta > 3.0 {
delta = 3.0;
} else if delta < 1.2 {
delta = 1.0 + (delta - 1.0) / 8.0;
if delta < 1.05 && !already_hit {
just_hit = true;
}
} else if delta < 1.5 {
delta = 1.0 + (delta - 1.0) / 4.0;
}
let t = (rps * delta).ceil();
if t >= (1u64 << 31) as f32 {
rps as usize
} else {
t as usize
}
} else if busy > target {
let mut delta = target / busy;
if delta < 0.3 {
delta = 0.3;
} else if delta > 0.9 {
delta += (1.0 - delta) / 8.0;
if delta > 0.95 && !already_hit {
just_hit = true;
}
} else if delta > 0.8 {
delta += (1.0 - delta) / 4.0;
}
(rps * delta).floor().max(0.0) as usize
} else {
if !already_hit {
just_hit = true;
}
rps as usize
};
live_rate.store(new_rate, Ordering::Relaxed);
if just_hit {
target_hit.store(true, Ordering::Relaxed);
}
just_hit
}
fn auto_scale_rps(
busy_state: &mut ReadBusyState,
live_rate: &AtomicUsize,
target_hit: &AtomicBool,
target_pct: usize,
) -> bool {
let Some(busy) = read_busy(busy_state) else {
return false;
};
scale_rps_for_busy(busy, live_rate, target_hit, target_pct)
}
fn reset_rps_accumulators(
whole: &mut PlatStats,
per_epoch: &mut std::collections::BTreeMap<u32, PlatStats>,
) {
*whole = PlatStats::default();
per_epoch.clear();
}
fn control_loop(
progress: &AtomicU64,
stop: &AtomicBool,
config: &SchbenchConfig,
live_rate: &AtomicUsize,
target_hit: &AtomicBool,
phase_epoch: Option<&AtomicU32>,
) -> (PlatStats, std::collections::BTreeMap<u32, PlatStats>) {
let mut rps_stats = PlatStats::default();
let mut per_epoch_rps: std::collections::BTreeMap<u32, PlatStats> =
std::collections::BTreeMap::new();
let mut last_loop = 0u64;
let mut last_t = monotonic_nanos();
let mut busy_state = ReadBusyState::default();
while !stop.load(Ordering::Acquire) {
let sleep_target = monotonic_nanos().saturating_add(1_000_000_000);
loop {
if stop.load(Ordering::Acquire) {
break;
}
let now = monotonic_nanos();
if now >= sleep_target {
break;
}
std::thread::sleep(std::time::Duration::from_nanos(
(sleep_target - now).min(STOP_POLL_QUANTUM_NS),
));
}
let now = monotonic_nanos();
let lc = progress.load(Ordering::Relaxed);
let dt = now.saturating_sub(last_t);
if let Some(sample) = control_loop_rps_sample(lc.saturating_sub(last_loop), dt) {
if config.auto_rps == 0 || target_hit.load(Ordering::Relaxed) {
rps_stats.add_lat(sample);
let epoch = phase_epoch.map_or(0, |e| e.load(Ordering::Relaxed));
per_epoch_rps.entry(epoch).or_default().add_lat(sample);
}
}
last_loop = lc;
last_t = now;
if config.auto_rps != 0 {
let just_hit = auto_scale_rps(&mut busy_state, live_rate, target_hit, config.auto_rps);
if just_hit {
reset_rps_accumulators(&mut rps_stats, &mut per_epoch_rps);
}
}
}
(rps_stats, per_epoch_rps)
}
fn control_loop_rps_sample(delta_loops: u64, dt_ns: u64) -> Option<u32> {
if dt_ns < STOP_POLL_QUANTUM_NS {
return None;
}
let rps = delta_loops as f64 * 1e9 / dt_ns as f64;
Some((rps as u64).min(u32::MAX as u64) as u32)
}
pub(crate) fn run(
config: &SchbenchConfig,
stop: &AtomicBool,
progress: &AtomicU64,
phase_epoch: Option<&AtomicU32>,
) -> SchbenchOutcome {
let (allowed_count, lock_array_size) = resolve_cpu_topology();
let worker_threads = config.resolve_worker_count(allowed_count);
let locks = if config.skip_locking {
None
} else {
Some(PerCpuLocks::new(lock_array_size))
};
if let Some(p) = config.split_percent {
assert!(
p <= 100,
"schbench split_percent must be in 0..=100, got {p} (schbench.c:362-365 rejects out of range)"
);
}
let shared_matrix_size = config.shared_matrix_size();
let shared: std::sync::Arc<[core::sync::atomic::AtomicU64]> = if shared_matrix_size > 0 {
(0..3 * shared_matrix_size * shared_matrix_size)
.map(|_| core::sync::atomic::AtomicU64::new(0))
.collect()
} else {
std::sync::Arc::from([] as [core::sync::atomic::AtomicU64; 0])
};
let start = monotonic_nanos();
let mut all_wakeup = PlatStats::default();
let mut all_request = PlatStats::default();
let mut total_msg_sched_delay = 0u64;
let mut total_worker_sched_delay = 0u64;
let mut all_phases: std::collections::BTreeMap<u32, SchbenchPhaseStats> =
std::collections::BTreeMap::new();
let live_rate = AtomicUsize::new(config.rps_per_message_thread());
let target_hit = AtomicBool::new(false);
let mut rps_stats = PlatStats::default();
std::thread::scope(|outer| {
let handles: Vec<_> = (0..config.message_threads)
.map(|_| {
let locks = locks.as_ref();
let live_rate = &live_rate;
let shared = &shared;
outer.spawn(move || {
run_one_message_thread(
worker_threads,
locks,
config,
stop,
progress,
phase_epoch,
live_rate,
&shared[..],
shared_matrix_size,
)
})
})
.collect();
let control = outer
.spawn(|| control_loop(progress, stop, config, &live_rate, &target_hit, phase_epoch));
for h in handles {
let mtr = h.join().expect("schbench message thread panicked");
all_wakeup.combine(&mtr.whole_wakeup);
all_request.combine(&mtr.whole_request);
total_msg_sched_delay = total_msg_sched_delay.saturating_add(mtr.msg_sched_delay_ns);
total_worker_sched_delay =
total_worker_sched_delay.saturating_add(mtr.workers_sched_delay_sum);
for (epoch, sps) in mtr.phases {
all_phases.entry(epoch).or_default().merge(&sps);
}
}
let (whole_rps, control_per_epoch_rps) =
control.join().expect("schbench control thread panicked");
rps_stats = whole_rps;
for (epoch, hist) in control_per_epoch_rps {
all_phases.entry(epoch).or_default().rps.combine(&hist);
}
});
let final_rps_goal = live_rate.load(Ordering::Relaxed) * config.message_threads;
let loop_count = progress.load(Ordering::Relaxed);
let elapsed_ns = monotonic_nanos().saturating_sub(start);
let achieved_rps = if elapsed_ns > 0 {
loop_count as f64 / (elapsed_ns as f64 / 1e9)
} else {
0.0
};
let sched_delay_msg_ns = total_msg_sched_delay / (config.message_threads.max(1) as u64);
let total_workers = (config.message_threads * worker_threads).max(1) as u64;
let sched_delay_worker_ns = total_worker_sched_delay / total_workers;
SchbenchOutcome {
whole_run: SchbenchResult {
wakeup: all_wakeup.percentiles(),
request: all_request.percentiles(),
rps: rps_stats.percentiles(),
loop_count,
nr_workers: total_workers as usize,
achieved_rps,
final_rps_goal,
sched_delay_msg_ns,
sched_delay_worker_ns,
},
phases: all_phases.into_iter().collect(),
}
}
#[cfg(test)]
mod tests {
use super::*;
struct TestNode {
next: AtomicPtr<TestNode>,
}
impl Linked for TestNode {
fn next_link(&self) -> &AtomicPtr<Self> {
&self.next
}
}
impl TestNode {
fn new() -> Self {
Self {
next: AtomicPtr::new(ptr::null_mut()),
}
}
}
#[test]
fn resolve_worker_count_divides_cpuset_across_message_threads() {
let c = SchbenchConfig::default()
.worker_threads(3)
.message_threads(4);
assert_eq!(c.resolve_worker_count(8), 3);
let c = SchbenchConfig::default().message_threads(1);
assert_eq!(c.resolve_worker_count(8), 8);
let c = SchbenchConfig::default().message_threads(2);
assert_eq!(c.resolve_worker_count(8), 4);
let c = SchbenchConfig::default().message_threads(3);
assert_eq!(c.resolve_worker_count(8), 3);
let c = SchbenchConfig::default().message_threads(0);
assert_eq!(c.resolve_worker_count(8), 8);
}
#[test]
fn matrix_size_matches_schbench_formula() {
assert_eq!(
SchbenchConfig {
cache_footprint_kib: 256,
operations: 5,
..Default::default()
}
.matrix_size(),
104
);
assert_eq!(
SchbenchConfig {
operations: 0,
..Default::default()
}
.matrix_size(),
0
);
}
#[test]
fn split_matrix_sizes_match_schbench_formula() {
let cfg = |split| SchbenchConfig {
cache_footprint_kib: 256,
operations: 5,
split_percent: split,
..Default::default()
};
assert_eq!(cfg(None).private_matrix_size(), 104);
assert_eq!(cfg(None).shared_matrix_size(), 0);
assert_eq!(cfg(Some(0)).private_matrix_size(), 0);
assert_eq!(cfg(Some(0)).shared_matrix_size(), 104);
assert_eq!(cfg(Some(25)).private_matrix_size(), 52);
assert_eq!(cfg(Some(25)).shared_matrix_size(), 90);
assert_eq!(cfg(Some(100)).private_matrix_size(), 104);
assert_eq!(cfg(Some(100)).shared_matrix_size(), 0);
let none_ops = SchbenchConfig {
operations: 0,
split_percent: Some(50),
..Default::default()
};
assert_eq!(none_ops.private_matrix_size(), 0);
assert_eq!(none_ops.shared_matrix_size(), 0);
}
#[test]
fn ops_split_matches_schbench_integer_division() {
assert_eq!(ops_split(5, 0), (5, 0)); assert_eq!(ops_split(5, 25), (4, 1)); assert_eq!(ops_split(5, 50), (3, 2)); assert_eq!(ops_split(5, 100), (0, 5)); }
#[test]
fn engine_split_runs_shared_matrix_concurrently() {
let config = SchbenchConfig {
message_threads: 1,
worker_threads: 2,
cache_footprint_kib: 16,
operations: 8,
sleep_usec: 0,
skip_locking: false,
requests_per_sec: 0,
auto_rps: 0,
split_percent: Some(50),
pipe_transfer_bytes: 0,
};
assert!(config.shared_matrix_size() > 0, "shared matrix present");
assert!(config.private_matrix_size() > 0, "private matrix present");
let stop = AtomicBool::new(false);
let progress = AtomicU64::new(0);
let outcome = std::thread::scope(|s| {
let runner = s.spawn(|| run(&config, &stop, &progress, None));
while progress.load(Ordering::Relaxed) < 50 {
core::hint::spin_loop();
}
stop.store(true, Ordering::Release);
runner.join().expect("run panicked")
});
assert!(
outcome.whole_run.loop_count >= 50,
"engine did split work: {}",
outcome.whole_run.loop_count
);
}
#[test]
fn pipe_fill_writes_every_byte_and_sizes_the_page() {
let td = ThreadData::new(64);
assert_eq!(td.pipe_bytes, 64);
unsafe { td.pipe_fill(2) };
assert_eq!(
unsafe { &*td.pipe_page.get() }.len(),
64,
"page sized to pipe_bytes"
);
assert!(
unsafe { &*td.pipe_page.get() }.iter().all(|&b| b == 2),
"worker fill (2) touches every byte"
);
unsafe { td.pipe_fill(1) };
assert!(
unsafe { &*td.pipe_page.get() }.iter().all(|&b| b == 1),
"waker fill (1) overwrites every byte"
);
let none = ThreadData::new(0);
assert_eq!(none.pipe_bytes, 0);
assert_eq!(unsafe { &*none.pipe_page.get() }.len(), 0);
}
#[test]
fn pipe_transfer_bytes_clamps_to_one_mib() {
assert_eq!(PIPE_TRANSFER_BUFFER, 1024 * 1024);
assert_eq!(
(2 * PIPE_TRANSFER_BUFFER).min(PIPE_TRANSFER_BUFFER),
PIPE_TRANSFER_BUFFER,
"over-cap pipe size clamps to the 1 MiB buffer"
);
assert_eq!(4096_usize.min(PIPE_TRANSFER_BUFFER), 4096);
}
#[test]
fn pipe_transfer_report_is_per_worker_scales_and_clamps() {
let r = pipe_transfer_report(2048.0, 1, 2);
assert_eq!(r.ops_per_sec, 1024.0);
assert_eq!((r.scaled, r.unit), (1.0, "KB"));
let one = pipe_transfer_report(1.0, 1, 1);
assert_eq!((one.ops_per_sec, one.scaled, one.unit), (1.0, 1.0, "B"));
assert_eq!(pipe_transfer_report(1.0, 1 << 20, 1).unit, "MB");
let over = pipe_transfer_report(1.0, 2 << 20, 1);
let cap = pipe_transfer_report(1.0, 1 << 20, 1);
assert_eq!(
(over.scaled, over.unit),
(cap.scaled, cap.unit),
"over-cap pipe size clamps to 1 MiB"
);
assert_eq!(pipe_transfer_report(f64::MAX, 1 << 20, 1).unit, "EB");
assert_eq!(pipe_transfer_report(100.0, 4096, 0).ops_per_sec, 100.0);
}
#[test]
fn engine_pipe_mode_transfers_without_matrix_work() {
let config = SchbenchConfig::default()
.message_threads(1)
.worker_threads(2)
.sleep_usec(0)
.operations(50)
.cache_footprint_kib(256)
.pipe_transfer_bytes(4096);
assert!(config.pipe_transfer_bytes > 0, "pipe mode engaged");
assert!(
config.matrix_size() > 0,
"matrix work would be non-trivial if do_work ran (regression guard)"
);
let stop = AtomicBool::new(false);
let progress = AtomicU64::new(0);
let outcome = std::thread::scope(|s| {
let runner = s.spawn(|| run(&config, &stop, &progress, None));
while progress.load(Ordering::Relaxed) < 50 {
core::hint::spin_loop();
}
stop.store(true, Ordering::Release);
runner.join().expect("run panicked")
});
assert!(
outcome.whole_run.loop_count >= 50,
"pipe engine transferred {} cycles",
outcome.whole_run.loop_count
);
assert!(
outcome.whole_run.request.nr_samples * 4 < outcome.whole_run.loop_count,
"pipe mode skips matrix work: {} request samples vs {} cycles",
outcome.whole_run.request.nr_samples,
outcome.whole_run.loop_count
);
}
#[test]
#[should_panic(expected = "split_percent must be in 0..=100")]
fn engine_panics_on_out_of_range_split() {
let config = SchbenchConfig {
message_threads: 1,
worker_threads: 1,
cache_footprint_kib: 16,
operations: 1,
sleep_usec: 0,
skip_locking: false,
requests_per_sec: 0,
auto_rps: 0,
split_percent: Some(101),
pipe_transfer_bytes: 0,
};
let stop = AtomicBool::new(true);
let progress = AtomicU64::new(0);
let _ = run(&config, &stop, &progress, None);
}
#[test]
fn stack_add_splice_is_lifo() {
let a = TestNode::new();
let b = TestNode::new();
let stack: TreiberStack<TestNode> = TreiberStack::new();
assert!(stack.splice().is_null(), "empty stack splices to null");
stack.add(&a as *const _ as *mut _);
stack.add(&b as *const _ as *mut _);
let head = stack.splice();
assert_eq!(head.cast_const(), &b as *const TestNode);
let second = unsafe { (*head).next.load(Ordering::Acquire) };
assert_eq!(second.cast_const(), &a as *const TestNode);
assert!(unsafe { (*second).next.load(Ordering::Acquire) }.is_null());
assert!(stack.splice().is_null(), "splice emptied the stack");
}
#[test]
fn stack_concurrent_add_loses_no_nodes() {
const THREADS: usize = 8;
const PER_THREAD: usize = 2000;
let nodes: Vec<TestNode> = (0..THREADS * PER_THREAD).map(|_| TestNode::new()).collect();
let stack = TreiberStack::new();
std::thread::scope(|s| {
for chunk in nodes.chunks(PER_THREAD) {
let stack = &stack;
s.spawn(move || {
for n in chunk {
stack.add(n as *const TestNode as *mut TestNode);
}
});
}
});
let mut seen = std::collections::HashSet::new();
let mut cur = stack.splice();
while !cur.is_null() {
assert!(seen.insert(cur), "node observed twice");
cur = unsafe { (*cur).next.load(Ordering::Acquire) };
}
assert_eq!(
seen.len(),
THREADS * PER_THREAD,
"no node lost under contention"
);
}
#[test]
fn engine_runs_and_produces_latency_samples() {
let config = SchbenchConfig {
message_threads: 1,
worker_threads: 2,
cache_footprint_kib: 16,
operations: 1,
sleep_usec: 0,
skip_locking: false,
requests_per_sec: 0,
auto_rps: 0,
split_percent: None,
pipe_transfer_bytes: 0,
};
let stop = AtomicBool::new(false);
let progress = AtomicU64::new(0);
let outcome = std::thread::scope(|s| {
let runner = s.spawn(|| run(&config, &stop, &progress, None));
while progress.load(Ordering::Relaxed) < 50 {
core::hint::spin_loop();
}
stop.store(true, Ordering::Release);
runner.join().expect("run panicked")
});
let result = &outcome.whole_run;
assert!(
result.loop_count >= 50,
"engine did work: {}",
result.loop_count
);
assert!(result.wakeup.nr_samples > 0, "wakeup samples recorded");
assert!(result.request.nr_samples > 0, "request samples recorded");
assert!(result.achieved_rps > 0.0, "positive achieved rps");
assert_eq!(
outcome.phases.len(),
1,
"non-phasic => single baseline phase"
);
assert_eq!(outcome.phases[0].0, 0, "the lone phase is BASELINE epoch 0");
}
#[test]
fn engine_terminates_when_lone_worker_stops() {
let config = SchbenchConfig {
message_threads: 1,
worker_threads: 1,
cache_footprint_kib: 256,
operations: 5,
sleep_usec: 0,
skip_locking: false,
requests_per_sec: 0,
auto_rps: 0,
split_percent: None,
pipe_transfer_bytes: 0,
};
let stop = AtomicBool::new(false);
let progress = AtomicU64::new(0);
let outcome = std::thread::scope(|s| {
let runner = s.spawn(|| run(&config, &stop, &progress, None));
while progress.load(Ordering::Relaxed) < 10 {
core::hint::spin_loop();
}
stop.store(true, Ordering::Release);
runner.join().expect("run panicked")
});
assert!(
outcome.whole_run.loop_count >= 10,
"engine did work and returned: {}",
outcome.whole_run.loop_count
);
}
#[test]
fn engine_rps_mode_injects_drains_and_terminates() {
let config = SchbenchConfig {
message_threads: 1,
worker_threads: 2,
cache_footprint_kib: 16,
operations: 1,
sleep_usec: 0,
skip_locking: false,
requests_per_sec: 10_000,
auto_rps: 0,
split_percent: None,
pipe_transfer_bytes: 0,
};
let stop = AtomicBool::new(false);
let progress = AtomicU64::new(0);
let outcome = std::thread::scope(|s| {
let runner = s.spawn(|| run(&config, &stop, &progress, None));
while progress.load(Ordering::Relaxed) < 50 {
core::hint::spin_loop();
}
stop.store(true, Ordering::Release);
runner.join().expect("run panicked")
});
let result = &outcome.whole_run;
assert!(
result.loop_count >= 50,
"RPS workers serviced requests: {}",
result.loop_count
);
assert!(
result.request.nr_samples > 0,
"RPS request-latency samples recorded"
);
}
#[test]
fn rps_injector_pacing_sleep_is_bounded_for_prompt_shutdown() {
let config = SchbenchConfig {
message_threads: 1,
worker_threads: 1,
cache_footprint_kib: 16,
operations: 1,
sleep_usec: 0,
skip_locking: false,
requests_per_sec: 1, auto_rps: 0,
split_percent: None,
pipe_transfer_bytes: 0,
};
let stop = AtomicBool::new(false);
let progress = AtomicU64::new(0);
let start = std::time::Instant::now();
std::thread::scope(|s| {
let runner = s.spawn(|| run(&config, &stop, &progress, None));
std::thread::sleep(std::time::Duration::from_millis(20));
stop.store(true, Ordering::Release);
let _ = runner.join().expect("run panicked");
});
let elapsed = start.elapsed();
assert!(
elapsed < std::time::Duration::from_millis(500),
"RPS injector joined in {elapsed:?}; the paced sleep must be bounded \
(interval ~1s — an unbounded sleep would block the join that long)",
);
}
#[test]
fn engine_rps_below_message_threads_falls_to_default() {
let config = SchbenchConfig {
message_threads: 2,
worker_threads: 1,
cache_footprint_kib: 16,
operations: 1,
sleep_usec: 0,
skip_locking: false,
requests_per_sec: 1, auto_rps: 0,
split_percent: None,
pipe_transfer_bytes: 0,
};
let stop = AtomicBool::new(false);
let progress = AtomicU64::new(0);
let outcome = std::thread::scope(|s| {
let runner = s.spawn(|| run(&config, &stop, &progress, None));
while progress.load(Ordering::Relaxed) < 50 {
core::hint::spin_loop();
}
stop.store(true, Ordering::Release);
runner.join().expect("run panicked")
});
assert!(
outcome.whole_run.loop_count >= 50,
"sub-1-per-thread RPS ran default-mode work: {}",
outcome.whole_run.loop_count
);
}
#[test]
fn auto_scale_grows_when_idle_and_shrinks_when_busy() {
let rate = AtomicUsize::new(100);
let hit = AtomicBool::new(false);
assert!(!scale_rps_for_busy(10.0, &rate, &hit, 80));
assert_eq!(rate.load(Ordering::Acquire), 300);
assert!(!hit.load(Ordering::Acquire));
let rate = AtomicUsize::new(300);
let hit = AtomicBool::new(false);
assert!(!scale_rps_for_busy(100.0, &rate, &hit, 20));
assert_eq!(rate.load(Ordering::Acquire), 90);
assert!(!hit.load(Ordering::Acquire));
let rate = AtomicUsize::new(123);
let hit = AtomicBool::new(false);
assert!(scale_rps_for_busy(50.0, &rate, &hit, 50));
assert_eq!(rate.load(Ordering::Acquire), 123);
assert!(hit.load(Ordering::Acquire));
}
#[test]
fn auto_scale_target_hit_uses_post_damping_delta() {
let rate = AtomicUsize::new(1000);
let hit = AtomicBool::new(false);
assert!(scale_rps_for_busy(90.0, &rate, &hit, 100));
assert!(hit.load(Ordering::Acquire));
let rate = AtomicUsize::new(1000);
let hit = AtomicBool::new(false);
assert!(scale_rps_for_busy(100.0, &rate, &hit, 95));
assert!(hit.load(Ordering::Acquire));
let rate = AtomicUsize::new(1000);
let hit = AtomicBool::new(true);
assert!(!scale_rps_for_busy(50.0, &rate, &hit, 50));
}
#[test]
fn reset_rps_accumulators_clears_both_in_lockstep() {
let mut whole = PlatStats::default();
whole.add_lat(500);
let mut per_epoch: std::collections::BTreeMap<u32, PlatStats> =
std::collections::BTreeMap::new();
per_epoch.entry(1).or_default().add_lat(500);
reset_rps_accumulators(&mut whole, &mut per_epoch);
assert_eq!(whole.sample_count(), 0, "whole-run rps reset");
assert!(per_epoch.is_empty(), "per-epoch rps reset in lockstep");
}
#[test]
fn control_loop_rps_sample_floors_sub_quantum_window() {
assert_eq!(control_loop_rps_sample(3, 0), None);
assert_eq!(control_loop_rps_sample(1, 1_000), None);
assert_eq!(control_loop_rps_sample(5, STOP_POLL_QUANTUM_NS - 1), None);
assert_eq!(control_loop_rps_sample(1, STOP_POLL_QUANTUM_NS), Some(20));
assert_eq!(
control_loop_rps_sample(1_000, STOP_POLL_QUANTUM_NS),
Some(20_000)
);
assert_eq!(control_loop_rps_sample(5_000, 1_000_000_000), Some(5_000));
}
#[test]
fn engine_auto_rps_mode_runs_and_terminates() {
let config = SchbenchConfig {
message_threads: 1,
worker_threads: 2,
cache_footprint_kib: 16,
operations: 1,
sleep_usec: 0,
skip_locking: false,
requests_per_sec: 10_000,
auto_rps: 50,
split_percent: None,
pipe_transfer_bytes: 0,
};
let stop = AtomicBool::new(false);
let progress = AtomicU64::new(0);
let outcome = std::thread::scope(|s| {
let runner = s.spawn(|| run(&config, &stop, &progress, None));
while progress.load(Ordering::Relaxed) < 50 {
core::hint::spin_loop();
}
stop.store(true, Ordering::Release);
runner.join().expect("run panicked")
});
assert!(
outcome.whole_run.loop_count >= 50,
"auto-RPS injector serviced requests: {}",
outcome.whole_run.loop_count
);
}
#[test]
fn engine_splits_stats_across_phase_epochs() {
let config = SchbenchConfig {
message_threads: 1,
worker_threads: 2,
cache_footprint_kib: 16,
operations: 1,
sleep_usec: 0,
skip_locking: false,
requests_per_sec: 0,
auto_rps: 0,
split_percent: None,
pipe_transfer_bytes: 0,
};
let stop = AtomicBool::new(false);
let progress = AtomicU64::new(0);
let epoch = AtomicU32::new(1); let outcome = std::thread::scope(|s| {
let runner = s.spawn(|| run(&config, &stop, &progress, Some(&epoch)));
while progress.load(Ordering::Relaxed) < 50 {
core::hint::spin_loop();
}
let after_phase1 = progress.load(Ordering::Relaxed);
epoch.store(2, Ordering::Release); while progress.load(Ordering::Relaxed) < after_phase1 + 50 {
core::hint::spin_loop();
}
stop.store(true, Ordering::Release);
runner.join().expect("run panicked")
});
let by_epoch: std::collections::BTreeMap<u32, &SchbenchPhaseStats> =
outcome.phases.iter().map(|(e, s)| (*e, s)).collect();
let p1 = by_epoch.get(&1).expect("phase 1 present");
let p2 = by_epoch.get(&2).expect("phase 2 present");
assert!(p1.loop_count > 0, "phase 1 ran cycles: {}", p1.loop_count);
assert!(p2.loop_count > 0, "phase 2 ran cycles: {}", p2.loop_count);
assert!(
p1.worker_pcount > 0,
"phase 1 worker run-delay split populated"
);
assert!(
p2.worker_pcount > 0,
"phase 2 worker run-delay split populated"
);
assert!(p1.msg_pcount > 0, "phase 1 msg run-delay split populated");
assert!(p2.msg_pcount > 0, "phase 2 msg run-delay split populated");
let loop_sum: u64 = outcome.phases.iter().map(|(_, s)| s.loop_count).sum();
assert_eq!(
loop_sum, outcome.whole_run.loop_count,
"per-phase loop_count partitions the whole-run count"
);
let phase_request_sum: u64 = outcome
.phases
.iter()
.map(|(_, s)| s.request.percentiles().nr_samples)
.sum();
assert_eq!(
outcome.whole_run.request.nr_samples, phase_request_sum,
"whole-run request count == Σ per-phase counts"
);
let phase_wakeup_sum: u64 = outcome
.phases
.iter()
.map(|(_, s)| s.wakeup.percentiles().nr_samples)
.sum();
assert_eq!(
outcome.whole_run.wakeup.nr_samples, phase_wakeup_sum,
"whole-run wakeup count == Σ per-phase counts"
);
assert!(
p1.rps.sample_count() <= 1,
"brief phase 1 gets at most one straddle rps tick, got {}",
p1.rps.sample_count()
);
let phase_rps_sum: u64 = outcome
.phases
.iter()
.map(|(_, s)| s.rps.sample_count())
.sum();
assert_eq!(
outcome.whole_run.rps.nr_samples, phase_rps_sum,
"whole-run rps count == Σ per-phase rps counts"
);
}
#[test]
fn engine_per_phase_rps_populates_and_sums_to_whole_run() {
let config = SchbenchConfig {
message_threads: 1,
worker_threads: 2,
cache_footprint_kib: 16,
operations: 1,
sleep_usec: 0,
skip_locking: false,
requests_per_sec: 0,
auto_rps: 0,
split_percent: None,
pipe_transfer_bytes: 0,
};
let stop = AtomicBool::new(false);
let progress = AtomicU64::new(0);
let epoch = AtomicU32::new(1);
let outcome = std::thread::scope(|s| {
let runner = s.spawn(|| run(&config, &stop, &progress, Some(&epoch)));
std::thread::sleep(std::time::Duration::from_millis(2200));
epoch.store(2, Ordering::Release);
std::thread::sleep(std::time::Duration::from_millis(2200));
stop.store(true, Ordering::Release);
runner.join().expect("run panicked")
});
let by_epoch: std::collections::BTreeMap<u32, &SchbenchPhaseStats> =
outcome.phases.iter().map(|(e, s)| (*e, s)).collect();
let p1 = by_epoch.get(&1).expect("phase 1 present");
let p2 = by_epoch.get(&2).expect("phase 2 present");
assert!(
p1.rps.sample_count() > 0,
"phase 1 per-phase rps populated: {}",
p1.rps.sample_count()
);
assert!(
p2.rps.sample_count() > 0,
"phase 2 per-phase rps populated: {}",
p2.rps.sample_count()
);
let phase_rps_sum: u64 = outcome
.phases
.iter()
.map(|(_, s)| s.rps.sample_count())
.sum();
assert_eq!(
phase_rps_sum, outcome.whole_run.rps.nr_samples,
"Σ per-epoch rps samples == whole-run rps samples"
);
}
#[test]
fn schbench_config_serde_roundtrips() {
let cfg = SchbenchConfig::default()
.message_threads(3)
.worker_threads(7)
.cache_footprint_kib(512)
.operations(9)
.sleep_usec(250)
.skip_locking(true)
.split_percent(Some(33))
.pipe_transfer_bytes(4096);
let json = serde_json::to_string(&cfg).expect("SchbenchConfig must serialize");
let back: SchbenchConfig =
serde_json::from_str(&json).expect("SchbenchConfig must deserialize");
assert_eq!(cfg, back, "config roundtrips unchanged");
}
#[test]
fn worktype_schbench_registration_and_serde() {
use crate::workload::WorkType;
let wt = WorkType::schbench(
SchbenchConfig::default()
.message_threads(2)
.worker_threads(4),
);
assert_eq!(wt.name(), "Schbench");
assert_eq!(
WorkType::from_name("Schbench"),
Some(WorkType::Schbench {
config: SchbenchConfig::default()
})
);
let json = serde_json::to_string(&wt).expect("WorkType::Schbench must serialize");
let back: WorkType =
serde_json::from_str(&json).expect("WorkType::Schbench must deserialize");
assert_eq!(wt, back);
}
#[test]
fn schbench_config_reachable_via_prelude() {
let cfg: crate::prelude::SchbenchConfig = crate::prelude::SchbenchConfig::default();
assert_eq!(cfg, SchbenchConfig::default());
}
#[test]
fn read_schedstat_raw_parses_own_and_handles_missing() {
let _own = read_schedstat_raw(gettid_self());
assert_eq!(
read_schedstat_raw(-1),
(0, 0),
"absent schedstat yields (0,0), no panic"
);
}
#[test]
fn schedstat_raw_parse_and_mean_pcount_guard() {
assert_eq!(parse_schedstat_raw("123456 50 5"), (50, 5));
assert_eq!(mean_sched_delay((50, 5)), 10); assert_eq!(mean_sched_delay((50, 0)), 0); assert_eq!(mean_sched_delay(parse_schedstat_raw("0 0 0")), 0); assert_eq!(mean_sched_delay((0, 5)), 0); }
#[test]
#[should_panic(expected = "schedstat field 3")]
fn parse_schedstat_raw_short_line_panics() {
parse_schedstat_raw("100 50");
}
#[test]
#[should_panic(expected = "schedstat field 2")]
fn parse_schedstat_raw_nonnumeric_panics() {
parse_schedstat_raw("alpha beta gamma");
}
}