Newer
Older
/// }
/// }
///
/// # fn main() { executor::block_on(async move {
/// let spawner = sp_core::testing::TaskExecutor::new();
/// let all_subsystems = AllSubsystems::<()>::dummy().replace_candidate_validation(ValidationSubsystem);
/// all_subsystems,
/// spawner,
/// ).unwrap();
///
/// let timer = Delay::new(Duration::from_millis(50)).fuse();
///
/// let overseer_fut = overseer.run().fuse();
/// pin_mut!(timer);
/// pin_mut!(overseer_fut);
///
/// select! {
/// _ = overseer_fut => (),
/// _ = timer => (),
/// }
/// #
/// # }); }
/// ```
pub fn new<CV, CB, CS, SD, AD, BS, BD, P, PoVD, RA, AS, NB, CA, CG, CP>(
leaves: impl IntoIterator<Item = BlockInfo>,
all_subsystems: AllSubsystems<CV, CB, CS, SD, AD, BS, BD, P, PoVD, RA, AS, NB, CA, CG, CP>,
prometheus_registry: Option<&prometheus::Registry>,
) -> SubsystemResult<(Self, OverseerHandler)>
where
CV: Subsystem<OverseerSubsystemContext<CandidateValidationMessage>> + Send,
CB: Subsystem<OverseerSubsystemContext<CandidateBackingMessage>> + Send,
CS: Subsystem<OverseerSubsystemContext<CandidateSelectionMessage>> + Send,
SD: Subsystem<OverseerSubsystemContext<StatementDistributionMessage>> + Send,
AD: Subsystem<OverseerSubsystemContext<AvailabilityDistributionMessage>> + Send,
BS: Subsystem<OverseerSubsystemContext<BitfieldSigningMessage>> + Send,
BD: Subsystem<OverseerSubsystemContext<BitfieldDistributionMessage>> + Send,
P: Subsystem<OverseerSubsystemContext<ProvisionerMessage>> + Send,
PoVD: Subsystem<OverseerSubsystemContext<PoVDistributionMessage>> + Send,
RA: Subsystem<OverseerSubsystemContext<RuntimeApiMessage>> + Send,
AS: Subsystem<OverseerSubsystemContext<AvailabilityStoreMessage>> + Send,
NB: Subsystem<OverseerSubsystemContext<NetworkBridgeMessage>> + Send,
CA: Subsystem<OverseerSubsystemContext<ChainApiMessage>> + Send,
CG: Subsystem<OverseerSubsystemContext<CollationGenerationMessage>> + Send,
CP: Subsystem<OverseerSubsystemContext<CollatorProtocolMessage>> + Send,
let (events_tx, events_rx) = mpsc::channel(CHANNEL_CAPACITY);
let handler = OverseerHandler {
events_tx: events_tx.clone(),
};
let mut running_subsystems_rx = StreamUnordered::new();
let mut running_subsystems = FuturesUnordered::new();
let candidate_validation_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.candidate_validation,
)?;
let candidate_backing_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
all_subsystems.candidate_backing,
)?;
let candidate_selection_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.candidate_selection,
)?;
let statement_distribution_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.statement_distribution,
)?;
let availability_distribution_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.availability_distribution,
)?;
let bitfield_signing_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.bitfield_signing,
)?;
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
let bitfield_distribution_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.bitfield_distribution,
)?;
let provisioner_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.provisioner,
)?;
let pov_distribution_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.pov_distribution,
)?;
let runtime_api_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.runtime_api,
)?;
let availability_store_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.availability_store,
)?;
let network_bridge_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.network_bridge,
let chain_api_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.chain_api,
)?;
let collation_generation_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.collation_generation,
)?;
let collator_protocol_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
all_subsystems.collator_protocol,
)?;
let leaves = leaves
.into_iter()
.map(|BlockInfo { hash, parent_hash: _, number }| (hash, number))
.collect();
let active_leaves = HashMap::new();
let metrics = <Metrics as metrics::Metrics>::register(prometheus_registry);
let activation_external_listeners = HashMap::new();
candidate_validation_subsystem,
candidate_selection_subsystem,
statement_distribution_subsystem,
availability_distribution_subsystem,
bitfield_signing_subsystem,
bitfield_distribution_subsystem,
provisioner_subsystem,
pov_distribution_subsystem,
runtime_api_subsystem,
availability_store_subsystem,
network_bridge_subsystem,
collation_generation_subsystem,
collator_protocol_subsystem,
s,
running_subsystems,
running_subsystems_rx,
events_rx,
activation_external_listeners,
leaves,
active_leaves,
};
Ok((this, handler))
}
// Stop the overseer.
async fn stop(mut self) {
if let Some(ref mut s) = self.candidate_validation_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.candidate_backing_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.candidate_selection_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.statement_distribution_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.availability_distribution_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.bitfield_signing_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.bitfield_distribution_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.provisioner_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.pov_distribution_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.runtime_api_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.availability_store_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.network_bridge_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.chain_api_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.collator_protocol_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
if let Some(ref mut s) = self.collation_generation_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
let mut stop_delay = Delay::new(Duration::from_secs(STOP_DELAY)).fuse();
loop {
select! {
_ = self.running_subsystems.next() => {
if self.running_subsystems.is_empty() {
break;
}
},
_ = stop_delay => break,
complete => break,
}
}
}
/// Run the `Overseer`.
pub async fn run(mut self) -> SubsystemResult<()> {
let leaves = std::mem::take(&mut self.leaves);
let mut update = ActiveLeavesUpdate::default();
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?;
loop {
while let Poll::Ready(Some(msg)) = poll!(&mut self.events_rx.next()) {
match msg {
Event::MsgToSubsystem(msg) => {
self.route_message(msg).await;
}
Event::Stop => {
self.stop().await;
return Ok(());
}
Event::BlockImported(block) => {
self.block_imported(block).await?;
}
Event::BlockFinalized(block) => {
self.block_finalized(block).await?;
}
Event::ExternalRequest(request) => {
self.handle_external_request(request);
}
}
}
while let Poll::Ready(Some((StreamYield::Item(msg), _))) = poll!(
&mut self.running_subsystems_rx.next()
) {
match msg {
ToOverseer::SubsystemMessage(msg) => self.route_message(msg).await,
ToOverseer::SpawnJob { name, s } => {
self.spawn_job(name, s);
ToOverseer::SpawnBlockingJob { name, s } => {
self.spawn_blocking_job(name, s);
}
}
}
// Some subsystem exited? It's time to panic.
if let Poll::Ready(Some(finished)) = poll!(self.running_subsystems.next()) {
log::error!(target: LOG_TARGET, "Subsystem finished unexpectedly {:?}", finished);
self.stop().await;
return Err(SubsystemError);
}
// Looks like nothing is left to be polled, let's take a break.
pending!();
}
}
async fn block_imported(&mut self, block: BlockInfo) -> SubsystemResult<()> {
let mut update = ActiveLeavesUpdate::default();
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);
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(())
}
async fn block_finalized(&mut self, block: BlockInfo) -> SubsystemResult<()> {
let mut update = ActiveLeavesUpdate::default();
self.active_leaves.retain(|h, n| {
update.deactivated.push(*h);
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?;
Ok(())
}
async fn broadcast_signal(&mut self, signal: OverseerSignal) -> SubsystemResult<()> {
if let Some(ref mut s) = self.candidate_validation_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.candidate_backing_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.candidate_selection_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.statement_distribution_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.availability_distribution_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.bitfield_distribution_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.bitfield_signing_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.provisioner_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.pov_distribution_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.runtime_api_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.availability_store_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.network_bridge_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.chain_api_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.collator_protocol_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
if let Some(ref mut s) = self.collation_generation_subsystem.instance {
s.tx.send(FromOverseer::Signal(signal.clone())).await?;
}
Ok(())
}
async fn route_message(&mut self, msg: AllMessages) {
self.metrics.on_message_relayed();
AllMessages::CandidateValidation(msg) => {
if let Some(ref mut s) = self.candidate_validation_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
AllMessages::CandidateBacking(msg) => {
if let Some(ref mut s) = self.candidate_backing_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
AllMessages::CandidateSelection(msg) => {
if let Some(ref mut s) = self.candidate_selection_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
AllMessages::StatementDistribution(msg) => {
if let Some(ref mut s) = self.statement_distribution_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
AllMessages::AvailabilityDistribution(msg) => {
if let Some(ref mut s) = self.availability_distribution_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
AllMessages::BitfieldDistribution(msg) => {
if let Some(ref mut s) = self.bitfield_distribution_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
AllMessages::BitfieldSigning(msg) => {
if let Some(ref mut s) = self.bitfield_signing_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication{ msg }).await;
}
}
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
AllMessages::Provisioner(msg) => {
if let Some(ref mut s) = self.provisioner_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
AllMessages::PoVDistribution(msg) => {
if let Some(ref mut s) = self.pov_distribution_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
AllMessages::RuntimeApi(msg) => {
if let Some(ref mut s) = self.runtime_api_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
AllMessages::AvailabilityStore(msg) => {
if let Some(ref mut s) = self.availability_store_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
AllMessages::NetworkBridge(msg) => {
if let Some(ref mut s) = self.network_bridge_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
AllMessages::ChainApi(msg) => {
if let Some(ref mut s) = self.chain_api_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
AllMessages::CollationGeneration(msg) => {
if let Some(ref mut s) = self.collation_generation_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
AllMessages::CollatorProtocol(msg) => {
if let Some(ref mut s) = self.collator_protocol_subsystem.instance {
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
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);
fn spawn_blocking_job(&mut self, name: &'static str, j: BoxFuture<'static, ()>) {
self.s.spawn_blocking(name, j);
}
fn spawn<S: SpawnNamed, M: Send + 'static>(
futures: &mut FuturesUnordered<BoxFuture<'static, ()>>,
streams: &mut StreamUnordered<mpsc::Receiver<ToOverseer>>,
s: impl Subsystem<OverseerSubsystemContext<M>>,
) -> SubsystemResult<OverseenSubsystem<M>> {
let (to_tx, to_rx) = mpsc::channel(CHANNEL_CAPACITY);
let (from_tx, from_rx) = mpsc::channel(CHANNEL_CAPACITY);
let ctx = OverseerSubsystemContext { rx: to_rx, tx: from_tx };
let SpawnedSubsystem { future, name } = s.start(ctx);
let (tx, rx) = oneshot::channel();
let fut = Box::pin(async move {
future.await;
let _ = tx.send(());
});
spawner.spawn(name, fut);
futures.push(Box::pin(rx.map(|_| ())));
let instance = Some(SubsystemInstance {
tx: to_tx,
});
Ok(OverseenSubsystem {
instance,
})
}
use std::sync::atomic;
use futures::{executor, pin_mut, select, channel::mpsc, FutureExt};
use polkadot_primitives::v1::{BlockData, CollatorPair, PoV};
use polkadot_subsystem::messages::RuntimeApiRequest;
use polkadot_node_primitives::{Collation, CollationGenerationConfig};
use polkadot_node_network_protocol::{PeerId, ReputationChange, NetworkBridgeEvent};
use super::*;
struct TestSubsystem1(mpsc::Sender<usize>);
impl<C> Subsystem<C> for TestSubsystem1
where C: SubsystemContext<Message=CandidateValidationMessage>
{
fn start(self, mut ctx: C) -> SpawnedSubsystem {
let mut sender = self.0;
SpawnedSubsystem {
name: "test-subsystem-1",
future: Box::pin(async move {
let mut i = 0;
loop {
match ctx.recv().await {
Ok(FromOverseer::Communication { .. }) => {
let _ = sender.send(i).await;
i += 1;
continue;
}
Ok(FromOverseer::Signal(OverseerSignal::Conclude)) => return,
Err(_) => return,
_ => (),
}
}
struct TestSubsystem2(mpsc::Sender<usize>);
impl<C> Subsystem<C> for TestSubsystem2
where C: SubsystemContext<Message=CandidateBackingMessage>
{
fn start(self, mut ctx: C) -> SpawnedSubsystem {
let sender = self.0.clone();
SpawnedSubsystem {
name: "test-subsystem-2",
future: Box::pin(async move {
let _sender = sender;
let mut c: usize = 0;
loop {
if c < 10 {
let (tx, _) = oneshot::channel();
ctx.send_message(
AllMessages::CandidateValidation(
CandidateValidationMessage::ValidateFromChainState(
Default::default(),
PoV {
block_data: BlockData(Vec::new()),
}.into(),
tx,
)
).await.unwrap();
c += 1;
match ctx.try_recv().await {
Ok(Some(FromOverseer::Signal(OverseerSignal::Conclude))) => {
break;
}
Ok(Some(_)) => {
continue;
}
Err(_) => return,
_ => (),
}
pending!();
impl<C> Subsystem<C> for TestSubsystem4
where C: SubsystemContext<Message=CandidateBackingMessage>
{
fn start(self, mut _ctx: C) -> SpawnedSubsystem {
SpawnedSubsystem {
name: "test-subsystem-4",
future: Box::pin(async move {
// Do nothing and exit.
}),
}
// Checks that a minimal configuration of two jobs can run and exchange messages.
#[test]
fn overseer_works() {
let spawner = sp_core::testing::TaskExecutor::new();
let (s1_tx, mut s1_rx) = mpsc::channel::<usize>(64);
let (s2_tx, mut s2_rx) = mpsc::channel::<usize>(64);
let all_subsystems = AllSubsystems::<()>::dummy()
.replace_candidate_validation(TestSubsystem1(s1_tx))
.replace_candidate_backing(TestSubsystem2(s2_tx));
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
spawner,
).unwrap();
let overseer_fut = overseer.run().fuse();
pin_mut!(overseer_fut);
let mut s1_results = Vec::new();
let mut s2_results = Vec::new();
loop {
select! {
a = overseer_fut => break,
s1_next = s1_rx.next() => {
match s1_next {
Some(msg) => {
s1_results.push(msg);
if s1_results.len() == 10 {
handler.stop().await.unwrap();
}
}
None => break,
}
},
s2_next = s2_rx.next() => {
match s2_next {
Some(msg) => s2_results.push(s2_next),
None => break,
}
},
complete => break,
}
}
assert_eq!(s1_results, (0..10).collect::<Vec<_>>());
});
}
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
// Checks activated/deactivated metrics are updated properly.
#[test]
fn overseer_metrics_work() {
let spawner = sp_core::testing::TaskExecutor::new();
executor::block_on(async move {
let first_block_hash = [1; 32].into();
let second_block_hash = [2; 32].into();
let third_block_hash = [3; 32].into();
let first_block = BlockInfo {
hash: first_block_hash,
parent_hash: [0; 32].into(),
number: 1,
};
let second_block = BlockInfo {
hash: second_block_hash,
parent_hash: first_block_hash,
number: 2,
};
let third_block = BlockInfo {
hash: third_block_hash,
parent_hash: second_block_hash,
number: 3,
};
let all_subsystems = AllSubsystems::<()>::dummy();
let registry = prometheus::Registry::new();
let (overseer, mut handler) = Overseer::new(
vec![first_block],
all_subsystems,
Some(®istry),
spawner,
).unwrap();
let overseer_fut = overseer.run().fuse();
pin_mut!(overseer_fut);
handler.block_imported(second_block).await.unwrap();
handler.block_imported(third_block).await.unwrap();
handler.send_msg(AllMessages::CandidateValidation(test_candidate_validation_msg())).await.unwrap();
handler.stop().await.unwrap();
select! {
res = overseer_fut => {
assert!(res.is_ok());
let metrics = extract_metrics(®istry);
assert_eq!(metrics["activated"], 3);
assert_eq!(metrics["deactivated"], 2);
assert_eq!(metrics["relayed"], 1);
},
complete => (),
}
});
}
fn extract_metrics(registry: &prometheus::Registry) -> HashMap<&'static str, u64> {
let gather = registry.gather();
assert_eq!(gather[0].get_name(), "parachain_activated_heads_total");
assert_eq!(gather[1].get_name(), "parachain_deactivated_heads_total");
assert_eq!(gather[2].get_name(), "parachain_messages_relayed_total");
let activated = gather[0].get_metric()[0].get_counter().get_value() as u64;
let deactivated = gather[1].get_metric()[0].get_counter().get_value() as u64;
let relayed = gather[2].get_metric()[0].get_counter().get_value() as u64;
let mut result = HashMap::new();
result.insert("activated", activated);
result.insert("deactivated", deactivated);
result.insert("relayed", relayed);
result
// Spawn a subsystem that immediately exits.
//
// Should immediately conclude the overseer itself with an error.
#[test]
asynchronous rob
committed
fn overseer_panics_on_subsystem_exit() {
let spawner = sp_core::testing::TaskExecutor::new();
executor::block_on(async move {
let (s1_tx, _) = mpsc::channel(64);
let all_subsystems = AllSubsystems::<()>::dummy()
.replace_candidate_validation(TestSubsystem1(s1_tx))
.replace_candidate_backing(TestSubsystem4);
spawner,
).unwrap();
let overseer_fut = overseer.run().fuse();
pin_mut!(overseer_fut);
select! {
res = overseer_fut => assert!(res.is_err()),
complete => (),
}
})
}
struct TestSubsystem5(mpsc::Sender<OverseerSignal>);
impl<C> Subsystem<C> for TestSubsystem5
where C: SubsystemContext<Message=CandidateValidationMessage>
{
fn start(self, mut ctx: C) -> SpawnedSubsystem {
let mut sender = self.0.clone();
SpawnedSubsystem {
name: "test-subsystem-5",
future: Box::pin(async move {
loop {
match ctx.try_recv().await {
Ok(Some(FromOverseer::Signal(OverseerSignal::Conclude))) => break,
Ok(Some(FromOverseer::Signal(s))) => {
sender.send(s).await.unwrap();
continue;
},
Ok(Some(_)) => continue,
Err(_) => return,
_ => (),
}
pending!();
}
}
struct TestSubsystem6(mpsc::Sender<OverseerSignal>);
impl<C> Subsystem<C> for TestSubsystem6
where C: SubsystemContext<Message=CandidateBackingMessage>
{
fn start(self, mut ctx: C) -> SpawnedSubsystem {
let mut sender = self.0.clone();
SpawnedSubsystem {
name: "test-subsystem-6",
future: Box::pin(async move {
loop {
match ctx.try_recv().await {
Ok(Some(FromOverseer::Signal(OverseerSignal::Conclude))) => break,
Ok(Some(FromOverseer::Signal(s))) => {
sender.send(s).await.unwrap();
continue;
},
Ok(Some(_)) => continue,
Err(_) => return,
_ => (),
}
pending!();
}
}
// Tests that starting with a defined set of leaves and receiving
// notifications on imported blocks triggers expected `StartWork` and `StopWork` heartbeats.
#[test]
fn overseer_start_stop_works() {
let spawner = sp_core::testing::TaskExecutor::new();
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964
executor::block_on(async move {
let first_block_hash = [1; 32].into();
let second_block_hash = [2; 32].into();
let third_block_hash = [3; 32].into();
let first_block = BlockInfo {
hash: first_block_hash,
parent_hash: [0; 32].into(),
number: 1,
};
let second_block = BlockInfo {
hash: second_block_hash,
parent_hash: first_block_hash,
number: 2,
};
let third_block = BlockInfo {
hash: third_block_hash,
parent_hash: second_block_hash,
number: 3,
};
let (tx_5, mut rx_5) = mpsc::channel(64);
let (tx_6, mut rx_6) = mpsc::channel(64);
let all_subsystems = AllSubsystems::<()>::dummy()
.replace_candidate_validation(TestSubsystem5(tx_5))
.replace_candidate_backing(TestSubsystem6(tx_6));
let (overseer, mut handler) = Overseer::new(
vec![first_block],
spawner,
).unwrap();
let overseer_fut = overseer.run().fuse();
pin_mut!(overseer_fut);
let mut ss5_results = Vec::new();
let mut ss6_results = Vec::new();
handler.block_imported(second_block).await.unwrap();
handler.block_imported(third_block).await.unwrap();
let expected_heartbeats = vec![
OverseerSignal::ActiveLeaves(ActiveLeavesUpdate::start_work(first_block_hash)),
OverseerSignal::ActiveLeaves(ActiveLeavesUpdate {
activated: [second_block_hash].as_ref().into(),
deactivated: [first_block_hash].as_ref().into(),
}),
OverseerSignal::ActiveLeaves(ActiveLeavesUpdate {
activated: [third_block_hash].as_ref().into(),
deactivated: [second_block_hash].as_ref().into(),
}),
];
loop {
select! {
res = overseer_fut => {
assert!(res.is_ok());
break;