From 78b0de21c7e5e094192905454a1154f72a8ff2ed Mon Sep 17 00:00:00 2001
From: Cameron Bytheway <bytheway.cameron@gmail.com>
Date: Fri, 25 Feb 2022 10:31:58 -0700
Subject: [PATCH] netbench: add scenario executable (#1198)

---
 netbench/netbench-scenarios/Cargo.toml        |  11 +
 netbench/netbench-scenarios/README.md         | 169 ++++++++++++++
 netbench/netbench-scenarios/src/config.rs     | 211 ++++++++++++++++++
 netbench/netbench-scenarios/src/lib.rs        | 129 +++++++++++
 netbench/netbench-scenarios/src/main.rs       |   4 +
 netbench/netbench-scenarios/src/prelude.rs    |  46 ++++
 .../src/request_response.rs                   |  97 ++++++++
 netbench/netbench/Cargo.toml                  |   2 +
 netbench/netbench/src/scenario.rs             |   5 +
 netbench/netbench/src/units/byte.rs           | 105 ++++++++-
 netbench/netbench/src/units/rate.rs           |  49 +++-
 ...bench__units__byte__tests__parse_test.snap |  98 ++++++++
 ...bench__units__rate__tests__parse_test.snap |  48 ++++
 13 files changed, 972 insertions(+), 2 deletions(-)
 create mode 100644 netbench/netbench-scenarios/Cargo.toml
 create mode 100644 netbench/netbench-scenarios/README.md
 create mode 100644 netbench/netbench-scenarios/src/config.rs
 create mode 100644 netbench/netbench-scenarios/src/lib.rs
 create mode 100644 netbench/netbench-scenarios/src/main.rs
 create mode 100644 netbench/netbench-scenarios/src/prelude.rs
 create mode 100644 netbench/netbench-scenarios/src/request_response.rs
 create mode 100644 netbench/netbench/src/units/snapshots/netbench__units__byte__tests__parse_test.snap
 create mode 100644 netbench/netbench/src/units/snapshots/netbench__units__rate__tests__parse_test.snap

diff --git a/netbench/netbench-scenarios/Cargo.toml b/netbench/netbench-scenarios/Cargo.toml
new file mode 100644
index 0000000000..c54e8f831e
--- /dev/null
+++ b/netbench/netbench-scenarios/Cargo.toml
@@ -0,0 +1,11 @@
+[package]
+name = "netbench-scenarios"
+version = "0.1.0"
+authors = ["AWS s2n"]
+edition = "2018"
+license = "Apache-2.0"
+
+[dependencies]
+clap = { version = "2", features = ["color", "suggestions"] }
+humantime = "2"
+netbench = { path = "../netbench" }
diff --git a/netbench/netbench-scenarios/README.md b/netbench/netbench-scenarios/README.md
new file mode 100644
index 0000000000..0e921d2b48
--- /dev/null
+++ b/netbench/netbench-scenarios/README.md
@@ -0,0 +1,169 @@
+# netbench-scenarios
+
+### Executable
+
+The executable includes a single default scenario: [`request_response`](https://github.com/aws/s2n-quic/blob/main/netbench/netbench-scenarios/src/request_response.rs). This sends `N` number of bytes to the server, which responds with `M` number of bytes. Several options are available for configuration:
+
+```shell
+$ cargo run --bin netbench-scenarios -- --help
+
+netbench scenarios
+
+USAGE:
+    netbench-scenarios [FLAGS] [OPTIONS] [OUT_DIR]
+
+FLAGS:
+    -h, --help
+            Prints help information
+
+        --request_response.parallel
+            Specifies if the requests should be performed in parallel
+
+    -V, --version
+            Prints version information
+
+
+OPTIONS:
+        --request_response.client_receive_rate <RATE>
+            The rate at which the client receives data [default: NONE]
+
+        --request_response.client_send_rate <RATE>
+            The rate at which the client sends data [default: NONE]
+
+        --request_response.count <COUNT>
+            The number of requests to make [default: 1]
+
+        --request_response.request_size <BYTES>
+            The size of the client's request to the server [default: 1KB]
+
+        --request_response.response_delay <TIME>
+            How long the server will take to respond to the request [default: 0s]
+
+        --request_response.response_size <BYTES>
+            The size of the server's response to the client [default: 10MB]
+
+        --request_response.server_receive_rate <RATE>
+            The rate at which the server receives data [default: NONE]
+
+        --request_response.server_send_rate <RATE>
+            The rate at which the server sends data [default: NONE]
+
+
+ARGS:
+    <OUT_DIR>
+             [default: target/netbench]
+
+FORMATS:
+    BYTES
+        42b         ->    42 bits
+        42          ->    42 bytes
+        42B         ->    42 bytes
+        42K         ->    42000 bytes
+        42Kb        ->    42000 bits
+        42KB        ->    42000 bytes
+        42KiB       ->    43008 bytes
+
+    COUNT
+        42          ->    42 units
+
+    RATE
+        42bps       ->    42 bits per second
+        42Mbps      ->    42 megabits per second
+        42MBps      ->    42 megabytes per second
+        42MiBps     ->    42 mebibytes per second
+        42MB/50ms   ->    42 megabytes per 50 milliseconds
+
+    TIME
+        42ms         ->    42 milliseconds
+        42s          ->    42 seconds
+        1s42ms       ->    1 second + 42 milliseconds
+```
+
+Moving forward, we can add any useful scenarios to this list.
+
+### Library
+
+For workflows that want to build their own scenarios, it can depend on the library and set up their `main.rs` as follows:
+
+```rust
+netbench_scenario::scenarios!(my_scenario_a, my_scenario_b);
+```
+
+They would then create a `my_scenario_a.rs` and `my_scenario_b.rs`:
+
+```rust
+// my_scenario_a.rs
+use netbench_scenario::prelude::*;
+
+config!({
+    /// The size of the client's request to the server
+    let request_size: Byte = 1.kilobytes();
+
+    /// The size of the server's response to the client
+    let response_size: Byte = 10.megabytes();
+});
+
+pub fn scenario(config: Config) -> Scenario {
+    let Config {
+        request_size,
+        response_size,
+    } = config;
+
+    Scenario::build(|scenario| {
+        let server = scenario.create_server();
+
+        scenario.create_client(|client| {
+            client.connect_to(server, |conn| {
+                conn.open_bidirectional_stream(
+                    |local| {
+                        local.send(request_size);
+                        local.receive(response_size);
+                    },
+                    |remote| {
+                        remote.receive(request_size);
+                        remote.send(response_size);
+                    },
+                );
+            });
+        });
+    })
+}
+```
+
+They can then run their scenario generator:
+
+```shell
+$ cargo run -- --help
+
+netbench scenarios
+
+USAGE:
+    netbench-scenarios [FLAGS] [OPTIONS] [OUT_DIR]
+
+FLAGS:
+    -h, --help
+            Prints help information
+
+    -V, --version
+            Prints version information
+
+
+OPTIONS:
+        --my_scenario_a.request_size <BYTES>
+            The size of the client's request to the server [default: 1KB]
+
+        --my_scenario_a.response_size <BYTES>
+            The size of the server's response to the client [default: 10MB]
+
+
+ARGS:
+    <OUT_DIR>
+             [default: target/netbench]
+
+```
+
+```
+$ cargo run
+created: target/netbench/my_scenario_a.json
+created: target/netbench/my_scenario_b.json
+```
diff --git a/netbench/netbench-scenarios/src/config.rs b/netbench/netbench-scenarios/src/config.rs
new file mode 100644
index 0000000000..16f211194d
--- /dev/null
+++ b/netbench/netbench-scenarios/src/config.rs
@@ -0,0 +1,211 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use netbench::{units::*, Error, Result};
+use std::collections::BTreeMap;
+
+#[derive(Debug, Default)]
+pub struct Registry {
+    definitions: BTreeMap<&'static str, Definition>,
+}
+
+impl Registry {
+    pub fn define<T>(&mut self, name: &'static str, docs: &'static [&'static str], default: &T)
+    where
+        T: TryFromValue,
+    {
+        self.definitions
+            .insert(name, Definition::new::<T>(docs, default.display()));
+    }
+
+    pub fn clap_args(&self) -> impl Iterator<Item = clap::Arg> + '_ {
+        self.definitions.iter().map(|(name, def)| {
+            clap::Arg::with_name(name)
+                .long(name)
+                .value_name(def.value_name)
+                .help(def.help)
+                .default_value(&def.default)
+                .takes_value(def.takes_value)
+                .long_help(&def.long_help)
+        })
+    }
+
+    pub fn load_overrides(&self, matches: &clap::ArgMatches) -> Overrides {
+        let mut overrides = Overrides::default();
+        for (name, _) in self.definitions.iter() {
+            if let Some(value) = matches.value_of(name) {
+                overrides
+                    .values
+                    .insert(name, Override::String(value.to_string()));
+            } else if matches.is_present(name) {
+                overrides.values.insert(name, Override::Enabled);
+            }
+        }
+        overrides
+    }
+}
+
+#[derive(Default)]
+pub struct Overrides {
+    values: BTreeMap<&'static str, Override>,
+    errors: BTreeMap<&'static str, Error>,
+}
+
+impl Overrides {
+    pub fn resolve<T>(&mut self, name: &'static str, default: T) -> T
+    where
+        T: TryFromValue,
+    {
+        if let Some(value) = self.values.get(name) {
+            match T::try_from_value(value) {
+                Ok(value) => {
+                    return value;
+                }
+                Err(err) => {
+                    self.errors.insert(name, err);
+                }
+            }
+        }
+
+        default
+    }
+
+    pub fn errors(&self) -> impl Iterator<Item = String> + '_ {
+        self.errors
+            .iter()
+            .map(|(name, error)| format!("{}: {}\n", name, error))
+    }
+}
+
+#[derive(Debug)]
+struct Definition {
+    help: &'static str,
+    long_help: String,
+    default: String,
+    value_name: &'static str,
+    takes_value: bool,
+}
+
+impl Definition {
+    fn new<T: TryFromValue>(docs: &[&'static str], default: String) -> Self {
+        Self {
+            help: docs[0],
+            long_help: docs.iter().map(|v| v.trim()).collect::<Vec<_>>().join("\n"),
+            default,
+            value_name: T::VALUE_NAME,
+            takes_value: T::TAKES_VALUE,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum Override {
+    Enabled,
+    String(String),
+}
+
+pub trait TryFromValue: Sized {
+    const VALUE_NAME: &'static str;
+    const TAKES_VALUE: bool = true;
+
+    fn try_from_value(value: &Override) -> Result<Self>;
+    fn display(&self) -> String;
+}
+
+impl TryFromValue for bool {
+    const VALUE_NAME: &'static str = "BOOL";
+    const TAKES_VALUE: bool = false;
+
+    fn try_from_value(value: &Override) -> Result<Self> {
+        match value {
+            Override::Enabled => Ok(true),
+            Override::String(v) => match v.as_str() {
+                "true" | "TRUE" | "1" | "yes" | "YES" => Ok(true),
+                "false" | "FALSE" | "0" | "no" | "NO" => Ok(false),
+                _ => Err(format!("invalid bool: {:?}", v).into()),
+            },
+        }
+    }
+
+    fn display(&self) -> String {
+        self.to_string()
+    }
+}
+
+impl TryFromValue for u64 {
+    const VALUE_NAME: &'static str = "COUNT";
+
+    fn try_from_value(value: &Override) -> Result<Self> {
+        match value {
+            Override::Enabled => Err("missing value".into()),
+            Override::String(v) => Ok(v.parse()?),
+        }
+    }
+
+    fn display(&self) -> String {
+        self.to_string()
+    }
+}
+
+impl TryFromValue for Duration {
+    const VALUE_NAME: &'static str = "TIME";
+
+    fn try_from_value(value: &Override) -> Result<Self> {
+        match value {
+            Override::Enabled => Err("missing value".into()),
+            Override::String(v) => {
+                let v: humantime::Duration = v.parse()?;
+                Ok(*v)
+            }
+        }
+    }
+
+    fn display(&self) -> String {
+        if *self == Self::ZERO {
+            return "0s".to_owned();
+        }
+        format!("{:?}", self)
+    }
+}
+
+impl<T: TryFromValue> TryFromValue for Option<T> {
+    const VALUE_NAME: &'static str = T::VALUE_NAME;
+
+    fn try_from_value(value: &Override) -> Result<Self> {
+        if matches!(value, Override::String(v) if v == "NONE") {
+            return Ok(None);
+        }
+
+        T::try_from_value(value).map(Some)
+    }
+
+    fn display(&self) -> String {
+        if let Some(value) = self.as_ref() {
+            value.display()
+        } else {
+            "NONE".to_owned()
+        }
+    }
+}
+
+macro_rules! try_from_value {
+    ($name:ty, $value_name:literal) => {
+        impl TryFromValue for $name {
+            const VALUE_NAME: &'static str = $value_name;
+
+            fn try_from_value(value: &Override) -> Result<Self> {
+                match value {
+                    Override::Enabled => Err("missing value".into()),
+                    Override::String(v) => Ok(v.parse()?),
+                }
+            }
+
+            fn display(&self) -> String {
+                self.to_string()
+            }
+        }
+    };
+}
+
+try_from_value!(Byte, "BYTES");
+try_from_value!(Rate, "RATE");
diff --git a/netbench/netbench-scenarios/src/lib.rs b/netbench/netbench-scenarios/src/lib.rs
new file mode 100644
index 0000000000..464d9c8ea6
--- /dev/null
+++ b/netbench/netbench-scenarios/src/lib.rs
@@ -0,0 +1,129 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use clap::{App, Arg};
+pub use netbench::Result;
+use std::path::Path;
+
+pub mod config;
+pub mod prelude;
+
+pub trait Configs {
+    fn registry() -> config::Registry;
+    fn new(overrides: &mut config::Overrides) -> Self;
+    fn write(self, out_dir: &Path) -> Result<()>;
+}
+
+#[doc(hidden)]
+pub fn main<C: Configs>() -> Result<()> {
+    let app = App::new("netbench scenarios")
+        .after_help(LONG_ABOUT.trim())
+        .arg(
+            Arg::with_name("out_dir")
+                .value_name("OUT_DIR")
+                .default_value("target/netbench")
+                .takes_value(true),
+        );
+
+    let map = C::registry();
+    let args = map.clap_args().collect::<Vec<_>>();
+    let matches = app.args(&args).get_matches();
+    let mut overrides = map.load_overrides(&matches);
+
+    let configs = C::new(&mut overrides);
+
+    let mut has_error = false;
+    for error in overrides.errors() {
+        eprintln!("{}", error);
+        has_error = true;
+    }
+
+    if has_error {
+        return Err("bailing due to errors".into());
+    }
+
+    let out_dir = matches.value_of("out_dir").unwrap();
+    let out_dir = Path::new(out_dir);
+
+    std::fs::create_dir_all(out_dir)?;
+    configs.write(out_dir)?;
+
+    Ok(())
+}
+
+const LONG_ABOUT: &'static str = r#"
+FORMATS:
+    BYTES
+        42b         ->    42 bits
+        42          ->    42 bytes
+        42B         ->    42 bytes
+        42K         ->    42000 bytes
+        42Kb        ->    42000 bits
+        42KB        ->    42000 bytes
+        42KiB       ->    43008 bytes
+
+    COUNT
+        42          ->    42 units
+
+    RATE
+        42bps       ->    42 bits per second
+        42Mbps      ->    42 megabits per second
+        42MBps      ->    42 megabytes per second
+        42MiBps     ->    42 mebibytes per second
+        42MB/50ms   ->    42 megabytes per 50 milliseconds
+
+    TIME
+        42ms         ->    42 milliseconds
+        42s          ->    42 seconds
+        1s42ms       ->    1 second + 42 milliseconds
+"#;
+
+#[macro_export]
+macro_rules! scenarios {
+    ($($name:ident),* $(,)?) => {
+        $(
+            mod $name;
+        )*
+
+        #[derive(Debug)]
+        pub struct Configs {
+            $(
+                $name: $name::Config,
+            )*
+        }
+
+        impl $crate::Configs for Configs {
+            fn registry() -> $crate::config::Registry {
+                let mut registry = $crate::config::Registry::default();
+                $(
+                    $name::Config::register(&mut registry);
+                )*
+                registry
+            }
+
+            fn new(overrides: &mut $crate::config::Overrides) -> Self {
+                Self {
+                    $(
+                        $name: $name::Config::new(overrides),
+                    )*
+                }
+            }
+
+            fn write(self, out_dir: &std::path::Path) -> $crate::Result<()> {
+                $({
+                    let path = out_dir.join(concat!(stringify!($name), ".json"));
+                    let mut f = std::fs::File::create(&path)?;
+                    let s = $name::scenario(self.$name);
+                    s.write(&mut f)?;
+                    eprintln!("created: {}", path.display());
+                })*
+
+                Ok(())
+            }
+        }
+
+        fn main() -> $crate::Result<()> {
+            $crate::main::<Configs>()
+        }
+    };
+}
diff --git a/netbench/netbench-scenarios/src/main.rs b/netbench/netbench-scenarios/src/main.rs
new file mode 100644
index 0000000000..71c4eebe34
--- /dev/null
+++ b/netbench/netbench-scenarios/src/main.rs
@@ -0,0 +1,4 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+netbench_scenarios::scenarios!(request_response);
diff --git a/netbench/netbench-scenarios/src/prelude.rs b/netbench/netbench-scenarios/src/prelude.rs
new file mode 100644
index 0000000000..759fbe163d
--- /dev/null
+++ b/netbench/netbench-scenarios/src/prelude.rs
@@ -0,0 +1,46 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+pub use crate::config;
+pub use netbench::{
+    scenario::{builder, Scenario},
+    units::*,
+    Result,
+};
+
+#[macro_export]
+macro_rules! config {
+    ({$($(#[doc = $doc:literal])* let $name:ident: $ty:ty = $default:expr;)*}) => {
+        #[derive(Debug)]
+        pub struct Config {
+            $(
+                $(#[doc = $doc])*
+                $name: $ty,
+            )*
+        }
+
+        impl Config {
+            pub fn register(registry: &mut $crate::config::Registry) {
+                $(
+                    {
+                        let name = concat!(module_path!(), ".", stringify!($name)).split("::").last().unwrap();
+                        static DOCS: &[&str] = &[$($doc),*];
+                        let default: $ty = $default;
+                        registry.define(name, DOCS, &default)
+                    }
+                )*
+            }
+
+            pub fn new(overrides: &mut $crate::config::Overrides) -> Config {
+                Config {
+                    $(
+                        $name: {
+                            let name = concat!(module_path!(), ".", stringify!($name)).split("::").last().unwrap();
+                            overrides.resolve(name, $default)
+                        },
+                    )*
+                }
+            }
+        }
+    };
+}
diff --git a/netbench/netbench-scenarios/src/request_response.rs b/netbench/netbench-scenarios/src/request_response.rs
new file mode 100644
index 0000000000..3592372c7a
--- /dev/null
+++ b/netbench/netbench-scenarios/src/request_response.rs
@@ -0,0 +1,97 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use netbench_scenarios::prelude::*;
+
+config!({
+    /// The size of the client's request to the server
+    let request_size: Byte = 1.kilobytes();
+
+    /// The size of the server's response to the client
+    let response_size: Byte = 10.megabytes();
+
+    /// How long the server will take to respond to the request
+    let response_delay: Duration = 0.seconds();
+
+    /// The number of requests to make
+    let count: u64 = 1;
+
+    /// Specifies if the requests should be performed in parallel
+    let parallel: bool = false;
+
+    /// The rate at which the client sends data
+    let client_send_rate: Option<Rate> = None;
+
+    /// The rate at which the client receives data
+    let client_receive_rate: Option<Rate> = None;
+
+    /// The rate at which the server sends data
+    let server_send_rate: Option<Rate> = None;
+
+    /// The rate at which the server receives data
+    let server_receive_rate: Option<Rate> = None;
+});
+
+pub fn scenario(config: Config) -> Scenario {
+    let Config {
+        request_size,
+        response_size,
+        count,
+        parallel,
+        client_send_rate,
+        client_receive_rate,
+        server_send_rate,
+        server_receive_rate,
+        response_delay,
+    } = config;
+
+    let request = |conn: &mut builder::connection::Builder<builder::Client>| {
+        conn.open_bidirectional_stream(
+            |local| {
+                if let Some(rate) = client_send_rate {
+                    local.set_send_rate(rate);
+                }
+                if let Some(rate) = client_receive_rate {
+                    local.set_receive_rate(rate);
+                }
+                local.send(request_size);
+                local.receive(response_size);
+            },
+            |remote| {
+                if let Some(rate) = server_send_rate {
+                    remote.set_send_rate(rate);
+                }
+                if let Some(rate) = server_receive_rate {
+                    remote.set_receive_rate(rate);
+                }
+                remote.receive(request_size);
+
+                if response_delay != Duration::ZERO {
+                    remote.sleep(response_delay);
+                }
+
+                remote.send(response_size);
+            },
+        );
+    };
+
+    Scenario::build(|scenario| {
+        let server = scenario.create_server();
+
+        scenario.create_client(|client| {
+            client.connect_to(server, |conn| {
+                if parallel {
+                    conn.scope(|scope| {
+                        for _ in 0..count {
+                            scope.spawn(request);
+                        }
+                    });
+                } else {
+                    for _ in 0..count {
+                        request(conn);
+                    }
+                }
+            });
+        });
+    })
+}
diff --git a/netbench/netbench/Cargo.toml b/netbench/netbench/Cargo.toml
index 97a43a2eb8..3575184ee2 100644
--- a/netbench/netbench/Cargo.toml
+++ b/netbench/netbench/Cargo.toml
@@ -12,6 +12,8 @@ default = ["s2n-quic"]
 bytes = "1"
 enum-primitive-derive = "0.2"
 futures = "0.3"
+humansize = "1"
+humantime = "2"
 num-traits = "0.2"
 s2n-quic-core = { version = "0.1", path = "../../quic/s2n-quic-core", features = ["testing"] }
 s2n-quic = { version = "1", path = "../../quic/s2n-quic", optional = true }
diff --git a/netbench/netbench/src/scenario.rs b/netbench/netbench/src/scenario.rs
index eb26c75a0e..fe09ddc1a8 100644
--- a/netbench/netbench/src/scenario.rs
+++ b/netbench/netbench/src/scenario.rs
@@ -35,6 +35,11 @@ impl Scenario {
         let scenario = serde_json::from_reader(&mut file)?;
         Ok(scenario)
     }
+
+    pub fn write<W: std::io::Write>(&self, out: &mut W) -> std::io::Result<()> {
+        serde_json::to_writer_pretty(out, self)?;
+        Ok(())
+    }
 }
 
 #[derive(Clone, Debug, Default, Deserialize, Serialize, Hash)]
diff --git a/netbench/netbench/src/units/byte.rs b/netbench/netbench/src/units/byte.rs
index e4acde8ca4..b13a06552e 100644
--- a/netbench/netbench/src/units/byte.rs
+++ b/netbench/netbench/src/units/byte.rs
@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::units::{Duration, Rate};
-use core::ops;
+use core::{fmt, ops};
 use serde::{Deserialize, Serialize};
 
 #[derive(
@@ -78,6 +78,75 @@ impl ops::Deref for Byte {
     }
 }
 
+impl fmt::Display for Byte {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let opts = humansize::file_size_opts::FileSizeOpts {
+            space: false,
+            ..humansize::file_size_opts::DECIMAL
+        };
+        humansize::FileSize::file_size(&self.0, opts)
+            .unwrap()
+            .fmt(f)
+    }
+}
+
+impl core::str::FromStr for Byte {
+    type Err = crate::Error;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        // try parsing without the suffix
+        if let Ok(v) = s.parse() {
+            return Ok(Self(v));
+        }
+
+        let number_index = s
+            .char_indices()
+            .find_map(|(idx, c)| {
+                if !(c.is_numeric() || c == '.') {
+                    Some(idx)
+                } else {
+                    None
+                }
+            })
+            .unwrap_or(s.len());
+
+        let mut v = Self(s[..number_index].parse()?);
+
+        let mut suffix = s[number_index..].trim();
+        let mut is_bits = false;
+
+        if let Some(s) = suffix.strip_suffix('B') {
+            suffix = s;
+        } else if let Some(s) = suffix.strip_suffix('b') {
+            is_bits = true;
+            suffix = s;
+        }
+
+        v.0 *= *match suffix.trim() {
+            "" => 1.bytes(),
+            "K" | "k" => 1.kilobytes(),
+            "Ki" | "ki" => 1.kibibytes(),
+            "M" | "m" => 1.megabytes(),
+            "Mi" | "mi" => 1.mebibytes(),
+            "G" | "g" => 1.gigabytes(),
+            "Gi" | "gi" => 1.gibibytes(),
+            "T" | "t" => 1.terabytes(),
+            "Ti" | "ti" => 1.tebibytes(),
+            _ => return Err(format!("invalid bytes: {:?}", s).into()),
+        };
+
+        if is_bits {
+            // round up to the nearest byte
+            if v.0 % 8 != 0 {
+                v += 8.bytes();
+            }
+            v.0 /= 8;
+        }
+
+        Ok(v)
+    }
+}
+
 pub trait ByteExt {
     fn bytes(&self) -> Byte;
     fn kilobytes(&self) -> Byte {
@@ -98,6 +167,12 @@ pub trait ByteExt {
     fn gibibytes(&self) -> Byte {
         self.mebibytes() * 1024
     }
+    fn terabytes(&self) -> Byte {
+        self.gigabytes() * 1000
+    }
+    fn tebibytes(&self) -> Byte {
+        self.gibibytes() * 1024
+    }
 }
 
 impl ByteExt for u64 {
@@ -124,4 +199,32 @@ mod tests {
             42.kibibytes() + 42.bytes()
         ]);
     }
+
+    fn p(s: &str) -> crate::Result<Byte> {
+        s.parse()
+    }
+
+    #[test]
+    fn parse_test() {
+        assert_debug_snapshot!([
+            p("42b"),
+            p("42B"),
+            p("42Kb"),
+            p("42KB"),
+            p("42Kib"),
+            p("42KiB"),
+            p("42Mb"),
+            p("42MB"),
+            p("42Mib"),
+            p("42MiB"),
+            p("42Gb"),
+            p("42GB"),
+            p("42Gib"),
+            p("42GiB"),
+            p("42Tb"),
+            p("42TB"),
+            p("42Tib"),
+            p("42TiB"),
+        ]);
+    }
 }
diff --git a/netbench/netbench/src/units/rate.rs b/netbench/netbench/src/units/rate.rs
index f74a4b280b..8baecc6e5a 100644
--- a/netbench/netbench/src/units/rate.rs
+++ b/netbench/netbench/src/units/rate.rs
@@ -1,7 +1,8 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::units::{duration_format, Byte, Duration};
+use crate::units::{duration_format, Byte, Duration, DurationExt};
+use core::fmt;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
@@ -12,6 +13,37 @@ pub struct Rate {
     pub period: Duration,
 }
 
+impl fmt::Display for Rate {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        if self.period == 1.seconds() {
+            return write!(f, "{}ps", self.bytes);
+        }
+        write!(f, "{}/{:?}", self.bytes, self.period)
+    }
+}
+
+impl core::str::FromStr for Rate {
+    type Err = crate::Error;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        if let Some(bytes) = s.strip_suffix("ps") {
+            return Ok(Self {
+                bytes: bytes.parse()?,
+                period: 1.seconds(),
+            });
+        }
+
+        if let Some((bytes, period)) = s.split_once('/') {
+            Ok(Self {
+                bytes: bytes.trim().parse()?,
+                period: *humantime::Duration::from_str(period.trim())?,
+            })
+        } else {
+            Err(format!("invalid rate: {}", s).into())
+        }
+    }
+}
+
 #[derive(Debug, Default)]
 pub(crate) struct Rates {
     pub send: HashMap<u64, Rate>,
@@ -31,4 +63,19 @@ mod tests {
             42.gigabytes() / 42.minutes()
         ]);
     }
+
+    fn p(s: &str) -> crate::Result<Rate> {
+        s.parse()
+    }
+
+    #[test]
+    fn parse_test() {
+        assert_debug_snapshot!([
+            p("42bps"),
+            p("42Bps"),
+            p("42KBps"),
+            p("42Kb / 50ms"),
+            p("42Mb/5ms"),
+        ]);
+    }
 }
diff --git a/netbench/netbench/src/units/snapshots/netbench__units__byte__tests__parse_test.snap b/netbench/netbench/src/units/snapshots/netbench__units__byte__tests__parse_test.snap
new file mode 100644
index 0000000000..0e9f011df3
--- /dev/null
+++ b/netbench/netbench/src/units/snapshots/netbench__units__byte__tests__parse_test.snap
@@ -0,0 +1,98 @@
+---
+source: netbench/netbench/src/units/byte.rs
+assertion_line: 209
+expression: "[p(\"42b\"), p(\"42B\"), p(\"42Kb\"), p(\"42KB\"), p(\"42Kib\"), p(\"42KiB\"), p(\"42Mb\"),\n p(\"42MB\"), p(\"42Mib\"), p(\"42MiB\"), p(\"42Gb\"), p(\"42GB\"), p(\"42Gib\"),\n p(\"42GiB\"), p(\"42Tb\"), p(\"42TB\"), p(\"42Tib\"), p(\"42TiB\")]"
+
+---
+[
+    Ok(
+        Byte(
+            6,
+        ),
+    ),
+    Ok(
+        Byte(
+            42,
+        ),
+    ),
+    Ok(
+        Byte(
+            5250,
+        ),
+    ),
+    Ok(
+        Byte(
+            42000,
+        ),
+    ),
+    Ok(
+        Byte(
+            5376,
+        ),
+    ),
+    Ok(
+        Byte(
+            43008,
+        ),
+    ),
+    Ok(
+        Byte(
+            5250000,
+        ),
+    ),
+    Ok(
+        Byte(
+            42000000,
+        ),
+    ),
+    Ok(
+        Byte(
+            5505024,
+        ),
+    ),
+    Ok(
+        Byte(
+            44040192,
+        ),
+    ),
+    Ok(
+        Byte(
+            5250000000,
+        ),
+    ),
+    Ok(
+        Byte(
+            42000000000,
+        ),
+    ),
+    Ok(
+        Byte(
+            5637144576,
+        ),
+    ),
+    Ok(
+        Byte(
+            45097156608,
+        ),
+    ),
+    Ok(
+        Byte(
+            5250000000000,
+        ),
+    ),
+    Ok(
+        Byte(
+            42000000000000,
+        ),
+    ),
+    Ok(
+        Byte(
+            5772436045824,
+        ),
+    ),
+    Ok(
+        Byte(
+            46179488366592,
+        ),
+    ),
+]
diff --git a/netbench/netbench/src/units/snapshots/netbench__units__rate__tests__parse_test.snap b/netbench/netbench/src/units/snapshots/netbench__units__rate__tests__parse_test.snap
new file mode 100644
index 0000000000..1a83ffdbbc
--- /dev/null
+++ b/netbench/netbench/src/units/snapshots/netbench__units__rate__tests__parse_test.snap
@@ -0,0 +1,48 @@
+---
+source: netbench/netbench/src/units/rate.rs
+assertion_line: 73
+expression: "[p(\"42bps\"), p(\"42Bps\"), p(\"42KBps\"), p(\"42Kb / 50ms\"), p(\"42Mb/5ms\")]"
+
+---
+[
+    Ok(
+        Rate {
+            bytes: Byte(
+                6,
+            ),
+            period: 1s,
+        },
+    ),
+    Ok(
+        Rate {
+            bytes: Byte(
+                42,
+            ),
+            period: 1s,
+        },
+    ),
+    Ok(
+        Rate {
+            bytes: Byte(
+                42000,
+            ),
+            period: 1s,
+        },
+    ),
+    Ok(
+        Rate {
+            bytes: Byte(
+                5250,
+            ),
+            period: 50ms,
+        },
+    ),
+    Ok(
+        Rate {
+            bytes: Byte(
+                5250000,
+            ),
+            period: 5ms,
+        },
+    ),
+]