Implement bitswap in the network behaviour using libp2p_bitswap.

client/network-gossip/src/bridge.rs

@@ -210,7 +210,8 @@ impl<B: BlockT> Future for GossipEngine<B> {
 
 								this.forwarding_state = ForwardingState::Busy(to_forward.into());
 							},
-							Event::Dht(_) => {}
+							Event::Dht(_) => {},
+							Event::Bitswap(_) => {},
 						}
 						// The network event stream closed. Do the same for [`GossipValidator`].
 						Poll::Ready(None) => return Poll::Ready(()),

client/network/Cargo.toml

@@ -32,6 +32,7 @@ futures-timer = "3.0.2"
 futures_codec = "0.4.0"
 hex = "0.4.0"
 ip_network = "0.3.4"
+libp2p-bitswap = "0.7.0"
 linked-hash-map = "0.5.2"
 linked_hash_set = "0.1.3"
 log = "0.4.8"
@@ -57,6 +58,7 @@ sp-core = { version = "2.0.0-rc6", path = "../../primitives/core" }
 sp-runtime = { version = "2.0.0-rc6", path = "../../primitives/runtime" }
 sp-utils = { version = "2.0.0-rc6", path = "../../primitives/utils" }
 thiserror = "1"
+tiny-cid = "0.2.5"
 unsigned-varint = { version = "0.4.0", features = ["futures", "futures-codec"] }
 void = "1.0.2"
 wasm-timer = "0.2"

client/network/src/behaviour.rs

@@ -54,6 +54,8 @@ pub struct Behaviour<B: BlockT, H: ExHashT> {
 	peer_info: peer_info::PeerInfoBehaviour,
 	/// Discovers nodes of the network.
 	discovery: DiscoveryBehaviour,
+	/// Exchanges blocks of data with other nodes.
+	bitswap: libp2p_bitswap::Bitswap,
 	/// Generic request-reponse protocols.
 	request_responses: request_responses::RequestResponsesBehaviour,
 	/// Block request handling.
@@ -172,6 +174,23 @@ pub enum BehaviourOut<B: BlockT> {
 	/// Events generated by a DHT as a response to get_value or put_value requests as well as the
 	/// request duration.
 	Dht(DhtEvent, Duration),
+
+	/// Event generated by bitswap.
+	Bitswap(BitswapEvent)
+}
+
+/// An event generated by bitswap.
+#[derive(Clone, Debug)]
+pub enum BitswapEvent {
+	/// A block has been received.
+	///
+	/// There are no guarantees about the recieved block:
+	/// * The CID and block contents might not match.
+	/// * We might have receieved a block without having sent out a want request first.
+	/// * It is possible to receieve this message multiple times with the same block.
+	ReceivedBlock(PeerId, tiny_cid::Cid, Box<[u8]>),
+	/// A peer wants the block with the hash of `cid`, and with a given priority.
+	ReceivedWant(PeerId, tiny_cid::Cid, libp2p_bitswap::Priority),
 }
 
 impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
@@ -197,6 +216,7 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 			finality_proof_requests,
 			light_client_handler,
 			events: VecDeque::new(),
+			bitswap: libp2p_bitswap::Bitswap::new(),
 			role,
 		})
 	}
@@ -300,6 +320,37 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 	pub fn light_client_request(&mut self, r: light_client_handler::Request<B>) -> Result<(), light_client_handler::Error> {
 		self.light_client_handler.request(r)
 	}
+
+	/// Count the number of connected peers who want the bitswap block with the hash of `cid`.
+	pub fn bitswap_num_peers_want(&self, cid: &tiny_cid::Cid) -> usize {
+		self.bitswap.peers_want(cid).count()
+	}
+
+	/// Determine whether a specific peer wants a bitswap block with the hash of `cid`.
+	pub fn bitswap_peer_wants_cid(&self, peer_id: &PeerId, cid: &tiny_cid::Cid) -> bool {
+		self.bitswap.peers_want(cid).find(|id| **id == *peer_id).is_some()
+	}
+
+	/// Send a bitswap block to a specific peer, regardless of whether they want it or not.
+	pub fn bitswap_send_block(&mut self, peer_id: &PeerId, cid: tiny_cid::Cid, block: Box<[u8]>) {
+		self.bitswap.send_block(peer_id, cid, block)
+	}
+
+	/// Send a bitswap block to all connected peers who want it.
+	pub fn bitswap_send_block_all(&mut self, cid: &tiny_cid::Cid, block: &[u8]) {
+		self.bitswap.send_block_all(cid, block)
+	}
+
+	/// Send out a bitswap message telling connected peers that we want the block with the hash of
+	/// `cid`. We can also say how this block should be prioritised.
+	pub fn bitswap_want_block(&mut self, cid: tiny_cid::Cid, priority: libp2p_bitswap::Priority) {
+		self.bitswap.want_block(cid, priority)
+	}
+
+	/// Send out a bitswap message cancelling a previously wanted block with the hash of `cid`.
+	pub fn bitswap_cancel_block(&mut self, cid: &tiny_cid::Cid) {
+		self.bitswap.cancel_block(cid)
+	}
 }
 
 fn reported_roles_to_observed_role(local_role: &Role, remote: &PeerId, roles: Roles) -> ObservedRole {
@@ -537,6 +588,25 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<DiscoveryOut>
 	}
 }
 
+impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<libp2p_bitswap::BitswapEvent>
+	for Behaviour<B, H> {
+	fn inject_event(&mut self, event: libp2p_bitswap::BitswapEvent) {
+		match event {
+			libp2p_bitswap::BitswapEvent::ReceivedBlock(peer_id, cid, data) => {
+				self.events.push_back(
+					BehaviourOut::Bitswap(BitswapEvent::ReceivedBlock(peer_id, cid, data))
+				);
+			},
+			libp2p_bitswap::BitswapEvent::ReceivedWant(peer_id, cid, priority) => {
+				self.events.push_back(
+					BehaviourOut::Bitswap(BitswapEvent::ReceivedWant(peer_id, cid, priority))
+				);
+			}
+			libp2p_bitswap::BitswapEvent::ReceivedCancel(..) => {},
+		}
+	}
+}
+
 impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 	fn poll<TEv>(&mut self, _: &mut Context, _: &mut impl PollParameters) -> Poll<NetworkBehaviourAction<TEv, BehaviourOut<B>>> {
 		if let Some(event) = self.events.pop_front() {

client/network/src/lib.rs

@@ -266,6 +266,7 @@ pub mod network_state;
 
 #[doc(inline)]
 pub use libp2p::{multiaddr, Multiaddr, PeerId};
+pub use behaviour::BitswapEvent;
 pub use protocol::{event::{DhtEvent, Event, ObservedRole}, sync::SyncState, PeerInfo};
 pub use service::{
 	NetworkService, NetworkWorker, RequestFailure, OutboundFailure, NotificationSender,

client/network/src/protocol/event.rs

@@ -22,6 +22,8 @@ use libp2p::core::PeerId;
 use libp2p::kad::record::Key;
 use sp_runtime::ConsensusEngineId;
 
+pub use crate::behaviour::BitswapEvent;
+
 /// Events generated by DHT as a response to get_value and put_value requests.
 #[derive(Debug, Clone)]
 #[must_use]
@@ -45,6 +47,8 @@ pub enum DhtEvent {
 pub enum Event {
 	/// Event generated by a DHT.
 	Dht(DhtEvent),
+	/// Event generated by Bitswap.
+	Bitswap(BitswapEvent),
 
 	/// Opened a substream with the given node with the given notifications protocol.
 	///

client/network/src/service.rs

@@ -597,6 +597,16 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
 	pub fn add_reserved_peer(&self, peer: String) -> Result<(), String> {
 		self.service.add_reserved_peer(peer)
 	}
+
+	/// Count the number of connected peers who want the bitswap block with the hash of `cid`.
+	pub fn bitswap_num_peers_want(&self, cid: &tiny_cid::Cid) -> usize {
+		self.network_service.bitswap_num_peers_want(cid)
+	}
+
+	/// Determine whether a specific peer wants a bitswap block with the hash of `cid`.
+	pub fn bitswap_peer_wants_cid(&self, peer_id: &PeerId, cid: &tiny_cid::Cid) -> bool {
+		self.network_service.bitswap_peer_wants_cid(peer_id, cid)
+	}
 }
 
 impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
@@ -1027,6 +1037,35 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
 			.to_worker
 			.unbounded_send(ServiceToWorkerMsg::OwnBlockImported(hash, number));
 	}
+
+	/// Send a bitswap block to a specific peer, regardless of whether they want it or not.
+	pub fn bitswap_send_block(&self, peer_id: PeerId, cid: tiny_cid::Cid, data: Box<[u8]>) {
+		let _ = self
+			.to_worker
+			.unbounded_send(ServiceToWorkerMsg::BitswapSendBlock(peer_id, cid, data));
+	}
+
+	/// Send a bitswap block to all connected peers who want it.
+	pub fn bitswap_send_block_all(&self, cid: tiny_cid::Cid, data: Box<[u8]>) {
+		let _ = self
+			.to_worker
+			.unbounded_send(ServiceToWorkerMsg::BitswapSendBlockAll(cid, data));
+	}
+
+	/// Send out a bitswap message telling connected peers that we want the block with the hash of
+	/// `cid`. We can also say how this block should be prioritised.
+	pub fn bitswap_want_block(&self, cid: tiny_cid::Cid, priority: libp2p_bitswap::Priority) {
+		let _ = self
+			.to_worker
+			.unbounded_send(ServiceToWorkerMsg::BitswapWantBlock(cid, priority));
+	}
+
+	/// Send out a bitswap message cancelling a previously wanted block with the hash of `cid`.
+	pub fn bitswap_cancel_block(&self, cid: tiny_cid::Cid) {
+		let _ = self
+			.to_worker
+			.unbounded_send(ServiceToWorkerMsg::BitswapCancelBlock(cid));
+	}
 }
 
 impl<B: BlockT + 'static, H: ExHashT> sp_consensus::SyncOracle
@@ -1159,6 +1198,10 @@ enum ServiceToWorkerMsg<B: BlockT, H: ExHashT> {
 	DisconnectPeer(PeerId),
 	UpdateChain,
 	OwnBlockImported(B::Hash, NumberFor<B>),
+	BitswapSendBlock(PeerId, tiny_cid::Cid, Box<[u8]>),
+	BitswapSendBlockAll(tiny_cid::Cid, Box<[u8]>),
+	BitswapWantBlock(tiny_cid::Cid, libp2p_bitswap::Priority),
+	BitswapCancelBlock(tiny_cid::Cid),
 }
 
 /// Main network worker. Must be polled in order for the network to advance.
@@ -1301,6 +1344,14 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
 					this.network_service.user_protocol_mut().update_chain(),
 				ServiceToWorkerMsg::OwnBlockImported(hash, number) =>
 					this.network_service.user_protocol_mut().own_block_imported(hash, number),
+				ServiceToWorkerMsg::BitswapSendBlock(peer_id, cid, block) =>
+					this.network_service.bitswap_send_block(&peer_id, cid, block),
+				ServiceToWorkerMsg::BitswapSendBlockAll(cid, block) =>
+					this.network_service.bitswap_send_block_all(&cid, &block),
+				ServiceToWorkerMsg::BitswapWantBlock(cid, priority) =>
+					this.network_service.bitswap_want_block(cid, priority),
+				ServiceToWorkerMsg::BitswapCancelBlock(cid) =>
+					this.network_service.bitswap_cancel_block(&cid),
 			}
 		}
 
@@ -1506,6 +1557,9 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
 
 					this.event_streams.send(Event::Dht(event));
 				},
+				Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::Bitswap(ev))) => {
+					this.event_streams.send(Event::Bitswap(ev));
+				},
 				Poll::Ready(SwarmEvent::ConnectionEstablished { peer_id, endpoint, num_established }) => {
 					trace!(target: "sub-libp2p", "Libp2p => Connected({:?})", peer_id);
 
@@ -1531,14 +1585,14 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
 						let reason = match cause {
 							Some(ConnectionError::IO(_)) => "transport-error",
 							Some(ConnectionError::Handler(NodeHandlerWrapperError::Handler(EitherError::A(EitherError::A(
-								EitherError::A(EitherError::A(EitherError::A(EitherError::B(
-								EitherError::A(PingFailure::Timeout)))))))))) => "ping-timeout",
+								EitherError::A(EitherError::A(EitherError::A(EitherError::A(EitherError::B(
+								EitherError::A(PingFailure::Timeout))))))))))) => "ping-timeout",
 							Some(ConnectionError::Handler(NodeHandlerWrapperError::Handler(EitherError::A(EitherError::A(
-								EitherError::A(EitherError::A(EitherError::A(EitherError::A(
-								NotifsHandlerError::Legacy(LegacyConnectionKillError)))))))))) => "force-closed",
+								EitherError::A(EitherError::A(EitherError::A(EitherError::A(EitherError::A(
+								NotifsHandlerError::Legacy(LegacyConnectionKillError))))))))))) => "force-closed",
 							Some(ConnectionError::Handler(NodeHandlerWrapperError::Handler(EitherError::A(EitherError::A(
-								EitherError::A(EitherError::A(EitherError::A(EitherError::A(
-								NotifsHandlerError::SyncNotificationsClogged))))))))) => "sync-notifications-clogged",
+								EitherError::A(EitherError::A(EitherError::A(EitherError::A(EitherError::A(
+								NotifsHandlerError::SyncNotificationsClogged)))))))))) => "sync-notifications-clogged",
 							Some(ConnectionError::Handler(NodeHandlerWrapperError::Handler(_))) => "protocol-error",
 							Some(ConnectionError::Handler(NodeHandlerWrapperError::KeepAliveTimeout)) => "keep-alive-timeout",
 							None => "actively-closed",

client/network/src/service/out_events.rs

@@ -227,6 +227,11 @@ impl Metrics {
 				self.events_total
 					.with_label_values(&["dht", "sent", name])
 					.inc_by(num);
+			},
+			Event::Bitswap(_) => {
+				self.events_total
+					.with_label_values(&["bitswap", "sent", name])
+					.inc_by(num);
 			}
 			Event::NotificationStreamOpened { engine_id, .. } => {
 				self.events_total
@@ -258,6 +263,11 @@ impl Metrics {
 					.with_label_values(&["dht", "received", name])
 					.inc();
 			}
+			Event::Bitswap(_) => {
+				self.events_total
+					.with_label_values(&["bitswap", "received", name])
+					.inc();
+			}
 			Event::NotificationStreamOpened { engine_id, .. } => {
 				self.events_total
 					.with_label_values(&[&format!("notif-open-{:?}", engine_id), "received", name])

client/network/src/service/tests.rs

@@ -271,6 +271,8 @@ fn notifications_state_consistent() {
 				// Add new events here.
 				future::Either::Left(Event::Dht(_)) => {}
 				future::Either::Right(Event::Dht(_)) => {}
+				future::Either::Left(Event::Bitswap(_)) => {}
+				future::Either::Right(Event::Bitswap(_)) => {}
 			};
 		}
 	});