use comfy_table::modifiers::UTF8_ROUND_CORNERS;
use comfy_table::presets::UTF8_FULL;
use comfy_table::{Attribute, Cell, ContentArrangement, Row, Table};
use statrs::statistics::{Max, Median, Min, OrderStatistics};
#[cfg(any(windows, unix))]
use std::cell::UnsafeCell;
use std::collections::HashMap;
use std::{
sync::{
Arc, Barrier,
atomic::{AtomicBool, Ordering},
},
thread,
time::Duration,
};
trait RwLock<T> {
fn new(v: T) -> Self;
fn read<F, R>(&self, f: F) -> R
where
F: FnOnce(&T) -> R;
fn write<F, R>(&self, f: F) -> R
where
F: FnOnce(&mut T) -> R;
fn name() -> &'static str;
}
impl<T> RwLock<T> for std::sync::RwLock<T> {
fn new(v: T) -> Self {
Self::new(v)
}
fn read<F, R>(&self, f: F) -> R
where
F: FnOnce(&T) -> R,
{
f(&*self.read().unwrap())
}
fn write<F, R>(&self, f: F) -> R
where
F: FnOnce(&mut T) -> R,
{
f(&mut *self.write().unwrap())
}
fn name() -> &'static str {
"std::sync::RwLock"
}
}
impl<T> RwLock<T> for parking_lot::RwLock<T> {
fn new(v: T) -> Self {
Self::new(v)
}
fn read<F, R>(&self, f: F) -> R
where
F: FnOnce(&T) -> R,
{
f(&*self.read())
}
fn write<F, R>(&self, f: F) -> R
where
F: FnOnce(&mut T) -> R,
{
f(&mut *self.write())
}
fn name() -> &'static str {
"parking_lot::RwLock"
}
}
impl<T: Copy> RwLock<T> for seqlock::SeqLock<T> {
fn new(v: T) -> Self {
Self::new(v)
}
fn read<F, R>(&self, f: F) -> R
where
F: FnOnce(&T) -> R,
{
f(&self.read())
}
fn write<F, R>(&self, f: F) -> R
where
F: FnOnce(&mut T) -> R,
{
f(&mut *self.lock_write())
}
fn name() -> &'static str {
"seqlock::SeqLock"
}
}
impl<T: Copy> RwLock<T> for snaplock::SnapshotLock<T> {
fn new(v: T) -> Self {
Self::new(v)
}
fn read<F, R>(&self, f: F) -> R
where
F: FnOnce(&T) -> R,
{
f(&self.read())
}
fn write<F, R>(&self, f: F) -> R
where
F: FnOnce(&mut T) -> R,
{
f(&mut *self.write())
}
fn name() -> &'static str {
"snaplock::SnapshotLock"
}
}
impl<T: Copy> RwLock<T> for crossbeam::sync::ShardedLock<T> {
fn new(v: T) -> Self {
Self::new(v)
}
fn read<F, R>(&self, f: F) -> R
where
F: FnOnce(&T) -> R,
{
f(&self.read().unwrap())
}
fn write<F, R>(&self, f: F) -> R
where
F: FnOnce(&mut T) -> R,
{
f(&mut self.write().unwrap())
}
fn name() -> &'static str {
"crossbeam::sync::ShardedLock"
}
}
#[cfg(not(windows))]
type SrwLock<T> = std::sync::RwLock<T>;
#[cfg(windows)]
use windows_sys::Win32::System::Threading::{
AcquireSRWLockExclusive, AcquireSRWLockShared, InitializeSRWLock, ReleaseSRWLockExclusive,
ReleaseSRWLockShared, SRWLOCK,
};
#[cfg(windows)]
struct SrwLock<T>(UnsafeCell<T>, UnsafeCell<SRWLOCK>);
#[cfg(windows)]
unsafe impl<T> Sync for SrwLock<T> {}
#[cfg(windows)]
unsafe impl<T: Send> Send for SrwLock<T> {}
#[cfg(windows)]
impl<T> RwLock<T> for SrwLock<T> {
fn new(v: T) -> Self {
let mut h: SRWLOCK = SRWLOCK {
Ptr: std::ptr::null_mut(),
};
unsafe {
InitializeSRWLock(&mut h);
}
SrwLock(UnsafeCell::new(v), UnsafeCell::new(h))
}
fn read<F, R>(&self, f: F) -> R
where
F: FnOnce(&T) -> R,
{
unsafe {
AcquireSRWLockShared(self.1.get());
let res = f(&*self.0.get());
ReleaseSRWLockShared(self.1.get());
res
}
}
fn write<F, R>(&self, f: F) -> R
where
F: FnOnce(&mut T) -> R,
{
unsafe {
AcquireSRWLockExclusive(self.1.get());
let res = f(&mut *self.0.get());
ReleaseSRWLockExclusive(self.1.get());
res
}
}
fn name() -> &'static str {
"windows_sys_srwlock"
}
}
#[cfg(not(unix))]
type PthreadRwLock<T> = std::sync::RwLock<T>;
#[cfg(unix)]
struct PthreadRwLock<T>(UnsafeCell<T>, UnsafeCell<libc::pthread_rwlock_t>);
#[cfg(unix)]
unsafe impl<T> Sync for PthreadRwLock<T> {}
#[cfg(unix)]
impl<T> RwLock<T> for PthreadRwLock<T> {
fn new(v: T) -> Self {
PthreadRwLock(
UnsafeCell::new(v),
UnsafeCell::new(libc::PTHREAD_RWLOCK_INITIALIZER),
)
}
fn read<F, R>(&self, f: F) -> R
where
F: FnOnce(&T) -> R,
{
unsafe {
libc::pthread_rwlock_rdlock(self.1.get());
let res = f(&*self.0.get());
libc::pthread_rwlock_unlock(self.1.get());
res
}
}
fn write<F, R>(&self, f: F) -> R
where
F: FnOnce(&mut T) -> R,
{
unsafe {
libc::pthread_rwlock_wrlock(self.1.get());
let res = f(&mut *self.0.get());
libc::pthread_rwlock_unlock(self.1.get());
res
}
}
fn name() -> &'static str {
"pthread_rwlock_t"
}
}
#[cfg(unix)]
impl<T> Drop for PthreadRwLock<T> {
fn drop(&mut self) {
unsafe {
libc::pthread_rwlock_destroy(self.1.get());
}
}
}
fn run_benchmark<M: RwLock<f64> + Send + Sync + 'static>(
num_writer_threads: usize,
num_reader_threads: usize,
work_per_critical_section: usize,
work_between_critical_sections: usize,
seconds_per_test: usize,
) -> (Vec<usize>, Vec<usize>) {
println!(
"Running benchmark w{num_writer_threads}r{num_reader_threads} {}",
M::name()
);
let lock = Arc::new(([0u8; 300], M::new(0.0), [0u8; 300]));
let keep_going = Arc::new(AtomicBool::new(true));
let barrier = Arc::new(Barrier::new(num_reader_threads + num_writer_threads));
let mut writers = vec![];
let mut readers = vec![];
for _ in 0..num_writer_threads {
let barrier = barrier.clone();
let lock = lock.clone();
let keep_going = keep_going.clone();
writers.push(thread::spawn(move || {
let mut local_value = 0.0;
let mut value = 0.0;
let mut iterations = 0usize;
barrier.wait();
while keep_going.load(Ordering::Relaxed) {
lock.1.write(|shared_value| {
for _ in 0..work_per_critical_section {
*shared_value += value;
*shared_value *= 1.01;
value = *shared_value;
}
});
for _ in 0..work_between_critical_sections {
local_value += value;
local_value *= 1.01;
value = local_value;
}
iterations += 1;
}
(iterations, value)
}));
}
for _ in 0..num_reader_threads {
let barrier = barrier.clone();
let lock = lock.clone();
let keep_going = keep_going.clone();
readers.push(thread::spawn(move || {
let mut local_value = 0.0;
let mut value = 0.0;
let mut iterations = 0usize;
barrier.wait();
while keep_going.load(Ordering::Relaxed) {
lock.1.read(|shared_value| {
for _ in 0..work_per_critical_section {
local_value += value;
local_value *= *shared_value;
value = local_value;
}
});
for _ in 0..work_between_critical_sections {
local_value += value;
local_value *= 1.01;
value = local_value;
}
iterations += 1;
}
(iterations, value)
}));
}
thread::sleep(Duration::new(seconds_per_test as u64, 0));
keep_going.store(false, Ordering::Relaxed);
let run_writers = writers
.into_iter()
.map(|x| x.join().unwrap().0)
.collect::<Vec<usize>>();
let run_readers = readers
.into_iter()
.map(|x| x.join().unwrap().0)
.collect::<Vec<usize>>();
(run_writers, run_readers)
}
macro_rules! run_benches {
(
threads: [
$(w=$writers:expr; r=$readers:expr),*
],
work_critical: $wcritical:expr,
work_between: $wbetween:expr,
seconds: $seconds:expr,
scores: $scores:ident,
{ $($(?if $predicate:expr =>)? $name:expr; $n:ty),* }
) => {
let mut $scores: HashMap<(usize, usize), Vec<(&'static str, Vec<usize>, Vec<usize>)>> = HashMap::default();
let writers_readers: Vec<(usize, usize)> = vec![$(
($writers, $readers),
)*];
$(
if true $(&& $predicate)? {
for key @ (writers, readers) in writers_readers.clone().into_iter() {
let (w, r) = run_benchmark::<$n>(
writers,
readers,
$wcritical,
$wbetween,
$seconds,
);
match $scores.get_mut(&key) {
Some(s) => { s.push(($name, w, r)); }
None => { $scores.insert(key, vec![($name, w, r)]); }
}
}
}
)*
};
}
struct RunStats {
name: String,
wmin: f64,
wmedian: f64,
wp99: f64,
wmax: f64,
rmin: f64,
rmedian: f64,
rp99: f64,
rmax: f64,
}
impl RunStats {
pub fn new<W: IntoIterator<Item = usize>, R: IntoIterator<Item = usize>>(
name: String,
writer: W,
reader: R,
) -> Self {
let wfloats = writer.into_iter().map(|x| x as f64).collect::<Vec<f64>>();
let mut wstats = statrs::statistics::Data::new(wfloats);
let rfloats = reader.into_iter().map(|x| x as f64).collect::<Vec<f64>>();
let mut rstats = statrs::statistics::Data::new(rfloats);
Self {
name,
wmin: wstats.min(),
wmedian: wstats.median(),
wmax: wstats.max(),
wp99: wstats.percentile(99),
rmin: rstats.min(),
rmedian: rstats.median(),
rmax: rstats.max(),
rp99: rstats.percentile(99),
}
}
pub fn header() -> Row {
vec![
Cell::new("Name").add_attribute(Attribute::Bold),
Cell::new("wMin").add_attribute(Attribute::Bold),
Cell::new("wMax").add_attribute(Attribute::Bold),
Cell::new("rMin").add_attribute(Attribute::Bold),
Cell::new("rMax").add_attribute(Attribute::Bold),
]
.into()
}
pub fn row(self) -> Row {
vec![
Cell::new(self.name).add_attribute(Attribute::Bold),
Cell::new(self.wmin),
Cell::new(self.wmax),
Cell::new(self.rmin),
Cell::new(self.rmax),
]
.into()
}
}
fn render_scores(scores: HashMap<(usize, usize), Vec<(&'static str, Vec<usize>, Vec<usize>)>>) {
for ((w, r), scores) in scores.into_iter() {
let mut table = Table::new();
table
.load_preset(UTF8_FULL)
.set_content_arrangement(ContentArrangement::Dynamic)
.set_header(RunStats::header());
for (lock, writer_scores, reader_scores) in scores {
table.add_row(
RunStats::new(format!("w{w}r{r} / {lock}"), writer_scores, reader_scores).row(),
);
}
println!("{table}");
}
}
fn main() {
run_benches! {
threads: [
w=10; r=10,
w=5; r=50,
w=50; r=5
],
work_critical: 10000,
work_between: 5000,
seconds: 10,
scores: scores,
{
"parking_lot::RwLock"; parking_lot::RwLock<f64>,
"std::sync::RwLock"; std::sync::RwLock<f64>,
"seqlock::SeqLock"; seqlock::SeqLock<f64>,
"snaplock::SnapshotLock"; snaplock::SnapshotLock<f64>,
"crossbeam::sync::ShardedLock"; crossbeam::sync::ShardedLock<f64>,
?if cfg!(windows) => "SrwLock"; SrwLock<f64>,
?if cfg!(unix) => "PthreadRwLock"; PthreadRwLock<f64>
}
}
render_scores(scores)
}