github/reth
1
1
Fork 0
mirror of https://github.com/paradigmxyz/reth.git synced 2026-02-19 03:04:27 -05:00
Code Issues Packages Projects Releases Wiki Activity

refactor: dedup runtime initializations (#22263)

Browse Source 
Co-authored-by: Alexey Shekhirin <github@shekhirin.com>
This commit is contained in:
DaniPopes
2026-02-17 18:35:31 +01:00
committed by GitHub
parent 6ff4f947c8
commit 0ba685386d
72 changed files with 275 additions and 318 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
crates/cli/commands/src/common.rs
View File

@@ -89,13 +89,15 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
/// Initializes environment according to [`AccessRights`] and returns an instance of
/// [`Environment`].
///
/// Internally builds a [`reth_tasks::Runtime`] attached to the current tokio handle for
/// parallel storage I/O.
pub fn init<N: CliNodeTypes>(&self, access: AccessRights) -> eyre::Result<Environment<N>>
/// The provided `runtime` is used for parallel storage I/O.
pub fn init<N: CliNodeTypes>(
&self,
access: AccessRights,
runtime: reth_tasks::Runtime,
) -> eyre::Result<Environment<N>>
where
C: ChainSpecParser<ChainSpec = N::ChainSpec>,
{
let runtime = reth_tasks::Runtime::with_existing_handle(tokio::runtime::Handle::current())?;
let data_dir = self.datadir.clone().resolve_datadir(self.chain.chain());
let db_path = data_dir.db();
let sf_path = data_dir.static_files();

3
crates/cli/commands/src/db/mod.rs
View File

@@ -83,7 +83,8 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C>
/// provided command.
macro_rules! db_exec {
($env:expr, $tool:ident, $N:ident, $access_rights:expr, $command:block) => {
let Environment { provider_factory, .. } = $env.init::<$N>($access_rights)?;
let Environment { provider_factory, .. } =
$env.init::<$N>($access_rights, ctx.task_executor.clone())?;
let $tool = DbTool::new(provider_factory)?;
$command;

4
crates/cli/commands/src/export_era.rs
View File

@@ -44,11 +44,11 @@ pub struct ExportArgs {
impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> ExportEraCommand<C> {
/// Execute `export-era` command
pub async fn execute<N>(self) -> eyre::Result<()>
pub async fn execute<N>(self, runtime: reth_tasks::Runtime) -> eyre::Result<()>
where
N: CliNodeTypes<ChainSpec = C::ChainSpec>,
{
let Environment { provider_factory, .. } = self.env.init::<N>(AccessRights::RO)?;
let Environment { provider_factory, .. } = self.env.init::<N>(AccessRights::RO, runtime)?;
// Either specified path or default to `<data-dir>/<chain>/era1-export/`
let data_dir = match &self.export.path {

5
crates/cli/commands/src/import.rs
View File

@@ -47,6 +47,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> ImportComm
pub async fn execute<N, Comp>(
self,
components: impl FnOnce(Arc<N::ChainSpec>) -> Comp,
runtime: reth_tasks::Runtime,
) -> eyre::Result<()>
where
N: CliNodeTypes<ChainSpec = C::ChainSpec>,
@@ -54,7 +55,8 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> ImportComm
{
info!(target: "reth::cli", "reth {} starting", version_metadata().short_version);
let Environment { provider_factory, config, .. } = self.env.init::<N>(AccessRights::RW)?;
let Environment { provider_factory, config, .. } =
self.env.init::<N>(AccessRights::RW, runtime.clone())?;
let components = components(provider_factory.chain_spec());
@@ -85,6 +87,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> ImportComm
&config,
executor.clone(),
consensus.clone(),
runtime.clone(),
)
.await?;

17
crates/cli/commands/src/import_core.rs
View File

@@ -87,6 +87,7 @@ pub async fn import_blocks_from_file<N>(
config: &Config,
executor: impl ConfigureEvm<Primitives = N::Primitives> + 'static,
consensus: Arc<impl FullConsensus<N::Primitives> + 'static>,
runtime: reth_tasks::Runtime,
) -> eyre::Result<ImportResult>
where
N: ProviderNodeTypes,
@@ -139,7 +140,7 @@ where
total_decoded_blocks += file_client.headers_len();
total_decoded_txns += file_client.total_transactions();
let (mut pipeline, events, _runtime) = build_import_pipeline_impl(
let (mut pipeline, events) = build_import_pipeline_impl(
config,
provider_factory.clone(),
&consensus,
@@ -147,6 +148,7 @@ where
static_file_producer.clone(),
import_config.no_state,
executor.clone(),
runtime.clone(),
)?;
// override the tip
@@ -257,6 +259,7 @@ where
///
/// If configured to execute, all stages will run. Otherwise, only stages that don't require state
/// will run.
#[expect(clippy::too_many_arguments)]
pub fn build_import_pipeline_impl<N, C, E>(
config: &Config,
provider_factory: ProviderFactory<N>,
@@ -265,11 +268,8 @@ pub fn build_import_pipeline_impl<N, C, E>(
static_file_producer: StaticFileProducer<ProviderFactory<N>>,
disable_exec: bool,
evm_config: E,
) -> eyre::Result<(
Pipeline<N>,
impl futures::Stream<Item = NodeEvent<N::Primitives>> + use<N, C, E>,
reth_tasks::Runtime,
)>
runtime: reth_tasks::Runtime,
) -> eyre::Result<(Pipeline<N>, impl futures::Stream<Item = NodeEvent<N::Primitives>> + use<N, C, E>)>
where
N: ProviderNodeTypes,
C: FullConsensus<N::Primitives> + 'static,
@@ -285,9 +285,6 @@ where
.sealed_header(last_block_number)?
.ok_or_else(|| ProviderError::HeaderNotFound(last_block_number.into()))?;
let runtime = reth_tasks::Runtime::with_existing_handle(tokio::runtime::Handle::current())
.expect("failed to create runtime");
let mut header_downloader = ReverseHeadersDownloaderBuilder::new(config.stages.headers)
.build(file_client.clone(), consensus.clone())
.into_task_with(&runtime);
@@ -333,5 +330,5 @@ where
let events = pipeline.events().map(Into::into);
Ok((pipeline, events, runtime))
Ok((pipeline, events))
}

5
crates/cli/commands/src/import_era.rs
View File

@@ -64,13 +64,14 @@ impl TryFromChain for ChainKind {
impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> ImportEraCommand<C> {
/// Execute `import-era` command
pub async fn execute<N>(self) -> eyre::Result<()>
pub async fn execute<N>(self, runtime: reth_tasks::Runtime) -> eyre::Result<()>
where
N: CliNodeTypes<ChainSpec = C::ChainSpec>,
{
info!(target: "reth::cli", "reth {} starting", version_metadata().short_version);
let Environment { provider_factory, config, .. } = self.env.init::<N>(AccessRights::RW)?;
let Environment { provider_factory, config, .. } =
self.env.init::<N>(AccessRights::RW, runtime)?;
let mut hash_collector = Collector::new(config.stages.etl.file_size, config.stages.etl.dir);

7
crates/cli/commands/src/init_cmd.rs
View File

@@ -18,10 +18,13 @@ pub struct InitCommand<C: ChainSpecParser> {
impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> InitCommand<C> {
/// Execute the `init` command
pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec>>(self) -> eyre::Result<()> {
pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec>>(
self,
runtime: reth_tasks::Runtime,
) -> eyre::Result<()> {
info!(target: "reth::cli", "reth init starting");
let Environment { provider_factory, .. } = self.env.init::<N>(AccessRights::RW)?;
let Environment { provider_factory, .. } = self.env.init::<N>(AccessRights::RW, runtime)?;
let genesis_block_number = provider_factory.chain_spec().genesis_header().number();
let hash = provider_factory

5
crates/cli/commands/src/init_state/mod.rs
View File

@@ -65,7 +65,7 @@ pub struct InitStateCommand<C: ChainSpecParser> {
impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> InitStateCommand<C> {
/// Execute the `init` command
pub async fn execute<N>(self) -> eyre::Result<()>
pub async fn execute<N>(self, runtime: reth_tasks::Runtime) -> eyre::Result<()>
where
N: CliNodeTypes<
ChainSpec = C::ChainSpec,
@@ -74,7 +74,8 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> InitStateC
{
info!(target: "reth::cli", "Reth init-state starting");
let Environment { config, provider_factory, .. } = self.env.init::<N>(AccessRights::RW)?;
let Environment { config, provider_factory, .. } =
self.env.init::<N>(AccessRights::RW, runtime)?;
let static_file_provider = provider_factory.static_file_provider();
let provider_rw = provider_factory.database_provider_rw()?;

24
crates/cli/commands/src/p2p/mod.rs
View File

@@ -16,6 +16,7 @@ use reth_node_core::{
args::{DatadirArgs, NetworkArgs},
utils::get_single_header,
};
use reth_tasks::Runtime;
pub mod bootnode;
pub mod enode;
@@ -194,17 +195,18 @@ impl<C: ChainSpecParser> DownloadArgs<C> {
let rlpx_socket = (self.network.addr, self.network.port).into();
let boot_nodes = self.chain.bootnodes().unwrap_or_default();
let net = NetworkConfigBuilder::<N::NetworkPrimitives>::new(p2p_secret_key)
.peer_config(config.peers_config_with_basic_nodes_from_file(None))
.external_ip_resolver(self.network.nat.clone())
.network_id(self.network.network_id)
.boot_nodes(boot_nodes.clone())
.apply(|builder| {
self.network.discovery.apply_to_builder(builder, rlpx_socket, boot_nodes)
})
.build_with_noop_provider(self.chain.clone())
.manager()
.await?;
let net =
NetworkConfigBuilder::<N::NetworkPrimitives>::new(p2p_secret_key, Runtime::test())
.peer_config(config.peers_config_with_basic_nodes_from_file(None))
.external_ip_resolver(self.network.nat.clone())
.network_id(self.network.network_id)
.boot_nodes(boot_nodes.clone())
.apply(|builder| {
self.network.discovery.apply_to_builder(builder, rlpx_socket, boot_nodes)
})
.build_with_noop_provider(self.chain.clone())
.manager()
.await?;
let handle = net.handle().clone();
tokio::task::spawn(net);

2
crates/cli/commands/src/prune.rs
View File

@@ -36,7 +36,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> PruneComma
self,
ctx: CliContext,
) -> eyre::Result<()> {
let env = self.env.init::<N>(AccessRights::RW)?;
let env = self.env.init::<N>(AccessRights::RW, ctx.task_executor.clone())?;
let provider_factory = env.provider_factory;
let config = env.config.prune;
let data_dir = env.data_dir;

8
crates/cli/commands/src/re_execute.rs
View File

@@ -60,11 +60,15 @@ impl<C: ChainSpecParser> Command<C> {
impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>> Command<C> {
/// Execute `re-execute` command
pub async fn execute<N>(self, components: impl CliComponentsBuilder<N>) -> eyre::Result<()>
pub async fn execute<N>(
self,
components: impl CliComponentsBuilder<N>,
runtime: reth_tasks::Runtime,
) -> eyre::Result<()>
where
N: CliNodeTypes<ChainSpec = C::ChainSpec>,
{
let Environment { provider_factory, .. } = self.env.init::<N>(AccessRights::RO)?;
let Environment { provider_factory, .. } = self.env.init::<N>(AccessRights::RO, runtime)?;
let components = components(provider_factory.chain_spec());

4
crates/cli/commands/src/stage/drop.rs
View File

@@ -37,11 +37,11 @@ pub struct Command<C: ChainSpecParser> {
impl<C: ChainSpecParser> Command<C> {
/// Execute `db` command
pub async fn execute<N: CliNodeTypes>(self) -> eyre::Result<()>
pub async fn execute<N: CliNodeTypes>(self, runtime: reth_tasks::Runtime) -> eyre::Result<()>
where
C: ChainSpecParser<ChainSpec = N::ChainSpec>,
{
let Environment { provider_factory, .. } = self.env.init::<N>(AccessRights::RW)?;
let Environment { provider_factory, .. } = self.env.init::<N>(AccessRights::RW, runtime)?;
let tool = DbTool::new(provider_factory)?;

3
crates/cli/commands/src/stage/dump/execution.rs
View File

@@ -16,6 +16,7 @@ use reth_stages::{stages::ExecutionStage, Stage, StageCheckpoint, UnwindInput};
use std::sync::Arc;
use tracing::info;
#[expect(clippy::too_many_arguments)]
pub(crate) async fn dump_execution_stage<N, E, C>(
db_tool: &DbTool<N>,
from: u64,
@@ -24,6 +25,7 @@ pub(crate) async fn dump_execution_stage<N, E, C>(
should_run: bool,
evm_config: E,
consensus: C,
runtime: reth_tasks::Runtime,
) -> eyre::Result<()>
where
N: ProviderNodeTypes<DB = DatabaseEnv>,
@@ -37,7 +39,6 @@ where
unwind_and_copy(db_tool, from, tip_block_number, &output_db, evm_config.clone())?;
if should_run {
let runtime = reth_tasks::Runtime::with_existing_handle(tokio::runtime::Handle::current())?;
dry_run(
ProviderFactory::<N>::new(
output_db,

2
crates/cli/commands/src/stage/dump/hashing_account.rs
View File

@@ -18,6 +18,7 @@ pub(crate) async fn dump_hashing_account_stage<N: ProviderNodeTypes<DB = Databas
to: BlockNumber,
output_datadir: ChainPath<DataDirPath>,
should_run: bool,
runtime: reth_tasks::Runtime,
) -> Result<()> {
let (output_db, tip_block_number) = setup(from, to, &output_datadir.db(), db_tool)?;
@@ -33,7 +34,6 @@ pub(crate) async fn dump_hashing_account_stage<N: ProviderNodeTypes<DB = Databas
unwind_and_copy(db_tool, from, tip_block_number, &output_db)?;
if should_run {
let runtime = reth_tasks::Runtime::with_existing_handle(tokio::runtime::Handle::current())?;
dry_run(
ProviderFactory::<N>::new(
output_db,

2
crates/cli/commands/src/stage/dump/hashing_storage.rs
View File

@@ -17,13 +17,13 @@ pub(crate) async fn dump_hashing_storage_stage<N: ProviderNodeTypes<DB = Databas
to: u64,
output_datadir: ChainPath<DataDirPath>,
should_run: bool,
runtime: reth_tasks::Runtime,
) -> Result<()> {
let (output_db, tip_block_number) = setup(from, to, &output_datadir.db(), db_tool)?;
unwind_and_copy(db_tool, from, tip_block_number, &output_db)?;
if should_run {
let runtime = reth_tasks::Runtime::with_existing_handle(tokio::runtime::Handle::current())?;
dry_run(
ProviderFactory::<N>::new(
output_db,

3
crates/cli/commands/src/stage/dump/merkle.rs
View File

@@ -24,6 +24,7 @@ use reth_stages::{
};
use tracing::info;
#[expect(clippy::too_many_arguments)]
pub(crate) async fn dump_merkle_stage<N>(
db_tool: &DbTool<N>,
from: BlockNumber,
@@ -32,6 +33,7 @@ pub(crate) async fn dump_merkle_stage<N>(
should_run: bool,
evm_config: impl ConfigureEvm<Primitives = N::Primitives>,
consensus: impl FullConsensus<N::Primitives> + 'static,
runtime: reth_tasks::Runtime,
) -> Result<()>
where
N: ProviderNodeTypes<DB = DatabaseEnv>,
@@ -57,7 +59,6 @@ where
unwind_and_copy(db_tool, (from, to), tip_block_number, &output_db, evm_config, consensus)?;
if should_run {
let runtime = reth_tasks::Runtime::with_existing_handle(tokio::runtime::Handle::current())?;
dry_run(
ProviderFactory::<N>::new(
output_db,

37
crates/cli/commands/src/stage/dump/mod.rs
View File

@@ -72,30 +72,36 @@ pub struct StageCommand {
}
macro_rules! handle_stage {
($stage_fn:ident, $tool:expr, $command:expr) => {{
($stage_fn:ident, $tool:expr, $command:expr, $runtime:expr) => {{
let StageCommand { output_datadir, from, to, dry_run, .. } = $command;
let output_datadir =
output_datadir.with_chain($tool.chain().chain(), DatadirArgs::default());
$stage_fn($tool, *from, *to, output_datadir, *dry_run).await?
$stage_fn($tool, *from, *to, output_datadir, *dry_run, $runtime).await?
}};
($stage_fn:ident, $tool:expr, $command:expr, $executor:expr, $consensus:expr) => {{
($stage_fn:ident, $tool:expr, $command:expr, $executor:expr, $consensus:expr, $runtime:expr) => {{
let StageCommand { output_datadir, from, to, dry_run, .. } = $command;
let output_datadir =
output_datadir.with_chain($tool.chain().chain(), DatadirArgs::default());
$stage_fn($tool, *from, *to, output_datadir, *dry_run, $executor, $consensus).await?
$stage_fn($tool, *from, *to, output_datadir, *dry_run, $executor, $consensus, $runtime)
.await?
}};
}
impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C> {
/// Execute `dump-stage` command
pub async fn execute<N, Comp, F>(self, components: F) -> eyre::Result<()>
pub async fn execute<N, Comp, F>(
self,
components: F,
runtime: reth_tasks::Runtime,
) -> eyre::Result<()>
where
N: CliNodeTypes<ChainSpec = C::ChainSpec>,
Comp: CliNodeComponents<N>,
F: FnOnce(Arc<C::ChainSpec>) -> Comp,
{
let Environment { provider_factory, .. } = self.env.init::<N>(AccessRights::RO)?;
let Environment { provider_factory, .. } =
self.env.init::<N>(AccessRights::RO, runtime.clone())?;
let tool = DbTool::new(provider_factory)?;
let components = components(tool.chain());
let evm_config = components.evm_config().clone();
@@ -103,12 +109,23 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C>
match &self.command {
Stages::Execution(cmd) => {
handle_stage!(dump_execution_stage, &tool, cmd, evm_config, consensus)
handle_stage!(
dump_execution_stage,
&tool,
cmd,
evm_config,
consensus,
runtime.clone()
)
}
Stages::StorageHashing(cmd) => {
handle_stage!(dump_hashing_storage_stage, &tool, cmd, runtime.clone())
}
Stages::AccountHashing(cmd) => {
handle_stage!(dump_hashing_account_stage, &tool, cmd, runtime.clone())
}
Stages::StorageHashing(cmd) => handle_stage!(dump_hashing_storage_stage, &tool, cmd),
Stages::AccountHashing(cmd) => handle_stage!(dump_hashing_account_stage, &tool, cmd),
Stages::Merkle(cmd) => {
handle_stage!(dump_merkle_stage, &tool, cmd, evm_config, consensus)
handle_stage!(dump_merkle_stage, &tool, cmd, evm_config, consensus, runtime.clone())
}
}

7
crates/cli/commands/src/stage/mod.rs
View File

@@ -49,11 +49,12 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
N: CliNodeTypes<ChainSpec = C::ChainSpec>,
Comp: CliNodeComponents<N>,
{
let executor = ctx.task_executor.clone();
match self.command {
Subcommands::Run(command) => command.execute::<N, _, _>(ctx, components).await,
Subcommands::Drop(command) => command.execute::<N>().await,
Subcommands::Dump(command) => command.execute::<N, _, _>(components).await,
Subcommands::Unwind(command) => command.execute::<N, _, _>(components).await,
Subcommands::Drop(command) => command.execute::<N>(executor).await,
Subcommands::Dump(command) => command.execute::<N, _, _>(components, executor).await,
Subcommands::Unwind(command) => command.execute::<N, _, _>(components, executor).await,
}
}
}

5
crates/cli/commands/src/stage/run.rs
View File

@@ -119,8 +119,9 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
// Does not do anything on windows.
let _ = fdlimit::raise_fd_limit();
let runtime = ctx.task_executor.clone();
let Environment { provider_factory, config, data_dir } =
self.env.init::<N>(AccessRights::RW)?;
self.env.init::<N>(AccessRights::RW, ctx.task_executor.clone())?;
let mut provider_rw = provider_factory.database_provider_rw()?;
let components = components(provider_factory.chain_spec());
@@ -171,6 +172,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
provider_factory.chain_spec(),
p2p_secret_key,
default_peers_path,
runtime.clone(),
)
.build(provider_factory.clone())
.start_network()
@@ -226,6 +228,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
provider_factory.chain_spec(),
p2p_secret_key,
default_peers_path,
runtime.clone(),
)
.build(provider_factory.clone())
.start_network()

4
crates/cli/commands/src/stage/unwind.rs
View File

@@ -46,12 +46,14 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C>
pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec>, F, Comp>(
self,
components: F,
runtime: reth_tasks::Runtime,
) -> eyre::Result<()>
where
Comp: CliNodeComponents<N>,
F: FnOnce(Arc<C::ChainSpec>) -> Comp,
{
let Environment { provider_factory, config, .. } = self.env.init::<N>(AccessRights::RW)?;
let Environment { provider_factory, config, .. } =
self.env.init::<N>(AccessRights::RW, runtime)?;
let target = self.command.unwind_target(provider_factory.clone())?;

5
crates/cli/runner/src/lib.rs
View File

@@ -47,6 +47,11 @@ impl CliRunner {
self
}
/// Returns a clone of the underlying [`Runtime`](reth_tasks::Runtime).
pub fn runtime(&self) -> reth_tasks::Runtime {
self.runtime.clone()
}
/// Executes an async block on the runtime and blocks until completion.
pub fn block_on<F, T>(&self, fut: F) -> T
where