perf(net): add monoio echo bench

Author: Berrysoft

compio/benches/fs.rs

@@ -14,21 +14,9 @@ use tempfile::NamedTempFile;
 use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
 
 #[cfg(target_os = "linux")]
-struct MonoioRuntime(std::cell::RefCell<monoio::Runtime<monoio::IoUringDriver>>);
-
+mod monoio_wrap;
 #[cfg(target_os = "linux")]
-impl criterion::async_executor::AsyncExecutor for MonoioRuntime {
-    fn block_on<T>(&self, future: impl futures_util::Future<Output = T>) -> T {
-        self.0.borrow_mut().block_on(future)
-    }
-}
-
-#[cfg(target_os = "linux")]
-impl criterion::async_executor::AsyncExecutor for &MonoioRuntime {
-    fn block_on<T>(&self, future: impl futures_util::Future<Output = T>) -> T {
-        self.0.borrow_mut().block_on(future)
-    }
-}
+use monoio_wrap::MonoioRuntime;
 
 criterion_group!(fs, read, write);
 criterion_main!(fs);
@@ -118,11 +106,7 @@ fn read_compio_join(b: &mut Bencher, (path, offsets): &(&Path, &[u64])) {
 
 #[cfg(target_os = "linux")]
 fn read_monoio(b: &mut Bencher, (path, offsets): &(&Path, &[u64])) {
-    let runtime = MonoioRuntime(std::cell::RefCell::new(
-        monoio::RuntimeBuilder::<monoio::IoUringDriver>::new()
-            .build()
-            .unwrap(),
-    ));
+    let runtime = MonoioRuntime::new();
     b.to_async(&runtime).iter_custom(|iter| async move {
         let file = monoio::fs::File::open(path).await.unwrap();
 
@@ -197,11 +181,7 @@ fn read_all_compio(b: &mut Bencher, (path, len): &(&Path, u64)) {
 
 #[cfg(target_os = "linux")]
 fn read_all_monoio(b: &mut Bencher, (path, len): &(&Path, u64)) {
-    let runtime = MonoioRuntime(std::cell::RefCell::new(
-        monoio::RuntimeBuilder::<monoio::IoUringDriver>::new()
-            .build()
-            .unwrap(),
-    ));
+    let runtime = MonoioRuntime::new();
     b.to_async(&runtime).iter_custom(|iter| async move {
         let file = monoio::fs::File::open(path).await.unwrap();
         let mut buffer = Box::new([0u8; BUFFER_SIZE]);
@@ -363,11 +343,7 @@ fn write_compio_join(b: &mut Bencher, (path, offsets, content): &(&Path, &[u64],
 
 #[cfg(target_os = "linux")]
 fn write_monoio(b: &mut Bencher, (path, offsets, content): &(&Path, &[u64], &[u8])) {
-    let runtime = MonoioRuntime(std::cell::RefCell::new(
-        monoio::RuntimeBuilder::<monoio::IoUringDriver>::new()
-            .build()
-            .unwrap(),
-    ));
+    let runtime = MonoioRuntime::new();
     let content = content.to_vec();
     b.to_async(&runtime).iter_custom(|iter| {
         let mut content = content.clone();
@@ -393,11 +369,7 @@ fn write_monoio(b: &mut Bencher, (path, offsets, content): &(&Path, &[u64], &[u8
 
 #[cfg(target_os = "linux")]
 fn write_monoio_join(b: &mut Bencher, (path, offsets, content): &(&Path, &[u64], &[u8])) {
-    let runtime = MonoioRuntime(std::cell::RefCell::new(
-        monoio::RuntimeBuilder::<monoio::IoUringDriver>::new()
-            .build()
-            .unwrap(),
-    ));
+    let runtime = MonoioRuntime::new();
     let content = content.to_vec();
     b.to_async(&runtime).iter_custom(|iter| {
         let content = content.clone();
@@ -505,11 +477,7 @@ fn write_all_compio(b: &mut Bencher, (path, content): &(&Path, &[u8])) {
 
 #[cfg(target_os = "linux")]
 fn write_all_monoio(b: &mut Bencher, (path, content): &(&Path, &[u8])) {
-    let runtime = MonoioRuntime(std::cell::RefCell::new(
-        monoio::RuntimeBuilder::<monoio::IoUringDriver>::new()
-            .build()
-            .unwrap(),
-    ));
+    let runtime = MonoioRuntime::new();
     let content = content.to_vec();
     b.to_async(&runtime).iter_custom(|iter| {
         let mut content = content.clone();

compio/benches/monoio_wrap/mod.rs

@@ -0,0 +1,33 @@
+use std::{cell::RefCell, future::Future};
+
+use criterion::async_executor::AsyncExecutor;
+
+pub struct MonoioRuntime(RefCell<monoio::Runtime<monoio::IoUringDriver>>);
+
+impl Default for MonoioRuntime {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl MonoioRuntime {
+    pub fn new() -> Self {
+        Self(RefCell::new(
+            monoio::RuntimeBuilder::<monoio::IoUringDriver>::new()
+                .build()
+                .unwrap(),
+        ))
+    }
+}
+
+impl AsyncExecutor for MonoioRuntime {
+    fn block_on<T>(&self, future: impl Future<Output = T>) -> T {
+        self.0.borrow_mut().block_on(future)
+    }
+}
+
+impl AsyncExecutor for &MonoioRuntime {
+    fn block_on<T>(&self, future: impl Future<Output = T>) -> T {
+        self.0.borrow_mut().block_on(future)
+    }
+}

compio/benches/net.rs

@@ -3,6 +3,11 @@ use std::{net::Ipv4Addr, rc::Rc, time::Instant};
 use criterion::{criterion_group, criterion_main, Bencher, Criterion, Throughput};
 use rand::{thread_rng, RngCore};
 
+#[cfg(target_os = "linux")]
+mod monoio_wrap;
+#[cfg(target_os = "linux")]
+use monoio_wrap::MonoioRuntime;
+
 criterion_group!(net, echo);
 criterion_main!(net);
 
@@ -90,6 +95,53 @@ fn echo_compio(b: &mut Bencher, content: Rc<[u8; BUFFER_SIZE]>) {
     })
 }
 
+#[cfg(target_os = "linux")]
+fn echo_monoio(b: &mut Bencher, content: &[u8; BUFFER_SIZE]) {
+    use monoio::io::{AsyncReadRentExt, AsyncWriteRentExt};
+
+    let runtime = MonoioRuntime::new();
+    b.to_async(&runtime).iter_custom(|iter| {
+        let content = Box::new(*content);
+        async move {
+            let listener = monoio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
+            let addr = listener.local_addr().unwrap();
+
+            let start = Instant::now();
+            for _i in 0..iter {
+                let (mut tx, (mut rx, _)) = futures_util::try_join!(
+                    monoio::net::TcpStream::connect(addr),
+                    listener.accept()
+                )
+                .unwrap();
+
+                let client = async {
+                    let mut content = content.clone();
+                    let mut buffer = Box::new([0u8; BUFFER_SIZE]);
+                    let mut res;
+                    for _i in 0..BUFFER_COUNT {
+                        (res, content) = tx.write_all(content).await;
+                        res.unwrap();
+                        (res, buffer) = tx.read_exact(buffer).await;
+                        res.unwrap();
+                    }
+                };
+                let server = async move {
+                    let mut buffer = Box::new([0u8; BUFFER_SIZE]);
+                    let mut res;
+                    for _i in 0..BUFFER_COUNT {
+                        (res, buffer) = rx.read_exact(buffer).await;
+                        res.unwrap();
+                        (res, buffer) = rx.write_all(buffer).await;
+                        res.unwrap();
+                    }
+                };
+                futures_util::join!(client, server);
+            }
+            start.elapsed()
+        }
+    })
+}
+
 fn echo(c: &mut Criterion) {
     let mut rng = thread_rng();
 
@@ -102,6 +154,8 @@ fn echo(c: &mut Criterion) {
 
     group.bench_function("tokio", |b| echo_tokio(b, &content));
     group.bench_function("compio", |b| echo_compio(b, content.clone()));
+    #[cfg(target_os = "linux")]
+    group.bench_function("monoio", |b| echo_monoio(b, &content));
 
     group.finish();
 }