Use multi-thread executor to execute network futures (#5)

nervosnetwork · Nov 20, 2018 · 3620f9c · 3620f9c
1 parent a3a16ae
commit 3620f9c
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 11 deletions.
diff --git a/network/src/ckb_service.rs b/network/src/ckb_service.rs
@@ -40,8 +40,10 @@ impl CKBService {
             match network.ckb_protocol_connec(&peer_id, protocol_id, endpoint, addresses.clone()) {
                 Ok(protocol_connec) => protocol_connec,
                 Err(err) => {
-                    return Box::new(future::err(IoError::new(IoErrorKind::Other, err)))
-                        as Box<Future<Item = (), Error = IoError> + Send>
+                    return Box::new(future::err(IoError::new(
+                        IoErrorKind::Other,
+                        format!("handle ckb_protocol connection error: {}", err),
+                    ))) as Box<Future<Item = (), Error = IoError> + Send>
                 }
             };
         if protocol_connec.state() == UniqueConnecState::Full {
@@ -133,13 +135,19 @@ impl CKBService {
             peer_store.report(&peer_id, Behaviour::Connect);
             peer_store.report_status(&peer_id, Status::Connected);
         }
-        protocol_handler.connected(
-            Box::new(DefaultCKBProtocolContext::new(
-                Arc::clone(&network),
-                protocol_id,
-            )),
-            peer_index,
-        );
+        {
+            let handle_connected = future::lazy(move || {
+                protocol_handler.connected(
+                    Box::new(DefaultCKBProtocolContext::new(
+                        Arc::clone(&network),
+                        protocol_id,
+                    )),
+                    peer_index,
+                );
+                Ok(())
+            });
+            tokio::spawn(handle_connected);
+        }
         Box::new(protocol_future) as Box<_>
     }
 }

diff --git a/network/src/discovery_service.rs b/network/src/discovery_service.rs
@@ -235,6 +235,8 @@ impl DiscoveryService {
         //};
 
         let handling_future = Box::new(
+            // why use loop_fn?????????????
+            // does client disconnect after discovery?
             future::loop_fn(kademlia_stream, {
                 let peer_id = peer_id.clone();
                 let kad_system = Arc::clone(&self.kad_system);

diff --git a/network/src/network_service.rs b/network/src/network_service.rs
@@ -14,7 +14,7 @@ use std::boxed::Box;
 use std::io::{Error as IoError, ErrorKind as IoErrorKind};
 use std::sync::Arc;
 use std::thread;
-use tokio::runtime::current_thread;
+use tokio::runtime;
 
 pub struct NetworkService {
     network: Arc<Network>,
@@ -82,7 +82,9 @@ impl NetworkService {
                 let network_future =
                     Network::build_network_future(network, &config, close_rx).unwrap();
                 init_tx.send(()).unwrap();
-                match current_thread::block_on_all(network_future) {
+                // here we use default config
+                let network_runtime = runtime::Runtime::new().unwrap();
+                match network_runtime.block_on_all(network_future) {
                     Ok(_) => info!(target: "network", "network service exit"),
                     Err(err) => panic!("network service exit unexpected {}", err),
                 }