Add benchmark for txfetcher (#134)

## 📝 Summary

This PR adds a benchmark for the txfetcher service. It only benches
normal Ethereum transactions.

## 💡 Motivation and Context

<!--- (Optional) Why is this change required? What problem does it
solve? Remove this section if not applicable. -->

---

## ✅ I have completed the following steps:

* [x] Run `make lint`
* [x] Run `make test`
* [x] Added tests (if applicable)

.github/workflows/bench.yaml

@@ -47,6 +47,11 @@ jobs:
           echo "BASE_SHA_SHORT=${BASE_SHA_SHORT}" >> $fn_vars
           cat $fn_vars
 
+      - name: Install Foundry toolchain
+        uses: foundry-rs/foundry-toolchain@v1
+        with:
+          version: nightly
+
       # RUN BENCHMARKS
       - run: make bench-in-ci
 

crates/rbuilder/Cargo.toml

@@ -127,7 +127,7 @@ built = { version = "0.7.1", features = ["git2", "chrono"] }
 
 [dev-dependencies]
 tempfile = "3.8"
-criterion = { version = "0.5.1", features = ["html_reports"] }
+criterion = { version = "0.5.1", features = ["html_reports", "async_tokio"] }
 
 [[bench]]
 name = "bench_main"

crates/rbuilder/benches/bench_main.rs

@@ -4,4 +4,5 @@ mod benchmarks;
 
 criterion_main! {
     benchmarks::mev_boost::serialization,
+    benchmarks::txpool_fetcher::txpool,
 }

crates/rbuilder/benches/benchmarks/mod.rs

@@ -1 +1,2 @@
 pub mod mev_boost;
+pub mod txpool_fetcher;

crates/rbuilder/benches/benchmarks/txpool_fetcher.rs

@@ -0,0 +1,71 @@
+use alloy_network::{EthereumWallet, TransactionBuilder};
+use alloy_node_bindings::Anvil;
+use alloy_primitives::U256;
+use alloy_provider::{Provider, ProviderBuilder};
+use alloy_rpc_types::TransactionRequest;
+use alloy_signer_local::PrivateKeySigner;
+use criterion::{criterion_group, Criterion};
+use rbuilder::live_builder::order_input::{
+    txpool_fetcher::subscribe_to_txpool_with_blobs, OrderInputConfig,
+};
+use std::time::Duration;
+use tokio::sync::mpsc;
+use tokio_util::sync::CancellationToken;
+
+async fn txpool_receive_util(count: u32) {
+    let anvil = Anvil::new()
+        .args(["--ipc", "/tmp/anvil.ipc"])
+        .try_spawn()
+        .unwrap();
+
+    let (sender, mut receiver) = mpsc::channel(10);
+    subscribe_to_txpool_with_blobs(
+        OrderInputConfig::default_e2e(),
+        sender,
+        CancellationToken::new(),
+    )
+    .await
+    .unwrap();
+
+    let signer: PrivateKeySigner = anvil.keys()[0].clone().into();
+    let wallet = EthereumWallet::from(signer);
+
+    let provider = ProviderBuilder::new()
+        .with_recommended_fillers()
+        .wallet(wallet)
+        .on_http(anvil.endpoint().parse().unwrap());
+
+    let alice = anvil.addresses()[0];
+    let eip1559_est = provider.estimate_eip1559_fees(None).await.unwrap();
+
+    let tx = TransactionRequest::default()
+        .with_to(alice)
+        .with_value(U256::from(1))
+        .with_max_fee_per_gas(eip1559_est.max_fee_per_gas)
+        .with_max_priority_fee_per_gas(eip1559_est.max_priority_fee_per_gas);
+
+    tokio::spawn(async move {
+        for _ in 0..count {
+            let _ = provider.send_transaction(tx.clone()).await.unwrap();
+        }
+    });
+
+    for _ in 0..count {
+        let _ = receiver.recv().await.unwrap();
+    }
+}
+
+fn bench_txpool_receive(c: &mut Criterion) {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    let mut group = c.benchmark_group("Txpool fetcher");
+
+    group.measurement_time(Duration::from_secs(20));
+    group.bench_function("txn_fetcher_normal_10", |b| {
+        b.to_async(&rt).iter(|| txpool_receive_util(10));
+    });
+    group.bench_function("txn_fetcher_normal_100", |b| {
+        b.to_async(&rt).iter(|| txpool_receive_util(100));
+    });
+}
+
+criterion_group!(txpool, bench_txpool_receive,);

crates/rbuilder/src/live_builder/order_input/mod.rs

@@ -138,6 +138,19 @@ impl OrderInputConfig {
             input_channel_buffer_size: 10_000,
         }
     }
+
+    pub fn default_e2e() -> Self {
+        Self {
+            ipc_path: PathBuf::from("/tmp/anvil.ipc"),
+            results_channel_timeout: Duration::new(5, 0),
+            ignore_cancellable_orders: false,
+            ignore_blobs: false,
+            input_channel_buffer_size: 10,
+            serve_max_connections: 4096,
+            server_ip: Ipv4Addr::new(127, 0, 0, 1),
+            server_port: 0,
+        }
+    }
 }
 
 /// Commands we can get from RPC or mempool fetcher.

crates/rbuilder/src/live_builder/order_input/txpool_fetcher.rs

@@ -122,21 +122,6 @@ mod test {
     use alloy_provider::{Provider, ProviderBuilder};
     use alloy_rpc_types::TransactionRequest;
     use alloy_signer_local::PrivateKeySigner;
-    use std::{net::Ipv4Addr, path::PathBuf};
-    use tokio::time::Duration;
-
-    fn default_config() -> OrderInputConfig {
-        OrderInputConfig {
-            ipc_path: PathBuf::from("/tmp/anvil.ipc"),
-            results_channel_timeout: Duration::new(5, 0),
-            ignore_cancellable_orders: false,
-            ignore_blobs: false,
-            input_channel_buffer_size: 10,
-            serve_max_connections: 4096,
-            server_ip: Ipv4Addr::new(127, 0, 0, 1),
-            server_port: 0,
-        }
-    }
 
     #[tokio::test]
     /// Test that the fetcher can retrieve transactions (both normal and blob) from the txpool
@@ -147,9 +132,13 @@ mod test {
             .unwrap();
 
         let (sender, mut receiver) = mpsc::channel(10);
-        subscribe_to_txpool_with_blobs(default_config(), sender, CancellationToken::new())
-            .await
-            .unwrap();
+        subscribe_to_txpool_with_blobs(
+            OrderInputConfig::default_e2e(),
+            sender,
+            CancellationToken::new(),
+        )
+        .await
+        .unwrap();
 
         let signer: PrivateKeySigner = anvil.keys()[0].clone().into();
         let wallet = EthereumWallet::from(signer);