From 8eecdd61b835202df2409e23d117aaac21413a20 Mon Sep 17 00:00:00 2001 From: Andronik Ordian Date: Thu, 20 Aug 2020 15:43:36 +0200 Subject: [PATCH] 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 | 4 + polkadot/node/overseer/src/lib.rs | 146 ++++++++++++++++++------- 2 files changed, 110 insertions(+), 40 deletions(-) diff --git a/polkadot/node/core/proposer/src/lib.rs b/polkadot/node/core/proposer/src/lib.rs index d43810227b..8fd432a1be 100644 --- a/polkadot/node/core/proposer/src/lib.rs +++ b/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 diff --git a/polkadot/node/overseer/src/lib.rs b/polkadot/node/overseer/src/lib.rs index 1d4c7bcaf4..8228682fcd 100644 --- a/polkadot/node/overseer/src/lib.rs +++ b/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 SubsystemContext for OverseerSubsystemContext { 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 + Send>>) @@ -308,24 +321,18 @@ impl SubsystemContext for OverseerSubsystemContext { 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(&mut self, msgs: T) -> SubsystemResult<()> where T: IntoIterator + 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 { /// Events that are sent to the overseer from the outside world events_rx: mpsc::Receiver, + /// External listeners waiting for a hash to be in the active-leave set. + activation_external_listeners: HashMap>>, + /// 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, /// 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 = ::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); }