reth_bench/bench/
new_payload_only.rs

1//! Runs the `reth bench` command, sending only newPayload, without a forkchoiceUpdated call.
2
3use crate::{
4    bench::{
5        context::BenchContext,
6        output::{
7            NewPayloadResult, TotalGasOutput, TotalGasRow, GAS_OUTPUT_SUFFIX,
8            NEW_PAYLOAD_OUTPUT_SUFFIX,
9        },
10    },
11    valid_payload::{block_to_new_payload, call_new_payload},
12};
13use alloy_provider::Provider;
14use clap::Parser;
15use csv::Writer;
16use eyre::{Context, OptionExt};
17use reth_cli_runner::CliContext;
18use reth_node_core::args::BenchmarkArgs;
19use std::time::{Duration, Instant};
20use tracing::{debug, info};
21
22/// `reth benchmark new-payload-only` command
23#[derive(Debug, Parser)]
24pub struct Command {
25    /// The RPC url to use for getting data.
26    #[arg(long, value_name = "RPC_URL", verbatim_doc_comment)]
27    rpc_url: String,
28
29    /// The size of the block buffer (channel capacity) for prefetching blocks from the RPC
30    /// endpoint.
31    #[arg(
32        long = "rpc-block-buffer-size",
33        value_name = "RPC_BLOCK_BUFFER_SIZE",
34        default_value = "20",
35        verbatim_doc_comment
36    )]
37    rpc_block_buffer_size: usize,
38
39    #[command(flatten)]
40    benchmark: BenchmarkArgs,
41}
42
43impl Command {
44    /// Execute `benchmark new-payload-only` command
45    pub async fn execute(self, _ctx: CliContext) -> eyre::Result<()> {
46        let BenchContext {
47            benchmark_mode,
48            block_provider,
49            auth_provider,
50            mut next_block,
51            is_optimism,
52            ..
53        } = BenchContext::new(&self.benchmark, self.rpc_url).await?;
54
55        let total_blocks = benchmark_mode.total_blocks();
56        let buffer_size = self.rpc_block_buffer_size;
57
58        // Use a oneshot channel to propagate errors from the spawned task
59        let (error_sender, mut error_receiver) = tokio::sync::oneshot::channel();
60        let (sender, mut receiver) = tokio::sync::mpsc::channel(buffer_size);
61
62        tokio::task::spawn(async move {
63            while benchmark_mode.contains(next_block) {
64                let block_res = block_provider
65                    .get_block_by_number(next_block.into())
66                    .full()
67                    .await
68                    .wrap_err_with(|| format!("Failed to fetch block by number {next_block}"));
69                let block = match block_res.and_then(|opt| opt.ok_or_eyre("Block not found")) {
70                    Ok(block) => block,
71                    Err(e) => {
72                        tracing::error!(target: "reth-bench", "Failed to fetch block {next_block}: {e}");
73                        let _ = error_sender.send(e);
74                        break;
75                    }
76                };
77
78                next_block += 1;
79                if let Err(e) = sender.send(block).await {
80                    tracing::error!(target: "reth-bench", "Failed to send block data: {e}");
81                    break;
82                }
83            }
84        });
85
86        let mut results = Vec::new();
87        let mut blocks_processed = 0u64;
88        let total_benchmark_duration = Instant::now();
89        let mut total_wait_time = Duration::ZERO;
90
91        while let Some(block) = {
92            let wait_start = Instant::now();
93            let result = receiver.recv().await;
94            total_wait_time += wait_start.elapsed();
95            result
96        } {
97            let block_number = block.header.number;
98            let transaction_count = block.transactions.len() as u64;
99            let gas_used = block.header.gas_used;
100
101            debug!(target: "reth-bench", number=?block.header.number, "Sending payload to engine");
102
103            let (version, params) = block_to_new_payload(block, is_optimism)?;
104
105            let start = Instant::now();
106            call_new_payload(&auth_provider, version, params).await?;
107
108            let new_payload_result = NewPayloadResult { gas_used, latency: start.elapsed() };
109            blocks_processed += 1;
110            let progress = match total_blocks {
111                Some(total) => format!("{blocks_processed}/{total}"),
112                None => format!("{blocks_processed}"),
113            };
114            info!(target: "reth-bench", progress, %new_payload_result);
115
116            // current duration since the start of the benchmark minus the time
117            // waiting for blocks
118            let current_duration = total_benchmark_duration.elapsed() - total_wait_time;
119
120            // record the current result
121            let row =
122                TotalGasRow { block_number, transaction_count, gas_used, time: current_duration };
123            results.push((row, new_payload_result));
124        }
125
126        // Check if the spawned task encountered an error
127        if let Ok(error) = error_receiver.try_recv() {
128            return Err(error);
129        }
130
131        let (gas_output_results, new_payload_results): (_, Vec<NewPayloadResult>) =
132            results.into_iter().unzip();
133
134        // write the csv output to files
135        if let Some(path) = self.benchmark.output {
136            // first write the new payload results to a file
137            let output_path = path.join(NEW_PAYLOAD_OUTPUT_SUFFIX);
138            info!(target: "reth-bench", "Writing newPayload call latency output to file: {:?}", output_path);
139            let mut writer = Writer::from_path(output_path)?;
140            for result in new_payload_results {
141                writer.serialize(result)?;
142            }
143            writer.flush()?;
144
145            // now write the gas output to a file
146            let output_path = path.join(GAS_OUTPUT_SUFFIX);
147            info!(target: "reth-bench", "Writing total gas output to file: {:?}", output_path);
148            let mut writer = Writer::from_path(output_path)?;
149            for row in &gas_output_results {
150                writer.serialize(row)?;
151            }
152            writer.flush()?;
153
154            info!(target: "reth-bench", "Finished writing benchmark output files to {:?}.", path);
155        }
156
157        // accumulate the results and calculate the overall Ggas/s
158        let gas_output = TotalGasOutput::new(gas_output_results)?;
159        info!(
160            target: "reth-bench",
161            total_duration=?gas_output.total_duration,
162            total_gas_used=?gas_output.total_gas_used,
163            blocks_processed=?gas_output.blocks_processed,
164            "Total Ggas/s: {:.4}",
165            gas_output.total_gigagas_per_second()
166        );
167
168        Ok(())
169    }
170}
reth_bench/bench/new_payload_only.rs

reth_bench/bench/
new_payload_only.rs
