chore: run commands as blocking task (#2286)

bin/reth/src/chain/init.rs

@@ -41,7 +41,7 @@ pub struct InitCommand {
 
 impl InitCommand {
     /// Execute the `init` command
-    pub async fn execute(&self) -> eyre::Result<()> {
+    pub async fn execute(self) -> eyre::Result<()> {
         info!(target: "reth::cli", "reth init starting");
 
         // add network name to db directory
@@ -52,7 +52,7 @@ impl InitCommand {
         info!(target: "reth::cli", "Database opened");
 
         info!(target: "reth::cli", "Writing genesis block");
-        let hash = init_genesis(db, self.chain.clone())?;
+        let hash = init_genesis(db, self.chain)?;
 
         info!(target: "reth::cli", hash = ?hash, "Genesis block written");
         Ok(())

bin/reth/src/cli.rs

@@ -1,6 +1,4 @@
 //! CLI definition and entrypoint to executable
-use std::str::FromStr;
-
 use crate::{
     chain, config, db,
     dirs::{LogsDir, PlatformPath},
@@ -14,6 +12,7 @@ use reth_tracing::{
     tracing_subscriber::{filter::Directive, registry::LookupSpan},
     BoxedLayer, FileWorkerGuard,
 };
+use std::str::FromStr;
 
 /// Parse CLI options, set up logging and run the chosen command.
 pub fn run() -> eyre::Result<()> {
@@ -29,12 +28,16 @@ pub fn run() -> eyre::Result<()> {
 
     match opt.command {
         Commands::Node(command) => runner.run_command_until_exit(|ctx| command.execute(ctx)),
-        Commands::Init(command) => runner.run_until_ctrl_c(command.execute()),
-        Commands::Import(command) => runner.run_until_ctrl_c(command.execute()),
-        Commands::Db(command) => runner.run_until_ctrl_c(command.execute()),
-        Commands::Stage(command) => runner.run_until_ctrl_c(command.execute()),
-        Commands::DumpStage(command) => runner.run_until_ctrl_c(command.execute()),
-        Commands::DropStage(command) => runner.run_until_ctrl_c(command.execute()),
+        Commands::Init(command) => runner.run_blocking_until_ctrl_c(command.execute()),
+        Commands::Import(command) => runner.run_blocking_until_ctrl_c(command.execute()),
+        Commands::Db(command) => runner.run_blocking_until_ctrl_c(command.execute()),
+        Commands::Stage(command) => runner.run_blocking_until_ctrl_c(command.execute()),
+        Commands::DumpStage(command) => {
+            // TODO: This should be run_blocking_until_ctrl_c as well, but fails to compile due to
+            // weird compiler GAT issues.
+            runner.run_until_ctrl_c(command.execute())
+        }
+        Commands::DropStage(command) => runner.run_blocking_until_ctrl_c(command.execute()),
         Commands::P2P(command) => runner.run_until_ctrl_c(command.execute()),
         Commands::TestVectors(command) => runner.run_until_ctrl_c(command.execute()),
         Commands::TestEthChain(command) => runner.run_until_ctrl_c(command.execute()),

bin/reth/src/db/mod.rs

@@ -1,6 +1,4 @@
 //! Database debugging tool
-use std::sync::Arc;
-
 use crate::{
     dirs::{DbPath, MaybePlatformPath},
     utils::DbTool,
@@ -12,6 +10,7 @@ use human_bytes::human_bytes;
 use reth_db::{database::Database, tables};
 use reth_primitives::ChainSpec;
 use reth_staged_sync::utils::chainspec::genesis_value_parser;
+use std::sync::Arc;
 use tracing::error;
 
 /// DB List TUI
@@ -85,7 +84,7 @@ pub struct ListArgs {
 
 impl Command {
     /// Execute `db` command
-    pub async fn execute(&self) -> eyre::Result<()> {
+    pub async fn execute(self) -> eyre::Result<()> {
         // add network name to db directory
         let db_path = self.db.unwrap_or_chain_default(self.chain.chain);
 

bin/reth/src/drop_stage.rs

@@ -54,7 +54,7 @@ pub struct Command {
 
 impl Command {
     /// Execute `db` command
-    pub async fn execute(&self) -> eyre::Result<()> {
+    pub async fn execute(self) -> eyre::Result<()> {
         // add network name to db directory
         let db_path = self.db.unwrap_or_chain_default(self.chain.chain);
 

bin/reth/src/dump_stage/mod.rs

@@ -1,7 +1,18 @@
 //! Database debugging tool
-mod hashing_storage;
+use crate::{
+    dirs::{DbPath, MaybePlatformPath, PlatformPath},
+    utils::DbTool,
+};
+use clap::Parser;
+use reth_db::{
+    cursor::DbCursorRO, database::Database, table::TableImporter, tables, transaction::DbTx,
+};
+use reth_primitives::ChainSpec;
+use reth_staged_sync::utils::{chainspec::genesis_value_parser, init::init_db};
 use std::sync::Arc;
+use tracing::info;
 
+mod hashing_storage;
 use hashing_storage::dump_hashing_storage_stage;
 
 mod hashing_account;
@@ -12,18 +23,6 @@ use execution::dump_execution_stage;
 
 mod merkle;
 use merkle::dump_merkle_stage;
-use reth_primitives::ChainSpec;
-
-use crate::{
-    dirs::{DbPath, MaybePlatformPath, PlatformPath},
-    utils::DbTool,
-};
-use clap::Parser;
-use reth_db::{
-    cursor::DbCursorRO, database::Database, table::TableImporter, tables, transaction::DbTx,
-};
-use reth_staged_sync::utils::{chainspec::genesis_value_parser, init::init_db};
-use tracing::info;
 
 /// `reth dump-stage` command
 #[derive(Debug, Parser)]
@@ -98,7 +97,7 @@ pub struct StageCommand {
 
 impl Command {
     /// Execute `dump-stage` command
-    pub async fn execute(&self) -> eyre::Result<()> {
+    pub async fn execute(self) -> eyre::Result<()> {
         // add network name to db directory
         let db_path = self.db.unwrap_or_chain_default(self.chain.chain);
 

bin/reth/src/runner.rs

@@ -60,6 +60,23 @@ impl CliRunner {
         tokio_runtime.block_on(run_until_ctrl_c(fut))?;
         Ok(())
     }
+
+    /// Executes a regular future as a spawned blocking task until completion or until external
+    /// signal received.
+    ///
+    /// See [Runtime::spawn_blocking](tokio::runtime::Runtime::spawn_blocking) .
+    pub fn run_blocking_until_ctrl_c<F, E>(self, fut: F) -> Result<(), E>
+    where
+        F: Future<Output = Result<(), E>> + Send + 'static,
+        E: Send + Sync + From<std::io::Error> + 'static,
+    {
+        let tokio_runtime = tokio_runtime()?;
+        let handle = tokio_runtime.handle().clone();
+        let fut = tokio_runtime.handle().spawn_blocking(move || handle.block_on(fut));
+        tokio_runtime
+            .block_on(run_until_ctrl_c(async move { fut.await.expect("Failed to join task") }))?;
+        Ok(())
+    }
 }
 
 /// [CliRunner] configuration when executing commands asynchronously

bin/reth/src/stage/mod.rs

@@ -94,7 +94,7 @@ pub struct Command {
 
 impl Command {
     /// Execute `stage` command
-    pub async fn execute(&self) -> eyre::Result<()> {
+    pub async fn execute(self) -> eyre::Result<()> {
         // Raise the fd limit of the process.
         // Does not do anything on windows.
         fdlimit::raise_fd_limit();