Rename aquatic_load_tester to aquatic_bencher

crates/bencher/src/common.rs

@@ -0,0 +1,258 @@
+use std::{fmt::Display, ops::Range, thread::available_parallelism};
+
+use itertools::Itertools;
+
+#[derive(Debug, Clone)]
+pub struct TaskSetCpuList(pub Vec<TaskSetCpuIndicator>);
+
+impl TaskSetCpuList {
+    pub fn as_cpu_list(&self) -> String {
+        let indicator = self.0.iter().map(|indicator| match indicator {
+            TaskSetCpuIndicator::Single(i) => i.to_string(),
+            TaskSetCpuIndicator::Range(range) => {
+                format!(
+                    "{}-{}",
+                    range.start,
+                    range.clone().into_iter().last().unwrap()
+                )
+            }
+        });
+
+        Itertools::intersperse_with(indicator, || ",".to_string())
+            .into_iter()
+            .collect()
+    }
+
+    pub fn new(
+        mode: CpuMode,
+        direction: CpuDirection,
+        requested_cpus: usize,
+    ) -> anyhow::Result<Self> {
+        let available_parallelism: usize = available_parallelism()?.into();
+
+        Ok(Self::new_with_available_parallelism(
+            available_parallelism,
+            mode,
+            direction,
+            requested_cpus,
+        ))
+    }
+
+    fn new_with_available_parallelism(
+        available_parallelism: usize,
+        mode: CpuMode,
+        direction: CpuDirection,
+        requested_cpus: usize,
+    ) -> Self {
+        match direction {
+            CpuDirection::Asc => match mode {
+                CpuMode::Split => {
+                    let middle = available_parallelism / 2;
+
+                    let range_a = 0..(middle.min(requested_cpus));
+                    let range_b = middle..(available_parallelism.min(middle + requested_cpus));
+
+                    Self(vec![
+                        range_a.try_into().unwrap(),
+                        range_b.try_into().unwrap(),
+                    ])
+                }
+                CpuMode::All => {
+                    let range = 0..(available_parallelism.min(requested_cpus));
+
+                    Self(vec![range.try_into().unwrap()])
+                }
+            },
+            CpuDirection::Desc => match mode {
+                CpuMode::Split => {
+                    let middle = available_parallelism / 2;
+
+                    let range_a = middle.saturating_sub(requested_cpus)..middle;
+                    let range_b = available_parallelism
+                        .saturating_sub(requested_cpus)
+                        .max(middle)..available_parallelism;
+
+                    Self(vec![
+                        range_a.try_into().unwrap(),
+                        range_b.try_into().unwrap(),
+                    ])
+                }
+                CpuMode::All => {
+                    let range =
+                        available_parallelism.saturating_sub(requested_cpus)..available_parallelism;
+
+                    Self(vec![range.try_into().unwrap()])
+                }
+            },
+        }
+    }
+}
+
+impl TryFrom<Vec<Range<usize>>> for TaskSetCpuList {
+    type Error = String;
+
+    fn try_from(value: Vec<Range<usize>>) -> Result<Self, Self::Error> {
+        let mut output = Vec::new();
+
+        for range in value {
+            output.push(range.try_into()?);
+        }
+
+        Ok(Self(output))
+    }
+}
+
+#[derive(Debug, Clone)]
+pub enum TaskSetCpuIndicator {
+    Single(usize),
+    Range(Range<usize>),
+}
+
+impl TryFrom<Range<usize>> for TaskSetCpuIndicator {
+    type Error = String;
+
+    fn try_from(value: Range<usize>) -> Result<Self, Self::Error> {
+        match value.len() {
+            0 => Err("Empty ranges not supported".into()),
+            1 => Ok(TaskSetCpuIndicator::Single(value.start)),
+            _ => Ok(TaskSetCpuIndicator::Range(value)),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, clap::ValueEnum)]
+pub enum CpuMode {
+    Split,
+    All,
+}
+
+impl Display for CpuMode {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::All => f.write_str("all"),
+            Self::Split => f.write_str("split"),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub enum CpuDirection {
+    Asc,
+    Desc,
+}
+
+pub fn simple_load_test_runs(cpu_mode: CpuMode, workers: &[usize]) -> Vec<(usize, TaskSetCpuList)> {
+    workers
+        .into_iter()
+        .copied()
+        .map(|workers| {
+            (
+                workers,
+                TaskSetCpuList::new(cpu_mode, CpuDirection::Desc, workers).unwrap(),
+            )
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_task_set_cpu_list_split_asc() {
+        let f = TaskSetCpuList::new_with_available_parallelism;
+
+        assert_eq!(
+            f(8, CpuMode::Split, CpuDirection::Asc, 1).as_cpu_list(),
+            "0,4"
+        );
+        assert_eq!(
+            f(8, CpuMode::Split, CpuDirection::Asc, 2).as_cpu_list(),
+            "0-1,4-5"
+        );
+        assert_eq!(
+            f(8, CpuMode::Split, CpuDirection::Asc, 4).as_cpu_list(),
+            "0-3,4-7"
+        );
+        assert_eq!(
+            f(8, CpuMode::Split, CpuDirection::Asc, 8).as_cpu_list(),
+            "0-3,4-7"
+        );
+        assert_eq!(
+            f(8, CpuMode::Split, CpuDirection::Asc, 9).as_cpu_list(),
+            "0-3,4-7"
+        );
+    }
+
+    #[test]
+    fn test_task_set_cpu_list_split_desc() {
+        let f = TaskSetCpuList::new_with_available_parallelism;
+
+        assert_eq!(
+            f(8, CpuMode::Split, CpuDirection::Desc, 1).as_cpu_list(),
+            "3,7"
+        );
+        assert_eq!(
+            f(8, CpuMode::Split, CpuDirection::Desc, 2).as_cpu_list(),
+            "2-3,6-7"
+        );
+        assert_eq!(
+            f(8, CpuMode::Split, CpuDirection::Desc, 4).as_cpu_list(),
+            "0-3,4-7"
+        );
+        assert_eq!(
+            f(8, CpuMode::Split, CpuDirection::Desc, 8).as_cpu_list(),
+            "0-3,4-7"
+        );
+        assert_eq!(
+            f(8, CpuMode::Split, CpuDirection::Desc, 9).as_cpu_list(),
+            "0-3,4-7"
+        );
+    }
+
+    #[test]
+    fn test_task_set_cpu_list_all_asc() {
+        let f = TaskSetCpuList::new_with_available_parallelism;
+
+        assert_eq!(f(8, CpuMode::All, CpuDirection::Asc, 1).as_cpu_list(), "0");
+        assert_eq!(
+            f(8, CpuMode::All, CpuDirection::Asc, 2).as_cpu_list(),
+            "0-1"
+        );
+        assert_eq!(
+            f(8, CpuMode::All, CpuDirection::Asc, 4).as_cpu_list(),
+            "0-3"
+        );
+        assert_eq!(
+            f(8, CpuMode::All, CpuDirection::Asc, 8).as_cpu_list(),
+            "0-7"
+        );
+        assert_eq!(
+            f(8, CpuMode::All, CpuDirection::Asc, 9).as_cpu_list(),
+            "0-7"
+        );
+    }
+
+    #[test]
+    fn test_task_set_cpu_list_all_desc() {
+        let f = TaskSetCpuList::new_with_available_parallelism;
+
+        assert_eq!(f(8, CpuMode::All, CpuDirection::Desc, 1).as_cpu_list(), "7");
+        assert_eq!(
+            f(8, CpuMode::All, CpuDirection::Desc, 2).as_cpu_list(),
+            "6-7"
+        );
+        assert_eq!(
+            f(8, CpuMode::All, CpuDirection::Desc, 4).as_cpu_list(),
+            "4-7"
+        );
+        assert_eq!(
+            f(8, CpuMode::All, CpuDirection::Desc, 8).as_cpu_list(),
+            "0-7"
+        );
+        assert_eq!(
+            f(8, CpuMode::All, CpuDirection::Desc, 9).as_cpu_list(),
+            "0-7"
+        );
+    }
+}

crates/bencher/src/main.rs

@@ -0,0 +1,42 @@
+pub mod common;
+pub mod protocols;
+pub mod run;
+pub mod set;
+
+use clap::{Parser, Subcommand};
+use common::CpuMode;
+
+#[derive(Parser)]
+#[command(author, version, about)]
+struct Args {
+    /// How to choose which virtual CPUs to allow trackers and load test
+    /// executables on
+    #[arg(long, default_value_t = CpuMode::Split)]
+    cpu_mode: CpuMode,
+    /// Minimum number of tracker cpu cores to run benchmarks for
+    #[arg(long)]
+    min_cores: Option<usize>,
+    /// Maximum number of tracker cpu cores to run benchmarks for
+    #[arg(long)]
+    max_cores: Option<usize>,
+    #[command(subcommand)]
+    command: Command,
+}
+
+#[derive(Subcommand)]
+enum Command {
+    /// Benchmark UDP BitTorrent trackers aquatic_udp and opentracker
+    #[cfg(feature = "udp")]
+    Udp(protocols::udp::UdpCommand),
+}
+
+fn main() {
+    let args = Args::parse();
+
+    match args.command {
+        #[cfg(feature = "udp")]
+        Command::Udp(command) => command
+            .run(args.cpu_mode, args.min_cores, args.max_cores)
+            .unwrap(),
+    }
+}

crates/bencher/src/protocols/mod.rs

@@ -0,0 +1,2 @@
+#[cfg(feature = "udp")]
+pub mod udp;

crates/bencher/src/protocols/udp.rs

@@ -0,0 +1,332 @@
+use std::{
+    io::Write,
+    path::PathBuf,
+    process::{Child, Command, Stdio},
+    rc::Rc,
+};
+
+use clap::Parser;
+use indexmap::{indexmap, IndexMap};
+use tempfile::NamedTempFile;
+
+use crate::{
+    common::{simple_load_test_runs, CpuMode, TaskSetCpuList},
+    run::ProcessRunner,
+    set::{run_sets, SetConfig, Tracker},
+};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum UdpTracker {
+    Aquatic,
+    OpenTracker,
+}
+
+impl Tracker for UdpTracker {
+    fn name(&self) -> String {
+        match self {
+            Self::Aquatic => "aquatic_udp".into(),
+            Self::OpenTracker => "opentracker".into(),
+        }
+    }
+}
+
+#[derive(Parser, Debug)]
+pub struct UdpCommand {
+    /// Path to aquatic_udp_load_test binary
+    #[arg(long, default_value = "./target/release-debug/aquatic_udp_load_test")]
+    load_test: PathBuf,
+    /// Path to aquatic_udp binary
+    #[arg(long, default_value = "./target/release-debug/aquatic_udp")]
+    aquatic: PathBuf,
+    /// Path to opentracker binary
+    #[arg(long, default_value = "opentracker")]
+    opentracker: PathBuf,
+}
+
+impl UdpCommand {
+    pub fn run(
+        &self,
+        cpu_mode: CpuMode,
+        min_cores: Option<usize>,
+        max_cores: Option<usize>,
+    ) -> anyhow::Result<()> {
+        let mut sets = self.sets(cpu_mode);
+
+        if let Some(min_cores) = min_cores {
+            sets = sets.into_iter().filter(|(k, _)| *k >= min_cores).collect();
+        }
+        if let Some(max_cores) = max_cores {
+            sets = sets.into_iter().filter(|(k, _)| *k <= max_cores).collect();
+        }
+
+        run_sets(self, cpu_mode, sets, |workers| {
+            Box::new(AquaticUdpLoadTestRunner { workers })
+        });
+
+        Ok(())
+    }
+
+    fn sets(&self, cpu_mode: CpuMode) -> IndexMap<usize, SetConfig<UdpCommand, UdpTracker>> {
+        indexmap::indexmap! {
+            1 => SetConfig {
+                implementations: indexmap! {
+                    UdpTracker::Aquatic => vec![
+                        AquaticUdpRunner::new(1, 1),
+                    ],
+                    UdpTracker::OpenTracker => vec![
+                        OpenTrackerUdpRunner::new(0), // Handle requests within event loop
+                        OpenTrackerUdpRunner::new(1),
+                        OpenTrackerUdpRunner::new(2),
+                    ],
+                },
+                load_test_runs: simple_load_test_runs(cpu_mode, &[1, 2, 4, 6]),
+            },
+            2 => SetConfig {
+                implementations: indexmap! {
+                    UdpTracker::Aquatic => vec![
+                        AquaticUdpRunner::new(1, 1),
+                        AquaticUdpRunner::new(2, 1),
+                        AquaticUdpRunner::new(3, 1),
+                    ],
+                    UdpTracker::OpenTracker => vec![
+                        OpenTrackerUdpRunner::new(2),
+                        OpenTrackerUdpRunner::new(4),
+                    ],
+                },
+                load_test_runs: simple_load_test_runs(cpu_mode, &[1, 2, 4, 6]),
+            },
+            3 => SetConfig {
+                implementations: indexmap! {
+                    UdpTracker::Aquatic => vec![
+                        AquaticUdpRunner::new(2, 1),
+                        AquaticUdpRunner::new(3, 1),
+                    ],
+                    UdpTracker::OpenTracker => vec![
+                        OpenTrackerUdpRunner::new(3),
+                        OpenTrackerUdpRunner::new(6),
+                    ],
+                },
+                load_test_runs: simple_load_test_runs(cpu_mode, &[4, 6, 8]),
+            },
+            4 => SetConfig {
+                implementations: indexmap! {
+                    UdpTracker::Aquatic => vec![
+                        AquaticUdpRunner::new(3, 1),
+                        AquaticUdpRunner::new(6, 1),
+                    ],
+                    UdpTracker::OpenTracker => vec![
+                        OpenTrackerUdpRunner::new(4),
+                        OpenTrackerUdpRunner::new(8),
+                    ],
+                },
+                load_test_runs: simple_load_test_runs(cpu_mode, &[4, 6, 8]),
+            },
+            6 => SetConfig {
+                implementations: indexmap! {
+                    UdpTracker::Aquatic => vec![
+                        AquaticUdpRunner::new(5, 1),
+                        AquaticUdpRunner::new(10, 1),
+                        AquaticUdpRunner::new(4, 2),
+                        AquaticUdpRunner::new(8, 2),
+                    ],
+                    UdpTracker::OpenTracker => vec![
+                        OpenTrackerUdpRunner::new(6),
+                        OpenTrackerUdpRunner::new(12),
+                    ],
+                },
+                load_test_runs: simple_load_test_runs(cpu_mode, &[4, 6, 8, 12]),
+            },
+            8 => SetConfig {
+                implementations: indexmap! {
+                    UdpTracker::Aquatic => vec![
+                        AquaticUdpRunner::new(7, 1),
+                        AquaticUdpRunner::new(14, 1),
+                        AquaticUdpRunner::new(6, 2),
+                        AquaticUdpRunner::new(12, 2),
+                    ],
+                    UdpTracker::OpenTracker => vec![
+                        OpenTrackerUdpRunner::new(8),
+                        OpenTrackerUdpRunner::new(16),
+                    ],
+                },
+                load_test_runs: simple_load_test_runs(cpu_mode, &[4, 8, 12]),
+            },
+            12 => SetConfig {
+                implementations: indexmap! {
+                    UdpTracker::Aquatic => vec![
+                        AquaticUdpRunner::new(11, 1),
+                        AquaticUdpRunner::new(22, 1),
+                        AquaticUdpRunner::new(10, 2),
+                        AquaticUdpRunner::new(20, 2),
+                        AquaticUdpRunner::new(9, 3),
+                        AquaticUdpRunner::new(18, 3),
+                    ],
+                    UdpTracker::OpenTracker => vec![
+                        OpenTrackerUdpRunner::new(12),
+                        OpenTrackerUdpRunner::new(24),
+                    ],
+                },
+                load_test_runs: simple_load_test_runs(cpu_mode, &[4, 8, 12, 16]),
+            },
+            16 => SetConfig {
+                implementations: indexmap! {
+                    UdpTracker::Aquatic => vec![
+                        AquaticUdpRunner::new(15, 1),
+                        AquaticUdpRunner::new(30, 1),
+                        AquaticUdpRunner::new(15, 2),
+                        AquaticUdpRunner::new(30, 2),
+                        AquaticUdpRunner::new(13, 3),
+                        AquaticUdpRunner::new(26, 3),
+                        AquaticUdpRunner::new(12, 4),
+                        AquaticUdpRunner::new(24, 4),
+                    ],
+                    UdpTracker::OpenTracker => vec![
+                        OpenTrackerUdpRunner::new(16),
+                        OpenTrackerUdpRunner::new(32),
+                    ],
+                },
+                load_test_runs: simple_load_test_runs(cpu_mode, &[4, 8, 12, 16]),
+            },
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+struct AquaticUdpRunner {
+    socket_workers: usize,
+    swarm_workers: usize,
+}
+
+impl AquaticUdpRunner {
+    fn new(
+        socket_workers: usize,
+        swarm_workers: usize,
+    ) -> Rc<dyn ProcessRunner<Command = UdpCommand>> {
+        Rc::new(Self {
+            socket_workers,
+            swarm_workers,
+        })
+    }
+}
+
+impl ProcessRunner for AquaticUdpRunner {
+    type Command = UdpCommand;
+
+    fn run(
+        &self,
+        command: &Self::Command,
+        vcpus: &TaskSetCpuList,
+        tmp_file: &mut NamedTempFile,
+    ) -> anyhow::Result<Child> {
+        let mut c = aquatic_udp::config::Config::default();
+
+        c.socket_workers = self.socket_workers;
+        c.swarm_workers = self.swarm_workers;
+
+        let c = toml::to_string_pretty(&c)?;
+
+        tmp_file.write_all(c.as_bytes())?;
+
+        Ok(Command::new("taskset")
+            .arg("--cpu-list")
+            .arg(vcpus.as_cpu_list())
+            .arg(&command.aquatic)
+            .arg("-c")
+            .arg(tmp_file.path())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .spawn()?)
+    }
+
+    fn keys(&self) -> IndexMap<String, String> {
+        indexmap! {
+            "socket workers".to_string() => self.socket_workers.to_string(),
+            "swarm workers".to_string() => self.swarm_workers.to_string(),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+struct OpenTrackerUdpRunner {
+    workers: usize,
+}
+
+impl OpenTrackerUdpRunner {
+    fn new(workers: usize) -> Rc<dyn ProcessRunner<Command = UdpCommand>> {
+        Rc::new(Self { workers })
+    }
+}
+
+impl ProcessRunner for OpenTrackerUdpRunner {
+    type Command = UdpCommand;
+
+    fn run(
+        &self,
+        command: &Self::Command,
+        vcpus: &TaskSetCpuList,
+        tmp_file: &mut NamedTempFile,
+    ) -> anyhow::Result<Child> {
+        writeln!(
+            tmp_file,
+            "listen.udp.workers {}\nlisten.udp 0.0.0.0:3000",
+            self.workers
+        )?;
+
+        Ok(Command::new("taskset")
+            .arg("--cpu-list")
+            .arg(vcpus.as_cpu_list())
+            .arg(&command.opentracker)
+            .arg("-f")
+            .arg(tmp_file.path())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .spawn()?)
+    }
+
+    fn keys(&self) -> IndexMap<String, String> {
+        indexmap! {
+            "workers".to_string() => self.workers.to_string(),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+struct AquaticUdpLoadTestRunner {
+    workers: usize,
+}
+
+impl ProcessRunner for AquaticUdpLoadTestRunner {
+    type Command = UdpCommand;
+
+    fn run(
+        &self,
+        command: &Self::Command,
+        vcpus: &TaskSetCpuList,
+        tmp_file: &mut NamedTempFile,
+    ) -> anyhow::Result<Child> {
+        let mut c = aquatic_udp_load_test::config::Config::default();
+
+        c.workers = self.workers as u8;
+        c.duration = 60;
+
+        let c = toml::to_string_pretty(&c)?;
+
+        tmp_file.write_all(c.as_bytes())?;
+
+        Ok(Command::new("taskset")
+            .arg("--cpu-list")
+            .arg(vcpus.as_cpu_list())
+            .arg(&command.load_test)
+            .arg("-c")
+            .arg(tmp_file.path())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .spawn()?)
+    }
+
+    fn keys(&self) -> IndexMap<String, String> {
+        indexmap! {
+            "workers".to_string() => self.workers.to_string(),
+        }
+    }
+}

crates/bencher/src/run.rs

@@ -0,0 +1,316 @@
+use std::{
+    process::{Child, Command},
+    rc::Rc,
+    str::FromStr,
+    time::Duration,
+};
+
+use indexmap::IndexMap;
+use itertools::Itertools;
+use nonblock::NonBlockingReader;
+use once_cell::sync::Lazy;
+use regex::Regex;
+use tempfile::NamedTempFile;
+
+use crate::common::TaskSetCpuList;
+
+pub trait ProcessRunner: ::std::fmt::Debug {
+    type Command;
+
+    fn run(
+        &self,
+        command: &Self::Command,
+        vcpus: &TaskSetCpuList,
+        tmp_file: &mut NamedTempFile,
+    ) -> anyhow::Result<Child>;
+
+    fn keys(&self) -> IndexMap<String, String>;
+
+    fn info(&self) -> String {
+        self.keys()
+            .into_iter()
+            .map(|(k, v)| format!("{}: {}", k, v))
+            .join(", ")
+    }
+}
+
+#[derive(Debug)]
+pub struct RunConfig<C> {
+    pub tracker_runner: Rc<dyn ProcessRunner<Command = C>>,
+    pub tracker_vcpus: TaskSetCpuList,
+    pub load_test_runner: Box<dyn ProcessRunner<Command = C>>,
+    pub load_test_vcpus: TaskSetCpuList,
+}
+
+impl<C> RunConfig<C> {
+    pub fn run(self, command: &C) -> Result<RunSuccessResults, RunErrorResults<C>> {
+        let mut tracker_config_file = NamedTempFile::new().unwrap();
+        let mut load_test_config_file = NamedTempFile::new().unwrap();
+
+        let tracker =
+            match self
+                .tracker_runner
+                .run(command, &self.tracker_vcpus, &mut tracker_config_file)
+            {
+                Ok(handle) => ChildWrapper(handle),
+                Err(err) => {
+                    return Err(RunErrorResults::new(self).set_error(err.into(), "run tracker"))
+                }
+            };
+
+        ::std::thread::sleep(Duration::from_secs(1));
+
+        let mut load_tester = match self.load_test_runner.run(
+            command,
+            &self.load_test_vcpus,
+            &mut load_test_config_file,
+        ) {
+            Ok(handle) => ChildWrapper(handle),
+            Err(err) => {
+                return Err(RunErrorResults::new(self)
+                    .set_error(err.into(), "run load test")
+                    .set_tracker(tracker))
+            }
+        };
+
+        ::std::thread::sleep(Duration::from_secs(59));
+
+        let tracker_process_stats_res = Command::new("ps")
+            .arg("-p")
+            .arg(tracker.0.id().to_string())
+            .arg("-o")
+            .arg("%cpu,rss")
+            .arg("--noheader")
+            .output();
+
+        let tracker_process_stats = match tracker_process_stats_res {
+            Ok(output) if output.status.success() => {
+                ProcessStats::from_str(&String::from_utf8_lossy(&output.stdout)).unwrap()
+            }
+            Ok(_) => {
+                return Err(RunErrorResults::new(self)
+                    .set_error_context("run ps")
+                    .set_tracker(tracker)
+                    .set_load_test_outputs(load_tester));
+            }
+            Err(err) => {
+                return Err(RunErrorResults::new(self)
+                    .set_error(err.into(), "run ps")
+                    .set_tracker(tracker)
+                    .set_load_test_outputs(load_tester));
+            }
+        };
+
+        ::std::thread::sleep(Duration::from_secs(5));
+
+        let (load_test_stdout, load_test_stderr) = match load_tester.0.try_wait() {
+            Ok(Some(status)) if status.success() => read_child_outputs(load_tester),
+            Ok(Some(_)) => {
+                return Err(RunErrorResults::new(self)
+                    .set_error_context("wait for load tester")
+                    .set_tracker(tracker)
+                    .set_load_test_outputs(load_tester))
+            }
+            Ok(None) => {
+                if let Err(err) = load_tester.0.kill() {
+                    return Err(RunErrorResults::new(self)
+                        .set_error(err.into(), "kill load tester")
+                        .set_tracker(tracker)
+                        .set_load_test_outputs(load_tester));
+                }
+
+                ::std::thread::sleep(Duration::from_secs(1));
+
+                match load_tester.0.try_wait() {
+                    Ok(_) => {
+                        return Err(RunErrorResults::new(self)
+                            .set_error_context("load tester didn't finish in time")
+                            .set_load_test_outputs(load_tester))
+                    }
+                    Err(err) => {
+                        return Err(RunErrorResults::new(self)
+                            .set_error(err.into(), "wait for load tester after kill")
+                            .set_tracker(tracker));
+                    }
+                }
+            }
+            Err(err) => {
+                return Err(RunErrorResults::new(self)
+                    .set_error(err.into(), "wait for load tester")
+                    .set_tracker(tracker)
+                    .set_load_test_outputs(load_tester))
+            }
+        };
+
+        let load_test_stdout = if let Some(load_test_stdout) = load_test_stdout {
+            load_test_stdout
+        } else {
+            return Err(RunErrorResults::new(self)
+                .set_error_context("couldn't read load tester stdout")
+                .set_tracker(tracker)
+                .set_load_test_stderr(load_test_stderr));
+        };
+
+        let avg_responses = {
+            static RE: Lazy<Regex> = Lazy::new(|| {
+                Regex::new(r"Average responses per second: ([0-9]+\.?[0-9]+)").unwrap()
+            });
+
+            let opt_avg_responses = RE
+                .captures_iter(&load_test_stdout)
+                .next()
+                .map(|c| {
+                    let (_, [avg_responses]) = c.extract();
+
+                    avg_responses.to_string()
+                })
+                .and_then(|v| v.parse::<f32>().ok());
+
+            if let Some(avg_responses) = opt_avg_responses {
+                avg_responses
+            } else {
+                return Err(RunErrorResults::new(self)
+                    .set_error_context("couldn't extract avg_responses")
+                    .set_tracker(tracker)
+                    .set_load_test_stdout(Some(load_test_stdout))
+                    .set_load_test_stderr(load_test_stderr));
+            }
+        };
+
+        let results = RunSuccessResults {
+            tracker_process_stats,
+            avg_responses,
+        };
+
+        Ok(results)
+    }
+}
+
+pub struct RunSuccessResults {
+    pub tracker_process_stats: ProcessStats,
+    pub avg_responses: f32,
+}
+
+#[derive(Debug)]
+pub struct RunErrorResults<C> {
+    pub run_config: RunConfig<C>,
+    pub tracker_process_stats: Option<ProcessStats>,
+    pub tracker_stdout: Option<String>,
+    pub tracker_stderr: Option<String>,
+    pub load_test_stdout: Option<String>,
+    pub load_test_stderr: Option<String>,
+    pub error: Option<anyhow::Error>,
+    pub error_context: Option<String>,
+}
+
+impl<C> RunErrorResults<C> {
+    fn new(run_config: RunConfig<C>) -> Self {
+        Self {
+            run_config,
+            tracker_process_stats: Default::default(),
+            tracker_stdout: Default::default(),
+            tracker_stderr: Default::default(),
+            load_test_stdout: Default::default(),
+            load_test_stderr: Default::default(),
+            error: Default::default(),
+            error_context: Default::default(),
+        }
+    }
+
+    fn set_tracker(mut self, tracker: ChildWrapper) -> Self {
+        let (stdout, stderr) = read_child_outputs(tracker);
+
+        self.tracker_stdout = stdout;
+        self.tracker_stderr = stderr;
+
+        self
+    }
+
+    fn set_load_test_outputs(mut self, load_test: ChildWrapper) -> Self {
+        let (stdout, stderr) = read_child_outputs(load_test);
+
+        self.load_test_stdout = stdout;
+        self.load_test_stderr = stderr;
+
+        self
+    }
+
+    fn set_load_test_stdout(mut self, stdout: Option<String>) -> Self {
+        self.load_test_stdout = stdout;
+
+        self
+    }
+
+    fn set_load_test_stderr(mut self, stderr: Option<String>) -> Self {
+        self.load_test_stderr = stderr;
+
+        self
+    }
+
+    fn set_error(mut self, error: anyhow::Error, context: &str) -> Self {
+        self.error = Some(error);
+        self.error_context = Some(context.to_string());
+
+        self
+    }
+
+    fn set_error_context(mut self, context: &str) -> Self {
+        self.error_context = Some(context.to_string());
+
+        self
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct ProcessStats {
+    pub avg_cpu_utilization: f32,
+    pub peak_rss_kb: f32,
+}
+
+impl FromStr for ProcessStats {
+    type Err = ();
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let mut parts = s.trim().split_whitespace();
+
+        Ok(Self {
+            avg_cpu_utilization: parts.next().ok_or(())?.parse().map_err(|_| ())?,
+            peak_rss_kb: parts.next().ok_or(())?.parse().map_err(|_| ())?,
+        })
+    }
+}
+
+struct ChildWrapper(Child);
+
+impl Drop for ChildWrapper {
+    fn drop(&mut self) {
+        let _ = self.0.kill();
+
+        ::std::thread::sleep(Duration::from_secs(1));
+
+        let _ = self.0.try_wait();
+    }
+}
+
+fn read_child_outputs(mut child: ChildWrapper) -> (Option<String>, Option<String>) {
+    let stdout = child.0.stdout.take().map(|stdout| {
+        let mut buf = String::new();
+
+        let mut reader = NonBlockingReader::from_fd(stdout).unwrap();
+
+        reader.read_available_to_string(&mut buf).unwrap();
+
+        buf
+    });
+    let stderr = child.0.stderr.take().map(|stderr| {
+        let mut buf = String::new();
+
+        let mut reader = NonBlockingReader::from_fd(stderr).unwrap();
+
+        reader.read_available_to_string(&mut buf).unwrap();
+
+        buf
+    });
+
+    (stdout, stderr)
+}

crates/bencher/src/set.rs

@@ -0,0 +1,294 @@
+use std::rc::Rc;
+
+use indexmap::{IndexMap, IndexSet};
+use itertools::Itertools;
+
+use crate::{
+    common::{CpuDirection, CpuMode, TaskSetCpuList},
+    run::{ProcessRunner, ProcessStats, RunConfig},
+};
+
+pub trait Tracker: ::std::fmt::Debug + Copy + Clone + ::std::hash::Hash {
+    fn name(&self) -> String;
+}
+
+pub struct SetConfig<C, I> {
+    pub implementations: IndexMap<I, Vec<Rc<dyn ProcessRunner<Command = C>>>>,
+    pub load_test_runs: Vec<(usize, TaskSetCpuList)>,
+}
+
+pub fn run_sets<C, F, I>(
+    command: &C,
+    cpu_mode: CpuMode,
+    set_configs: IndexMap<usize, SetConfig<C, I>>,
+    load_test_gen: F,
+) where
+    C: ::std::fmt::Debug,
+    I: Tracker,
+    F: Fn(usize) -> Box<dyn ProcessRunner<Command = C>>,
+{
+    println!("# Load test report");
+
+    let results = set_configs
+        .into_iter()
+        .map(|(tracker_core_count, set_config)| {
+            let tracker_vcpus =
+                TaskSetCpuList::new(cpu_mode, CpuDirection::Asc, tracker_core_count).unwrap();
+
+            println!(
+                "## Tracker cores: {} (cpus: {})",
+                tracker_core_count,
+                tracker_vcpus.as_cpu_list()
+            );
+
+            let tracker_results = set_config
+                .implementations
+                .into_iter()
+                .map(|(implementation, tracker_runs)| {
+                    let tracker_run_results = tracker_runs
+                        .iter()
+                        .map(|tracker_run| {
+                            let load_test_run_results = set_config
+                                .load_test_runs
+                                .clone()
+                                .into_iter()
+                                .map(|(workers, load_test_vcpus)| {
+                                    LoadTestRunResults::produce(
+                                        command,
+                                        &load_test_gen,
+                                        implementation,
+                                        &tracker_run,
+                                        tracker_vcpus.clone(),
+                                        workers,
+                                        load_test_vcpus,
+                                    )
+                                })
+                                .collect();
+
+                            TrackerConfigurationResults {
+                                load_tests: load_test_run_results,
+                            }
+                        })
+                        .collect();
+
+                    ImplementationResults {
+                        name: implementation.name(),
+                        configurations: tracker_run_results,
+                    }
+                })
+                .collect();
+
+            TrackerCoreCountResults {
+                core_count: tracker_core_count,
+                implementations: tracker_results,
+            }
+        })
+        .collect::<Vec<_>>();
+
+    html_summary(&results);
+}
+
+pub struct TrackerCoreCountResults {
+    core_count: usize,
+    implementations: Vec<ImplementationResults>,
+}
+
+pub struct ImplementationResults {
+    name: String,
+    configurations: Vec<TrackerConfigurationResults>,
+}
+
+impl ImplementationResults {
+    fn best_result(&self) -> Option<LoadTestRunResultsSuccess> {
+        self.configurations
+            .iter()
+            .filter_map(|c| c.best_result())
+            .reduce(|acc, r| {
+                if r.average_responses > acc.average_responses {
+                    r
+                } else {
+                    acc
+                }
+            })
+    }
+}
+
+pub struct TrackerConfigurationResults {
+    load_tests: Vec<LoadTestRunResults>,
+}
+
+impl TrackerConfigurationResults {
+    fn best_result(&self) -> Option<LoadTestRunResultsSuccess> {
+        self.load_tests
+            .iter()
+            .filter_map(|r| match r {
+                LoadTestRunResults::Success(r) => Some(r.clone()),
+                LoadTestRunResults::Failure(_) => None,
+            })
+            .reduce(|acc, r| {
+                if r.average_responses > acc.average_responses {
+                    r
+                } else {
+                    acc
+                }
+            })
+    }
+}
+
+pub enum LoadTestRunResults {
+    Success(LoadTestRunResultsSuccess),
+    Failure(LoadTestRunResultsFailure),
+}
+
+impl LoadTestRunResults {
+    pub fn produce<C, F, I>(
+        command: &C,
+        load_test_gen: &F,
+        implementation: I,
+        tracker_process: &Rc<dyn ProcessRunner<Command = C>>,
+        tracker_vcpus: TaskSetCpuList,
+        workers: usize,
+        load_test_vcpus: TaskSetCpuList,
+    ) -> Self
+    where
+        C: ::std::fmt::Debug,
+        I: Tracker,
+        F: Fn(usize) -> Box<dyn ProcessRunner<Command = C>>,
+    {
+        println!(
+            "### {} run ({}) (load test workers: {}, cpus: {})",
+            implementation.name(),
+            tracker_process.info(),
+            workers,
+            load_test_vcpus.as_cpu_list()
+        );
+
+        let load_test_runner = load_test_gen(workers);
+        // let load_test_keys = load_test_runner.keys();
+
+        let run_config = RunConfig {
+            tracker_runner: tracker_process.clone(),
+            tracker_vcpus: tracker_vcpus.clone(),
+            load_test_runner,
+            load_test_vcpus,
+        };
+
+        match run_config.run(command) {
+            Ok(r) => {
+                println!("- Average responses per second: {}", r.avg_responses);
+                println!(
+                    "- Average tracker CPU utilization: {}%",
+                    r.tracker_process_stats.avg_cpu_utilization,
+                );
+                println!(
+                    "- Peak tracker RSS: {} kB",
+                    r.tracker_process_stats.peak_rss_kb
+                );
+
+                LoadTestRunResults::Success(LoadTestRunResultsSuccess {
+                    average_responses: r.avg_responses,
+                    // tracker_keys: tracker_process.keys(),
+                    tracker_info: tracker_process.info(),
+                    tracker_process_stats: r.tracker_process_stats,
+                    // load_test_keys,
+                })
+            }
+            Err(results) => {
+                println!("\nRun failed:\n{:#?}\n", results);
+
+                LoadTestRunResults::Failure(LoadTestRunResultsFailure {
+                    // load_test_keys
+                })
+            }
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct LoadTestRunResultsSuccess {
+    average_responses: f32,
+    // tracker_keys: IndexMap<String, String>,
+    tracker_info: String,
+    tracker_process_stats: ProcessStats,
+    // load_test_keys: IndexMap<String, String>,
+}
+
+pub struct LoadTestRunResultsFailure {
+    // load_test_keys: IndexMap<String, String>,
+}
+
+pub fn html_summary(results: &[TrackerCoreCountResults]) {
+    let mut all_implementation_names = IndexSet::new();
+
+    for core_count_results in results {
+        all_implementation_names.extend(
+            core_count_results
+                .implementations
+                .iter()
+                .map(|r| r.name.clone()),
+        );
+    }
+
+    let mut data_rows = Vec::new();
+
+    for core_count_results in results {
+        let best_results = core_count_results
+            .implementations
+            .iter()
+            .map(|implementation| (implementation.name.clone(), implementation.best_result()))
+            .collect::<IndexMap<_, _>>();
+
+        let best_results_for_all_implementations = all_implementation_names
+            .iter()
+            .map(|name| best_results.get(name).cloned().flatten())
+            .collect::<Vec<_>>();
+
+        let data_row = format!(
+            "
+            <tr>
+                <th>{}</th>
+                {}
+            </tr>
+            ",
+            core_count_results.core_count,
+            best_results_for_all_implementations
+                .into_iter()
+                .map(|result| {
+                    if let Some(r) = result {
+                        format!(
+                            r#"<td><span title="{}, avg cpu utilization: {}%">{}</span></td>"#,
+                            r.tracker_info,
+                            r.tracker_process_stats.avg_cpu_utilization,
+                            r.average_responses,
+                        )
+                    } else {
+                        "<td>-</td>".to_string()
+                    }
+                })
+                .join("\n"),
+        );
+
+        data_rows.push(data_row);
+    }
+
+    println!(
+        "
+        <table>
+            <thead>
+                <tr>
+                    <th>CPU cores</th>
+                    {}
+                </tr>
+            </thead>
+            <tbody>
+                {}
+            </tbody>
+        </table>
+        ",
+        all_implementation_names
+            .iter()
+            .map(|name| format!("<th>{name}</th>"))
+            .join("\n"),
+        data_rows.join("\n")
+    )
+}