Unverified Commit 8845df22 authored by Andronik Ordian's avatar Andronik Ordian Committed by GitHub
Browse files

Integrate all (dummy) subsystems with the Overseer (#1374)

* overseer: introduce a utility typemap

* it's ugly but it compiles

* move DummySubsystem to subsystem crate

* fix tests fallout

* use a struct for all subsystems

* more tests fallout

* add missing pov_distribution subsystem

* remove unused imports and bounds

* fix minimal-example
parent 6aa5bfba
Pipeline #100068 canceled with stages
in 3 minutes and 56 seconds
......@@ -28,11 +28,14 @@ use futures_timer::Delay;
use kv_log_macro as log;
use polkadot_primitives::parachain::{BlockData, PoVBlock};
use polkadot_overseer::Overseer;
use polkadot_overseer::{Overseer, AllSubsystems};
use polkadot_subsystem::{Subsystem, SubsystemContext, SpawnedSubsystem, FromOverseer};
use polkadot_subsystem::{
Subsystem, SubsystemContext, DummySubsystem,
SpawnedSubsystem, FromOverseer,
};
use polkadot_subsystem::messages::{
AllMessages, CandidateBackingMessage, CandidateValidationMessage
CandidateValidationMessage, CandidateBackingMessage, AllMessages,
};
struct Subsystem1;
......@@ -128,10 +131,22 @@ fn main() {
Delay::new(Duration::from_secs(1)).await;
});
let all_subsystems = AllSubsystems {
candidate_validation: Subsystem2,
candidate_backing: Subsystem1,
candidate_selection: DummySubsystem,
statement_distribution: DummySubsystem,
availability_distribution: DummySubsystem,
bitfield_distribution: DummySubsystem,
provisioner: DummySubsystem,
pov_distribution: DummySubsystem,
runtime_api: DummySubsystem,
availability_store: DummySubsystem,
network_bridge: DummySubsystem,
};
let (overseer, _handler) = Overseer::new(
vec![],
Subsystem2,
Subsystem1,
all_subsystems,
spawner,
).unwrap();
let overseer_fut = overseer.run().fuse();
......
......@@ -76,7 +76,11 @@ use polkadot_primitives::{Block, BlockNumber, Hash};
use client::{BlockImportNotification, BlockchainEvents, FinalityNotification};
use polkadot_subsystem::messages::{
CandidateValidationMessage, CandidateBackingMessage, AllMessages
CandidateValidationMessage, CandidateBackingMessage,
CandidateSelectionMessage, StatementDistributionMessage,
AvailabilityDistributionMessage, BitfieldDistributionMessage,
ProvisionerMessage, PoVDistributionMessage, RuntimeApiMessage,
AvailabilityStoreMessage, NetworkBridgeMessage, AllMessages,
};
pub use polkadot_subsystem::{
Subsystem, SubsystemContext, OverseerSignal, FromOverseer, SubsystemError, SubsystemResult,
......@@ -319,12 +323,40 @@ struct OverseenSubsystem<M> {
/// The `Overseer` itself.
pub struct Overseer<S: Spawn> {
/// A validation subsystem
validation_subsystem: OverseenSubsystem<CandidateValidationMessage>,
/// A candidate validation subsystem.
candidate_validation_subsystem: OverseenSubsystem<CandidateValidationMessage>,
/// A candidate backing subsystem
/// A candidate backing subsystem.
candidate_backing_subsystem: OverseenSubsystem<CandidateBackingMessage>,
/// A candidate selection subsystem.
candidate_selection_subsystem: OverseenSubsystem<CandidateSelectionMessage>,
/// A statement distribution subsystem.
statement_distribution_subsystem: OverseenSubsystem<StatementDistributionMessage>,
/// An availability distribution subsystem.
availability_distribution_subsystem: OverseenSubsystem<AvailabilityDistributionMessage>,
/// A bitfield distribution subsystem.
bitfield_distribution_subsystem: OverseenSubsystem<BitfieldDistributionMessage>,
/// A provisioner subsystem.
provisioner_subsystem: OverseenSubsystem<ProvisionerMessage>,
/// A PoV distribution subsystem.
pov_distribution_subsystem: OverseenSubsystem<PoVDistributionMessage>,
/// A runtime API subsystem.
runtime_api_subsystem: OverseenSubsystem<RuntimeApiMessage>,
/// An availability store subsystem.
availability_store_subsystem: OverseenSubsystem<AvailabilityStoreMessage>,
/// A network bridge subsystem.
network_bridge_subsystem: OverseenSubsystem<NetworkBridgeMessage>,
/// Spawner to spawn tasks to.
s: S,
......@@ -346,22 +378,48 @@ pub struct Overseer<S: Spawn> {
active_leaves: HashSet<(Hash, BlockNumber)>,
}
/// This struct is passed as an argument to create a new instance of an [`Overseer`].
///
/// As any entity that satisfies the interface may act as a [`Subsystem`] this allows
/// mocking in the test code:
///
/// Each [`Subsystem`] is supposed to implement some interface that is generic over
/// message type that is specific to this [`Subsystem`]. At the moment not all
/// subsystems are implemented and the rest can be mocked with the [`DummySubsystem`].
///
/// [`Subsystem`]: trait.Subsystem.html
/// [`DummySubsystem`]: struct.DummySubsystem.html
pub struct AllSubsystems<CV, CB, CS, SD, AD, BD, P, PoVD, RA, AS, NB> {
/// A candidate validation subsystem.
pub candidate_validation: CV,
/// A candidate backing subsystem.
pub candidate_backing: CB,
/// A candidate selection subsystem.
pub candidate_selection: CS,
/// A statement distribution subsystem.
pub statement_distribution: SD,
/// An availability distribution subsystem.
pub availability_distribution: AD,
/// A bitfield distribution subsystem.
pub bitfield_distribution: BD,
/// A provisioner subsystem.
pub provisioner: P,
/// A PoV distribution subsystem.
pub pov_distribution: PoVD,
/// A runtime API subsystem.
pub runtime_api: RA,
/// An availability store subsystem.
pub availability_store: AS,
/// A network bridge subsystem.
pub network_bridge: NB,
}
impl<S> Overseer<S>
where
S: Spawn,
{
/// Create a new intance of the `Overseer` with a fixed set of [`Subsystem`]s.
///
/// Each [`Subsystem`] is passed to this function as an explicit parameter
/// and is supposed to implement some interface that is generic over message type
/// that is specific to this [`Subsystem`]. At the moment there are only two
/// subsystems:
/// * Validation
/// * CandidateBacking
///
/// As any entity that satisfies the interface may act as a [`Subsystem`] this allows
/// mocking in the test code:
///
/// ```text
/// +------------------------------------+
/// | Overseer |
......@@ -388,16 +446,16 @@ where
/// # Example
///
/// The [`Subsystems`] may be any type as long as they implement an expected interface.
/// Here, we create two mock subsystems and start the `Overseer` with them. For the sake
/// of simplicity the termination of the example is done with a timeout.
/// Here, we create a mock validation subsystem and a few dummy ones and start the `Overseer` with them.
/// For the sake of simplicity the termination of the example is done with a timeout.
/// ```
/// # use std::time::Duration;
/// # use futures::{executor, pin_mut, select, FutureExt};
/// # use futures_timer::Delay;
/// # use polkadot_overseer::Overseer;
/// # use polkadot_overseer::{Overseer, AllSubsystems};
/// # use polkadot_subsystem::{
/// # Subsystem, SpawnedSubsystem, SubsystemContext,
/// # messages::{CandidateValidationMessage, CandidateBackingMessage},
/// # Subsystem, DummySubsystem, SpawnedSubsystem, SubsystemContext,
/// # messages::CandidateValidationMessage,
/// # };
///
/// struct ValidationSubsystem;
......@@ -417,28 +475,24 @@ where
/// }
/// }
///
/// struct CandidateBackingSubsystem;
/// impl<C> Subsystem<C> for CandidateBackingSubsystem
/// where C: SubsystemContext<Message=CandidateBackingMessage>
/// {
/// fn start(
/// self,
/// mut ctx: C,
/// ) -> SpawnedSubsystem {
/// SpawnedSubsystem(Box::pin(async move {
/// loop {
/// Delay::new(Duration::from_secs(1)).await;
/// }
/// }))
/// }
/// }
///
/// # fn main() { executor::block_on(async move {
/// let spawner = executor::ThreadPool::new().unwrap();
/// let all_subsystems = AllSubsystems {
/// candidate_validation: ValidationSubsystem,
/// candidate_backing: DummySubsystem,
/// candidate_selection: DummySubsystem,
/// statement_distribution: DummySubsystem,
/// availability_distribution: DummySubsystem,
/// bitfield_distribution: DummySubsystem,
/// provisioner: DummySubsystem,
/// pov_distribution: DummySubsystem,
/// runtime_api: DummySubsystem,
/// availability_store: DummySubsystem,
/// network_bridge: DummySubsystem,
/// };
/// let (overseer, _handler) = Overseer::new(
/// vec![],
/// ValidationSubsystem,
/// CandidateBackingSubsystem,
/// all_subsystems,
/// spawner,
/// ).unwrap();
///
......@@ -455,12 +509,24 @@ where
/// #
/// # }); }
/// ```
pub fn new(
pub fn new<CV, CB, CS, SD, AD, BD, P, PoVD, RA, AS, NB>(
leaves: impl IntoIterator<Item = BlockInfo>,
validation: impl Subsystem<OverseerSubsystemContext<CandidateValidationMessage>> + Send,
candidate_backing: impl Subsystem<OverseerSubsystemContext<CandidateBackingMessage>> + Send,
all_subsystems: AllSubsystems<CV, CB, CS, SD, AD, BD, P, PoVD, RA, AS, NB>,
mut s: S,
) -> SubsystemResult<(Self, OverseerHandler)> {
) -> 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,
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,
{
let (events_tx, events_rx) = mpsc::channel(CHANNEL_CAPACITY);
let handler = OverseerHandler {
......@@ -470,18 +536,81 @@ where
let mut running_subsystems_rx = StreamUnordered::new();
let mut running_subsystems = FuturesUnordered::new();
let validation_subsystem = spawn(
let candidate_validation_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
validation,
all_subsystems.candidate_validation,
)?;
let candidate_backing_subsystem = spawn(
&mut s,
&mut running_subsystems,
&mut running_subsystems_rx,
candidate_backing,
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_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 active_leaves = HashSet::new();
......@@ -492,8 +621,17 @@ where
.collect();
let this = Self {
validation_subsystem,
candidate_validation_subsystem,
candidate_backing_subsystem,
candidate_selection_subsystem,
statement_distribution_subsystem,
availability_distribution_subsystem,
bitfield_distribution_subsystem,
provisioner_subsystem,
pov_distribution_subsystem,
runtime_api_subsystem,
availability_store_subsystem,
network_bridge_subsystem,
s,
running_subsystems,
running_subsystems_rx,
......@@ -507,7 +645,7 @@ where
// Stop the overseer.
async fn stop(mut self) {
if let Some(ref mut s) = self.validation_subsystem.instance {
if let Some(ref mut s) = self.candidate_validation_subsystem.instance {
let _ = s.tx.send(FromOverseer::Signal(OverseerSignal::Conclude)).await;
}
......@@ -515,6 +653,42 @@ where
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_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_distribution_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;
}
let mut stop_delay = Delay::new(Duration::from_secs(STOP_DELAY)).fuse();
loop {
......@@ -616,11 +790,47 @@ where
}
async fn broadcast_signal(&mut self, signal: OverseerSignal) -> SubsystemResult<()> {
if let Some(ref mut s) = self.validation_subsystem.instance {
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.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)).await?;
}
......@@ -630,7 +840,7 @@ where
async fn route_message(&mut self, msg: AllMessages) {
match msg {
AllMessages::CandidateValidation(msg) => {
if let Some(ref mut s) = self.validation_subsystem.instance {
if let Some(ref mut s) = self.candidate_validation_subsystem.instance {
let _= s.tx.send(FromOverseer::Communication { msg }).await;
}
}
......@@ -639,11 +849,50 @@ where
let _ = s.tx.send(FromOverseer::Communication { msg }).await;
}
}
_ => {
// TODO: temporary catch-all until all subsystems are integrated with overseer.
// The overseer is not complete until this is an exhaustive match with all
// messages targeting an included subsystem.
// https://github.com/paritytech/polkadot/issues/1317
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::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;
}
}
}
}
......@@ -678,13 +927,16 @@ fn spawn<S: Spawn, M: Send + 'static>(
})
}
#[cfg(test)]
mod tests {
use futures::{executor, pin_mut, select, channel::mpsc, FutureExt};
use polkadot_primitives::parachain::{BlockData, PoVBlock};
use polkadot_subsystem::DummySubsystem;
use super::*;
struct TestSubsystem1(mpsc::Sender<usize>);
impl<C> Subsystem<C> for TestSubsystem1
......@@ -776,10 +1028,22 @@ mod tests {
let (s1_tx, mut s1_rx) = mpsc::channel(64);
let (s2_tx, mut s2_rx) = mpsc::channel(64);
let all_subsystems = AllSubsystems {
candidate_validation: TestSubsystem1(s1_tx),
candidate_backing: TestSubsystem2(s2_tx),
candidate_selection: DummySubsystem,
statement_distribution: DummySubsystem,
availability_distribution: DummySubsystem,
bitfield_distribution: DummySubsystem,
provisioner: DummySubsystem,
pov_distribution: DummySubsystem,
runtime_api: DummySubsystem,
availability_store: DummySubsystem,
network_bridge: DummySubsystem,
};
let (overseer, mut handler) = Overseer::new(
vec![],
TestSubsystem1(s1_tx),
TestSubsystem2(s2_tx),
all_subsystems,
spawner,
).unwrap();
let overseer_fut = overseer.run().fuse();
......@@ -826,10 +1090,22 @@ mod tests {
executor::block_on(async move {
let (s1_tx, _) = mpsc::channel(64);
let all_subsystems = AllSubsystems {
candidate_validation: TestSubsystem1(s1_tx),
candidate_backing: TestSubsystem4,
candidate_selection: DummySubsystem,
statement_distribution: DummySubsystem,
availability_distribution: DummySubsystem,
bitfield_distribution: DummySubsystem,
provisioner: DummySubsystem,
pov_distribution: DummySubsystem,
runtime_api: DummySubsystem,
availability_store: DummySubsystem,
network_bridge: DummySubsystem,
};
let (overseer, _handle) = Overseer::new(
vec![],
TestSubsystem1(s1_tx),
TestSubsystem4,
all_subsystems,
spawner,
).unwrap();
let overseer_fut = overseer.run().fuse();
......@@ -923,11 +1199,22 @@ mod tests {
let (tx_5, mut rx_5) = mpsc::channel(64);
let (tx_6, mut rx_6) = mpsc::channel(64);
let all_subsystems = AllSubsystems {
candidate_validation: TestSubsystem5(tx_5),
candidate_backing: TestSubsystem6(tx_6),
candidate_selection: DummySubsystem,
statement_distribution: DummySubsystem,
availability_distribution: DummySubsystem,
bitfield_distribution: DummySubsystem,
provisioner: DummySubsystem,
pov_distribution: DummySubsystem,
runtime_api: DummySubsystem,
availability_store: DummySubsystem,
network_bridge: DummySubsystem,
};
let (overseer, mut handler) = Overseer::new(
vec![first_block],
TestSubsystem5(tx_5),
TestSubsystem6(tx_6),
all_subsystems,
spawner,
).unwrap();
......@@ -1008,11 +1295,23 @@ mod tests {
let (tx_5, mut rx_5) = mpsc::channel(64);
let (tx_6, mut rx_6) = mpsc::channel(64);
let all_subsystems = AllSubsystems {
candidate_validation: TestSubsystem5(tx_5),
candidate_backing: TestSubsystem6(tx_6),
candidate_selection: DummySubsystem,