Upgrade simulator to stable futures

tests/node_test_rig/src/lib.rs

@@ -4,7 +4,6 @@
 
 use beacon_node::ProductionBeaconNode;
 use environment::RuntimeContext;
-use futures::Future;
 use std::path::PathBuf;
 use std::time::{SystemTime, UNIX_EPOCH};
 use tempdir::TempDir;
@@ -40,7 +39,9 @@ impl<E: EthSpec> LocalBeaconNode<E> {
         client_config.data_dir = datadir.path().into();
         client_config.network.network_dir = PathBuf::from(datadir.path()).join("network");
 
-        ProductionBeaconNode::new(context, client_config).await.map(move |client| Self {
+        ProductionBeaconNode::new(context, client_config)
+            .await
+            .map(move |client| Self {
                 client: client.into_inner(),
                 datadir,
             })
@@ -141,7 +142,9 @@ impl<E: EthSpec> LocalValidatorClient<E> {
     ) -> Result<Self, String> {
         config.data_dir = datadir.path().into();
 
-        ProductionValidatorClient::new(context, config).await.map(move |mut client| {
+        ProductionValidatorClient::new(context, config)
+            .await
+            .map(move |mut client| {
                 client
                     .start_service()
                     .expect("should start validator services");

tests/simulator/src/checks.rs

@@ -1,87 +1,81 @@
 use crate::local_network::LocalNetwork;
-use futures::{stream, Future, IntoFuture, Stream};
+use std::time::Duration;
-use std::time::{Duration, Instant};
-use tokio::timer::Delay;
 use types::{Epoch, EthSpec, Slot, Unsigned};
 
 /// Checks that all of the validators have on-boarded by the start of the second eth1 voting
 /// period.
-pub fn verify_initial_validator_count<E: EthSpec>(
+pub async fn verify_initial_validator_count<E: EthSpec>(
     network: LocalNetwork<E>,
     slot_duration: Duration,
     initial_validator_count: usize,
-) -> impl Future<Item = (), Error = String> {
+) -> Result<(), String> {
-    slot_delay(Slot::new(1), slot_duration)
+    slot_delay(Slot::new(1), slot_duration).await;
-        .and_then(move |()| verify_validator_count(network, initial_validator_count))
+    verify_validator_count(network, initial_validator_count).await?;
+    Ok(())
 }
 
 /// Checks that all of the validators have on-boarded by the start of the second eth1 voting
 /// period.
-pub fn verify_validator_onboarding<E: EthSpec>(
+pub async fn verify_validator_onboarding<E: EthSpec>(
     network: LocalNetwork<E>,
     slot_duration: Duration,
     expected_validator_count: usize,
-) -> impl Future<Item = (), Error = String> {
+) -> Result<(), String> {
     slot_delay(
         Slot::new(E::SlotsPerEth1VotingPeriod::to_u64()),
         slot_duration,
     )
-    .and_then(move |()| verify_validator_count(network, expected_validator_count))
+    .await;
+    verify_validator_count(network, expected_validator_count).await?;
+    Ok(())
 }
 
 /// Checks that the chain has made the first possible finalization.
 ///
 /// Intended to be run as soon as chain starts.
-pub fn verify_first_finalization<E: EthSpec>(
+pub async fn verify_first_finalization<E: EthSpec>(
     network: LocalNetwork<E>,
     slot_duration: Duration,
-) -> impl Future<Item = (), Error = String> {
+) -> Result<(), String> {
-    epoch_delay(Epoch::new(4), slot_duration, E::slots_per_epoch())
+    epoch_delay(Epoch::new(4), slot_duration, E::slots_per_epoch()).await;
-        .and_then(|()| verify_all_finalized_at(network, Epoch::new(2)))
+    verify_all_finalized_at(network, Epoch::new(2)).await?;
+    Ok(())
 }
 
 /// Delays for `epochs`, plus half a slot extra.
-pub fn epoch_delay(
+pub async fn epoch_delay(epochs: Epoch, slot_duration: Duration, slots_per_epoch: u64) {
-    epochs: Epoch,
-    slot_duration: Duration,
-    slots_per_epoch: u64,
-) -> impl Future<Item = (), Error = String> {
     let duration = slot_duration * (epochs.as_u64() * slots_per_epoch) as u32 + slot_duration / 2;
-
+    tokio::time::delay_for(duration).await
-    Delay::new(Instant::now() + duration).map_err(|e| format!("Epoch delay failed: {:?}", e))
 }
 
 /// Delays for `slots`, plus half a slot extra.
-fn slot_delay(slots: Slot, slot_duration: Duration) -> impl Future<Item = (), Error = String> {
+async fn slot_delay(slots: Slot, slot_duration: Duration) {
     let duration = slot_duration * slots.as_u64() as u32 + slot_duration / 2;
-
+    tokio::time::delay_for(duration).await;
-    Delay::new(Instant::now() + duration).map_err(|e| format!("Epoch delay failed: {:?}", e))
 }
 
 /// Verifies that all beacon nodes in the given network have a head state that has a finalized
 /// epoch of `epoch`.
-pub fn verify_all_finalized_at<E: EthSpec>(
+pub async fn verify_all_finalized_at<E: EthSpec>(
     network: LocalNetwork<E>,
     epoch: Epoch,
-) -> impl Future<Item = (), Error = String> {
+) -> Result<(), String> {
-    network
+    let epochs = {
-        .remote_nodes()
+        let mut epochs = Vec::new();
-        .into_future()
+        for remote_node in network.remote_nodes()? {
-        .and_then(|remote_nodes| {
+            epochs.push(
-            stream::unfold(remote_nodes.into_iter(), |mut iter| {
-                iter.next().map(|remote_node| {
                 remote_node
                     .http
                     .beacon()
                     .get_head()
+                    .await
                     .map(|head| head.finalized_slot.epoch(E::slots_per_epoch()))
-                        .map(|epoch| (epoch, iter))
+                    .map_err(|e| format!("Get head via http failed: {:?}", e))?,
-                        .map_err(|e| format!("Get head via http failed: {:?}", e))
+            );
-                })
+        }
-            })
+        epochs
-            .collect()
+    };
-        })
+
-        .and_then(move |epochs| {
     if epochs.iter().any(|node_epoch| *node_epoch != epoch) {
         Err(format!(
             "Nodes are not finalized at epoch {}. Finalized epochs: {:?}",
@@ -90,37 +84,36 @@ pub fn verify_all_finalized_at<E: EthSpec>(
     } else {
         Ok(())
     }
-        })
 }
 
 /// Verifies that all beacon nodes in the given `network` have a head state that contains
 /// `expected_count` validators.
-fn verify_validator_count<E: EthSpec>(
+async fn verify_validator_count<E: EthSpec>(
     network: LocalNetwork<E>,
     expected_count: usize,
-) -> impl Future<Item = (), Error = String> {
+) -> Result<(), String> {
-    network
+    let validator_counts = {
-        .remote_nodes()
+        let mut validator_counts = Vec::new();
-        .into_future()
+        for remote_node in network.remote_nodes()? {
-        .and_then(|remote_nodes| {
-            stream::unfold(remote_nodes.into_iter(), |mut iter| {
-                iter.next().map(|remote_node| {
             let beacon = remote_node.http.beacon();
-                    beacon
+
+            let head = beacon
                 .get_head()
-                        .map_err(|e| format!("Get head via http failed: {:?}", e))
+                .await
-                        .and_then(move |head| {
+                .map_err(|e| format!("Get head via http failed: {:?}", e))?;
-                            beacon
+
+            let vc = beacon
                 .get_state_by_root(head.state_root)
+                .await
                 .map(|(state, _root)| state)
-                                .map_err(|e| format!("Get state root via http failed: {:?}", e))
+                .map_err(|e| format!("Get state root via http failed: {:?}", e))?
-                        })
+                .validators
-                        .map(|state| (state.validators.len(), iter))
+                .len();
-                })
+            validator_counts.push(vc);
-            })
+        }
-            .collect()
+        validator_counts
-        })
+    };
-        .and_then(move |validator_counts| {
+
     if validator_counts
         .iter()
         .any(|count| *count != expected_count)
@@ -132,5 +125,4 @@ fn verify_validator_count<E: EthSpec>(
     } else {
         Ok(())
     }
-        })
 }

tests/simulator/src/eth1_sim.rs

@@ -1,13 +1,12 @@
 use crate::{checks, LocalNetwork, E};
 use clap::ArgMatches;
 use eth1_test_rig::GanacheEth1Instance;
-use futures::{future, stream, Future, Stream};
+use futures::prelude::*;
 use node_test_rig::{
     environment::EnvironmentBuilder, testing_client_config, ClientGenesis, ValidatorConfig,
 };
 use std::net::{IpAddr, Ipv4Addr};
-use std::time::{Duration, Instant};
+use std::time::Duration;
-use tokio::timer::Interval;
 
 pub fn run_eth1_sim(matches: &ArgMatches) -> Result<(), String> {
     let node_count = value_t!(matches, "nodes", usize).expect("missing nodes default");
@@ -50,42 +49,35 @@ pub fn run_eth1_sim(matches: &ArgMatches) -> Result<(), String> {
     let deposit_amount = env.eth2_config.spec.max_effective_balance;
 
     let context = env.core_context();
-    let executor = context.executor.clone();
 
-    let future = GanacheEth1Instance::new()
+    let main_future = async {
         /*
          * Deploy the deposit contract, spawn tasks to keep creating new blocks and deposit
          * validators.
          */
-        .map(move |ganache_eth1_instance| {
+        let ganache_eth1_instance = GanacheEth1Instance::new().await?;
         let deposit_contract = ganache_eth1_instance.deposit_contract;
         let ganache = ganache_eth1_instance.ganache;
         let eth1_endpoint = ganache.endpoint();
         let deposit_contract_address = deposit_contract.address();
 
         // Start a timer that produces eth1 blocks on an interval.
-            executor.spawn(
+        tokio::spawn(async move {
-                Interval::new(Instant::now(), eth1_block_time)
+            let mut interval = tokio::time::interval(eth1_block_time);
-                    .map_err(|_| eprintln!("Eth1 block timer failed"))
+            while let Some(_) = interval.next().await {
-                    .for_each(move |_| ganache.evm_mine().map_err(|_| ()))
+                let _ = ganache.evm_mine().await;
-                    .map_err(|_| eprintln!("Eth1 evm_mine failed"))
+            }
-                    .map(|_| ()),
+        });
-            );
 
         // Submit deposits to the deposit contract.
-            executor.spawn(
+        tokio::spawn(async move {
-                stream::unfold(0..total_validator_count, move |mut iter| {
+            for i in 0..total_validator_count {
-                    iter.next().map(|i| {
                 println!("Submitting deposit for validator {}...", i);
-                        deposit_contract
+                let _ = deposit_contract
                     .deposit_deterministic_async::<E>(i, deposit_amount)
-                            .map(|_| ((), iter))
+                    .await;
-                    })
+            }
-                })
+        });
-                .collect()
-                .map(|_| ())
-                .map_err(|e| eprintln!("Error submitting deposit: {}", e)),
-            );
 
         let mut beacon_config = testing_client_config();
 
@@ -100,37 +92,21 @@ pub fn run_eth1_sim(matches: &ArgMatches) -> Result<(), String> {
 
         beacon_config.network.enr_address = Some(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
 
-            beacon_config
-        })
         /*
          * Create a new `LocalNetwork` with one beacon node.
          */
-        .and_then(move |beacon_config| {
+        let network = LocalNetwork::new(context, beacon_config.clone()).await?;
-            LocalNetwork::new(context, beacon_config.clone())
-                .map(|network| (network, beacon_config))
-        })
         /*
          * One by one, add beacon nodes to the network.
          */
-        .and_then(move |(network, beacon_config)| {
-            let network_1 = network.clone();
 
-            stream::unfold(0..node_count - 1, move |mut iter| {
+        for _ in 0..node_count - 1 {
-                iter.next().map(|_| {
+            network.add_beacon_node(beacon_config.clone()).await?;
-                    network_1
+        }
-                        .add_beacon_node(beacon_config.clone())
-                        .map(|()| ((), iter))
-                })
-            })
-            .collect()
-            .map(|_| network)
-        })
         /*
          * One by one, add validator clients to the network. Each validator client is attached to
          * a single corresponding beacon node.
          */
-        .and_then(move |network| {
-            let network_1 = network.clone();
 
         // Note: presently the validator client future will only resolve once genesis time
         // occurs. This is great for this scenario, but likely to change in the future.
@@ -139,60 +115,46 @@ pub fn run_eth1_sim(matches: &ArgMatches) -> Result<(), String> {
         // that delays until genesis. Otherwise, all of the checks that start in the next
         // future will start too early.
 
-            stream::unfold(0..node_count, move |mut iter| {
+        for i in 0..node_count {
-                iter.next().map(|i| {
+            let indices =
-                    let indices = (i * validators_per_node..(i + 1) * validators_per_node)
+                (i * validators_per_node..(i + 1) * validators_per_node).collect::<Vec<_>>();
-                        .collect::<Vec<_>>();
+            network
-
-                    network_1
                 .add_validator_client(ValidatorConfig::default(), i, indices)
-                        .map(|()| ((), iter))
+                .await?;
-                })
+        }
-            })
+
-            .collect()
-            .map(|_| network)
-        })
         /*
          * Start the processes that will run checks on the network as it runs.
          */
-        .and_then(move |network| {
-            // The `final_future` either completes immediately or never completes, depending on the value
-            // of `end_after_checks`.
-            let final_future: Box<dyn Future<Item = (), Error = String> + Send> =
-                if end_after_checks {
-                    Box::new(future::ok(()).map_err(|()| "".to_string()))
-                } else {
-                    Box::new(future::empty().map_err(|()| "".to_string()))
-                };
 
-            future::ok(())
+        let _err = futures::join!(
             // Check that the chain finalizes at the first given opportunity.
-                .join(checks::verify_first_finalization(
+            checks::verify_first_finalization(network.clone(), slot_duration),
-                    network.clone(),
-                    slot_duration,
-                ))
             // Check that the chain starts with the expected validator count.
-                .join(checks::verify_initial_validator_count(
+            checks::verify_initial_validator_count(
                 network.clone(),
                 slot_duration,
                 initial_validator_count,
-                ))
+            ),
             // Check that validators greater than `spec.min_genesis_active_validator_count` are
             // onboarded at the first possible opportunity.
-                .join(checks::verify_validator_onboarding(
+            checks::verify_validator_onboarding(
                 network.clone(),
                 slot_duration,
                 total_validator_count,
-                ))
+            )
-                // End now or run forever, depending on the `end_after_checks` flag.
+        );
-                .join(final_future)
+
-                .map(|_| network)
+        // The `final_future` either completes immediately or never completes, depending on the value
-        })
+        // of `end_after_checks`.
+
+        if !end_after_checks {
+            future::pending::<()>().await;
+        }
         /*
          * End the simulation by dropping the network. This will kill all running beacon nodes and
          * validator clients.
          */
-        .map(|network| {
         println!(
             "Simulation complete. Finished with {} beacon nodes and {} validator clients",
             network.beacon_node_count(),
@@ -201,8 +163,9 @@ pub fn run_eth1_sim(matches: &ArgMatches) -> Result<(), String> {
 
         // Be explicit about dropping the network, as this kills all the nodes. This ensures
         // all the checks have adequate time to pass.
-            drop(network)
+        drop(network);
-        });
+        Ok::<(), String>(())
+    };
 
-    env.runtime().block_on(future)
+    Ok(env.runtime().block_on(main_future).unwrap())
 }

tests/simulator/src/local_network.rs

@@ -1,4 +1,3 @@
-use futures::{Future, IntoFuture};
 use node_test_rig::{
     environment::RuntimeContext, ClientConfig, LocalBeaconNode, LocalValidatorClient,
     RemoteBeaconNode, ValidatorConfig,
@@ -42,23 +41,24 @@ impl<E: EthSpec> Deref for LocalNetwork<E> {
 
 impl<E: EthSpec> LocalNetwork<E> {
     /// Creates a new network with a single `BeaconNode`.
-    pub fn new(
+    pub async fn new(
         context: RuntimeContext<E>,
         mut beacon_config: ClientConfig,
-    ) -> impl Future<Item = Self, Error = String> {
+    ) -> Result<Self, String> {
         beacon_config.network.discovery_port = BOOTNODE_PORT;
         beacon_config.network.libp2p_port = BOOTNODE_PORT;
         beacon_config.network.enr_udp_port = Some(BOOTNODE_PORT);
         beacon_config.network.enr_tcp_port = Some(BOOTNODE_PORT);
-        LocalBeaconNode::production(context.service_context("boot_node".into()), beacon_config).map(
+        let beacon_node =
-            |beacon_node| Self {
+            LocalBeaconNode::production(context.service_context("boot_node".into()), beacon_config)
+                .await?;
+        Ok(Self {
             inner: Arc::new(Inner {
                 context,
                 beacon_nodes: RwLock::new(vec![beacon_node]),
                 validator_clients: RwLock::new(vec![]),
             }),
-            },
+        })
-        )
     }
 
     /// Returns the number of beacon nodes in the network.
@@ -78,72 +78,65 @@ impl<E: EthSpec> LocalNetwork<E> {
     }
 
     /// Adds a beacon node to the network, connecting to the 0'th beacon node via ENR.
-    pub fn add_beacon_node(
+    pub async fn add_beacon_node(&self, mut beacon_config: ClientConfig) -> Result<(), String> {
-        &self,
-        mut beacon_config: ClientConfig,
-    ) -> impl Future<Item = (), Error = String> {
         let self_1 = self.clone();
         println!("Adding beacon node..");
-        self.beacon_nodes
+        {
-            .read()
+            let read_lock = self.beacon_nodes.read();
-            .first()
+
-            .map(|boot_node| {
+            let boot_node = read_lock.first().expect("should have at least one node");
+
             beacon_config.network.boot_nodes.push(
                 boot_node
                     .client
                     .enr()
                     .expect("bootnode must have a network"),
             );
-            })
+        }
-            .expect("should have at least one node");
 
         let index = self.beacon_nodes.read().len();
 
-        LocalBeaconNode::production(
+        let beacon_node = LocalBeaconNode::production(
             self.context.service_context(format!("node_{}", index)),
             beacon_config,
         )
-        .map(move |beacon_node| {
+        .await?;
         self_1.beacon_nodes.write().push(beacon_node);
-        })
+        Ok(())
     }
 
     /// Adds a validator client to the network, connecting it to the beacon node with index
     /// `beacon_node`.
-    pub fn add_validator_client(
+    pub async fn add_validator_client(
         &self,
         mut validator_config: ValidatorConfig,
         beacon_node: usize,
         keypair_indices: Vec<usize>,
-    ) -> impl Future<Item = (), Error = String> {
+    ) -> Result<(), String> {
         let index = self.validator_clients.read().len();
         let context = self.context.service_context(format!("validator_{}", index));
         let self_1 = self.clone();
-
+        let socket_addr = {
-        self.beacon_nodes
+            let read_lock = self.beacon_nodes.read();
-            .read()
+            let beacon_node = read_lock
                 .get(beacon_node)
-            .map(move |beacon_node| {
+                .ok_or_else(|| format!("No beacon node for index {}", beacon_node))?;
-                let socket_addr = beacon_node
+            beacon_node
                 .client
                 .http_listen_addr()
-                    .expect("Must have http started");
+                .expect("Must have http started")
+        };
 
         validator_config.http_server =
             format!("http://{}:{}", socket_addr.ip(), socket_addr.port());
-
+        let validator_client = LocalValidatorClient::production_with_insecure_keypairs(
-                validator_config
-            })
-            .ok_or_else(|| format!("No beacon node for index {}", beacon_node))
-            .into_future()
-            .and_then(move |validator_config| {
-                LocalValidatorClient::production_with_insecure_keypairs(
             context,
             validator_config,
             &keypair_indices,
         )
-            })
+        .await?;
-            .map(move |validator_client| self_1.validator_clients.write().push(validator_client))
+        self_1.validator_clients.write().push(validator_client);
+        Ok(())
     }
 
     /// For all beacon nodes in `Self`, return a HTTP client to access each nodes HTTP API.
@@ -157,13 +150,14 @@ impl<E: EthSpec> LocalNetwork<E> {
     }
 
     /// Return current epoch of bootnode.
-    pub fn bootnode_epoch(&self) -> impl Future<Item = Epoch, Error = String> {
+    pub async fn bootnode_epoch(&self) -> Result<Epoch, String> {
         let nodes = self.remote_nodes().expect("Failed to get remote nodes");
         let bootnode = nodes.first().expect("Should contain bootnode");
         bootnode
             .http
             .beacon()
             .get_head()
+            .await
             .map_err(|e| format!("Cannot get head: {:?}", e))
             .map(|head| head.finalized_slot.epoch(E::slots_per_epoch()))
     }

tests/simulator/src/no_eth1_sim.rs

@@ -1,6 +1,6 @@
 use crate::{checks, LocalNetwork};
 use clap::ArgMatches;
-use futures::{future, stream, Future, Stream};
+use futures::prelude::*;
 use node_test_rig::{
     environment::EnvironmentBuilder, testing_client_config, ClientGenesis, ValidatorConfig,
 };
@@ -63,29 +63,19 @@ pub fn run_no_eth1_sim(matches: &ArgMatches) -> Result<(), String> {
 
     beacon_config.network.enr_address = Some(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
 
-    let future = LocalNetwork::new(context, beacon_config.clone())
+    let main_future = async {
+        let network = LocalNetwork::new(context, beacon_config.clone()).await?;
         /*
          * One by one, add beacon nodes to the network.
          */
-        .and_then(move |network| {
-            let network_1 = network.clone();
 
-            stream::unfold(0..node_count - 1, move |mut iter| {
+        for _ in 0..node_count - 1 {
-                iter.next().map(|_| {
+            network.add_beacon_node(beacon_config.clone()).await?;
-                    network_1
+        }
-                        .add_beacon_node(beacon_config.clone())
-                        .map(|()| ((), iter))
-                })
-            })
-            .collect()
-            .map(|_| network)
-        })
         /*
          * One by one, add validator clients to the network. Each validator client is attached to
          * a single corresponding beacon node.
          */
-        .and_then(move |network| {
-            let network_1 = network.clone();
 
         // Note: presently the validator client future will only resolve once genesis time
         // occurs. This is great for this scenario, but likely to change in the future.
@@ -94,47 +84,29 @@ pub fn run_no_eth1_sim(matches: &ArgMatches) -> Result<(), String> {
         // that delays until genesis. Otherwise, all of the checks that start in the next
         // future will start too early.
 
-            stream::unfold(0..node_count, move |mut iter| {
+        for i in 0..node_count {
-                iter.next().map(|i| {
+            let indices =
-                    let indices = (i * validators_per_node..(i + 1) * validators_per_node)
+                (i * validators_per_node..(i + 1) * validators_per_node).collect::<Vec<_>>();
-                        .collect::<Vec<_>>();
+            network
-
-                    network_1
                 .add_validator_client(ValidatorConfig::default(), i, indices)
-                        .map(|()| ((), iter))
+                .await?;
-                })
+        }
-            })
-            .collect()
-            .map(|_| network)
-        })
         /*
          * Start the processes that will run checks on the network as it runs.
          */
-        .and_then(move |network| {
+        // Check that the chain finalizes at the first given opportunity.
+        checks::verify_first_finalization(network.clone(), slot_duration).await?;
+
         // The `final_future` either completes immediately or never completes, depending on the value
         // of `end_after_checks`.
-            let final_future: Box<dyn Future<Item = (), Error = String> + Send> =
-                if end_after_checks {
-                    Box::new(future::ok(()).map_err(|()| "".to_string()))
-                } else {
-                    Box::new(future::empty().map_err(|()| "".to_string()))
-                };
 
-            future::ok(())
+        if !end_after_checks {
-                // Check that the chain finalizes at the first given opportunity.
+            future::pending::<()>().await;
-                .join(checks::verify_first_finalization(
+        }
-                    network.clone(),
-                    slot_duration,
-                ))
-                // End now or run forever, depending on the `end_after_checks` flag.
-                .join(final_future)
-                .map(|_| network)
-        })
         /*
          * End the simulation by dropping the network. This will kill all running beacon nodes and
          * validator clients.
          */
-        .map(|network| {
         println!(
             "Simulation complete. Finished with {} beacon nodes and {} validator clients",
             network.beacon_node_count(),
@@ -143,8 +115,9 @@ pub fn run_no_eth1_sim(matches: &ArgMatches) -> Result<(), String> {
 
         // Be explicit about dropping the network, as this kills all the nodes. This ensures
         // all the checks have adequate time to pass.
-            drop(network)
+        drop(network);
-        });
+        Ok::<(), String>(())
+    };
 
-    env.runtime().block_on(future)
+    Ok(env.runtime().block_on(main_future).unwrap())
 }

tests/simulator/src/sync_sim.rs

@@ -1,14 +1,13 @@
 use crate::checks::{epoch_delay, verify_all_finalized_at};
 use crate::local_network::LocalNetwork;
 use clap::ArgMatches;
-use futures::{future, stream, Future, IntoFuture, Stream};
+use futures::prelude::*;
 use node_test_rig::ClientConfig;
 use node_test_rig::{
     environment::EnvironmentBuilder, testing_client_config, ClientGenesis, ValidatorConfig,
 };
 use std::net::{IpAddr, Ipv4Addr};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
-use tokio::timer::Interval;
 use types::{Epoch, EthSpec};
 
 pub fn run_syncing_sim(matches: &ArgMatches) -> Result<(), String> {
@@ -78,46 +77,41 @@ fn syncing_sim(
 
     beacon_config.network.enr_address = Some(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
 
-    let future = LocalNetwork::new(context, beacon_config.clone())
+    let main_future = async {
+        /*
+         * Create a new `LocalNetwork` with one beacon node.
+         */
+        let network = LocalNetwork::new(context, beacon_config.clone()).await?;
+
         /*
          * Add a validator client which handles all validators from the genesis state.
          */
-        .and_then(move |network| {
         network
             .add_validator_client(ValidatorConfig::default(), 0, (0..num_validators).collect())
-                .map(|_| network)
+            .await?;
-        })
-        /*
-         * Start the processes that will run checks on the network as it runs.
-         */
-        .and_then(move |network| {
-            // The `final_future` either completes immediately or never completes, depending on the value
-            // of `end_after_checks`.
-            let final_future: Box<dyn Future<Item = (), Error = String> + Send> =
-                if end_after_checks {
-                    Box::new(future::ok(()).map_err(|()| "".to_string()))
-                } else {
-                    Box::new(future::empty().map_err(|()| "".to_string()))
-                };
 
-            future::ok(())
         // Check all syncing strategies one after other.
-                .join(pick_strategy(
+        pick_strategy(
             &strategy,
             network.clone(),
             beacon_config.clone(),
             slot_duration,
             initial_delay,
             sync_timeout,
-                ))
+        )
-                .join(final_future)
+        .await?;
-                .map(|_| network)
+
-        })
+        // The `final_future` either completes immediately or never completes, depending on the value
+        // of `end_after_checks`.
+
+        if !end_after_checks {
+            future::pending::<()>().await;
+        }
+
         /*
          * End the simulation by dropping the network. This will kill all running beacon nodes and
          * validator clients.
          */
-        .map(|network| {
         println!(
             "Simulation complete. Finished with {} beacon nodes and {} validator clients",
             network.beacon_node_count(),
@@ -126,62 +120,75 @@ fn syncing_sim(
 
         // Be explicit about dropping the network, as this kills all the nodes. This ensures
         // all the checks have adequate time to pass.
-            drop(network)
+        drop(network);
-        });
+        Ok::<(), String>(())
+    };
 
-    env.runtime().block_on(future)
+    env.runtime().block_on(main_future)
 }
 
-pub fn pick_strategy<E: EthSpec>(
+pub async fn pick_strategy<E: EthSpec>(
     strategy: &str,
     network: LocalNetwork<E>,
     beacon_config: ClientConfig,
     slot_duration: Duration,
     initial_delay: u64,
     sync_timeout: u64,
-) -> Box<dyn Future<Item = (), Error = String> + Send + 'static> {
+) -> Result<(), String> {
     match strategy {
-        "one-node" => Box::new(verify_one_node_sync(
+        "one-node" => {
+            verify_one_node_sync(
                 network,
                 beacon_config,
                 slot_duration,
                 initial_delay,
                 sync_timeout,
-        )),
+            )
-        "two-nodes" => Box::new(verify_two_nodes_sync(
+            .await
+        }
+        "two-nodes" => {
+            verify_two_nodes_sync(
                 network,
                 beacon_config,
                 slot_duration,
                 initial_delay,
                 sync_timeout,
-        )),
+            )
-        "mixed" => Box::new(verify_in_between_sync(
+            .await
+        }
+        "mixed" => {
+            verify_in_between_sync(
                 network,
                 beacon_config,
                 slot_duration,
                 initial_delay,
                 sync_timeout,
-        )),
+            )
-        "all" => Box::new(verify_syncing(
+            .await
+        }
+        "all" => {
+            verify_syncing(
                 network,
                 beacon_config,
                 slot_duration,
                 initial_delay,
                 sync_timeout,
-        )),
+            )
-        _ => Box::new(Err("Invalid strategy".into()).into_future()),
+            .await
+        }
+        _ => Err("Invalid strategy".into()),
     }
 }
 
 /// Verify one node added after `initial_delay` epochs is in sync
 /// after `sync_timeout` epochs.
-pub fn verify_one_node_sync<E: EthSpec>(
+pub async fn verify_one_node_sync<E: EthSpec>(
     network: LocalNetwork<E>,
     beacon_config: ClientConfig,
     slot_duration: Duration,
     initial_delay: u64,
     sync_timeout: u64,
-) -> impl Future<Item = (), Error = String> {
+) -> Result<(), String> {
     let epoch_duration = slot_duration * (E::slots_per_epoch() as u32);
     let network_c = network.clone();
     // Delay for `initial_delay` epochs before adding another node to start syncing
@@ -190,35 +197,34 @@ pub fn verify_one_node_sync<E: EthSpec>(
         slot_duration,
         E::slots_per_epoch(),
     )
-    .and_then(move |_| {
+    .await;
     // Add a beacon node
-        network.add_beacon_node(beacon_config).map(|_| network)
+    network.add_beacon_node(beacon_config).await?;
-    })
-    .and_then(move |network| {
     // Check every `epoch_duration` if nodes are synced
     // limited to at most `sync_timeout` epochs
-        Interval::new_interval(epoch_duration)
+    let mut interval = tokio::time::interval(epoch_duration);
-            .take(sync_timeout)
+    let mut count = 0;
-            .map_err(|_| "Failed to create interval".to_string())
+    while let Some(_) = interval.next().await {
-            .take_while(move |_| check_still_syncing(&network_c))
+        if count >= sync_timeout || !check_still_syncing(&network_c).await? {
-            .for_each(|_| Ok(())) // consume the stream
+            break;
-            .map(|_| network)
+        }
-    })
+        count += 1;
-    .and_then(move |network| network.bootnode_epoch().map(|e| (e, network)))
+    }
-    .and_then(move |(epoch, network)| {
+    let epoch = network.bootnode_epoch().await?;
-        verify_all_finalized_at(network, epoch).map_err(|e| format!("One node sync error: {}", e))
+    verify_all_finalized_at(network, epoch)
-    })
+        .map_err(|e| format!("One node sync error: {}", e))
+        .await
 }
 
 /// Verify two nodes added after `initial_delay` epochs are in sync
 /// after `sync_timeout` epochs.
-pub fn verify_two_nodes_sync<E: EthSpec>(
+pub async fn verify_two_nodes_sync<E: EthSpec>(
     network: LocalNetwork<E>,
     beacon_config: ClientConfig,
     slot_duration: Duration,
     initial_delay: u64,
     sync_timeout: u64,
-) -> impl Future<Item = (), Error = String> {
+) -> Result<(), String> {
     let epoch_duration = slot_duration * (E::slots_per_epoch() as u32);
     let network_c = network.clone();
     // Delay for `initial_delay` epochs before adding another node to start syncing
@@ -227,41 +233,36 @@ pub fn verify_two_nodes_sync<E: EthSpec>(
         slot_duration,
         E::slots_per_epoch(),
     )
-    .and_then(move |_| {
+    .await;
     // Add beacon nodes
-        network
+    network.add_beacon_node(beacon_config.clone()).await?;
-            .add_beacon_node(beacon_config.clone())
+    network.add_beacon_node(beacon_config).await?;
-            .map(|_| (network, beacon_config))
-            .and_then(|(network, beacon_config)| {
-                network.add_beacon_node(beacon_config).map(|_| network)
-            })
-    })
-    .and_then(move |network| {
     // Check every `epoch_duration` if nodes are synced
     // limited to at most `sync_timeout` epochs
-        Interval::new_interval(epoch_duration)
+    let mut interval = tokio::time::interval(epoch_duration);
-            .take(sync_timeout)
+    let mut count = 0;
-            .map_err(|_| "Failed to create interval".to_string())
+    while let Some(_) = interval.next().await {
-            .take_while(move |_| check_still_syncing(&network_c))
+        if count >= sync_timeout || !check_still_syncing(&network_c).await? {
-            .for_each(|_| Ok(())) // consume the stream
+            break;
-            .map(|_| network)
+        }
-    })
+        count += 1;
-    .and_then(move |network| network.bootnode_epoch().map(|e| (e, network)))
+    }
-    .and_then(move |(epoch, network)| {
+    let epoch = network.bootnode_epoch().await?;
-        verify_all_finalized_at(network, epoch).map_err(|e| format!("Two node sync error: {}", e))
+    verify_all_finalized_at(network, epoch)
-    })
+        .map_err(|e| format!("One node sync error: {}", e))
+        .await
 }
 
 /// Add 2 syncing nodes after `initial_delay` epochs,
 /// Add another node after `sync_timeout - 5` epochs and verify all are
 /// in sync after `sync_timeout + 5` epochs.
-pub fn verify_in_between_sync<E: EthSpec>(
+pub async fn verify_in_between_sync<E: EthSpec>(
     network: LocalNetwork<E>,
     beacon_config: ClientConfig,
     slot_duration: Duration,
     initial_delay: u64,
     sync_timeout: u64,
-) -> impl Future<Item = (), Error = String> {
+) -> Result<(), String> {
     let epoch_duration = slot_duration * (E::slots_per_epoch() as u32);
     let network_c = network.clone();
     // Delay for `initial_delay` epochs before adding another node to start syncing
@@ -271,52 +272,43 @@ pub fn verify_in_between_sync<E: EthSpec>(
         slot_duration,
         E::slots_per_epoch(),
     )
-    .and_then(move |_| {
+    .await;
-        // Add a beacon node
+    // Add two beacon nodes
-        network
+    network.add_beacon_node(beacon_config.clone()).await?;
-            .add_beacon_node(beacon_config.clone())
+    network.add_beacon_node(beacon_config).await?;
-            .map(|_| (network, beacon_config))
-            .and_then(|(network, beacon_config)| {
-                network.add_beacon_node(beacon_config).map(|_| network)
-            })
-    })
-    .and_then(move |network| {
     // Delay before adding additional syncing nodes.
     epoch_delay(
         Epoch::new(sync_timeout - 5),
         slot_duration,
         E::slots_per_epoch(),
     )
-        .map(|_| network)
+    .await;
-    })
-    .and_then(move |network| {
     // Add a beacon node
-        network.add_beacon_node(config1.clone()).map(|_| network)
+    network.add_beacon_node(config1.clone()).await?;
-    })
-    .and_then(move |network| {
     // Check every `epoch_duration` if nodes are synced
     // limited to at most `sync_timeout` epochs
-        Interval::new_interval(epoch_duration)
+    let mut interval = tokio::time::interval(epoch_duration);
-            .take(sync_timeout + 5)
+    let mut count = 0;
-            .map_err(|_| "Failed to create interval".to_string())
+    while let Some(_) = interval.next().await {
-            .take_while(move |_| check_still_syncing(&network_c))
+        if count >= sync_timeout || !check_still_syncing(&network_c).await? {
-            .for_each(|_| Ok(())) // consume the stream
+            break;
-            .map(|_| network)
+        }
-    })
+        count += 1;
-    .and_then(move |network| network.bootnode_epoch().map(|e| (e, network)))
+    }
-    .and_then(move |(epoch, network)| {
+    let epoch = network.bootnode_epoch().await?;
-        verify_all_finalized_at(network, epoch).map_err(|e| format!("In between sync error: {}", e))
+    verify_all_finalized_at(network, epoch)
-    })
+        .map_err(|e| format!("One node sync error: {}", e))
+        .await
 }
 
 /// Run syncing strategies one after other.
-pub fn verify_syncing<E: EthSpec>(
+pub async fn verify_syncing<E: EthSpec>(
     network: LocalNetwork<E>,
     beacon_config: ClientConfig,
     slot_duration: Duration,
     initial_delay: u64,
     sync_timeout: u64,
-) -> impl Future<Item = (), Error = String> {
+) -> Result<(), String> {
     verify_one_node_sync(
         network.clone(),
         beacon_config.clone(),
@@ -324,8 +316,8 @@ pub fn verify_syncing<E: EthSpec>(
         initial_delay,
         sync_timeout,
     )
-    .map(|_| println!("Completed one node sync"))
+    .await?;
-    .and_then(move |_| {
+    println!("Completed one node sync");
     verify_two_nodes_sync(
         network.clone(),
         beacon_config.clone(),
@@ -333,12 +325,8 @@ pub fn verify_syncing<E: EthSpec>(
         initial_delay,
         sync_timeout,
     )
-        .map(|_| {
+    .await?;
     println!("Completed two node sync");
-            (network, beacon_config)
-        })
-    })
-    .and_then(move |(network, beacon_config)| {
     verify_in_between_sync(
         network,
         beacon_config,
@@ -346,31 +334,24 @@ pub fn verify_syncing<E: EthSpec>(
         initial_delay,
         sync_timeout,
     )
-        .map(|_| println!("Completed in between sync"))
+    .await?;
-    })
+    println!("Completed in between sync");
+    Ok(())
 }
 
-pub fn check_still_syncing<E: EthSpec>(
+pub async fn check_still_syncing<E: EthSpec>(network: &LocalNetwork<E>) -> Result<bool, String> {
-    network: &LocalNetwork<E>,
-) -> impl Future<Item = bool, Error = String> {
-    network
-        .remote_nodes()
-        .into_future()
     // get syncing status of nodes
-        .and_then(|remote_nodes| {
+    let mut status = Vec::new();
-            stream::unfold(remote_nodes.into_iter(), |mut iter| {
+    for remote_node in network.remote_nodes()? {
-                iter.next().map(|remote_node| {
+        status.push(
             remote_node
                 .http
                 .node()
                 .syncing_status()
+                .await
                 .map(|status| status.is_syncing)
-                        .map(|status| (status, iter))
+                .map_err(|e| format!("Get syncing status via http failed: {:?}", e))?,
-                        .map_err(|e| format!("Get syncing status via http failed: {:?}", e))
+        )
-                })
+    }
-            })
+    Ok(status.iter().any(|is_syncing| *is_syncing))
-            .collect()
-        })
-        .and_then(move |status| Ok(status.iter().any(|is_syncing| *is_syncing)))
-        .map_err(|e| format!("Failed syncing check: {:?}", e))
 }