Bitswap Integration

blockchain/chain_sync/Cargo.toml

@@ -31,6 +31,7 @@ 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,17 +1,17 @@
 // 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::Sender;
 use blocks::{FullTipset, Tipset, TipsetKeys};
+use flo_stream::Subscriber;
 use forest_libp2p::{
     blocksync::{BlockSyncRequest, BlockSyncResponse, BLOCKS, MESSAGES},
     hello::HelloRequest,
-    rpc::{RPCRequest, RPCResponse, RequestId},
+    rpc::RequestId,
     NetworkEvent, NetworkMessage,
 };
+use futures::channel::oneshot::channel as oneshot_channel;
 use libp2p::core::PeerId;
 use log::trace;
 use std::time::Duration;
@@ -27,22 +27,14 @@ 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>,
 }
 
 impl SyncNetworkContext {
-    pub fn new(
-        network_send: Sender<NetworkMessage>,
-        rpc_receiver: RPCReceiver,
-        receiver: Receiver<NetworkEvent>,
-    ) -> Self {
+    pub fn new(network_send: Sender<NetworkMessage>, receiver: Subscriber<NetworkEvent>) -> Self {
         Self {
             network_send,
-            rpc_receiver,
             receiver,
             request_id: RequestId(1),
         }
@@ -97,16 +89,26 @@ 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?;
+        let request_id = self.request_id;
+        self.request_id.0 += 1;
 
-        if let RPCResponse::BlockSync(bs_res) = rpc_res {
-            Ok(bs_res)
-        } else {
-            Err("Invalid response type")
+        let (tx, rx) = oneshot_channel();
+
+        self.network_send
+            .send(NetworkMessage::BlockSyncRequest {
+                peer_id,
+                request,
+                id: request_id,
+                response_channel: tx,
+            })
+            .await;
+
+        match future::timeout(Duration::from_secs(RPC_TIMEOUT), rx).await {
+            Ok(Ok(bs_res)) => Ok(bs_res),
+            Ok(Err(e)) => Err(format!("RPC error: {}", e.to_string())),
+            Err(_) => Err("Connection timed out".to_string()),
         }
     }
 
@@ -115,42 +117,12 @@ impl SyncNetworkContext {
         trace!("Sending Hello Message {:?}", request);
         // TODO update to await response when we want to handle the latency
         self.network_send
-            .send(NetworkMessage::RPC {
+            .send(NetworkMessage::HelloRequest {
                 peer_id,
-                request: RPCRequest::Hello(request),
+                request,
                 id: self.request_id,
             })
             .await;
         self.request_id.0 += 1;
     }
-
-    /// Send any RPC request to the network and await the response
-    pub async fn send_rpc_request(
-        &mut self,
-        peer_id: PeerId,
-        request: RPCRequest,
-    ) -> Result<RPCResponse, &'static str> {
-        let request_id = self.request_id;
-        self.request_id.0 += 1;
-        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"),
-            }
-        }
-    }
 }

blockchain/chain_sync/src/network_handler.rs

@@ -3,67 +3,42 @@
 
 use super::peer_manager::PeerManager;
 use async_std::prelude::*;
-use async_std::sync::{Receiver, Sender};
+use async_std::sync::Receiver;
 use async_std::task;
-use forest_libp2p::rpc::{RPCResponse, RequestId};
+use flo_stream::{MessagePublisher, Publisher};
 use forest_libp2p::NetworkEvent;
 use log::trace;
 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>,
 }
 
 impl NetworkHandler {
     pub(crate) fn new(
         receiver: Receiver<NetworkEvent>,
-        rpc_send: RPCSender,
-        event_send: Sender<NetworkEvent>,
+        event_send: Publisher<NetworkEvent>,
     ) -> Self {
         Self {
             receiver,
-            rpc_send,
             event_send,
         }
     }
 
     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();
         task::spawn(async move {
-            loop {
-                match receiver.next().await {
-                    // Handle specifically RPC responses and send to that channel
-                    Some(NetworkEvent::BlockSyncResponse {
-                        request_id,
-                        response,
-                    }) => {
-                        rpc_send
-                            .send((request_id, RPCResponse::BlockSync(response)))
-                            .await
-                    }
-                    // Pass any non RPC responses through event channel
-                    Some(event) => {
-                        // Update peer on this thread before sending hello
-                        if let NetworkEvent::HelloRequest { channel, .. } = &event {
-                            // 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
-                        }
-                    }
-                    None => break,
+            while let Some(event) = receiver.next().await {
+                // Update peer on this thread before sending hello
+                if let NetworkEvent::HelloRequest { channel, .. } = &event {
+                    // TODO should probably add peer with their tipset/ not handled seperately
+                    peer_manager.add_peer(channel.peer.clone(), None).await;
+                }
+                if let NetworkEvent::BitswapBlock { .. } = &event {
+                    event_send.publish(event).await
                 }
             }
         });