Implement new benchmark engine
Sayan Nandan
10 months ago
parent
daf0f32c30
commit
250a2b3c16
@ -1,279 +0,0 @@
|
||||
/*
|
||||
* Created on Fri Nov 17 2023
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2023, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
use {
|
||||
crossbeam_channel::{unbounded, Receiver, Sender},
|
||||
std::{
|
||||
fmt,
|
||||
marker::PhantomData,
|
||||
thread::{self, JoinHandle},
|
||||
time::Instant,
|
||||
},
|
||||
};
|
||||
|
||||
pub type TaskPoolResult<T, Th> = Result<T, TaskpoolError<Th>>;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum TaskpoolError<Th: ThreadedTask> {
|
||||
InitError(Th::TaskWorkerInitError),
|
||||
BombardError(&'static str),
|
||||
WorkerError(Th::TaskWorkerWorkError),
|
||||
}
|
||||
|
||||
impl<Th: ThreadedTask> fmt::Display for TaskpoolError<Th>
|
||||
where
|
||||
Th::TaskWorkerInitError: fmt::Display,
|
||||
Th::TaskWorkerWorkError: fmt::Display,
|
||||
{
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
Self::InitError(e) => write!(f, "failed to init worker pool. {e}"),
|
||||
Self::BombardError(e) => write!(f, "failed to post work to pool. {e}"),
|
||||
Self::WorkerError(e) => write!(f, "failed running worker task. {e}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub trait ThreadedTask: Send + Sync + 'static {
|
||||
/// the per-thread item that does the actual work
|
||||
///
|
||||
/// NB: this is not to be confused with the actual underlying thread pool worker
|
||||
type TaskWorker: Send + Sync;
|
||||
/// when attempting initialization of the per-thread task worker, if an error is thrown, this is the type
|
||||
/// you're looking for
|
||||
type TaskWorkerInitError: Send + Sync;
|
||||
/// when attempting to run a single unit of work, if any error occurs this is the error type that is to be returned
|
||||
type TaskWorkerWorkError: Send + Sync;
|
||||
/// when attempting to close a worker, if an error occurs this is the error type that is returned
|
||||
type TaskWorkerTerminateError: Send + Sync;
|
||||
/// the task that is sent to each worker
|
||||
type TaskInput: Send + Sync;
|
||||
// fn
|
||||
/// initialize the worker
|
||||
fn initialize_worker(&self) -> Result<Self::TaskWorker, Self::TaskWorkerInitError>;
|
||||
/// drive the worker to complete a task and return the time
|
||||
fn drive_worker_timed(
|
||||
worker: &mut Self::TaskWorker,
|
||||
task: Self::TaskInput,
|
||||
) -> Result<(Instant, Instant), Self::TaskWorkerWorkError>;
|
||||
fn terminate_worker(
|
||||
&self,
|
||||
worker: &mut Self::TaskWorker,
|
||||
) -> Result<(), Self::TaskWorkerTerminateError>;
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct ThreadWorker<Th> {
|
||||
handle: JoinHandle<()>,
|
||||
_m: PhantomData<Th>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
enum WorkerTask<Th: ThreadedTask> {
|
||||
Task(Th::TaskInput),
|
||||
Exit,
|
||||
}
|
||||
|
||||
impl<Th: ThreadedTask> ThreadWorker<Th> {
|
||||
fn new(
|
||||
hl_worker: Th::TaskWorker,
|
||||
task_rx: Receiver<WorkerTask<Th>>,
|
||||
res_tx: Sender<Result<(Instant, Instant), Th::TaskWorkerWorkError>>,
|
||||
) -> Self {
|
||||
Self {
|
||||
handle: thread::spawn(move || {
|
||||
let mut worker = hl_worker;
|
||||
loop {
|
||||
let task = match task_rx.recv().unwrap() {
|
||||
WorkerTask::Exit => {
|
||||
drop(task_rx);
|
||||
return;
|
||||
}
|
||||
WorkerTask::Task(t) => t,
|
||||
};
|
||||
res_tx
|
||||
.send(Th::drive_worker_timed(&mut worker, task))
|
||||
.unwrap();
|
||||
}
|
||||
}),
|
||||
_m: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct Taskpool<Th: ThreadedTask> {
|
||||
workers: Vec<ThreadWorker<Th>>,
|
||||
_config: Th,
|
||||
task_tx: Sender<WorkerTask<Th>>,
|
||||
res_rx: Receiver<Result<(Instant, Instant), Th::TaskWorkerWorkError>>,
|
||||
record_real_start: Instant,
|
||||
record_real_stop: Instant,
|
||||
stat_run_avg_ns: f64,
|
||||
stat_run_tail_ns: u128,
|
||||
stat_run_head_ns: u128,
|
||||
}
|
||||
|
||||
// TODO(@ohsayan): prepare histogram for report; for now there's no use of the head and tail latencies
|
||||
#[derive(Default, Debug)]
|
||||
pub struct RuntimeStats {
|
||||
pub qps: f64,
|
||||
pub avg_per_query_ns: f64,
|
||||
pub head_ns: u128,
|
||||
pub tail_ns: u128,
|
||||
}
|
||||
|
||||
impl<Th: ThreadedTask> Taskpool<Th> {
|
||||
pub fn stat_avg(&self) -> f64 {
|
||||
self.stat_run_avg_ns
|
||||
}
|
||||
pub fn stat_tail(&self) -> u128 {
|
||||
self.stat_run_tail_ns
|
||||
}
|
||||
pub fn stat_head(&self) -> u128 {
|
||||
self.stat_run_head_ns
|
||||
}
|
||||
pub fn stat_elapsed(&self) -> u128 {
|
||||
self.record_real_stop
|
||||
.duration_since(self.record_real_start)
|
||||
.as_nanos()
|
||||
}
|
||||
}
|
||||
|
||||
fn qps(query_count: usize, time_taken_in_nanos: u128) -> f64 {
|
||||
const NANOS_PER_SECOND: u128 = 1_000_000_000;
|
||||
let time_taken_in_nanos_f64 = time_taken_in_nanos as f64;
|
||||
let query_count_f64 = query_count as f64;
|
||||
(query_count_f64 / time_taken_in_nanos_f64) * NANOS_PER_SECOND as f64
|
||||
}
|
||||
|
||||
impl<Th: ThreadedTask> Taskpool<Th> {
|
||||
pub fn new(size: usize, config: Th) -> TaskPoolResult<Self, Th> {
|
||||
let (task_tx, task_rx) = unbounded();
|
||||
let (res_tx, res_rx) = unbounded();
|
||||
let mut workers = Vec::with_capacity(size);
|
||||
for _ in 0..size {
|
||||
let con = config
|
||||
.initialize_worker()
|
||||
.map_err(TaskpoolError::InitError)?;
|
||||
workers.push(ThreadWorker::new(con, task_rx.clone(), res_tx.clone()));
|
||||
}
|
||||
Ok(Self {
|
||||
workers,
|
||||
_config: config,
|
||||
task_tx,
|
||||
res_rx,
|
||||
stat_run_avg_ns: 0.0,
|
||||
record_real_start: Instant::now(),
|
||||
record_real_stop: Instant::now(),
|
||||
stat_run_head_ns: u128::MAX,
|
||||
stat_run_tail_ns: u128::MIN,
|
||||
})
|
||||
}
|
||||
pub fn blocking_bombard(
|
||||
&mut self,
|
||||
vec: Vec<Th::TaskInput>,
|
||||
) -> TaskPoolResult<RuntimeStats, Th> {
|
||||
let expected = vec.len();
|
||||
let mut received = 0usize;
|
||||
for task in vec {
|
||||
match self.task_tx.send(WorkerTask::Task(task)) {
|
||||
Ok(()) => {}
|
||||
Err(_) => {
|
||||
// stop bombarding, we hit an error
|
||||
return Err(TaskpoolError::BombardError(
|
||||
"all worker threads exited. this indicates a catastrophic failure",
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
while received != expected {
|
||||
match self.res_rx.recv() {
|
||||
Err(_) => {
|
||||
// all workers exited. that is catastrophic
|
||||
return Err(TaskpoolError::BombardError(
|
||||
"detected all worker threads crashed during run check",
|
||||
));
|
||||
}
|
||||
Ok(r) => self.recompute_stats(&mut received, r)?,
|
||||
};
|
||||
}
|
||||
// compute stats
|
||||
let ret = Ok(RuntimeStats {
|
||||
qps: qps(received, self.stat_elapsed()),
|
||||
avg_per_query_ns: self.stat_avg(),
|
||||
head_ns: self.stat_head(),
|
||||
tail_ns: self.stat_tail(),
|
||||
});
|
||||
// reset stats
|
||||
self.stat_run_avg_ns = 0.0;
|
||||
self.record_real_start = Instant::now();
|
||||
self.record_real_stop = Instant::now();
|
||||
self.stat_run_head_ns = u128::MAX;
|
||||
self.stat_run_tail_ns = u128::MIN;
|
||||
// return
|
||||
ret
|
||||
}
|
||||
fn recompute_stats(
|
||||
&mut self,
|
||||
received: &mut usize,
|
||||
result: Result<(Instant, Instant), <Th as ThreadedTask>::TaskWorkerWorkError>,
|
||||
) -> Result<(), TaskpoolError<Th>> {
|
||||
*received += 1;
|
||||
let (start, stop) = match result {
|
||||
Ok(time) => time,
|
||||
Err(e) => return Err(TaskpoolError::WorkerError(e)),
|
||||
};
|
||||
// adjust real start
|
||||
if start < self.record_real_start {
|
||||
self.record_real_start = start;
|
||||
}
|
||||
if stop > self.record_real_stop {
|
||||
self.record_real_stop = stop;
|
||||
}
|
||||
let current_time = stop.duration_since(start).as_nanos();
|
||||
self.stat_run_avg_ns = self.stat_run_avg_ns
|
||||
+ ((current_time as f64 - self.stat_run_avg_ns) / *received as f64);
|
||||
if current_time > self.stat_run_tail_ns {
|
||||
self.stat_run_tail_ns = current_time;
|
||||
}
|
||||
if current_time < self.stat_run_head_ns {
|
||||
self.stat_run_head_ns = current_time;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl<Th: ThreadedTask> Drop for Taskpool<Th> {
|
||||
fn drop(&mut self) {
|
||||
for _ in 0..self.workers.len() {
|
||||
self.task_tx.send(WorkerTask::Exit).unwrap();
|
||||
}
|
||||
for worker in self.workers.drain(..) {
|
||||
worker.handle.join().unwrap()
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,406 @@
|
||||
/*
|
||||
* Created on Sun Nov 19 2023
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2023, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
use {
|
||||
crossbeam_channel::{unbounded, Receiver, Sender},
|
||||
std::{
|
||||
fmt::{self, Display},
|
||||
sync::atomic::{AtomicBool, AtomicU64, Ordering},
|
||||
thread::{self, JoinHandle},
|
||||
time::{Duration, Instant},
|
||||
},
|
||||
};
|
||||
|
||||
pub type BombardResult<T, Bt> = Result<T, BombardError<Bt>>;
|
||||
|
||||
/*
|
||||
state mgmt
|
||||
*/
|
||||
|
||||
#[derive(Debug)]
|
||||
/// The pool state. Be warned **ONLY ONE POOL AT A TIME!**
|
||||
struct GPState {
|
||||
current: AtomicU64,
|
||||
state: AtomicBool,
|
||||
occupied: AtomicBool,
|
||||
}
|
||||
|
||||
impl GPState {
|
||||
#[inline(always)]
|
||||
fn get() -> &'static Self {
|
||||
static STATE: GPState = GPState::zero();
|
||||
&STATE
|
||||
}
|
||||
const fn zero() -> Self {
|
||||
Self {
|
||||
current: AtomicU64::new(0),
|
||||
state: AtomicBool::new(true),
|
||||
occupied: AtomicBool::new(false),
|
||||
}
|
||||
}
|
||||
fn occupy(&self) {
|
||||
assert!(!self.occupied.swap(true, Ordering::Release));
|
||||
}
|
||||
fn vacate(&self) {
|
||||
assert!(self.occupied.swap(false, Ordering::Release));
|
||||
}
|
||||
fn guard<T>(f: impl FnOnce() -> T) -> T {
|
||||
let slf = Self::get();
|
||||
slf.occupy();
|
||||
let ret = f();
|
||||
slf.vacate();
|
||||
ret
|
||||
}
|
||||
fn post_failure(&self) {
|
||||
self.state.store(false, Ordering::Release)
|
||||
}
|
||||
fn post_target(&self, target: u64) {
|
||||
self.current.store(target, Ordering::Release)
|
||||
}
|
||||
/// WARNING: this is not atomic! only sensible to run a quiescent state
|
||||
fn post_reset(&self) {
|
||||
self.current.store(0, Ordering::Release);
|
||||
self.state.store(true, Ordering::Release);
|
||||
}
|
||||
fn update_target(&self) -> u64 {
|
||||
let mut current = self.current.load(Ordering::Acquire);
|
||||
loop {
|
||||
if current == 0 {
|
||||
return 0;
|
||||
}
|
||||
match self.current.compare_exchange(
|
||||
current,
|
||||
current - 1,
|
||||
Ordering::Release,
|
||||
Ordering::Acquire,
|
||||
) {
|
||||
Ok(last) => {
|
||||
return last;
|
||||
}
|
||||
Err(new) => {
|
||||
current = new;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
fn load_okay(&self) -> bool {
|
||||
self.state.load(Ordering::Acquire)
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
task spec
|
||||
*/
|
||||
|
||||
/// A threaded bombard task specification which drives a global pool of threads towards a common goal
|
||||
pub trait ThreadedBombardTask: Send + Sync + 'static {
|
||||
/// The per-task worker that is initialized once in every thread (not to be confused with the actual thread worker!)
|
||||
type Worker: Send + Sync;
|
||||
/// The task that the [`ThreadedBombardTask::TaskWorker`] performs
|
||||
type WorkerTask: Send + Sync;
|
||||
type WorkerTaskSpec: Clone + Send + Sync + 'static;
|
||||
/// Errors while running a task
|
||||
type WorkerTaskError: Send + Sync;
|
||||
/// Errors while initializing a task worker
|
||||
type WorkerInitError: Send + Sync;
|
||||
/// Initialize a task worker
|
||||
fn worker_init(&self) -> Result<Self::Worker, Self::WorkerInitError>;
|
||||
fn generate_task(spec: &Self::WorkerTaskSpec, current: u64) -> Self::WorkerTask;
|
||||
/// Drive a single subtask
|
||||
fn worker_drive_timed(
|
||||
worker: &mut Self::Worker,
|
||||
task: Self::WorkerTask,
|
||||
) -> Result<u128, Self::WorkerTaskError>;
|
||||
}
|
||||
|
||||
/*
|
||||
worker
|
||||
*/
|
||||
|
||||
#[derive(Debug)]
|
||||
enum WorkerResult<Bt: ThreadedBombardTask> {
|
||||
Completed(WorkerLocalStats),
|
||||
Errored(Bt::WorkerTaskError),
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct WorkerLocalStats {
|
||||
start: Instant,
|
||||
elapsed: u128,
|
||||
head: u128,
|
||||
tail: u128,
|
||||
}
|
||||
|
||||
impl WorkerLocalStats {
|
||||
fn new(start: Instant, elapsed: u128, head: u128, tail: u128) -> Self {
|
||||
Self {
|
||||
start,
|
||||
elapsed,
|
||||
head,
|
||||
tail,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
enum WorkerTask<Bt: ThreadedBombardTask> {
|
||||
Task(Bt::WorkerTaskSpec),
|
||||
Exit,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct Worker {
|
||||
handle: JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl Worker {
|
||||
fn start<Bt: ThreadedBombardTask>(
|
||||
id: usize,
|
||||
driver: Bt::Worker,
|
||||
rx_work: Receiver<WorkerTask<Bt>>,
|
||||
tx_res: Sender<WorkerResult<Bt>>,
|
||||
) -> Self {
|
||||
Self {
|
||||
handle: thread::Builder::new()
|
||||
.name(format!("worker-{id}"))
|
||||
.spawn(move || {
|
||||
let mut worker_driver = driver;
|
||||
'blocking_wait: loop {
|
||||
let task = match rx_work.recv().unwrap() {
|
||||
WorkerTask::Exit => return,
|
||||
WorkerTask::Task(spec) => spec,
|
||||
};
|
||||
// check global state
|
||||
let mut global_okay = GPState::get().load_okay();
|
||||
let mut global_position = GPState::get().update_target();
|
||||
// init local state
|
||||
let mut local_start = None;
|
||||
let mut local_elapsed = 0u128;
|
||||
let mut local_head = u128::MAX;
|
||||
let mut local_tail = 0;
|
||||
// bombard
|
||||
while (global_position != 0) & global_okay {
|
||||
let task = Bt::generate_task(&task, global_position);
|
||||
if local_start.is_none() {
|
||||
local_start = Some(Instant::now());
|
||||
}
|
||||
let this_elapsed =
|
||||
match Bt::worker_drive_timed(&mut worker_driver, task) {
|
||||
Ok(elapsed) => elapsed,
|
||||
Err(e) => {
|
||||
GPState::get().post_failure();
|
||||
tx_res.send(WorkerResult::Errored(e)).unwrap();
|
||||
continue 'blocking_wait;
|
||||
}
|
||||
};
|
||||
local_elapsed += this_elapsed;
|
||||
if this_elapsed < local_head {
|
||||
local_head = this_elapsed;
|
||||
}
|
||||
if this_elapsed > local_tail {
|
||||
local_tail = this_elapsed;
|
||||
}
|
||||
global_position = GPState::get().update_target();
|
||||
global_okay = GPState::get().load_okay();
|
||||
}
|
||||
if global_okay {
|
||||
// we're done
|
||||
tx_res
|
||||
.send(WorkerResult::Completed(WorkerLocalStats::new(
|
||||
local_start.unwrap(),
|
||||
local_elapsed,
|
||||
local_head,
|
||||
local_tail,
|
||||
)))
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
})
|
||||
.expect("failed to start thread"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
pool
|
||||
*/
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum BombardError<Bt: ThreadedBombardTask> {
|
||||
InitError(Bt::WorkerInitError),
|
||||
WorkerTaskError(Bt::WorkerTaskError),
|
||||
AllWorkersOffline,
|
||||
}
|
||||
|
||||
impl<Bt: ThreadedBombardTask> fmt::Display for BombardError<Bt>
|
||||
where
|
||||
Bt::WorkerInitError: fmt::Display,
|
||||
Bt::WorkerTaskError: Display,
|
||||
{
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
Self::AllWorkersOffline => write!(
|
||||
f,
|
||||
"bombard failed because all workers went offline indicating catastrophic failure"
|
||||
),
|
||||
Self::WorkerTaskError(e) => write!(f, "worker task failed. {e}"),
|
||||
Self::InitError(e) => write!(f, "worker init failed. {e}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct RuntimeStats {
|
||||
pub qps: f64,
|
||||
pub head: u128,
|
||||
pub tail: u128,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct BombardPool<Bt: ThreadedBombardTask> {
|
||||
workers: Vec<(Worker, Sender<WorkerTask<Bt>>)>,
|
||||
rx_res: Receiver<WorkerResult<Bt>>,
|
||||
_config: Bt,
|
||||
}
|
||||
|
||||
impl<Bt: ThreadedBombardTask> BombardPool<Bt> {
|
||||
fn qps(query_count: usize, time_taken_in_nanos: u128) -> f64 {
|
||||
const NANOS_PER_SECOND: u128 = 1_000_000_000;
|
||||
let time_taken_in_nanos_f64 = time_taken_in_nanos as f64;
|
||||
let query_count_f64 = query_count as f64;
|
||||
(query_count_f64 / time_taken_in_nanos_f64) * NANOS_PER_SECOND as f64
|
||||
}
|
||||
pub fn new(size: usize, config: Bt) -> BombardResult<Self, Bt> {
|
||||
assert_ne!(size, 0, "pool can't be empty");
|
||||
let mut workers = Vec::with_capacity(size);
|
||||
let (tx_res, rx_res) = unbounded();
|
||||
for id in 0..size {
|
||||
let (tx_work, rx_work) = unbounded();
|
||||
let driver = config.worker_init().map_err(BombardError::InitError)?;
|
||||
workers.push((Worker::start(id, driver, rx_work, tx_res.clone()), tx_work));
|
||||
}
|
||||
Ok(Self {
|
||||
workers,
|
||||
rx_res,
|
||||
_config: config,
|
||||
})
|
||||
}
|
||||
/// Bombard queries to the workers
|
||||
pub fn blocking_bombard(
|
||||
&mut self,
|
||||
task_description: Bt::WorkerTaskSpec,
|
||||
count: usize,
|
||||
) -> BombardResult<RuntimeStats, Bt> {
|
||||
GPState::guard(|| {
|
||||
GPState::get().post_target(count as _);
|
||||
let mut global_start = None;
|
||||
let mut global_stop = None;
|
||||
let mut global_head = u128::MAX;
|
||||
let mut global_tail = 0u128;
|
||||
let messages: Vec<<Bt as ThreadedBombardTask>::WorkerTaskSpec> =
|
||||
(0..self.workers.len())
|
||||
.into_iter()
|
||||
.map(|_| task_description.clone())
|
||||
.collect();
|
||||
for ((_, sender), msg) in self.workers.iter().zip(messages) {
|
||||
sender.send(WorkerTask::Task(msg)).unwrap();
|
||||
}
|
||||
// wait for all workers to complete
|
||||
let mut received = 0;
|
||||
while received != self.workers.len() {
|
||||
let results = match self.rx_res.recv() {
|
||||
Err(_) => return Err(BombardError::AllWorkersOffline),
|
||||
Ok(r) => r,
|
||||
};
|
||||
let WorkerLocalStats {
|
||||
start: this_start,
|
||||
elapsed,
|
||||
head,
|
||||
tail,
|
||||
} = match results {
|
||||
WorkerResult::Completed(r) => r,
|
||||
WorkerResult::Errored(e) => return Err(BombardError::WorkerTaskError(e)),
|
||||
};
|
||||
// update start if required
|
||||
match global_start.as_mut() {
|
||||
None => {
|
||||
global_start = Some(this_start);
|
||||
}
|
||||
Some(start) => {
|
||||
if this_start < *start {
|
||||
*start = this_start;
|
||||
}
|
||||
}
|
||||
}
|
||||
let this_task_stopped_at =
|
||||
this_start + Duration::from_nanos(elapsed.try_into().unwrap());
|
||||
match global_stop.as_mut() {
|
||||
None => {
|
||||
global_stop = Some(this_task_stopped_at);
|
||||
}
|
||||
Some(stop) => {
|
||||
if this_task_stopped_at > *stop {
|
||||
// this task stopped later than the previous one
|
||||
*stop = this_task_stopped_at;
|
||||
}
|
||||
}
|
||||
}
|
||||
if head < global_head {
|
||||
global_head = head;
|
||||
}
|
||||
if tail > global_tail {
|
||||
global_tail = tail;
|
||||
}
|
||||
received += 1;
|
||||
}
|
||||
// reset global pool state
|
||||
GPState::get().post_reset();
|
||||
// compute results
|
||||
let global_elapsed = global_stop
|
||||
.unwrap()
|
||||
.duration_since(global_start.unwrap())
|
||||
.as_nanos();
|
||||
Ok(RuntimeStats {
|
||||
qps: Self::qps(count, global_elapsed),
|
||||
head: global_head,
|
||||
tail: global_tail,
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl<Bt: ThreadedBombardTask> Drop for BombardPool<Bt> {
|
||||
fn drop(&mut self) {
|
||||
info!("taking all workers offline");
|
||||
for (_, sender) in self.workers.iter() {
|
||||
sender.send(WorkerTask::Exit).unwrap();
|
||||
}
|
||||
for (worker, _) in self.workers.drain(..) {
|
||||
worker.handle.join().unwrap();
|
||||
}
|
||||
info!("all workers now offline");
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue