use a global runtime instead of a local one and new future

jormungandr/Cargo.toml

@@ -56,6 +56,7 @@ structopt = "^0.2"
 thiserror = "1.0"
 tokio      = "^0.1.16"
 tokio-threadpool = "0.1"
+tokio-compat = "^0.1.4"
 bech32 = "0.7"
 
 [build-dependencies]

jormungandr/src/main.rs

@@ -57,10 +57,10 @@ pub mod utils;
 
 use stats_counter::StatsCounter;
 
-fn start() -> Result<(), start_up::Error> {
-    let initialized_node = initialize_node()?;
+async fn start() -> Result<(), start_up::Error> {
+    let initialized_node = initialize_node().await?;
 
-    let bootstrapped_node = bootstrap(initialized_node)?;
+    let bootstrapped_node = bootstrap(initialized_node).await?;
 
     start_services(bootstrapped_node)
 }
@@ -322,7 +322,7 @@ fn start_services(bootstrapped_node: BootstrappedNode) -> Result<(), start_up::E
 /// * network / peer discoveries (?)
 ///
 ///
-fn bootstrap(initialized_node: InitializedNode) -> Result<BootstrappedNode, start_up::Error> {
+async fn bootstrap(initialized_node: InitializedNode) -> Result<BootstrappedNode, start_up::Error> {
     let InitializedNode {
         settings,
         block0,
@@ -353,7 +353,8 @@ fn bootstrap(initialized_node: InitializedNode) -> Result<BootstrappedNode, star
         blockchain.clone(),
         blockchain_tip.clone(),
         &bootstrap_logger,
-    )?;
+    )
+    .await?;
 
     let explorer_db = if settings.explorer {
         Some(explorer::ExplorerDB::bootstrap(
@@ -394,7 +395,7 @@ pub struct InitializedNode {
     pub diagnostic: Diagnostic,
 }
 
-fn initialize_node() -> Result<InitializedNode, start_up::Error> {
+async fn initialize_node() -> Result<InitializedNode, start_up::Error> {
     let command_line = CommandLine::load();
 
     if command_line.full_version {
@@ -460,7 +461,8 @@ fn initialize_node() -> Result<InitializedNode, start_up::Error> {
         &settings,
         &storage,
         &init_logger, /* add network to fetch block0 */
-    )?;
+    )
+    .await?;
 
     Ok(InitializedNode {
         settings,
@@ -476,19 +478,21 @@ fn initialize_node() -> Result<InitializedNode, start_up::Error> {
 fn main() {
     use std::error::Error;
 
-    if let Err(error) = start() {
-        eprintln!("{}", error);
-        let mut source = error.source();
-        while let Some(err) = source {
-            eprintln!(" |-> {}", err);
-            source = err.source();
+    tokio_compat::run_std(async {
+        if let Err(error) = start().await {
+            eprintln!("{}", error);
+            let mut source = error.source();
+            while let Some(err) = source {
+                eprintln!(" |-> {}", err);
+                source = err.source();
+            }
+
+            // TODO: https://github.com/rust-lang/rust/issues/43301
+            //
+            // as soon as #43301 is stabilized it would be nice to no use
+            // `exit` but the more appropriate:
+            // https://doc.rust-lang.org/stable/std/process/trait.Termination.html
+            std::process::exit(error.code());
         }
-
-        // TODO: https://github.com/rust-lang/rust/issues/43301
-        //
-        // as soon as #43301 is stabilized it would be nice to no use
-        // `exit` but the more appropriate:
-        // https://doc.rust-lang.org/stable/std/process/trait.Termination.html
-        std::process::exit(error.code());
-    }
+    })
 }

jormungandr/src/network/bootstrap.rs

@@ -9,7 +9,7 @@ use network_grpc::client::Connection;
 use slog::Logger;
 use thiserror::Error;
 use tokio::prelude::*;
-use tokio::runtime::Runtime;
+use tokio_compat::prelude::*;
 
 use std::fmt::Debug;
 use std::io;
@@ -39,20 +39,18 @@ pub enum Error {
     ChainSelectionFailed { source: BlockchainError },
 }
 
-pub fn bootstrap_from_peer(
+pub async fn bootstrap_from_peer(
     peer: Peer,
     blockchain: Blockchain,
     branch: Tip,
     logger: Logger,
 ) -> Result<Arc<Ref>, Error> {
     info!(logger, "connecting to bootstrap peer {}", peer.connection);
 
-    let runtime = Runtime::new().map_err(|e| Error::RuntimeInit { source: e })?;
-
     let blockchain2 = blockchain.clone();
     let logger2 = logger.clone();
 
-    let bootstrap = grpc::connect(peer.address(), None, runtime.executor())
+    let bootstrap = grpc::connect(peer.address(), None)
         .map_err(|e| Error::Connect { source: e })
         .and_then(|client: Connection<BlockConfig>| {
             client
@@ -74,7 +72,7 @@ pub fn bootstrap_from_peer(
                 .map(|()| tip)
         });
 
-    runtime.block_on_all(bootstrap)
+    bootstrap.compat().await
 }
 
 fn bootstrap_from_stream<S>(

jormungandr/src/network/client/connect.rs

@@ -34,7 +34,7 @@ pub fn connect(
         channels,
         logger: state.logger,
     });
-    let cf = grpc::connect(addr, Some(node_id), state.global.executor.clone());
+    let cf = grpc::connect(addr, Some(node_id));
     let handle = ConnectHandle { receiver };
     let future = ConnectFuture {
         sender: Some(sender),

jormungandr/src/network/grpc/client.rs

@@ -11,16 +11,14 @@ use network_core::error as core_error;
 use network_grpc::client::Connect;
 use slog::Logger;
 use thiserror::Error;
-use tokio::runtime::{Runtime, TaskExecutor};
+use tokio_compat::prelude::*;
 
 use std::io;
 use std::net::{IpAddr, SocketAddr};
 use std::slice;
 
 #[derive(Error, Debug)]
 pub enum FetchBlockError {
-    #[error("runtime initialization failed")]
-    RuntimeInit { source: io::Error },
     #[error("connection to peer failed")]
     Connect { source: ConnectError },
     #[error("connection broken")]
@@ -34,15 +32,18 @@ pub enum FetchBlockError {
 }
 
 pub type Connection = network_grpc::client::Connection<BlockConfig>;
-pub type ConnectFuture =
-    network_grpc::client::ConnectFuture<BlockConfig, HttpConnector, TaskExecutor>;
+pub type ConnectFuture = network_grpc::client::ConnectFuture<
+    BlockConfig,
+    HttpConnector,
+    tokio::executor::DefaultExecutor,
+>;
 pub type ConnectError = network_grpc::client::ConnectError<io::Error>;
 
-pub fn connect(addr: SocketAddr, node_id: Option<Id>, executor: TaskExecutor) -> ConnectFuture {
+pub fn connect(addr: SocketAddr, node_id: Option<Id>) -> ConnectFuture {
     let uri = destination_uri(addr);
     let mut connector = HttpConnector::new(2);
     connector.set_nodelay(true);
-    let mut builder = Connect::with_executor(connector, executor);
+    let mut builder = Connect::new(connector);
     if let Some(id) = node_id {
         builder.node_id(id);
     }
@@ -60,14 +61,13 @@ fn destination_uri(addr: SocketAddr) -> Uri {
 
 // Fetches a block from a network peer in a one-off, blocking call.
 // This function is used during node bootstrap to fetch the genesis block.
-pub fn fetch_block(
+pub async fn fetch_block(
     peer: Peer,
     hash: HeaderHash,
     logger: &Logger,
 ) -> Result<Block, FetchBlockError> {
     info!(logger, "fetching block {}", hash);
-    let runtime = Runtime::new().map_err(|e| FetchBlockError::RuntimeInit { source: e })?;
-    let fetch = connect(peer.address(), None, runtime.executor())
+    let fetch = connect(peer.address(), None)
         .map_err(|err| FetchBlockError::Connect { source: err })
         .and_then(move |client: Connection| {
             client
@@ -88,5 +88,6 @@ pub fn fetch_block(
             None => Err(FetchBlockError::NoBlocks),
             Some(block) => Ok(block),
         });
-    runtime.block_on_all(fetch)
+
+    fetch.compat().await
 }