Bitswap Integration

blockchain/chain_sync/Cargo.toml

@@ -31,6 +31,8 @@ num-traits = "0.2"
 filecoin-proofs-api = "4.0.1"
 fil_types = { path = "../../types" }
 commcid = { path = "../../utils/commcid" }
+flo_stream = "0.4"
+
 
 [dev-dependencies]
 test_utils = { version = "0.1.0", path = "../../utils/test_utils/", features = ["test_constructors"] }

blockchain/chain_sync/src/errors.rs

@@ -71,6 +71,12 @@ impl From<&str> for Error {
     }
 }
 
+impl From<String> for Error {
+    fn from(e: String) -> Error {
+        Error::Other(e)
+    }
+}
+
 impl From<std::num::TryFromIntError> for Error {
     fn from(e: std::num::TryFromIntError) -> Error {
         Error::Other(e.to_string())

blockchain/chain_sync/src/network_context.rs

@@ -1,19 +1,24 @@
 // Copyright 2020 ChainSafe Systems
 // SPDX-License-Identifier: Apache-2.0, MIT
 
-use super::network_handler::RPCReceiver;
 use async_std::future;
-use async_std::prelude::*;
-use async_std::sync::{Receiver, Sender};
+use async_std::sync::{Mutex, Sender};
 use blocks::{FullTipset, Tipset, TipsetKeys};
+use flo_stream::Subscriber;
 use forest_libp2p::{
     blocksync::{BlockSyncRequest, BlockSyncResponse, BLOCKS, MESSAGES},
     hello::HelloRequest,
     rpc::{RPCRequest, RPCResponse, RequestId},
     NetworkEvent, NetworkMessage,
 };
+
+use futures::channel::oneshot::{
+    channel as oneshot_channel, Receiver as OneShotReceiver, Sender as OneShotSender,
+};
 use libp2p::core::PeerId;
 use log::trace;
+use std::collections::HashMap;
+use std::sync::Arc;
 use std::time::Duration;
 
 /// Timeout for response from an RPC request
@@ -27,23 +32,21 @@ pub struct SyncNetworkContext {
     /// Handles sequential request ID enumeration for requests
     request_id: RequestId,
 
-    /// Receiver channel for BlockSync responses
-    rpc_receiver: RPCReceiver,
-
     /// Receiver channel for network events
-    pub receiver: Receiver<NetworkEvent>,
+    pub receiver: Subscriber<NetworkEvent>,
+    request_table: Arc<Mutex<HashMap<RequestId, OneShotSender<RPCResponse>>>>,
 }
 
 impl SyncNetworkContext {
     pub fn new(
         network_send: Sender<NetworkMessage>,
-        rpc_receiver: RPCReceiver,
-        receiver: Receiver<NetworkEvent>,
+        receiver: Subscriber<NetworkEvent>,
+        request_table: Arc<Mutex<HashMap<RequestId, OneShotSender<RPCResponse>>>>,
     ) -> Self {
         Self {
             network_send,
-            rpc_receiver,
             receiver,
+            request_table,
             request_id: RequestId(1),
         }
     }
@@ -97,16 +100,18 @@ impl SyncNetworkContext {
         &mut self,
         peer_id: PeerId,
         request: BlockSyncRequest,
-    ) -> Result<BlockSyncResponse, &'static str> {
+    ) -> Result<BlockSyncResponse, String> {
         trace!("Sending BlockSync Request {:?}", request);
         let rpc_res = self
             .send_rpc_request(peer_id, RPCRequest::BlockSync(request))
-            .await?;
-
-        if let RPCResponse::BlockSync(bs_res) = rpc_res {
-            Ok(bs_res)
-        } else {
-            Err("Invalid response type")
+            .await;
+        match future::timeout(Duration::from_secs(RPC_TIMEOUT), rpc_res).await {
+            Ok(Ok(RPCResponse::BlockSync(bs_res))) => Ok(bs_res),
+            Ok(Ok(RPCResponse::Hello(_))) => {
+                unreachable!();
+            }
+            Ok(Err(e)) => Err(format!("RPC error: {}", e.to_string())),
+            Err(_) => Err("Connection Timedout".to_string()),
         }
     }
 
@@ -129,28 +134,19 @@ impl SyncNetworkContext {
         &mut self,
         peer_id: PeerId,
         request: RPCRequest,
-    ) -> Result<RPCResponse, &'static str> {
+    ) -> OneShotReceiver<RPCResponse> {
         let request_id = self.request_id;
         self.request_id.0 += 1;
+
+        let (tx, rx) = oneshot_channel();
+        self.request_table.lock().await.insert(request_id, tx);
         self.network_send
             .send(NetworkMessage::RPC {
                 peer_id,
                 request,
                 id: request_id,
             })
             .await;
-        loop {
-            match future::timeout(Duration::from_secs(RPC_TIMEOUT), self.rpc_receiver.next()).await
-            {
-                Ok(Some((id, response))) => {
-                    if id == request_id {
-                        return Ok(response);
-                    }
-                    // Ignore any other RPC responses for now
-                }
-                Ok(None) => return Err("RPC Stream closed"),
-                Err(_) => return Err("Connection timeout"),
-            }
-        }
+        rx
     }
 }

blockchain/chain_sync/src/network_handler.rs

@@ -3,40 +3,42 @@
 
 use super::peer_manager::PeerManager;
 use async_std::prelude::*;
-use async_std::sync::{Receiver, Sender};
+use async_std::sync::Mutex;
+use async_std::sync::Receiver;
 use async_std::task;
+use flo_stream::{MessagePublisher, Publisher};
 use forest_libp2p::rpc::{RPCResponse, RequestId};
 use forest_libp2p::NetworkEvent;
-use log::trace;
+use futures::channel::oneshot::Sender as OneShotSender;
+use log::{debug, trace};
+use std::collections::HashMap;
 use std::sync::Arc;
 
-pub(crate) type RPCReceiver = Receiver<(RequestId, RPCResponse)>;
-pub(crate) type RPCSender = Sender<(RequestId, RPCResponse)>;
-
 /// Handles network events from channel and splits based on request
 pub(crate) struct NetworkHandler {
-    rpc_send: RPCSender,
-    event_send: Sender<NetworkEvent>,
+    event_send: Publisher<NetworkEvent>,
     receiver: Receiver<NetworkEvent>,
+    /// keeps track of a mapping from rpc request id to oneshot senders
+    request_table: Arc<Mutex<HashMap<RequestId, OneShotSender<RPCResponse>>>>,
 }
 
 impl NetworkHandler {
     pub(crate) fn new(
         receiver: Receiver<NetworkEvent>,
-        rpc_send: RPCSender,
-        event_send: Sender<NetworkEvent>,
+        event_send: Publisher<NetworkEvent>,
+        request_table: Arc<Mutex<HashMap<RequestId, OneShotSender<RPCResponse>>>>,
     ) -> Self {
         Self {
             receiver,
-            rpc_send,
             event_send,
+            request_table,
         }
     }
 
     pub(crate) fn spawn(&self, peer_manager: Arc<PeerManager>) {
         let mut receiver = self.receiver.clone();
-        let rpc_send = self.rpc_send.clone();
-        let event_send = self.event_send.clone();
+        let mut event_send = self.event_send.republish();
+        let request_table = self.request_table.clone();
 
         task::spawn(async move {
             loop {
@@ -46,9 +48,15 @@ impl NetworkHandler {
                         request_id,
                         response,
                     }) => {
-                        rpc_send
-                            .send((request_id, RPCResponse::BlockSync(response)))
-                            .await
+                        let tx = request_table.lock().await.remove(&request_id);
+
+                        if let Some(tx) = tx {
+                            if let Err(e) = tx.send(RPCResponse::BlockSync(response)) {
+                                debug!("RPCResponse receive failed: {:?}", e)
+                            }
+                        } else {
+                            debug!("RPCResponse receive failed: channel not found");
+                        };
                     }
                     // Pass any non RPC responses through event channel
                     Some(event) => {
@@ -57,10 +65,8 @@ impl NetworkHandler {
                             // TODO should probably add peer with their tipset/ not handled seperately
                             peer_manager.add_peer(channel.peer.clone(), None).await;
                         }
-
-                        // TODO revisit, doing this to avoid blocking this thread but can handle better
-                        if !event_send.is_full() {
-                            event_send.send(event).await
+                        if let NetworkEvent::BitswapBlock { .. } = &event {
+                            event_send.publish(event).await
                         }
                     }
                     None => break,

blockchain/chain_sync/src/sync.rs

@@ -10,7 +10,7 @@ use super::peer_manager::PeerManager;
 use super::{Error, SyncNetworkContext};
 use address::{Address, Protocol};
 use amt::Amt;
-use async_std::sync::{channel, Receiver, Sender};
+use async_std::sync::{Mutex, Receiver, Sender};
 use async_std::task;
 use beacon::{Beacon, BeaconEntry};
 use blocks::{Block, BlockHeader, FullTipset, Tipset, TipsetKeys, TxMeta};
@@ -23,6 +23,7 @@ use crypto::DomainSeparationTag;
 use encoding::{Cbor, Error as EncodingError};
 use fil_types::SectorInfo;
 use filecoin_proofs_api::{post::verify_winning_post, ProverId, PublicReplicaInfo, SectorId};
+use flo_stream::{MessagePublisher, Publisher};
 use forest_libp2p::{
     hello::HelloRequest, BlockSyncRequest, NetworkEvent, NetworkMessage, MESSAGES,
 };
@@ -121,14 +122,14 @@ where
         let state_manager = Arc::new(StateManager::new(chain_store.db.clone()));
 
         // Split incoming channel to handle blocksync requests
-        let (rpc_send, rpc_rx) = channel(20);
-        let (event_send, event_rx) = channel(30);
-
-        let network = SyncNetworkContext::new(network_send, rpc_rx, event_rx);
+        let mut event_send = Publisher::new(30);
+        let req_table = Arc::new(Mutex::new(HashMap::new()));
+        let network =
+            SyncNetworkContext::new(network_send, event_send.subscribe(), req_table.clone());
 
         let peer_manager = Arc::new(PeerManager::default());
 
-        let net_handler = NetworkHandler::new(network_rx, rpc_send, event_send);
+        let net_handler = NetworkHandler::new(network_rx, event_send, req_table);
 
         Ok(Self {
             state: SyncState::Init,
@@ -1020,6 +1021,7 @@ fn cids_from_messages<T: Cbor>(messages: &[T]) -> Result<Vec<Cid>, EncodingError
 #[cfg(test)]
 mod tests {
     use super::*;
+    use async_std::sync::channel;
     use async_std::sync::Sender;
     use beacon::MockBeacon;
     use blocks::BlockHeader;
@@ -1093,8 +1095,6 @@ mod tests {
         let (mut cs, event_sender) = chain_syncer_setup(db);
 
         cs.net_handler.spawn(Arc::clone(&cs.peer_manager));
-        // send blocksync response to channel
-        send_blocksync_response(event_sender);
 
         // params for sync_headers_reverse
         let source = PeerId::random();
@@ -1104,9 +1104,12 @@ mod tests {
         task::block_on(async move {
             cs.peer_manager.add_peer(source.clone(), None).await;
             assert_eq!(cs.peer_manager.len().await, 1);
-
-            let return_set = cs.sync_headers_reverse(head, &to).await;
-            assert_eq!(return_set.unwrap().len(), 4);
+            // make blocksync request
+            let return_set = task::spawn(async move { cs.sync_headers_reverse(head, &to).await });
+            task::sleep(Duration::from_secs(2)).await;
+            // send blocksync response to channel
+            send_blocksync_response(event_sender);
+            assert_eq!(return_set.await.unwrap().len(), 4);
         });
     }
 

forest/src/daemon.rs

@@ -45,7 +45,8 @@ pub(super) async fn start(config: Config) {
         initialize_genesis(&config.genesis_file, &mut chain_store).unwrap();
 
     // Libp2p service setup
-    let p2p_service = Libp2pService::new(config.network, net_keypair, &network_name);
+    let p2p_service =
+        Libp2pService::new(config.network, Arc::clone(&db), net_keypair, &network_name);
     let network_rx = p2p_service.network_receiver();
     let network_send = p2p_service.network_sender();
 

node/forest_libp2p/Cargo.toml

@@ -23,6 +23,9 @@ fnv = "1.0.6"
 smallvec = "1.1.0"
 clock = { path = "../clock" }
 num-bigint = { path = "../../utils/bigint", package = "forest_bigint" }
+libp2p-bitswap = { git = "https://github.com/ChainSafe/libp2p-bitswap", rev = "c31e0ae31fa7d0b848bf4fb711106e59aac0fa5e" }
+libipld-core = "0.3.0"
+ipld_blockstore = { path = "../../ipld/blockstore" }
 async-trait = "0.1"
 
 [dev-dependencies]