proposer: wait for a hash to be in the active-leaves set (#1616)

* overseer: add ExternalRequest to Event

* proposer: wait for the hash to be activated

* update comments

* overseer: handle unbounded growth of listeners map

* overseer: fix compilation

* overseer: clean up dead listeners

* overseer: cosmetic changes

* overseer: cosmetic changes t.2

* overseer: add debug_assertions

* overseer: fix formatting

polkadot/node/core/proposer/src/lib.rs

@@ -119,6 +119,10 @@ where
 		let mut provisioner_inherent_data = async move {
 			let (sender, receiver) = futures::channel::oneshot::channel();
 
+			overseer.wait_for_activation(parent_header_hash, sender).await?;
+			receiver.await.map_err(Error::ClosedChannelFromProvisioner)?;
+
+			let (sender, receiver) = futures::channel::oneshot::channel();
 			// strictly speaking, we don't _have_ to .await this send_msg before opening the
 			// receiver; it's possible that the response there would be ready slightly before
 			// this call completes. IMO it's not worth the hassle or overhead of spawning a

polkadot/node/overseer/src/lib.rs

@@ -59,7 +59,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::task::Poll;
 use std::time::Duration;
-use std::collections::HashSet;
+use std::collections::{hash_map, HashMap};
 
 use futures::channel::{mpsc, oneshot};
 use futures::{
@@ -165,9 +165,18 @@ enum Event {
 	BlockImported(BlockInfo),
 	BlockFinalized(BlockInfo),
 	MsgToSubsystem(AllMessages),
+	ExternalRequest(ExternalRequest),
 	Stop,
 }
 
+/// Some request from outer world.
+enum ExternalRequest {
+	WaitForActivation {
+		hash: Hash,
+		response_channel: oneshot::Sender<()>,
+	},
+}
+
 /// A handler used to communicate with the [`Overseer`].
 ///
 /// [`Overseer`]: struct.Overseer.html
@@ -179,30 +188,36 @@ pub struct OverseerHandler {
 impl OverseerHandler {
 	/// Inform the `Overseer` that that some block was imported.
 	pub async fn block_imported(&mut self, block: BlockInfo) -> SubsystemResult<()> {
-		self.events_tx.send(Event::BlockImported(block)).await?;
-
-		Ok(())
+		self.events_tx.send(Event::BlockImported(block)).await.map_err(Into::into)
 	}
 
 	/// Send some message to one of the `Subsystem`s.
 	pub async fn send_msg(&mut self, msg: AllMessages) -> SubsystemResult<()> {
-		self.events_tx.send(Event::MsgToSubsystem(msg)).await?;
-
-		Ok(())
+		self.events_tx.send(Event::MsgToSubsystem(msg)).await.map_err(Into::into)
 	}
 
 	/// Inform the `Overseer` that that some block was finalized.
 	pub async fn block_finalized(&mut self, block: BlockInfo) -> SubsystemResult<()> {
-		self.events_tx.send(Event::BlockFinalized(block)).await?;
+		self.events_tx.send(Event::BlockFinalized(block)).await.map_err(Into::into)
+	}
 
-		Ok(())
+	/// Wait for a block with the given hash to be in the active-leaves set.
+	/// This method is used for external code like `Proposer` that doesn't subscribe to Overseer's signals.
+	///
+	/// The response channel responds if the hash was activated and is closed if the hash was deactivated.
+	/// Note that due the fact the overseer doesn't store the whole active-leaves set, only deltas,
+	/// the response channel may never return if the hash was deactivated before this call.
+	/// In this case, it's the caller's responsibility to ensure a timeout is set.
+	pub async fn wait_for_activation(&mut self, hash: Hash, response_channel: oneshot::Sender<()>) -> SubsystemResult<()> {
+		self.events_tx.send(Event::ExternalRequest(ExternalRequest::WaitForActivation {
+			hash,
+			response_channel
+		})).await.map_err(Into::into)
 	}
 
 	/// Tell `Overseer` to shutdown.
 	pub async fn stop(&mut self) -> SubsystemResult<()> {
-		self.events_tx.send(Event::Stop).await?;
-
-		Ok(())
+		self.events_tx.send(Event::Stop).await.map_err(Into::into)
 	}
 }
 
@@ -297,9 +312,7 @@ impl<M: Send + 'static> SubsystemContext for OverseerSubsystemContext<M> {
 		self.tx.send(ToOverseer::SpawnJob {
 			name,
 			s,
-		}).await?;
-
-		Ok(())
+		}).await.map_err(Into::into)
 	}
 
 	async fn spawn_blocking(&mut self, name: &'static str, s: Pin<Box<dyn Future<Output = ()> + Send>>)
@@ -308,24 +321,18 @@ impl<M: Send + 'static> SubsystemContext for OverseerSubsystemContext<M> {
 		self.tx.send(ToOverseer::SpawnBlockingJob {
 			name,
 			s,
-		}).await?;
-
-		Ok(())
+		}).await.map_err(Into::into)
 	}
 
 	async fn send_message(&mut self, msg: AllMessages) -> SubsystemResult<()> {
-		self.tx.send(ToOverseer::SubsystemMessage(msg)).await?;
-
-		Ok(())
+		self.tx.send(ToOverseer::SubsystemMessage(msg)).await.map_err(Into::into)
 	}
 
 	async fn send_messages<T>(&mut self, msgs: T) -> SubsystemResult<()>
 		where T: IntoIterator<Item = AllMessages> + Send, T::IntoIter: Send
 	{
 		let mut msgs = stream::iter(msgs.into_iter().map(ToOverseer::SubsystemMessage).map(Ok));
-		self.tx.send_all(&mut msgs).await?;
-
-		Ok(())
+		self.tx.send_all(&mut msgs).await.map_err(Into::into)
 	}
 }
 
@@ -399,13 +406,16 @@ pub struct Overseer<S: SpawnNamed> {
 	/// Events that are sent to the overseer from the outside world
 	events_rx: mpsc::Receiver<Event>,
 
+	/// External listeners waiting for a hash to be in the active-leave set.
+	activation_external_listeners: HashMap<Hash, Vec<oneshot::Sender<()>>>,
+
 	/// A set of leaves that `Overseer` starts working with.
 	///
 	/// Drained at the beginning of `run` and never used again.
 	leaves: Vec<(Hash, BlockNumber)>,
 
 	/// The set of the "active leaves".
-	active_leaves: HashSet<(Hash, BlockNumber)>,
+	active_leaves: HashMap<Hash, BlockNumber>,
 
 	/// Various Prometheus metrics.
 	metrics: Metrics,
@@ -749,9 +759,10 @@ where
 			.map(|BlockInfo { hash, parent_hash: _, number }| (hash, number))
 			.collect();
 
-		let active_leaves = HashSet::new();
+		let active_leaves = HashMap::new();
 
 		let metrics = <Metrics as metrics::Metrics>::register(prometheus_registry);
+		let activation_external_listeners = HashMap::new();
 
 		let this = Self {
 			candidate_validation_subsystem,
@@ -773,6 +784,7 @@ where
 			running_subsystems,
 			running_subsystems_rx,
 			events_rx,
+			activation_external_listeners,
 			leaves,
 			active_leaves,
 			metrics,
@@ -863,10 +875,10 @@ where
 		let leaves = std::mem::take(&mut self.leaves);
 		let mut update = ActiveLeavesUpdate::default();
 
-		for leaf in leaves.into_iter() {
-			update.activated.push(leaf.0);
-			self.active_leaves.insert(leaf);
-			self.metrics.on_head_activated();
+		for (hash, number) in leaves.into_iter() {
+			update.activated.push(hash);
+			self.active_leaves.insert(hash, number);
+			self.on_head_activated(&hash);
 		}
 
 		self.broadcast_signal(OverseerSignal::ActiveLeaves(update)).await?;
@@ -887,6 +899,9 @@ where
 					Event::BlockFinalized(block) => {
 						self.block_finalized(block).await?;
 					}
+					Event::ExternalRequest(request) => {
+						self.handle_external_request(request);
+					}
 				}
 			}
 
@@ -919,17 +934,27 @@ where
 	async fn block_imported(&mut self, block: BlockInfo) -> SubsystemResult<()> {
 		let mut update = ActiveLeavesUpdate::default();
 
-		if let Some(parent) = block.number.checked_sub(1).and_then(|number| self.active_leaves.take(&(block.parent_hash, number))) {
-			update.deactivated.push(parent.0);
-			self.metrics.on_head_deactivated();
+		if let Some(number) = self.active_leaves.remove(&block.parent_hash) {
+			if let Some(expected_parent_number) = block.number.checked_sub(1) {
+				debug_assert_eq!(expected_parent_number, number);
+			}
+			update.deactivated.push(block.parent_hash);
+			self.on_head_deactivated(&block.parent_hash);
 		}
 
-		if !self.active_leaves.contains(&(block.hash, block.number)) {
-			update.activated.push(block.hash);
-			self.active_leaves.insert((block.hash, block.number));
-			self.metrics.on_head_activated();
+		match self.active_leaves.entry(block.hash) {
+			hash_map::Entry::Vacant(entry) => {
+				update.activated.push(block.hash);
+				entry.insert(block.number);
+				self.on_head_activated(&block.hash);
+			},
+			hash_map::Entry::Occupied(entry) => {
+				debug_assert_eq!(*entry.get(), block.number);
+			}
 		}
 
+		self.clean_up_external_listeners();
+
 		self.broadcast_signal(OverseerSignal::ActiveLeaves(update)).await?;
 
 		Ok(())
@@ -937,18 +962,20 @@ where
 
 	async fn block_finalized(&mut self, block: BlockInfo) -> SubsystemResult<()> {
 		let mut update = ActiveLeavesUpdate::default();
-		let metrics = &self.metrics;
 
-		self.active_leaves.retain(|(h, n)| {
+		self.active_leaves.retain(|h, n| {
 			if *n <= block.number {
 				update.deactivated.push(*h);
-				metrics.on_head_deactivated();
 				false
 			} else {
 				true
 			}
 		});
 
+		for deactivated in &update.deactivated {
+			self.on_head_deactivated(deactivated)
+		}
+
 		self.broadcast_signal(OverseerSignal::ActiveLeaves(update)).await?;
 
 		self.broadcast_signal(OverseerSignal::BlockFinalized(block.hash)).await?;
@@ -1100,6 +1127,45 @@ where
 		}
 	}
 
+	fn on_head_activated(&mut self, hash: &Hash) {
+		self.metrics.on_head_activated();
+		if let Some(listeners) = self.activation_external_listeners.remove(hash) {
+			for listener in listeners {
+				// it's fine if the listener is no longer interested
+				let _ = listener.send(());
+			}
+		}
+	}
+
+	fn on_head_deactivated(&mut self, hash: &Hash) {
+		self.metrics.on_head_deactivated();
+		if let Some(listeners) = self.activation_external_listeners.remove(hash) {
+			// clean up and signal to listeners the block is deactivated
+			drop(listeners);
+		}
+	}
+
+	fn clean_up_external_listeners(&mut self) {
+		self.activation_external_listeners.retain(|_, v| {
+			// remove dead listeners
+			v.retain(|c| !c.is_canceled());
+			!v.is_empty()
+		})
+	}
+
+	fn handle_external_request(&mut self, request: ExternalRequest) {
+		match request {
+			ExternalRequest::WaitForActivation { hash, response_channel } => {
+				if self.active_leaves.get(&hash).is_some() {
+					// it's fine if the listener is no longer interested
+					let _ = response_channel.send(());
+				} else {
+					self.activation_external_listeners.entry(hash).or_default().push(response_channel);
+				}
+			}
+		}
+	}
+
 	fn spawn_job(&mut self, name: &'static str, j: BoxFuture<'static, ()>) {
 		self.s.spawn(name, j);
 	}