Reduce network bandwidth, improve parablock times: optimize approval-distribution

node/network/approval-distribution/src/lib.rs

@@ -216,7 +216,8 @@ impl State {
 					entry.known_by.remove(&peer_id);
 				})
 			},
-			NetworkBridgeEvent::NewGossipTopology(peers) => {
+			NetworkBridgeEvent::NewGossipTopology { our_neighbors_x, our_neighbors_y } => {
+				let peers: HashSet<_> = our_neighbors_x.union(&our_neighbors_y).cloned().collect();
 				let newly_added: Vec<PeerId> =
 					peers.difference(&self.gossip_peers).cloned().collect();
 				self.gossip_peers = peers;

node/network/bitfield-distribution/src/lib.rs

@@ -523,7 +523,8 @@ async fn handle_network_msg<Context>(
 			// get rid of superfluous data
 			state.peer_views.remove(&peer);
 		},
-		NetworkBridgeEvent::NewGossipTopology(peers) => {
+		NetworkBridgeEvent::NewGossipTopology { our_neighbors_x, our_neighbors_y } => {
+			let peers: HashSet<_> = our_neighbors_x.union(&our_neighbors_y).cloned().collect();
 			let newly_added: Vec<PeerId> = peers.difference(&state.gossip_peers).cloned().collect();
 			state.gossip_peers = peers;
 			for new_peer in newly_added {

node/network/bridge/src/lib.rs

@@ -590,30 +590,48 @@ where
 						).await;
 					}
 					NetworkBridgeMessage::NewGossipTopology {
-						our_neighbors,
+						our_neighbors_x,
+						our_neighbors_y,
 					} => {
 						gum::debug!(
 							target: LOG_TARGET,
 							action = "NewGossipTopology",
-							neighbors = our_neighbors.len(),
+							neighbors_x = our_neighbors_x.len(),
+							neighbors_y = our_neighbors_y.len(),
 							"Gossip topology has changed",
 						);
 
 						let ads = &mut authority_discovery_service;
-						let mut gossip_peers = HashSet::with_capacity(our_neighbors.len());
-						for authority in our_neighbors {
+						let mut gossip_peers_x = HashSet::with_capacity(our_neighbors_x.len());
+						let mut gossip_peers_y = HashSet::with_capacity(our_neighbors_y.len());
+
+						for authority in our_neighbors_x {
+							let addr = get_peer_id_by_authority_id(
+								ads,
+								authority.clone(),
+							).await;
+
+							if let Some(peer_id) = addr {
+								gossip_peers_x.insert(peer_id);
+							}
+						}
+
+						for authority in our_neighbors_y {
 							let addr = get_peer_id_by_authority_id(
 								ads,
 								authority.clone(),
 							).await;
 
 							if let Some(peer_id) = addr {
-								gossip_peers.insert(peer_id);
+								gossip_peers_y.insert(peer_id);
 							}
 						}
 
 						dispatch_validation_event_to_all_unbounded(
-							NetworkBridgeEvent::NewGossipTopology(gossip_peers),
+							NetworkBridgeEvent::NewGossipTopology {
+								our_neighbors_x: gossip_peers_x,
+								our_neighbors_y: gossip_peers_y,
+							},
 							ctx.sender(),
 						);
 					}

node/network/collator-protocol/src/collator_side/mod.rs

@@ -971,7 +971,7 @@ where
 		PeerMessage(remote, msg) => {
 			handle_incoming_peer_message(ctx, runtime, state, remote, msg).await?;
 		},
-		NewGossipTopology(..) => {
+		NewGossipTopology { .. } => {
 			// impossible!
 		},
 	}

node/network/collator-protocol/src/validator_side/mod.rs

@@ -1090,7 +1090,7 @@ where
 			state.peer_data.remove(&peer_id);
 			state.metrics.note_collator_peer_count(state.peer_data.len());
 		},
-		NewGossipTopology(..) => {
+		NewGossipTopology { .. } => {
 			// impossible!
 		},
 		PeerViewChange(peer_id, view) => {

node/network/gossip-support/src/lib.rs

@@ -378,7 +378,7 @@ where
 			},
 			NetworkBridgeEvent::OurViewChange(_) => {},
 			NetworkBridgeEvent::PeerViewChange(_, _) => {},
-			NetworkBridgeEvent::NewGossipTopology(_) => {},
+			NetworkBridgeEvent::NewGossipTopology { .. } => {},
 			NetworkBridgeEvent::PeerMessage(_, v) => {
 				match v {};
 			},
@@ -493,16 +493,29 @@ where
 		.expect("our_index < len; indices contains it; qed");
 
 	let neighbors = matrix_neighbors(our_shuffled_position, len);
-	let our_neighbors = neighbors.map(|i| authorities[indices[i]].clone()).collect();
+	let row_neighbors = neighbors.row_neighbors.map(|i| authorities[indices[i]].clone()).collect();
+	let column_neighbors =
+		neighbors.column_neighbors.map(|i| authorities[indices[i]].clone()).collect();
 
-	ctx.send_message(NetworkBridgeMessage::NewGossipTopology { our_neighbors })
-		.await;
+	ctx.send_message(NetworkBridgeMessage::NewGossipTopology {
+		our_neighbors_x: row_neighbors,
+		our_neighbors_y: column_neighbors,
+	})
+	.await;
 
 	Ok(())
 }
 
+struct MatrixNeighbors<R, C> {
+	row_neighbors: R,
+	column_neighbors: C,
+}
+
 /// Compute our row and column neighbors in a matrix
-fn matrix_neighbors(our_index: usize, len: usize) -> impl Iterator<Item = usize> {
+fn matrix_neighbors(
+	our_index: usize,
+	len: usize,
+) -> MatrixNeighbors<impl Iterator<Item = usize>, impl Iterator<Item = usize>> {
 	assert!(our_index < len, "our_index is computed using `enumerate`; qed");
 
 	// e.g. for size 11 the matrix would be
@@ -520,7 +533,10 @@ fn matrix_neighbors(our_index: usize, len: usize) -> impl Iterator<Item = usize>
 	let row_neighbors = our_row * sqrt..std::cmp::min(our_row * sqrt + sqrt, len);
 	let column_neighbors = (our_column..len).step_by(sqrt);
 
-	row_neighbors.chain(column_neighbors).filter(move |i| *i != our_index)
+	MatrixNeighbors {
+		row_neighbors: row_neighbors.filter(move |i| *i != our_index),
+		column_neighbors: column_neighbors.filter(move |i| *i != our_index),
+	}
 }
 
 impl<Context, AD> overseer::Subsystem<Context, SubsystemError> for GossipSupport<AD>

node/network/gossip-support/src/tests.rs

@@ -53,9 +53,18 @@ lazy_static! {
 		Sr25519Keyring::One.public().into(),
 		Sr25519Keyring::Two.public().into(),
 	];
-	static ref NEIGHBORS: Vec<AuthorityDiscoveryId> = vec![
-		Sr25519Keyring::Two.public().into(),
+
+	// [2 6]
+	// [4 5]
+	// [1 3]
+	// [0  ]
+
+	static ref ROW_NEIGHBORS: Vec<AuthorityDiscoveryId> = vec![
 		Sr25519Keyring::Charlie.public().into(),
+	];
+
+	static ref COLUMN_NEIGHBORS: Vec<AuthorityDiscoveryId> = vec![
+		Sr25519Keyring::Two.public().into(),
 		Sr25519Keyring::Eve.public().into(),
 	];
 }
@@ -209,11 +218,15 @@ async fn test_neighbors(overseer: &mut VirtualOverseer) {
 	assert_matches!(
 		overseer_recv(overseer).await,
 		AllMessages::NetworkBridge(NetworkBridgeMessage::NewGossipTopology {
-			our_neighbors,
+			our_neighbors_x,
+			our_neighbors_y,
 		}) => {
-			let mut got: Vec<_> = our_neighbors.into_iter().collect();
-			got.sort();
-			assert_eq!(got, NEIGHBORS.clone());
+			let mut got_row: Vec<_> = our_neighbors_x.into_iter().collect();
+			let mut got_column: Vec<_> = our_neighbors_y.into_iter().collect();
+			got_row.sort();
+			got_column.sort();
+			assert_eq!(got_row, ROW_NEIGHBORS.clone());
+			assert_eq!(got_column, COLUMN_NEIGHBORS.clone());
 		}
 	);
 }
@@ -504,18 +517,23 @@ fn issues_a_connection_request_when_last_request_was_mostly_unresolved() {
 
 #[test]
 fn test_matrix_neighbors() {
-	for (our_index, len, expected) in vec![
-		(0usize, 1usize, vec![]),
-		(1, 2, vec![0usize]),
-		(0, 9, vec![1, 2, 3, 6]),
-		(9, 10, vec![0, 3, 6]),
-		(10, 11, vec![1, 4, 7, 9]),
-		(7, 11, vec![1, 4, 6, 8, 10]),
+	for (our_index, len, expected_row, expected_column) in vec![
+		(0usize, 1usize, vec![], vec![]),
+		(1, 2, vec![], vec![0usize]),
+		(0, 9, vec![1, 2], vec![3, 6]),
+		(9, 10, vec![], vec![0, 3, 6]),
+		(10, 11, vec![9], vec![1, 4, 7]),
+		(7, 11, vec![6, 8], vec![1, 4, 10]),
 	]
 	.into_iter()
 	{
-		let mut result: Vec<_> = matrix_neighbors(our_index, len).collect();
-		result.sort();
-		assert_eq!(result, expected);
+		let matrix = matrix_neighbors(our_index, len);
+		let mut row_result: Vec<_> = matrix.row_neighbors.collect();
+		let mut column_result: Vec<_> = matrix.column_neighbors.collect();
+		row_result.sort();
+		column_result.sort();
+
+		assert_eq!(row_result, expected_row);
+		assert_eq!(column_result, expected_column);
 	}
 }

node/network/statement-distribution/src/lib.rs

@@ -1623,7 +1623,8 @@ async fn handle_network_update(
 				});
 			}
 		},
-		NetworkBridgeEvent::NewGossipTopology(new_peers) => {
+		NetworkBridgeEvent::NewGossipTopology { our_neighbors_x, our_neighbors_y } => {
+			let new_peers: HashSet<_> = our_neighbors_x.union(&our_neighbors_y).cloned().collect();
 			let _ = metrics.time_network_bridge_update_v1("new_gossip_topology");
 			let newly_added: Vec<PeerId> = new_peers.difference(gossip_peers).cloned().collect();
 			*gossip_peers = new_peers;

node/subsystem-types/src/messages.rs

@@ -378,9 +378,14 @@ pub enum NetworkBridgeMessage {
 	/// Inform the distribution subsystems about the new
 	/// gossip network topology formed.
 	NewGossipTopology {
-		/// Ids of our neighbors in the new gossip topology.
-		/// We're not necessarily connected to all of them, but we should.
-		our_neighbors: HashSet<AuthorityDiscoveryId>,
+		/// Ids of our neighbors in the X dimensions of the new gossip topology.
+		/// We're not necessarily connected to all of them, but we should
+		/// try to be.
+		our_neighbors_x: HashSet<AuthorityDiscoveryId>,
+		/// Ids of our neighbors in the X dimensions of the new gossip topology.
+		/// We're not necessarily connected to all of them, but we should
+		/// try to be.
+		our_neighbors_y: HashSet<AuthorityDiscoveryId>,
 	},
 }
 

node/subsystem-types/src/messages/network_bridge_event.rs

@@ -37,7 +37,12 @@ pub enum NetworkBridgeEvent<M> {
 	///
 	/// Note, that the distribution subsystems need to handle the last
 	/// view update of the newly added gossip peers manually.
-	NewGossipTopology(HashSet<PeerId>),
+	NewGossipTopology {
+		/// Neighbors in the 'X' dimension of the grid.
+		our_neighbors_x: HashSet<PeerId>,
+		/// Neighbors in the 'Y' dimension of the grid.
+		our_neighbors_y: HashSet<PeerId>,
+	},
 
 	/// Peer has sent a message.
 	PeerMessage(PeerId, M),
@@ -77,8 +82,11 @@ impl<M> NetworkBridgeEvent<M> {
 				NetworkBridgeEvent::PeerConnected(peer.clone(), role.clone(), authority_id.clone()),
 			NetworkBridgeEvent::PeerDisconnected(ref peer) =>
 				NetworkBridgeEvent::PeerDisconnected(peer.clone()),
-			NetworkBridgeEvent::NewGossipTopology(ref peers) =>
-				NetworkBridgeEvent::NewGossipTopology(peers.clone()),
+			NetworkBridgeEvent::NewGossipTopology { ref our_neighbors_x, ref our_neighbors_y } =>
+				NetworkBridgeEvent::NewGossipTopology {
+					our_neighbors_x: our_neighbors_x.clone(),
+					our_neighbors_y: our_neighbors_y.clone(),
+				},
 			NetworkBridgeEvent::PeerViewChange(ref peer, ref view) =>
 				NetworkBridgeEvent::PeerViewChange(peer.clone(), view.clone()),
 			NetworkBridgeEvent::OurViewChange(ref view) =>

roadmap/implementers-guide/src/types/network.md

@@ -154,7 +154,10 @@ enum NetworkBridgeEvent<M> {
 	///
 	/// Note, that the distribution subsystems need to handle the last
 	/// view update of the newly added gossip peers manually.
-	NewGossipTopology(HashSet<PeerId>),
+	NewGossipTopology {
+		our_neighbors_x: HashSet<PeerId>,
+		our_neighbors_y: HashSet<PeerId>,
+	}
 	/// We received a message from the given peer.
 	PeerMessage(PeerId, M),
 	/// The given peer has updated its description of its view.

roadmap/implementers-guide/src/types/overseer-protocol.md

@@ -553,9 +553,12 @@ enum NetworkBridgeMessage {
     /// Inform the distribution subsystems about the new
     /// gossip network topology formed.
     NewGossipTopology {
-        /// Ids of our neighbors in the new gossip topology.
-        /// We're not necessarily connected to all of them, but we should.
-        our_neighbors: HashSet<AuthorityDiscoveryId>,
+        /// Ids of our neighbors in the X dimension of the new gossip topology.
+        /// We're not necessarily connected to all of them, but we should try to be.
+        our_neighbors_x: HashSet<AuthorityDiscoveryId>,
+        /// Ids of our neighbors in the Y dimension of the new gossip topology.
+        /// We're not necessarily connected to all of them, but we should try to be.
+        our_neighbors_y: HashSet<AuthorityDiscoveryId>,
     }
 }
 ```