main: Fix hanging on startup (for real this time)
The root cause is still the same, but even
without the `.wait()` the future would
still block itself, wrapping everything in
a `tokio::spawn` fixed it.
@@ -491,6 +491,9 @@ fn async_main() -> impl Future<Item = (), Error = ()> + Send + 'static {
let stores_logger = logger.clone();
let stores_error_logger = logger.clone();
let stores_eth_adapters = eth_adapters.clone();
+ let contention_logger = logger.clone();
+
+ tokio::spawn(
futures::stream::futures_ordered(stores_eth_adapters.into_iter().map(
|(network_name, eth_adapter)| {
info!(
@@ -542,8 +545,11 @@ fn async_main() -> impl Future<Item = (), Error = ()> + Send + 'static {
generic_store.clone(),
node_id.clone(),
);
- let mut subscription_server =
- GraphQLSubscriptionServer::new(&logger, graphql_runner.clone(), generic_store.clone());
+ let mut subscription_server = GraphQLSubscriptionServer::new(
+ &logger,
+ graphql_runner.clone(),
+ generic_store.clone(),
+ );
let mut index_node_server = IndexNodeServer::new(
&logger_factory,
@@ -660,8 +666,8 @@ fn async_main() -> impl Future<Item = (), Error = ()> + Send + 'static {
let name = SubgraphName::new(name)
.expect("Subgraph name must contain only a-z, A-Z, 0-9, '-' and '_'");
- let subgraph_id =
- SubgraphDeploymentId::new(hash).expect("Subgraph hash must be a valid IPFS hash");
+ let subgraph_id = SubgraphDeploymentId::new(hash)
+ .expect("Subgraph hash must be a valid IPFS hash");
tokio::spawn(
subgraph_registrar
@@ -705,6 +711,10 @@ fn async_main() -> impl Future<Item = (), Error = ()> + Send + 'static {
.expect("Failed to start metrics server"),
);
+ future::ok(())
+ }),
+ );
+
// Periodically check for contention in the tokio threadpool. First spawn a
// task that simply responds to "ping" requests. Then spawn a separate
// thread to periodically ping it and check responsiveness.
@@ -713,7 +723,6 @@ fn async_main() -> impl Future<Item = (), Error = ()> + Send + 'static {
ping_receive
.for_each(move |pong_send| pong_send.clone().send(()).map(|_| ()).map_err(|_| ())),
);
- let contention_logger = logger.clone();
std::thread::spawn(move || loop {
std::thread::sleep(Duration::from_secs(1));
let (pong_send, pong_receive) = crossbeam_channel::bounded(1);
@@ -735,7 +744,6 @@ fn async_main() -> impl Future<Item = (), Error = ()> + Send + 'static {
});
future::empty()
- })
}
/// Parses an Ethereum connection string and returns the network name and Ethereum adapter.
