Merge #274

274: fix bogus empty bitswap messages r=ljedrz a=ljedrz

We've been seeing `(empty message)` bitswap for a while, but these logs were bogus - this PR fixes those, and adds a small bitswap stress test that can be used for future bitswap debugging.

Also removes some noisy logging and, as a drive-by, fixes a subscription-related log.

Co-authored-by: ljedrz <ljedrz@gmail.com>
Co-authored-by: ljedrz <ljedrz@users.noreply.github.com>

bitswap/src/behaviour.rs

@@ -7,7 +7,7 @@
 //! will allow providing and reciving IPFS blocks.
 use crate::block::Block;
 use crate::ledger::{Ledger, Message, Priority};
-use crate::protocol::BitswapConfig;
+use crate::protocol::{BitswapConfig, MessageWrapper};
 use cid::Cid;
 use fnv::FnvHashSet;
 use futures::channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
@@ -208,7 +208,7 @@ impl Bitswap {
 }
 
 impl NetworkBehaviour for Bitswap {
-    type ProtocolsHandler = OneShotHandler<BitswapConfig, Message, Message>;
+    type ProtocolsHandler = OneShotHandler<BitswapConfig, Message, MessageWrapper>;
     type OutEvent = BitswapEvent;
 
     fn new_handler(&mut self) -> Self::ProtocolsHandler {
@@ -234,7 +234,16 @@ impl NetworkBehaviour for Bitswap {
         //self.connected_peers.remove(peer_id);
     }
 
-    fn inject_event(&mut self, source: PeerId, _connection: ConnectionId, mut message: Message) {
+    fn inject_event(&mut self, source: PeerId, _connection: ConnectionId, message: MessageWrapper) {
+        let mut message = match message {
+            // we just sent an outgoing bitswap message, nothing to do here
+            // FIXME: we could commit any pending stats accounting for this peer now
+            // that the message may have sent, if we'd do such accounting
+            MessageWrapper::Tx => return,
+            // we've received a bitswap message, process it
+            MessageWrapper::Rx(msg) => msg,
+        };
+
         debug!("bitswap: inject_event from {}: {:?}", source, message);
 
         let current_wantlist = self.local_wantlist();

bitswap/src/protocol.rs

@@ -68,16 +68,37 @@ where
     type Future = FutureResult<Self::Output, Self::Error>;
 
     #[inline]
-    fn upgrade_outbound(self, mut socket: TSocket, info: Self::Info) -> Self::Future {
+    fn upgrade_outbound(self, mut socket: TSocket, _info: Self::Info) -> Self::Future {
         Box::pin(async move {
-            debug!("upgrade_outbound: {}", std::str::from_utf8(info).unwrap());
             let bytes = self.to_bytes();
-            upgrade::write_one(&mut socket, bytes).await?;
+            upgrade::write_one(&mut socket, bytes).await
-            Ok(())
         })
     }
 }
 
+/// An object to facilitate communication between the `OneShotHandler` and the `BitswapHandler`.
+#[derive(Debug)]
+pub enum MessageWrapper {
+    /// We received a `Message` from a remote.
+    Rx(Message),
+    /// We successfully sent a `Message`.
+    Tx,
+}
+
+impl From<Message> for MessageWrapper {
+    #[inline]
+    fn from(message: Message) -> Self {
+        Self::Rx(message)
+    }
+}
+
+impl From<()> for MessageWrapper {
+    #[inline]
+    fn from(_: ()) -> Self {
+        Self::Tx
+    }
+}
+
 #[cfg(test)]
 mod tests {
     /*

src/subscription.rs

@@ -128,7 +128,6 @@ impl<TRes: Debug + Clone + PartialEq> SubscriptionRegistry<TRes> {
     /// Finalizes all pending subscriptions of the specified kind with the given `result`.
     ///
     pub fn finish_subscription(&self, req_kind: RequestKind, result: TRes) {
-        debug!("Finishing the subscription to {}", req_kind);
         let mut subscriptions = task::block_on(async { self.subscriptions.lock().await });
         let related_subs = subscriptions.get_mut(&req_kind);
 
@@ -136,6 +135,8 @@ impl<TRes: Debug + Clone + PartialEq> SubscriptionRegistry<TRes> {
         // ones have an associated `SubscriptionFuture` and there can be multiple of them
         // depending on how many times the given request kind was filed
         if let Some(related_subs) = related_subs {
+            debug!("Finishing the subscription to {}", req_kind);
+
             for sub in related_subs.values_mut() {
                 if let Subscription::Pending { .. } = sub {
                     sub.wake(result.clone());

tests/bitswap_stress_test.rs

@@ -0,0 +1,54 @@
+use ipfs::{Block, Node};
+use libipld::cid::{Cid, Codec};
+use multihash::Sha2_256;
+
+fn filter(i: usize) -> bool {
+    i % 2 == 0
+}
+
+// this test is designed to trigger unfavorable conditions for the bitswap
+// protocol by putting blocks in every second node and attempting to get
+// them from the other nodes; intended to be used for debugging or stress
+// testing the bitswap protocol (though it would be advised to uncomment
+// the tracing_subscriber for stress-testing purposes)
+#[ignore]
+#[async_std::test]
+async fn bitswap_stress_test() {
+    tracing_subscriber::fmt::init();
+
+    let data = b"hello block\n".to_vec().into_boxed_slice();
+    let cid = Cid::new_v1(Codec::Raw, Sha2_256::digest(&data));
+
+    const NODE_COUNT: usize = 3;
+
+    let mut nodes = Vec::with_capacity(NODE_COUNT);
+    for i in 0..NODE_COUNT {
+        nodes.push(Node::new(i.to_string()).await);
+    }
+
+    for i in 0..NODE_COUNT {
+        for (j, peer) in nodes.iter().enumerate() {
+            if i != j {
+                let (_, mut addrs) = peer.identity().await.unwrap();
+                nodes[i].connect(addrs.pop().unwrap()).await.unwrap();
+            }
+        }
+    }
+
+    for (i, node) in nodes.iter().enumerate() {
+        if filter(i) {
+            node.put_block(Block {
+                cid: cid.clone(),
+                data: data.clone(),
+            })
+            .await
+            .unwrap();
+        }
+    }
+
+    for (i, node) in nodes.iter().enumerate() {
+        if !filter(i) {
+            node.get_block(&cid).await.unwrap();
+        }
+    }
+}