add flume performance test compare

Author: lispking

Cargo.lock

@@ -20,4 +20,5 @@ crossbeam-queue = "0.3"
 [dev-dependencies]
 tokio-test = "0.4"
 tracing-subscriber = "0.3"
+flume = "0.11.1"
 cargo-husky = { version = "1", features = ["precommit-hook", "run-cargo-test", "run-cargo-clippy", "run-cargo-fmt"] }

Cargo.toml

@@ -1,5 +1,7 @@
 mod tests;
 
+use tests::flume_bounded_io_test::run_flume_bounded_io_test;
+use tests::flume_bounded_test::run_flume_bounded_test;
 use tests::tokio_mpmc_io_test::run_tokio_mpmc_io_test;
 use tests::tokio_mpmc_test::run_tokio_mpmc_test;
 use tests::tokio_mpsc_io_test::run_tokio_mpsc_io_test;
@@ -20,6 +22,7 @@ async fn main() {
     )
     .await;
     run_tokio_mpsc_test(queue_size as u32, num_producers as u32).await;
+    run_flume_bounded_test(queue_size as u32, num_producers as u32).await;
 
     println!("============================================");
     println!("Starting IO tests");
@@ -30,4 +33,5 @@ async fn main() {
     )
     .await;
     run_tokio_mpsc_io_test(queue_size as u32, num_producers as u32).await;
+    run_flume_bounded_io_test(queue_size as u32, num_producers as u32).await;
 }

examples/performance-test.rs

@@ -0,0 +1,84 @@
+use std::fs::OpenOptions;
+use std::io::Write;
+use std::time::Instant;
+
+pub async fn run_flume_bounded_io_test(queue_size: u32, num_producers: u32) -> std::time::Duration {
+    let num_consumers = 1;
+    println!(
+        "Starting flume::bounded io performance test with queue size: {}, producers: {}, consumers: {}",
+        queue_size, num_producers, num_consumers
+    );
+
+    let (tx, rx) = flume::bounded(queue_size as usize);
+    let start_time = Instant::now();
+
+    // Spawn producer tasks
+    let mut producer_handles = vec![];
+    for i in 0..num_producers {
+        let tx_clone = tx.clone();
+        let handle = tokio::spawn(async move {
+            let messages_per_producer = queue_size / num_producers;
+            let start_message = i * messages_per_producer;
+            let end_message = start_message + messages_per_producer;
+            for msg in start_message..end_message {
+                if tx_clone.send(msg).is_err() {
+                    eprintln!("mpsc Producer {} failed to send message {}", i, msg);
+                    break;
+                }
+            }
+        });
+        producer_handles.push(handle);
+    }
+
+    // Spawn consumer tasks
+    let mut consumer_handles = vec![];
+    for i in 0..num_consumers {
+        let rx_clone = rx.clone();
+        let handle = tokio::spawn(async move {
+            let messages_per_consumer = queue_size / num_consumers;
+            let mut received_count = 0;
+            while received_count < messages_per_consumer {
+                match rx_clone.recv() {
+                    Ok(msg) => {
+                        received_count += 1;
+                        let mut file = OpenOptions::new()
+                            .append(true)
+                            .create(true)
+                            .open("output.txt")
+                            .expect("Failed to open file");
+                        writeln!(file, "Received message: {}", msg)
+                            .expect("Failed to write to file");
+                    }
+                    Err(e) => {
+                        eprintln!("Consumer {} failed to receive message: {:?}", i, e);
+                        break;
+                    }
+                }
+            }
+        });
+        consumer_handles.push(handle);
+    }
+
+    // Wait for all producers to finish
+    for handle in producer_handles {
+        let _ = handle.await;
+    }
+
+    // Wait for all consumers to finish
+    for handle in consumer_handles {
+        let _ = handle.await;
+    }
+
+    let duration = start_time.elapsed();
+
+    let _ = tokio::fs::remove_file("output.txt").await;
+    println!(
+        "flume::bounded performance test finished in: {:?}",
+        duration
+    );
+
+    // Drop the original sender to signal consumers that no more messages will be sent
+    drop(tx);
+
+    duration
+}

examples/tests/flume_bounded_io_test.rs

@@ -0,0 +1,74 @@
+use std::time::Instant;
+
+pub async fn run_flume_bounded_test(queue_size: u32, num_producers: u32) -> std::time::Duration {
+    let num_consumers = 1;
+    println!(
+        "Starting flume::bounded performance test with queue size: {}, producers: {}, consumers: {}",
+        queue_size, num_producers, num_consumers
+    );
+
+    let (tx, rx) = flume::bounded(queue_size as usize);
+    let start_time = Instant::now();
+
+    // Spawn producer tasks
+    let mut producer_handles = vec![];
+    for i in 0..num_producers {
+        let tx_clone = tx.clone();
+        let handle = tokio::spawn(async move {
+            let messages_per_producer = queue_size / num_producers;
+            let start_message = i * messages_per_producer;
+            let end_message = start_message + messages_per_producer;
+            for msg in start_message..end_message {
+                if tx_clone.send(msg).is_err() {
+                    eprintln!("mpsc Producer {} failed to send message {}", i, msg);
+                    break;
+                }
+            }
+        });
+        producer_handles.push(handle);
+    }
+
+    // Spawn consumer tasks
+    let mut consumer_handles = vec![];
+    for i in 0..num_consumers {
+        let rx_clone = rx.clone();
+        let handle = tokio::spawn(async move {
+            let messages_per_consumer = queue_size / num_consumers;
+            let mut received_count = 0;
+            while received_count < messages_per_consumer {
+                match rx_clone.recv() {
+                    Ok(_) => {
+                        received_count += 1;
+                    }
+                    Err(e) => {
+                        eprintln!("Consumer {} failed to receive message: {:?}", i, e);
+                        break;
+                    }
+                }
+            }
+        });
+        consumer_handles.push(handle);
+    }
+
+    // Wait for all producers to finish
+    for handle in producer_handles {
+        let _ = handle.await;
+    }
+
+    // Wait for all consumers to finish
+    for handle in consumer_handles {
+        let _ = handle.await;
+    }
+
+    let duration = start_time.elapsed();
+
+    println!(
+        "flume::bounded performance test finished in: {:?}",
+        duration
+    );
+
+    // Drop the original sender to signal consumers that no more messages will be sent
+    drop(tx);
+
+    duration
+}

examples/tests/flume_bounded_test.rs

@@ -1,3 +1,5 @@
+pub mod flume_bounded_io_test;
+pub mod flume_bounded_test;
 pub mod tokio_mpmc_io_test;
 pub mod tokio_mpmc_test;
 pub mod tokio_mpsc_io_test;