perf: add perf binaries

.gitignore

@@ -3,3 +3,7 @@
 
 target
 Cargo.lock
+
+perf.data
+perf.data.old
+flamegraph.svg

Cargo.toml

@@ -13,6 +13,7 @@ members = [
     "examples/stdio",
     "stress-tests",
     "tests-integration",
+    "perf",
 ]
 default-members = [
     "channels-io",
@@ -75,3 +76,11 @@ debug = 0
 [profile.release]
 incremental = true
 debug = 0
+
+[profile.perf]
+inherits = "release"
+incremental = false
+debug = true
+opt-level = 3
+lto = "fat"
+codegen-units = 1

justfile

@@ -0,0 +1,7 @@
+profile := "perf"
+package := "perf"
+
+# Produce a flamegraph of any binary of the `perf` package.
+flamegraph BIN OUTPUT="flamegraph.svg":
+    echo 1000 | sudo tee /proc/sys/kernel/perf_event_max_sample_rate >/dev/null
+    cargo flamegraph --deterministic -F 1000 --profile {{profile}} -o {{OUTPUT}} --package {{package}} --bin {{BIN}}

perf/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "perf"
+version = "0.1.0"
+edition = "2021"
+publish = false
+
+[lib]
+path = "src/lib.rs"
+
+[[bin]]
+name = "recv"
+path = "src/bin/recv.rs"
+
+[[bin]]
+name = "send"
+path = "src/bin/send.rs"
+
+[dependencies]
+channels = { workspace = true }

perf/src/bin/recv.rs

@@ -0,0 +1,55 @@
+use std::convert::Infallible;
+use std::hint::black_box;
+
+use channels::io::{Buf, Cursor, Native};
+use channels::receiver::{Config, Receiver};
+use channels::serdes::Deserializer;
+
+static PACKET: &[u8] = include_bytes!("./packet.bin");
+
+struct NoopDeserializer;
+
+impl Deserializer<()> for NoopDeserializer {
+	type Error = Infallible;
+
+	fn deserialize(
+		&mut self,
+		_buf: &mut [u8],
+	) -> Result<(), Self::Error> {
+		Ok(())
+	}
+}
+
+struct Repeat<T: AsRef<[u8]>> {
+	inner: Cursor<T>,
+}
+
+impl<T: AsRef<[u8]>> Buf for Repeat<T> {
+	fn remaining(&self) -> usize {
+		self.inner.remaining()
+	}
+
+	fn chunk(&self) -> &[u8] {
+		self.inner.chunk()
+	}
+
+	fn advance(&mut self, n: usize) {
+		let new_pos =
+			(self.inner.pos() + n) % self.inner.get().as_ref().len();
+		self.inner.set_pos(new_pos);
+	}
+}
+
+fn main() {
+	let mut rx: Receiver<(), Native<_>, NoopDeserializer> =
+		Receiver::<(), _, _>::builder()
+			.reader(Repeat { inner: Cursor::new(PACKET) }.reader())
+			.deserializer(NoopDeserializer)
+			.config(Config::default().with_verify_order(false))
+			.build();
+
+	perf::run_for_default_duration(|| {
+		#[allow(clippy::unit_arg)]
+		black_box(rx.recv_blocking().unwrap());
+	});
+}

perf/src/bin/send.rs

@@ -0,0 +1,29 @@
+use std::convert::Infallible;
+use std::hint::black_box;
+use std::io::empty;
+
+use channels::sender::{Config, Sender};
+use channels::serdes::Serializer;
+
+struct NoopSerializer;
+
+impl Serializer<()> for NoopSerializer {
+	type Error = Infallible;
+
+	fn serialize(&mut self, _: &()) -> Result<Vec<u8>, Self::Error> {
+		Ok(Vec::new())
+	}
+}
+
+fn main() {
+	let mut tx = Sender::<(), _, _>::builder()
+		.writer(empty())
+		.serializer(NoopSerializer)
+		.config(Config::default())
+		.build();
+
+	perf::run_for_default_duration(|| {
+		#[allow(clippy::unit_arg)]
+		black_box(tx.send_blocking(black_box(())).unwrap());
+	});
+}

perf/src/lib.rs

@@ -0,0 +1,38 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::thread;
+use std::time::Duration;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Run {
+	runs: u64,
+}
+
+/// Run _f_ repeatedly until `x` amount of time has passed.
+pub fn run_for<F>(x: Duration, mut f: F) -> Run
+where
+	F: FnMut(),
+{
+	let exit = AtomicBool::new(false);
+	thread::scope(|scope| {
+		scope.spawn(|| {
+			thread::sleep(x);
+			exit.store(true, Ordering::Relaxed);
+		});
+
+		let mut runs = 0;
+		while !exit.load(Ordering::Relaxed) {
+			f();
+			runs += 1;
+		}
+
+		Run { runs }
+	})
+}
+
+/// Run _f_ repeatedly for 30 seconds.
+pub fn run_for_default_duration<F>(f: F) -> Run
+where
+	F: FnMut(),
+{
+	run_for(Duration::from_secs(30), f)
+}