From c72e59c7c829fb5a1205d8da0f76f974b1f7c94c Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Tue, 14 Jul 2020 14:09:37 +0200 Subject: [PATCH 01/15] get conclude signal working properly; don't allocate a vector --- node/subsystem/src/util.rs | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/node/subsystem/src/util.rs b/node/subsystem/src/util.rs index 1b89bfb053a6..31906dff7574 100644 --- a/node/subsystem/src/util.rs +++ b/node/subsystem/src/util.rs @@ -540,17 +540,20 @@ where // // Forwarding the stream to a drain means we wait until all of the items in the stream // have completed. Contrast with `into_future`, which turns it into a future of `(head, rest_stream)`. + use futures::sink::drain; use futures::stream::StreamExt; use futures::stream::FuturesUnordered; - let unordered = jobs.running + if let Err(e) = jobs.running .drain() .map(|(_, handle)| handle.stop()) - .collect::>(); - // now wait for all the futures to complete; collect a vector of their results - // this is strictly less efficient than draining them into oblivion, but this compiles, and that doesn't - // https://github.com/paritytech/polkadot/pull/1376#pullrequestreview-446488645 - let _ = async move { unordered.collect::>() }.await; + .collect::>() + .map(Ok) + .forward(drain()) + .await + { + log::error!("failed to stop all jobs on conclude signal: {:?}", e); + } return true; } From 00bbe471af00bdc3bb69a2472ae7841d7a6dc0bc Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Tue, 14 Jul 2020 16:25:32 +0200 Subject: [PATCH 02/15] wip: add test suite / example / explanation for using utility subsystem Unfortunately, the test fails right now for reasons which seem very odd. Just have to keep poking at it. --- Cargo.lock | 2 + node/subsystem/Cargo.toml | 5 + node/subsystem/src/messages.rs | 6 ++ node/subsystem/src/util.rs | 184 ++++++++++++++++++++++++++++++++- 4 files changed, 195 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7f60ff2254f4..6d643def1b69 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4543,6 +4543,7 @@ dependencies = [ name = "polkadot-node-subsystem" version = "0.1.0" dependencies = [ + "assert_matches", "async-trait", "derive_more 0.99.9", "futures 0.3.5", @@ -4553,6 +4554,7 @@ dependencies = [ "polkadot-node-primitives", "polkadot-primitives", "polkadot-statement-table", + "polkadot-subsystem-test-helpers", "sc-keystore", "sc-network", "sp-core", diff --git a/node/subsystem/Cargo.toml b/node/subsystem/Cargo.toml index 188e7cbfa764..e5108feb2911 100644 --- a/node/subsystem/Cargo.toml +++ b/node/subsystem/Cargo.toml @@ -20,3 +20,8 @@ polkadot-statement-table = { path = "../../statement-table" } sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" } sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" } streamunordered = "0.5.1" + +[dev-dependencies] +assert_matches = "1.3.0" +futures = { version = "0.3.5", features = ["thread-pool"] } +polkadot-subsystem-test-helpers = { path = "../test-helpers/subsystem" } diff --git a/node/subsystem/src/messages.rs b/node/subsystem/src/messages.rs index d3c630cb56f0..2040b413488d 100644 --- a/node/subsystem/src/messages.rs +++ b/node/subsystem/src/messages.rs @@ -408,4 +408,10 @@ pub enum AllMessages { AvailabilityStore(AvailabilityStoreMessage), /// Message for the network bridge subsystem. NetworkBridge(NetworkBridgeMessage), + /// Test message + /// + /// This variant is only valid while testing, but makes the process of testing the + /// subsystem job manager much simpler. + #[cfg(test)] + Test(String), } diff --git a/node/subsystem/src/util.rs b/node/subsystem/src/util.rs index 31906dff7574..9b411edd05a0 100644 --- a/node/subsystem/src/util.rs +++ b/node/subsystem/src/util.rs @@ -323,8 +323,8 @@ pub trait JobTrait: Unpin { fn run( parent: Hash, run_args: Self::RunArgs, - rx_to: mpsc::Receiver, - tx_from: mpsc::Sender, + receiver: mpsc::Receiver, + sender: mpsc::Sender, ) -> Pin> + Send>>; /// Handle a message which has no relay parent, and therefore can't be dispatched to a particular job @@ -614,3 +614,183 @@ where })) } } + +#[cfg(test)] +mod tests { + use super::*; + + use assert_matches::assert_matches; + use crate::messages::{AllMessages, CandidateSelectionMessage}; + use futures::{ + executor::{self, ThreadPool}, + stream::{self, StreamExt}, + }; + use polkadot_subsystem_test_helpers::make_subsystem_context; + use std::collections::HashMap; + + // basic usage: in a nutshell, when you want to define a subsystem, just focus on what its jobs do; + // you can leave the subsystem itself to the job manager. + + // for purposes of demonstration, we're going to whip up a fake subsystem. + // this will 'select' candidates which are pre-loaded in the job + + // job structs are constructed within JobTrait::run + // most will want to retain the sender and receiver, as well as whatever other data they like + struct FakeCandidateSelectionJob { + receiver: mpsc::Receiver, + } + + // ToJob implementations require the following properties: + // + // - have a Stop variant (to impl ToJobTrait) + // - impl ToJobTrait + // - impl TryFrom + // + // Mostly, they are just a type-safe subset of AllMessages that this job is prepared to receive + enum ToJob { + CandidateSelection(CandidateSelectionMessage), + Stop, + } + + impl ToJobTrait for ToJob { + const STOP: Self = ToJob::Stop; + + fn relay_parent(&self) -> Option { + match self { + Self::CandidateSelection(csm) => csm.relay_parent(), + Self::Stop => None, + } + } + } + + impl TryFrom for ToJob { + type Error = (); + + fn try_from(msg: AllMessages) -> Result { + match msg { + AllMessages::CandidateSelection(csm) => Ok(ToJob::CandidateSelection(csm)), + _ => Err(()) + } + } + } + + // FromJob must be infallibly convertable into AllMessages. + // + // It exists to be a type-safe subset of AllMessages that this job is specified to send. + // + // Note: the Clone impl here is not generally required; it's just ueful for this test context because + // we include it in the RunArgs + #[derive(Clone)] + enum FromJob { + Test(String), + } + + impl From for AllMessages { + fn from(from_job: FromJob) -> AllMessages { + match from_job { + FromJob::Test(s) => AllMessages::Test(s), + } + } + } + + // Error will mostly be a wrapper to make the try operator more convenient; + // deriving From implementations for most variants is recommended. + // It must implement Debug for logging. + #[derive(Debug, derive_more::From)] + enum Error { + #[from] + Sending(mpsc::SendError) + } + + impl JobTrait for FakeCandidateSelectionJob { + type ToJob = ToJob; + type FromJob = FromJob; + type Error = Error; + // RunArgs can be anything that a particular job needs supplied from its external context + // in order to create the Job. In this case, they're a hashmap of parents to the mock outputs + // expected from that job. + // + // Note that it's not recommended to use something as heavy as a hashmap in production: the + // RunArgs get cloned so that each job gets its own owned copy. If you need that, wrap it in + // an Arc. Within a testing context, that efficiency is less important. + type RunArgs = HashMap>; + + const NAME: &'static str = "FakeCandidateSelectionJob"; + + /// Run a job for the parent block indicated + // + // this function is in charge of creating and executing the job's main loop + fn run( + parent: Hash, + mut run_args: Self::RunArgs, + receiver: mpsc::Receiver, + mut sender: mpsc::Sender, + ) -> Pin> + Send>> { + async move { + let job = FakeCandidateSelectionJob { + receiver, + }; + + // most jobs will have a request-response cycle at the heart of their run loop. + // however, in this case, we never receive valid messages, so we may as well + // just send all of our (mock) output messages now + let mock_output = run_args.remove(&parent).unwrap_or_default(); + let mut stream = stream::iter(mock_output.into_iter().map(Ok)); + sender.send_all(&mut stream).await?; + + // it isn't necessary to break run_loop into its own function, + // but it's convenient to separate the concerns in this way + job.run_loop().await + }.boxed() + } + } + + impl FakeCandidateSelectionJob { + async fn run_loop(mut self) -> Result<(), Error> { + while let Some(msg) = self.receiver.next().await { + match msg { + ToJob::CandidateSelection(_csm) => { + unimplemented!("we'd report the collator to the peer set manager here, but that's not implemented yet"); + } + ToJob::Stop => break, + } + } + + Ok(()) + } + } + + // with the job defined, it's straightforward to get a subsystem implementation. + type FakeCandidateSelectionSubsystem = JobManager; + + // this type lets us pretend to be the overseer + type OverseerHandle = polkadot_subsystem_test_helpers::TestSubsystemContextHandle; + + fn test_harness>(run_args: HashMap>, test: impl FnOnce(OverseerHandle) -> T) { + let pool = ThreadPool::new().unwrap(); + let (context, overseer_handle) = make_subsystem_context(pool.clone()); + + let subsystem = FakeCandidateSelectionSubsystem::run(context, run_args, pool); + let test_future = test(overseer_handle); + + futures::pin_mut!(test_future); + futures::pin_mut!(subsystem); + + executor::block_on(future::select(test_future, subsystem)); + } + + #[test] + fn starting_job_works() { + let relay_parent: Hash = [0; 32].into(); + let mut run_args = HashMap::new(); + let test_message = format!("greetings from {}", relay_parent); + run_args.insert(relay_parent.clone(), vec![FromJob::Test(test_message)]); + + test_harness(run_args, |overseer_handle| async move { + assert_matches!( + overseer_handle.recv().await, + AllMessages::Test(msg) if msg == test_message + ); + }); + } +} From 837a62ab64b120be2ad76a8e7e93a44e63daa4c7 Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Thu, 16 Jul 2020 12:47:30 +0200 Subject: [PATCH 03/15] explicitly import everything --- node/subsystem/src/util.rs | 23 +++++++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/node/subsystem/src/util.rs b/node/subsystem/src/util.rs index 9b411edd05a0..6214cde8ee15 100644 --- a/node/subsystem/src/util.rs +++ b/node/subsystem/src/util.rs @@ -617,16 +617,31 @@ where #[cfg(test)] mod tests { - use super::*; - use assert_matches::assert_matches; - use crate::messages::{AllMessages, CandidateSelectionMessage}; + use crate::{ + messages::{AllMessages, CandidateSelectionMessage}, + util::{ + JobManager, + JobTrait, + ToJobTrait, + }, + }; use futures::{ + channel::mpsc, executor::{self, ThreadPool}, + future, + Future, + FutureExt, stream::{self, StreamExt}, + SinkExt, }; + use polkadot_primitives::v1::Hash; use polkadot_subsystem_test_helpers::make_subsystem_context; - use std::collections::HashMap; + use std::{ + collections::HashMap, + convert::TryFrom, + pin::Pin, + }; // basic usage: in a nutshell, when you want to define a subsystem, just focus on what its jobs do; // you can leave the subsystem itself to the job manager. From 5601550ca47d16b2bbbbc175d03a42aac5aa181e Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Thu, 16 Jul 2020 15:18:53 +0200 Subject: [PATCH 04/15] fix subsystem-util test The root problem here was two-fold: - there was a circular dependency from subsystem -> test-helpers/subsystem -> subsystem - cfg(test) doesn't propagate between crates The solution: move the subsystem test helpers into a sub-module within subsystem. Publicly export them from the previous location so no other code breaks. Doing this has an additional benefit: it ensures that no production code can ever accidentally use the subsystem helpers, as they are compile- gated on cfg(test). --- Cargo.lock | 5 +- node/subsystem/Cargo.toml | 3 +- node/subsystem/src/lib.rs | 2 + node/subsystem/src/test_helpers.rs | 229 +++++++++++++++++++++++++ node/subsystem/src/util.rs | 15 +- node/test-helpers/subsystem/Cargo.toml | 3 - node/test-helpers/subsystem/src/lib.rs | 214 +---------------------- 7 files changed, 248 insertions(+), 223 deletions(-) create mode 100644 node/subsystem/src/test_helpers.rs diff --git a/Cargo.lock b/Cargo.lock index 6d643def1b69..3f66ec24054e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4550,11 +4550,11 @@ dependencies = [ "futures-timer 3.0.2", "log 0.4.8", "parity-scale-codec", + "parking_lot 0.10.2", "pin-project", "polkadot-node-primitives", "polkadot-primitives", "polkadot-statement-table", - "polkadot-subsystem-test-helpers", "sc-keystore", "sc-network", "sp-core", @@ -4982,9 +4982,6 @@ dependencies = [ name = "polkadot-subsystem-test-helpers" version = "0.1.0" dependencies = [ - "async-trait", - "futures 0.3.5", - "parking_lot 0.10.2", "polkadot-node-subsystem", ] diff --git a/node/subsystem/Cargo.toml b/node/subsystem/Cargo.toml index e5108feb2911..0e851770b84f 100644 --- a/node/subsystem/Cargo.toml +++ b/node/subsystem/Cargo.toml @@ -23,5 +23,6 @@ streamunordered = "0.5.1" [dev-dependencies] assert_matches = "1.3.0" +async-trait = "0.1" futures = { version = "0.3.5", features = ["thread-pool"] } -polkadot-subsystem-test-helpers = { path = "../test-helpers/subsystem" } +parking_lot = "0.10.0" diff --git a/node/subsystem/src/lib.rs b/node/subsystem/src/lib.rs index b6c3a79ef3d9..2005b79f78ef 100644 --- a/node/subsystem/src/lib.rs +++ b/node/subsystem/src/lib.rs @@ -35,6 +35,8 @@ use crate::messages::AllMessages; pub mod messages; pub mod util; +#[cfg(test)] +pub mod test_helpers; /// Signals sent by an overseer to a subsystem. #[derive(PartialEq, Clone, Debug)] diff --git a/node/subsystem/src/test_helpers.rs b/node/subsystem/src/test_helpers.rs new file mode 100644 index 000000000000..5a5b10d9444b --- /dev/null +++ b/node/subsystem/src/test_helpers.rs @@ -0,0 +1,229 @@ +// Copyright 2017-2020 Parity Technologies (UK) Ltd. +// This file is part of Polkadot. + +// Polkadot is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// Polkadot is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with Polkadot. If not, see . + +//! Utilities for testing subsystems. + +use crate::{SubsystemContext, FromOverseer, SubsystemResult, SubsystemError}; +use crate::messages::AllMessages; + +use futures::prelude::*; +use futures::channel::mpsc; +use futures::task::{Spawn, SpawnExt}; +use futures::poll; +use parking_lot::Mutex; + +use std::convert::Infallible; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll, Waker}; + +enum SinkState { + Empty { + read_waker: Option, + }, + Item { + item: T, + ready_waker: Option, + flush_waker: Option, + }, +} + +/// The sink half of a single-item sink that does not resolve until the item has been read. +pub struct SingleItemSink(Arc>>); + +/// The stream half of a single-item sink. +pub struct SingleItemStream(Arc>>); + +impl Sink for SingleItemSink { + type Error = Infallible; + + fn poll_ready( + self: Pin<&mut Self>, + cx: &mut Context, + ) -> Poll> { + let mut state = self.0.lock(); + match *state { + SinkState::Empty { .. } => Poll::Ready(Ok(())), + SinkState::Item { ref mut ready_waker, .. } => { + *ready_waker = Some(cx.waker().clone()); + Poll::Pending + } + } + } + + fn start_send( + self: Pin<&mut Self>, + item: T, + ) -> Result<(), Infallible> { + let mut state = self.0.lock(); + + match *state { + SinkState::Empty { ref mut read_waker } => { + if let Some(waker) = read_waker.take() { + waker.wake(); + } + } + _ => panic!("start_send called outside of empty sink state ensured by poll_ready"), + } + + *state = SinkState::Item { + item, + ready_waker: None, + flush_waker: None, + }; + + Ok(()) + } + + fn poll_flush( + self: Pin<&mut Self>, + cx: &mut Context, + ) -> Poll> { + let mut state = self.0.lock(); + match *state { + SinkState::Empty { .. } => Poll::Ready(Ok(())), + SinkState::Item { ref mut flush_waker, .. } => { + *flush_waker = Some(cx.waker().clone()); + Poll::Pending + } + } + } + + fn poll_close( + self: Pin<&mut Self>, + cx: &mut Context, + ) -> Poll> { + self.poll_flush(cx) + } +} + +impl Stream for SingleItemStream { + type Item = T; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll> { + let mut state = self.0.lock(); + + let read_waker = Some(cx.waker().clone()); + + match std::mem::replace(&mut *state, SinkState::Empty { read_waker }) { + SinkState::Empty { .. } => Poll::Pending, + SinkState::Item { item, ready_waker, flush_waker } => { + if let Some(waker) = ready_waker { + waker.wake(); + } + + if let Some(waker) = flush_waker { + waker.wake(); + } + + Poll::Ready(Some(item)) + } + } + } +} + +/// Create a single-item Sink/Stream pair. +/// +/// The sink's send methods resolve at the point which the stream reads the item, +/// not when the item is buffered. +pub fn single_item_sink() -> (SingleItemSink, SingleItemStream) { + let inner = Arc::new(Mutex::new(SinkState::Empty { read_waker: None })); + ( + SingleItemSink(inner.clone()), + SingleItemStream(inner), + ) +} + +/// A test subsystem context. +pub struct TestSubsystemContext { + tx: mpsc::UnboundedSender, + rx: SingleItemStream>, + spawn: S, +} + +#[async_trait::async_trait] +impl SubsystemContext for TestSubsystemContext { + type Message = M; + + async fn try_recv(&mut self) -> Result>, ()> { + match poll!(self.rx.next()) { + Poll::Ready(Some(msg)) => Ok(Some(msg)), + Poll::Ready(None) => Err(()), + Poll::Pending => Ok(None), + } + } + + async fn recv(&mut self) -> SubsystemResult> { + self.rx.next().await.ok_or(SubsystemError) + } + + async fn spawn(&mut self, s: Pin + Send>>) -> SubsystemResult<()> { + self.spawn.spawn(s).map_err(Into::into) + } + + async fn send_message(&mut self, msg: AllMessages) -> SubsystemResult<()> { + self.tx.send(msg).await.expect("test overseer no longer live"); + Ok(()) + } + + async fn send_messages(&mut self, msgs: T) -> SubsystemResult<()> + where T: IntoIterator + Send, T::IntoIter: Send + { + let mut iter = stream::iter(msgs.into_iter().map(Ok)); + self.tx.send_all(&mut iter).await.expect("test overseer no longer live"); + + Ok(()) + } +} + +/// A handle for interacting with the subsystem context. +pub struct TestSubsystemContextHandle { + tx: SingleItemSink>, + rx: mpsc::UnboundedReceiver, +} + +impl TestSubsystemContextHandle { + /// Send a message or signal to the subsystem. This resolves at the point in time where the + /// subsystem has _read_ the message. + pub async fn send(&mut self, from_overseer: FromOverseer) { + self.tx.send(from_overseer).await.expect("Test subsystem no longer live"); + } + + /// Receive the next message from the subsystem. + pub async fn recv(&mut self) -> AllMessages { + self.rx.next().await.expect("Test subsystem no longer live") + } +} + +/// Make a test subsystem context. +pub fn make_subsystem_context(spawn: S) + -> (TestSubsystemContext, TestSubsystemContextHandle) +{ + let (overseer_tx, overseer_rx) = single_item_sink(); + let (all_messages_tx, all_messages_rx) = mpsc::unbounded(); + + ( + TestSubsystemContext { + tx: all_messages_tx, + rx: overseer_rx, + spawn, + }, + TestSubsystemContextHandle { + tx: overseer_tx, + rx: all_messages_rx + }, + ) +} diff --git a/node/subsystem/src/util.rs b/node/subsystem/src/util.rs index 6214cde8ee15..5ffa5f71fd7d 100644 --- a/node/subsystem/src/util.rs +++ b/node/subsystem/src/util.rs @@ -620,6 +620,7 @@ mod tests { use assert_matches::assert_matches; use crate::{ messages::{AllMessages, CandidateSelectionMessage}, + test_helpers::{self, make_subsystem_context}, util::{ JobManager, JobTrait, @@ -636,7 +637,6 @@ mod tests { SinkExt, }; use polkadot_primitives::v1::Hash; - use polkadot_subsystem_test_helpers::make_subsystem_context; use std::{ collections::HashMap, convert::TryFrom, @@ -660,6 +660,7 @@ mod tests { // - have a Stop variant (to impl ToJobTrait) // - impl ToJobTrait // - impl TryFrom + // - impl From (from SubsystemContext::Message) // // Mostly, they are just a type-safe subset of AllMessages that this job is prepared to receive enum ToJob { @@ -689,6 +690,12 @@ mod tests { } } + impl From for ToJob { + fn from(csm: CandidateSelectionMessage) -> ToJob { + ToJob::CandidateSelection(csm) + } + } + // FromJob must be infallibly convertable into AllMessages. // // It exists to be a type-safe subset of AllMessages that this job is specified to send. @@ -779,7 +786,7 @@ mod tests { type FakeCandidateSelectionSubsystem = JobManager; // this type lets us pretend to be the overseer - type OverseerHandle = polkadot_subsystem_test_helpers::TestSubsystemContextHandle; + type OverseerHandle = test_helpers::TestSubsystemContextHandle; fn test_harness>(run_args: HashMap>, test: impl FnOnce(OverseerHandle) -> T) { let pool = ThreadPool::new().unwrap(); @@ -799,9 +806,9 @@ mod tests { let relay_parent: Hash = [0; 32].into(); let mut run_args = HashMap::new(); let test_message = format!("greetings from {}", relay_parent); - run_args.insert(relay_parent.clone(), vec![FromJob::Test(test_message)]); + run_args.insert(relay_parent.clone(), vec![FromJob::Test(test_message.clone())]); - test_harness(run_args, |overseer_handle| async move { + test_harness(run_args, |mut overseer_handle| async move { assert_matches!( overseer_handle.recv().await, AllMessages::Test(msg) if msg == test_message diff --git a/node/test-helpers/subsystem/Cargo.toml b/node/test-helpers/subsystem/Cargo.toml index 0fc26a24ea14..fa6f4038e6da 100644 --- a/node/test-helpers/subsystem/Cargo.toml +++ b/node/test-helpers/subsystem/Cargo.toml @@ -6,7 +6,4 @@ edition = "2018" description = "Helpers for testing subsystems" [dependencies] -futures = "0.3.5" -async-trait = "0.1" polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" } -parking_lot = "0.10.0" diff --git a/node/test-helpers/subsystem/src/lib.rs b/node/test-helpers/subsystem/src/lib.rs index c99a33c78d9b..68197f4d1fed 100644 --- a/node/test-helpers/subsystem/src/lib.rs +++ b/node/test-helpers/subsystem/src/lib.rs @@ -15,215 +15,7 @@ // along with Polkadot. If not, see . //! Utilities for testing subsystems. +//! +//! **DEPRECATED**: use `polkadot-node-subsystem::test_helpers` instead -use polkadot_subsystem::{SubsystemContext, FromOverseer, SubsystemResult, SubsystemError}; -use polkadot_subsystem::messages::AllMessages; - -use futures::prelude::*; -use futures::channel::mpsc; -use futures::task::{Spawn, SpawnExt}; -use futures::poll; -use parking_lot::Mutex; - -use std::convert::Infallible; -use std::pin::Pin; -use std::sync::Arc; -use std::task::{Context, Poll, Waker}; - -enum SinkState { - Empty { - read_waker: Option, - }, - Item { - item: T, - ready_waker: Option, - flush_waker: Option, - }, -} - -/// The sink half of a single-item sink that does not resolve until the item has been read. -pub struct SingleItemSink(Arc>>); - -/// The stream half of a single-item sink. -pub struct SingleItemStream(Arc>>); - -impl Sink for SingleItemSink { - type Error = Infallible; - - fn poll_ready( - self: Pin<&mut Self>, - cx: &mut Context, - ) -> Poll> { - let mut state = self.0.lock(); - match *state { - SinkState::Empty { .. } => Poll::Ready(Ok(())), - SinkState::Item { ref mut ready_waker, .. } => { - *ready_waker = Some(cx.waker().clone()); - Poll::Pending - } - } - } - - fn start_send( - self: Pin<&mut Self>, - item: T, - ) -> Result<(), Infallible> { - let mut state = self.0.lock(); - - match *state { - SinkState::Empty { ref mut read_waker } => { - if let Some(waker) = read_waker.take() { - waker.wake(); - } - } - _ => panic!("start_send called outside of empty sink state ensured by poll_ready"), - } - - *state = SinkState::Item { - item, - ready_waker: None, - flush_waker: None, - }; - - Ok(()) - } - - fn poll_flush( - self: Pin<&mut Self>, - cx: &mut Context, - ) -> Poll> { - let mut state = self.0.lock(); - match *state { - SinkState::Empty { .. } => Poll::Ready(Ok(())), - SinkState::Item { ref mut flush_waker, .. } => { - *flush_waker = Some(cx.waker().clone()); - Poll::Pending - } - } - } - - fn poll_close( - self: Pin<&mut Self>, - cx: &mut Context, - ) -> Poll> { - self.poll_flush(cx) - } -} - -impl Stream for SingleItemStream { - type Item = T; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll> { - let mut state = self.0.lock(); - - let read_waker = Some(cx.waker().clone()); - - match std::mem::replace(&mut *state, SinkState::Empty { read_waker }) { - SinkState::Empty { .. } => Poll::Pending, - SinkState::Item { item, ready_waker, flush_waker } => { - if let Some(waker) = ready_waker { - waker.wake(); - } - - if let Some(waker) = flush_waker { - waker.wake(); - } - - Poll::Ready(Some(item)) - } - } - } -} - -/// Create a single-item Sink/Stream pair. -/// -/// The sink's send methods resolve at the point which the stream reads the item, -/// not when the item is buffered. -pub fn single_item_sink() -> (SingleItemSink, SingleItemStream) { - let inner = Arc::new(Mutex::new(SinkState::Empty { read_waker: None })); - ( - SingleItemSink(inner.clone()), - SingleItemStream(inner), - ) -} - -/// A test subsystem context. -pub struct TestSubsystemContext { - tx: mpsc::UnboundedSender, - rx: SingleItemStream>, - spawn: S, -} - -#[async_trait::async_trait] -impl SubsystemContext for TestSubsystemContext { - type Message = M; - - async fn try_recv(&mut self) -> Result>, ()> { - match poll!(self.rx.next()) { - Poll::Ready(Some(msg)) => Ok(Some(msg)), - Poll::Ready(None) => Err(()), - Poll::Pending => Ok(None), - } - } - - async fn recv(&mut self) -> SubsystemResult> { - self.rx.next().await.ok_or(SubsystemError) - } - - async fn spawn(&mut self, s: Pin + Send>>) -> SubsystemResult<()> { - self.spawn.spawn(s).map_err(Into::into) - } - - async fn send_message(&mut self, msg: AllMessages) -> SubsystemResult<()> { - self.tx.send(msg).await.expect("test overseer no longer live"); - Ok(()) - } - - async fn send_messages(&mut self, msgs: T) -> SubsystemResult<()> - where T: IntoIterator + Send, T::IntoIter: Send - { - let mut iter = stream::iter(msgs.into_iter().map(Ok)); - self.tx.send_all(&mut iter).await.expect("test overseer no longer live"); - - Ok(()) - } -} - -/// A handle for interacting with the subsystem context. -pub struct TestSubsystemContextHandle { - tx: SingleItemSink>, - rx: mpsc::UnboundedReceiver, -} - -impl TestSubsystemContextHandle { - /// Send a message or signal to the subsystem. This resolves at the point in time where the - /// subsystem has _read_ the message. - pub async fn send(&mut self, from_overseer: FromOverseer) { - self.tx.send(from_overseer).await.expect("Test subsystem no longer live"); - } - - /// Receive the next message from the subsystem. - pub async fn recv(&mut self) -> AllMessages { - self.rx.next().await.expect("Test subsystem no longer live") - } -} - -/// Make a test subsystem context. -pub fn make_subsystem_context(spawn: S) - -> (TestSubsystemContext, TestSubsystemContextHandle) -{ - let (overseer_tx, overseer_rx) = single_item_sink(); - let (all_messages_tx, all_messages_rx) = mpsc::unbounded(); - - ( - TestSubsystemContext { - tx: all_messages_tx, - rx: overseer_rx, - spawn, - }, - TestSubsystemContextHandle { - tx: overseer_tx, - rx: all_messages_rx - }, - ) -} +pub use polkadot_subsystem::test_helpers::*; From 3ff4f66754ddcdc4655f334ecb6c46bec9f6cb9f Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Thu, 16 Jul 2020 16:21:51 +0200 Subject: [PATCH 05/15] fully commit to moving test helpers into a subsystem module --- Cargo.lock | 11 --------- Cargo.toml | 1 - node/core/backing/Cargo.toml | 1 - node/core/backing/src/lib.rs | 6 ++--- node/network/bridge/Cargo.toml | 1 - node/network/bridge/src/lib.rs | 8 +++---- node/network/pov-distribution/Cargo.toml | 1 - node/network/pov-distribution/src/lib.rs | 24 +++++++++---------- .../network/statement-distribution/Cargo.toml | 2 +- .../network/statement-distribution/src/lib.rs | 4 ++-- node/subsystem/Cargo.toml | 4 ++++ node/subsystem/src/lib.rs | 2 +- node/test-helpers/subsystem/Cargo.toml | 9 ------- node/test-helpers/subsystem/src/lib.rs | 21 ---------------- 14 files changed, 27 insertions(+), 68 deletions(-) delete mode 100644 node/test-helpers/subsystem/Cargo.toml delete mode 100644 node/test-helpers/subsystem/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 3f66ec24054e..81477b73509a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4453,7 +4453,6 @@ dependencies = [ "polkadot-node-primitives", "polkadot-node-subsystem", "polkadot-primitives", - "polkadot-subsystem-test-helpers", "sc-network", "sp-runtime", "streamunordered", @@ -4493,7 +4492,6 @@ dependencies = [ "polkadot-node-subsystem", "polkadot-primitives", "polkadot-statement-table", - "polkadot-subsystem-test-helpers", "sc-client-api", "sc-keystore", "sp-api", @@ -4609,7 +4607,6 @@ dependencies = [ "polkadot-node-primitives", "polkadot-node-subsystem", "polkadot-primitives", - "polkadot-subsystem-test-helpers", "sc-network", "sp-runtime", "streamunordered", @@ -4962,7 +4959,6 @@ dependencies = [ "polkadot-node-primitives", "polkadot-node-subsystem", "polkadot-primitives", - "polkadot-subsystem-test-helpers", "sp-keyring", "sp-runtime", "sp-staking", @@ -4978,13 +4974,6 @@ dependencies = [ "sp-core", ] -[[package]] -name = "polkadot-subsystem-test-helpers" -version = "0.1.0" -dependencies = [ - "polkadot-node-subsystem", -] - [[package]] name = "polkadot-test-runtime" version = "0.8.14" diff --git a/Cargo.toml b/Cargo.toml index b6d1aa53eaa2..91d27395fadb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -52,7 +52,6 @@ members = [ "node/service", "node/core/backing", "node/subsystem", - "node/test-helpers/subsystem", "node/test-service", "parachain/test-parachains", diff --git a/node/core/backing/Cargo.toml b/node/core/backing/Cargo.toml index 720b8af418d2..5a354482ce3c 100644 --- a/node/core/backing/Cargo.toml +++ b/node/core/backing/Cargo.toml @@ -22,5 +22,4 @@ bitvec = { version = "0.17.4", default-features = false, features = ["alloc"] } [dev-dependencies] sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "master" } futures = { version = "0.3.5", features = ["thread-pool"] } -subsystem-test = { package = "polkadot-subsystem-test-helpers", path = "../../test-helpers/subsystem" } assert_matches = "1.3.0" diff --git a/node/core/backing/src/lib.rs b/node/core/backing/src/lib.rs index e0309ec8428e..6237fa295acb 100644 --- a/node/core/backing/src/lib.rs +++ b/node/core/backing/src/lib.rs @@ -901,13 +901,13 @@ mod tests { } struct TestHarness { - virtual_overseer: subsystem_test::TestSubsystemContextHandle, + virtual_overseer: polkadot_subsystem::test_helpers::TestSubsystemContextHandle, } fn test_harness>(keystore: KeyStorePtr, test: impl FnOnce(TestHarness) -> T) { let pool = ThreadPool::new().unwrap(); - let (context, virtual_overseer) = subsystem_test::make_subsystem_context(pool.clone()); + let (context, virtual_overseer) = polkadot_subsystem::test_helpers::make_subsystem_context(pool.clone()); let subsystem = CandidateBackingSubsystem::run(context, keystore, pool.clone()); @@ -965,7 +965,7 @@ mod tests { // Tests that the subsystem performs actions that are requied on startup. async fn test_startup( - virtual_overseer: &mut subsystem_test::TestSubsystemContextHandle, + virtual_overseer: &mut polkadot_subsystem::test_helpers::TestSubsystemContextHandle, test_state: &TestState, ) { // Start work on some new parent. diff --git a/node/network/bridge/Cargo.toml b/node/network/bridge/Cargo.toml index 4f6c8631e2f9..8d63fefb7bdb 100644 --- a/node/network/bridge/Cargo.toml +++ b/node/network/bridge/Cargo.toml @@ -18,5 +18,4 @@ polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsys [dev-dependencies] parking_lot = "0.10.0" -subsystem-test = { package = "polkadot-subsystem-test-helpers", path = "../../test-helpers/subsystem" } assert_matches = "1.3.0" diff --git a/node/network/bridge/src/lib.rs b/node/network/bridge/src/lib.rs index 46b9dbd84024..88e0b40211f2 100644 --- a/node/network/bridge/src/lib.rs +++ b/node/network/bridge/src/lib.rs @@ -528,7 +528,7 @@ mod tests { use assert_matches::assert_matches; use polkadot_subsystem::messages::{StatementDistributionMessage, BitfieldDistributionMessage}; - use subsystem_test::{SingleItemSink, SingleItemStream}; + use polkadot_subsystem::test_helpers::{SingleItemSink, SingleItemStream}; // The subsystem's view of the network - only supports a single call to `event_stream`. struct TestNetwork { @@ -547,7 +547,7 @@ mod tests { TestNetwork, TestNetworkHandle, ) { - let (net_tx, net_rx) = subsystem_test::single_item_sink(); + let (net_tx, net_rx) = polkadot_subsystem::test_helpers::single_item_sink(); let (action_tx, action_rx) = mpsc::unbounded(); ( @@ -628,14 +628,14 @@ mod tests { struct TestHarness { network_handle: TestNetworkHandle, - virtual_overseer: subsystem_test::TestSubsystemContextHandle, + virtual_overseer: polkadot_subsystem::test_helpers::TestSubsystemContextHandle, } fn test_harness>(test: impl FnOnce(TestHarness) -> T) { let pool = ThreadPool::new().unwrap(); let (network, network_handle) = new_test_network(); - let (context, virtual_overseer) = subsystem_test::make_subsystem_context(pool); + let (context, virtual_overseer) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); let network_bridge = run_network( network, diff --git a/node/network/pov-distribution/Cargo.toml b/node/network/pov-distribution/Cargo.toml index a99e5e3d5604..15f48ad7a473 100644 --- a/node/network/pov-distribution/Cargo.toml +++ b/node/network/pov-distribution/Cargo.toml @@ -18,5 +18,4 @@ polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsys [dev-dependencies] parking_lot = "0.10.0" -subsystem-test = { package = "polkadot-subsystem-test-helpers", path = "../../test-helpers/subsystem" } assert_matches = "1.3.0" diff --git a/node/network/pov-distribution/src/lib.rs b/node/network/pov-distribution/src/lib.rs index 84d6e803d2c7..86d6a864fe24 100644 --- a/node/network/pov-distribution/src/lib.rs +++ b/node/network/pov-distribution/src/lib.rs @@ -617,7 +617,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); let mut descriptor = CandidateDescriptor::default(); descriptor.pov_hash = pov_hash; @@ -697,7 +697,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); let mut descriptor = CandidateDescriptor::default(); descriptor.pov_hash = pov_hash; @@ -775,7 +775,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); executor::block_on(async move { handle_network_update( @@ -847,7 +847,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); executor::block_on(async move { // Peer A answers our request before peer B. @@ -935,7 +935,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); executor::block_on(async move { // Peer A answers our request: right relay parent, awaited hash, wrong PoV. @@ -998,7 +998,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); executor::block_on(async move { // Peer A answers our request: right relay parent, awaited hash, wrong PoV. @@ -1059,7 +1059,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); executor::block_on(async move { // Peer A answers our request: right relay parent, awaited hash, wrong PoV. @@ -1117,7 +1117,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); executor::block_on(async move { let max_plausibly_awaited = n_validators * 2; @@ -1202,7 +1202,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); executor::block_on(async move { let pov_hash = make_pov(vec![1, 2, 3]).hash(); @@ -1264,7 +1264,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); executor::block_on(async move { let pov_hash = make_pov(vec![1, 2, 3]).hash(); @@ -1341,7 +1341,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); executor::block_on(async move { handle_network_update( @@ -1425,7 +1425,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); executor::block_on(async move { handle_network_update( diff --git a/node/network/statement-distribution/Cargo.toml b/node/network/statement-distribution/Cargo.toml index 2f8da8ee3d85..eaf24046cf67 100644 --- a/node/network/statement-distribution/Cargo.toml +++ b/node/network/statement-distribution/Cargo.toml @@ -21,6 +21,6 @@ indexmap = "1.4.0" [dev-dependencies] parking_lot = "0.10.0" -subsystem-test = { package = "polkadot-subsystem-test-helpers", path = "../../test-helpers/subsystem" } +polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem", features = ["test-helpers"] } assert_matches = "1.3.0" sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "master" } diff --git a/node/network/statement-distribution/src/lib.rs b/node/network/statement-distribution/src/lib.rs index cef499eae98b..fc3708be281e 100644 --- a/node/network/statement-distribution/src/lib.rs +++ b/node/network/statement-distribution/src/lib.rs @@ -1210,7 +1210,7 @@ mod tests { }; let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); let peer = PeerId::random(); executor::block_on(async move { @@ -1302,7 +1302,7 @@ mod tests { ].into_iter().collect(); let pool = ThreadPool::new().unwrap(); - let (mut ctx, mut handle) = subsystem_test::make_subsystem_context(pool); + let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool); executor::block_on(async move { let statement = { diff --git a/node/subsystem/Cargo.toml b/node/subsystem/Cargo.toml index 0e851770b84f..701a197c2fce 100644 --- a/node/subsystem/Cargo.toml +++ b/node/subsystem/Cargo.toml @@ -13,6 +13,7 @@ futures-timer = "3.0.2" keystore = { package = "sc-keystore", git = "https://github.com/paritytech/substrate", branch = "master" } log = "0.4.8" parity-scale-codec = "1.3.0" +parking_lot = { version = "0.10.0", optional = true } pin-project = "0.4.22" polkadot-node-primitives = { path = "../primitives" } polkadot-primitives = { path = "../../primitives" } @@ -26,3 +27,6 @@ assert_matches = "1.3.0" async-trait = "0.1" futures = { version = "0.3.5", features = ["thread-pool"] } parking_lot = "0.10.0" + +[features] +test-helpers = [ "parking_lot" ] diff --git a/node/subsystem/src/lib.rs b/node/subsystem/src/lib.rs index 2005b79f78ef..8f9ec35b3f35 100644 --- a/node/subsystem/src/lib.rs +++ b/node/subsystem/src/lib.rs @@ -35,7 +35,7 @@ use crate::messages::AllMessages; pub mod messages; pub mod util; -#[cfg(test)] +#[cfg(any(test, feature = "test-helpers"))] pub mod test_helpers; /// Signals sent by an overseer to a subsystem. diff --git a/node/test-helpers/subsystem/Cargo.toml b/node/test-helpers/subsystem/Cargo.toml deleted file mode 100644 index fa6f4038e6da..000000000000 --- a/node/test-helpers/subsystem/Cargo.toml +++ /dev/null @@ -1,9 +0,0 @@ -[package] -name = "polkadot-subsystem-test-helpers" -version = "0.1.0" -authors = ["Parity Technologies "] -edition = "2018" -description = "Helpers for testing subsystems" - -[dependencies] -polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" } diff --git a/node/test-helpers/subsystem/src/lib.rs b/node/test-helpers/subsystem/src/lib.rs deleted file mode 100644 index 68197f4d1fed..000000000000 --- a/node/test-helpers/subsystem/src/lib.rs +++ /dev/null @@ -1,21 +0,0 @@ -// Copyright 2017-2020 Parity Technologies (UK) Ltd. -// This file is part of Polkadot. - -// Polkadot is free software: you can redistribute it and/or modify -// it under the terms of the GNU General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. - -// Polkadot is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU General Public License for more details. - -// You should have received a copy of the GNU General Public License -// along with Polkadot. If not, see . - -//! Utilities for testing subsystems. -//! -//! **DEPRECATED**: use `polkadot-node-subsystem::test_helpers` instead - -pub use polkadot_subsystem::test_helpers::*; From 356c26fe2b282a98690997c7573b86d142cb6bbd Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Thu, 16 Jul 2020 17:07:59 +0200 Subject: [PATCH 06/15] add some more tests --- Cargo.lock | 10 ++++++++++ node/subsystem/Cargo.toml | 1 + node/subsystem/src/util.rs | 38 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 49 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index 81477b73509a..f592dbe9f8fd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4557,6 +4557,7 @@ dependencies = [ "sc-network", "sp-core", "streamunordered", + "testing_logger", ] [[package]] @@ -8400,6 +8401,15 @@ dependencies = [ "tiny-keccak 1.5.0", ] +[[package]] +name = "testing_logger" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d92b727cb45d33ae956f7f46b966b25f1bc712092aeef9dba5ac798fc89f720" +dependencies = [ + "log 0.4.8", +] + [[package]] name = "textwrap" version = "0.11.0" diff --git a/node/subsystem/Cargo.toml b/node/subsystem/Cargo.toml index 701a197c2fce..e027b4c7bc31 100644 --- a/node/subsystem/Cargo.toml +++ b/node/subsystem/Cargo.toml @@ -27,6 +27,7 @@ assert_matches = "1.3.0" async-trait = "0.1" futures = { version = "0.3.5", features = ["thread-pool"] } parking_lot = "0.10.0" +testing_logger = "0.1.1" [features] test-helpers = [ "parking_lot" ] diff --git a/node/subsystem/src/util.rs b/node/subsystem/src/util.rs index 5ffa5f71fd7d..9e162d9ddbb5 100644 --- a/node/subsystem/src/util.rs +++ b/node/subsystem/src/util.rs @@ -626,6 +626,8 @@ mod tests { JobTrait, ToJobTrait, }, + FromOverseer, + OverseerSignal, }; use futures::{ channel::mpsc, @@ -789,6 +791,11 @@ mod tests { type OverseerHandle = test_helpers::TestSubsystemContextHandle; fn test_harness>(run_args: HashMap>, test: impl FnOnce(OverseerHandle) -> T) { + // testing is a little weird: this module throws away quite a few errors where various traits don't offer the opportunity + // to propagate them. However, most of the ones we care about get logged. Therefore, we can use a testing logger to make + // assertions about the number of errors which have occurred. + testing_logger::setup(); + let pool = ThreadPool::new().unwrap(); let (context, overseer_handle) = make_subsystem_context(pool.clone()); @@ -809,10 +816,41 @@ mod tests { run_args.insert(relay_parent.clone(), vec![FromJob::Test(test_message.clone())]); test_harness(run_args, |mut overseer_handle| async move { + overseer_handle.send(FromOverseer::Signal(OverseerSignal::StartWork(relay_parent))).await; assert_matches!( overseer_handle.recv().await, AllMessages::Test(msg) if msg == test_message ); }); + + testing_logger::validate(|captured_logs| assert_eq!(captured_logs.len(), 0)); + } + + #[test] + fn stopping_running_job_works() { + let relay_parent: Hash = [0; 32].into(); + let run_args = HashMap::new(); + + test_harness(run_args, |mut overseer_handle| async move { + overseer_handle.send(FromOverseer::Signal(OverseerSignal::StartWork(relay_parent))).await; + overseer_handle.send(FromOverseer::Signal(OverseerSignal::StopWork(relay_parent))).await; + }); + + testing_logger::validate(|captured_logs| assert_eq!(captured_logs.len(), 0)); + } + + #[test] + fn stopping_non_running_job_fails() { + let relay_parent: Hash = [0; 32].into(); + let run_args = HashMap::new(); + + test_harness(run_args, |mut overseer_handle| async move { + overseer_handle.send(FromOverseer::Signal(OverseerSignal::StopWork(relay_parent))).await; + }); + + testing_logger::validate(|captured_logs| { + assert_eq!(captured_logs.len(), 1); + assert_eq!(captured_logs[0].level, log::Level::Error); + }); } } From f0c23de2a7b1f1ed94026a183caab4cd4c9e4712 Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Fri, 17 Jul 2020 13:54:53 +0200 Subject: [PATCH 07/15] get rid of log tests in favor of real error forwarding It's not obvious whether we'll ever really want to chase down these errors outside a testing context, but having the capability won't hurt. --- Cargo.lock | 10 -- node/subsystem/Cargo.toml | 1 - node/subsystem/src/lib.rs | 2 +- node/subsystem/src/util.rs | 183 +++++++++++++++++++++++++++++-------- 4 files changed, 145 insertions(+), 51 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f592dbe9f8fd..81477b73509a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4557,7 +4557,6 @@ dependencies = [ "sc-network", "sp-core", "streamunordered", - "testing_logger", ] [[package]] @@ -8401,15 +8400,6 @@ dependencies = [ "tiny-keccak 1.5.0", ] -[[package]] -name = "testing_logger" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6d92b727cb45d33ae956f7f46b966b25f1bc712092aeef9dba5ac798fc89f720" -dependencies = [ - "log 0.4.8", -] - [[package]] name = "textwrap" version = "0.11.0" diff --git a/node/subsystem/Cargo.toml b/node/subsystem/Cargo.toml index e027b4c7bc31..701a197c2fce 100644 --- a/node/subsystem/Cargo.toml +++ b/node/subsystem/Cargo.toml @@ -27,7 +27,6 @@ assert_matches = "1.3.0" async-trait = "0.1" futures = { version = "0.3.5", features = ["thread-pool"] } parking_lot = "0.10.0" -testing_logger = "0.1.1" [features] test-helpers = [ "parking_lot" ] diff --git a/node/subsystem/src/lib.rs b/node/subsystem/src/lib.rs index 8f9ec35b3f35..2bd61f0dfbff 100644 --- a/node/subsystem/src/lib.rs +++ b/node/subsystem/src/lib.rs @@ -73,7 +73,7 @@ pub enum FromOverseer { /// * Subsystems dying when they are not expected to /// * Subsystems not dying when they are told to die /// * etc. -#[derive(Debug)] +#[derive(Debug, PartialEq)] pub struct SubsystemError; impl From for SubsystemError { diff --git a/node/subsystem/src/util.rs b/node/subsystem/src/util.rs index 9e162d9ddbb5..f52fe14e87c1 100644 --- a/node/subsystem/src/util.rs +++ b/node/subsystem/src/util.rs @@ -22,7 +22,7 @@ use crate::{ messages::{AllMessages, RuntimeApiMessage, RuntimeApiRequest, SchedulerRoster}, - FromOverseer, SpawnedSubsystem, Subsystem, SubsystemContext, SubsystemResult, + FromOverseer, SpawnedSubsystem, Subsystem, SubsystemContext, SubsystemError, SubsystemResult, }; use futures::{ channel::{mpsc, oneshot}, @@ -67,12 +67,40 @@ pub enum Error { /// Attempted to spawn a new task, and failed #[from] Spawn(SpawnError), + /// A subsystem error + #[from] + Subsystem(SubsystemError), + /// The type system wants this even though it doesn't make sense + #[from] + Infallible(std::convert::Infallible), /// Attempted to convert from an AllMessages to a FromJob, and failed. SenderConversion(String), /// The local node is not a validator. NotAValidator, /// The desired job is not present in the jobs list. JobNotFound(Hash), + /// Already forwarding errors to another sender + AlreadyForwarding, +} + +impl PartialEq for Error { + fn eq(&self, other: &Self) -> bool { + match (self, other) { + // spawn errors are not comparable + (Self::Spawn(_), _) | (_, Self::Spawn(_)) => false, + + (Self::Oneshot(left), Self::Oneshot(right)) => left == right, + (Self::Mpsc(left), Self::Mpsc(right)) => left == right, + (Self::Subsystem(left), Self::Subsystem(right)) => left == right, + (Self::Infallible(left), Self::Infallible(right)) => left == right, + (Self::SenderConversion(left), Self::SenderConversion(right)) => left == right, + (Self::JobNotFound(left), Self::JobNotFound(right)) => left == right, + (Self::NotAValidator, Self::NotAValidator) | (Self::AlreadyForwarding, Self::AlreadyForwarding) => true, + + // if the variants don't match, the errors are unequal + _ => false + } + } } /// Request some data from the `RuntimeApi`. @@ -262,7 +290,7 @@ pub trait ToJobTrait: TryFrom { } /// A JobHandle manages a particular job for a subsystem. -pub struct JobHandle { +struct JobHandle { abort_handle: future::AbortHandle, to_job: mpsc::Sender, finished: oneshot::Receiver<()>, @@ -271,23 +299,23 @@ pub struct JobHandle { impl JobHandle { /// Send a message to the job. - pub async fn send_msg(&mut self, msg: ToJob) -> Result<(), Error> { + async fn send_msg(&mut self, msg: ToJob) -> Result<(), Error> { self.to_job.send(msg).await.map_err(Into::into) } - - /// Abort the job without waiting for a graceful shutdown - pub fn abort(self) { - self.abort_handle.abort(); - } } impl JobHandle { /// Stop this job gracefully. /// /// If it hasn't shut itself down after `JOB_GRACEFUL_STOP_DURATION`, abort it. - pub async fn stop(mut self) { + async fn stop(mut self) { // we don't actually care if the message couldn't be sent - let _ = self.to_job.send(ToJob::STOP).await; + if let Err(_) = self.to_job.send(ToJob::STOP).await { + // no need to wait further here: the job is either stalled or + // disconnected, and in either case, we can just abort it immediately + self.abort_handle.abort(); + return; + } let stop_timer = Delay::new(JOB_GRACEFUL_STOP_DURATION); match future::select(stop_timer, self.finished).await { @@ -310,7 +338,7 @@ pub trait JobTrait: Unpin { /// Message type from the job. Typically a subset of AllMessages. type FromJob: 'static + Into + Send; /// Job runtime error. - type Error: std::fmt::Debug; + type Error: 'static + std::fmt::Debug + PartialEq + Send; /// Extra arguments this job needs to run properly. /// /// If no extra information is needed, it is perfectly acceptable to set it to `()`. @@ -342,6 +370,18 @@ pub trait JobTrait: Unpin { } } +/// Error which can be returned by the jobs manager +/// +/// Wraps the utility error type and the job-specific error +#[derive(Debug, derive_more::From, PartialEq)] +pub enum JobsError { + /// utility error + #[from] + Utility(Error), + /// internal job error + Job(JobError), +} + /// Jobs manager for a subsystem /// /// - Spawns new jobs for a given relay-parent on demand. @@ -356,6 +396,7 @@ pub struct Jobs { #[pin] outgoing_msgs: StreamUnordered>, job: std::marker::PhantomData, + errors: Option, JobsError)>>, } impl Jobs { @@ -366,15 +407,31 @@ impl Jobs { running: HashMap::new(), outgoing_msgs: StreamUnordered::new(), job: std::marker::PhantomData, + errors: None, } } + /// Monitor errors which may occur during handling of a spawned job. + /// + /// By default, an error in a job is simply logged. Once this is called, + /// the error is forwarded onto the provided channel. + /// + /// Errors if the error channel already exists. + pub fn fwd_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { + if self.errors.is_some() { return Err(Error::AlreadyForwarding) } + self.errors = Some(tx); + Ok(()) + } + /// Spawn a new job for this `parent_hash`, with whatever args are appropriate. fn spawn_job(&mut self, parent_hash: Hash, run_args: Job::RunArgs) -> Result<(), Error> { let (to_job_tx, to_job_rx) = mpsc::channel(JOB_CHANNEL_CAPACITY); let (from_job_tx, from_job_rx) = mpsc::channel(JOB_CHANNEL_CAPACITY); let (finished_tx, finished) = oneshot::channel(); + // clone the error transmitter to move into the future + let err_tx = self.errors.clone(); + let (future, abort_handle) = future::abortable(async move { if let Err(e) = Job::run(parent_hash, run_args, to_job_rx, from_job_tx).await { log::error!( @@ -383,12 +440,23 @@ impl Jobs { parent_hash, e, ); + + if let Some(mut err_tx) = err_tx { + // if we can't send the notification of error on the error channel, then + // there's no point trying to propagate this error onto the channel too + let _ = err_tx.send((Some(parent_hash), JobsError::Job(e))).await; + } } }); - // discard output + // the spawn mechanism requires that the spawned future has no output let future = async move { + // job errors are already handled within the future, meaning + // that any errors here are due to the abortable mechanism. + // failure to abort isn't of interest. let _ = future.await; + // transmission failure here is only possible if the receiver is closed, + // which means the handle is dropped, which means we don't care anymore let _ = finished_tx.send(()); }; self.spawner.spawn(future)?; @@ -472,6 +540,7 @@ pub struct JobManager { run_args: Job::RunArgs, context: std::marker::PhantomData, job: std::marker::PhantomData, + errors: Option, JobsError)>>, } impl JobManager @@ -489,9 +558,22 @@ where run_args, context: std::marker::PhantomData, job: std::marker::PhantomData, + errors: None, } } + /// Monitor errors which may occur during handling of a spawned job. + /// + /// By default, an error in a job is simply logged. Once this is called, + /// the error is forwarded onto the provided channel. + /// + /// Errors if the error channel already exists. + pub fn fwd_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { + if self.errors.is_some() { return Err(Error::AlreadyForwarding) } + self.errors = Some(tx); + Ok(()) + } + /// Run this subsystem /// /// Conceptually, this is very simple: it just loops forever. @@ -500,23 +582,38 @@ where /// - On other incoming messages, if they can be converted into Job::ToJob and /// include a hash, then they're forwarded to the appropriate individual job. /// - On outgoing messages from the jobs, it forwards them to the overseer. - pub async fn run(mut ctx: Context, run_args: Job::RunArgs, spawner: Spawner) { + /// + /// If `err_tx` is not `None`, errors are forwarded onto that channel as they occur. + /// Otherwise, most are logged and then discarded. + pub async fn run(mut ctx: Context, run_args: Job::RunArgs, spawner: Spawner, mut err_tx: Option, JobsError)>>) { let mut jobs = Jobs::new(spawner.clone()); + if let Some(ref err_tx) = err_tx { + jobs.fwd_errors(err_tx.clone()).expect("we never call this twice in this context; qed"); + } loop { select! { - incoming = ctx.recv().fuse() => if Self::handle_incoming(incoming, &mut jobs, &run_args).await { break }, - outgoing = jobs.next().fuse() => if Self::handle_outgoing(outgoing, &mut ctx).await { break }, + incoming = ctx.recv().fuse() => if Self::handle_incoming(incoming, &mut jobs, &run_args, &mut err_tx).await { break }, + outgoing = jobs.next().fuse() => if Self::handle_outgoing(outgoing, &mut ctx, &mut err_tx).await { break }, complete => break, } } } + // if we have a channel on which to forward errors, do so + async fn fwd_err(hash: Option, err: JobsError, err_tx: &mut Option, JobsError)>>) { + if let Some(err_tx) = err_tx { + // if we can't send on the error transmission channel, we can't do anything about it + let _ = err_tx.send((hash, err)).await; + } + } + // handle an incoming message. return true if we should break afterwards. async fn handle_incoming( incoming: SubsystemResult>, jobs: &mut Jobs, run_args: &Job::RunArgs, + err_tx: &mut Option, JobsError)>> ) -> bool { use crate::FromOverseer::{Communication, Signal}; use crate::OverseerSignal::{Conclude, StartWork, StopWork}; @@ -525,12 +622,14 @@ where Ok(Signal(StartWork(hash))) => { if let Err(e) = jobs.spawn_job(hash, run_args.clone()) { log::error!("Failed to spawn a job: {:?}", e); + Self::fwd_err(Some(hash), e.into(), err_tx).await; return true; } } Ok(Signal(StopWork(hash))) => { if let Err(e) = jobs.stop_job(hash).await { log::error!("Failed to stop a job: {:?}", e); + Self::fwd_err(Some(hash), e.into(), err_tx).await; return true; } } @@ -553,6 +652,7 @@ where .await { log::error!("failed to stop all jobs on conclude signal: {:?}", e); + Self::fwd_err(None, Error::from(e).into(), err_tx).await; } return true; @@ -563,12 +663,14 @@ where Some(hash) => { if let Err(err) = jobs.send_msg(hash, to_job).await { log::error!("Failed to send a message to a job: {:?}", err); + Self::fwd_err(Some(hash), err.into(), err_tx).await; return true; } } None => { if let Err(err) = Job::handle_unanchored_msg(to_job) { log::error!("Failed to handle unhashed message: {:?}", err); + Self::fwd_err(None, JobsError::Job(err), err_tx).await; return true; } } @@ -577,6 +679,7 @@ where } Err(err) => { log::error!("error receiving message from subsystem context: {:?}", err); + Self::fwd_err(None, Error::from(err).into(), err_tx).await; return true; } } @@ -584,11 +687,12 @@ where } // handle an outgoing message. return true if we should break afterwards. - async fn handle_outgoing(outgoing: Option, ctx: &mut Context) -> bool { + async fn handle_outgoing(outgoing: Option, ctx: &mut Context, err_tx: &mut Option, JobsError)>>) -> bool { match outgoing { Some(msg) => { - // discard errors when sending the message upstream - let _ = ctx.send_message(msg.into()).await; + if let Err(e) = ctx.send_message(msg.into()).await { + Self::fwd_err(None, Error::from(e).into(), err_tx).await; + } } None => return true, } @@ -608,9 +712,10 @@ where fn start(self, ctx: Context) -> SpawnedSubsystem { let spawner = self.spawner.clone(); let run_args = self.run_args.clone(); + let errors = self.errors; SpawnedSubsystem(Box::pin(async move { - Self::run(ctx, run_args, spawner).await; + Self::run(ctx, run_args, spawner, errors).await; })) } } @@ -622,6 +727,8 @@ mod tests { messages::{AllMessages, CandidateSelectionMessage}, test_helpers::{self, make_subsystem_context}, util::{ + self, + JobsError, JobManager, JobTrait, ToJobTrait, @@ -720,7 +827,7 @@ mod tests { // Error will mostly be a wrapper to make the try operator more convenient; // deriving From implementations for most variants is recommended. // It must implement Debug for logging. - #[derive(Debug, derive_more::From)] + #[derive(Debug, derive_more::From, PartialEq)] enum Error { #[from] Sending(mpsc::SendError) @@ -790,17 +897,13 @@ mod tests { // this type lets us pretend to be the overseer type OverseerHandle = test_helpers::TestSubsystemContextHandle; - fn test_harness>(run_args: HashMap>, test: impl FnOnce(OverseerHandle) -> T) { - // testing is a little weird: this module throws away quite a few errors where various traits don't offer the opportunity - // to propagate them. However, most of the ones we care about get logged. Therefore, we can use a testing logger to make - // assertions about the number of errors which have occurred. - testing_logger::setup(); - + fn test_harness>(run_args: HashMap>, test: impl FnOnce(OverseerHandle, mpsc::Receiver<(Option, JobsError)>) -> T) { let pool = ThreadPool::new().unwrap(); let (context, overseer_handle) = make_subsystem_context(pool.clone()); + let (err_tx, err_rx) = mpsc::channel(16); - let subsystem = FakeCandidateSelectionSubsystem::run(context, run_args, pool); - let test_future = test(overseer_handle); + let subsystem = FakeCandidateSelectionSubsystem::run(context, run_args, pool, Some(err_tx)); + let test_future = test(overseer_handle, err_rx); futures::pin_mut!(test_future); futures::pin_mut!(subsystem); @@ -815,15 +918,16 @@ mod tests { let test_message = format!("greetings from {}", relay_parent); run_args.insert(relay_parent.clone(), vec![FromJob::Test(test_message.clone())]); - test_harness(run_args, |mut overseer_handle| async move { + test_harness(run_args, |mut overseer_handle, err_rx| async move { overseer_handle.send(FromOverseer::Signal(OverseerSignal::StartWork(relay_parent))).await; assert_matches!( overseer_handle.recv().await, AllMessages::Test(msg) if msg == test_message ); - }); - testing_logger::validate(|captured_logs| assert_eq!(captured_logs.len(), 0)); + let errs: Vec<_> = err_rx.collect().await; + assert_eq!(errs.len(), 0); + }); } #[test] @@ -831,12 +935,13 @@ mod tests { let relay_parent: Hash = [0; 32].into(); let run_args = HashMap::new(); - test_harness(run_args, |mut overseer_handle| async move { + test_harness(run_args, |mut overseer_handle, err_rx| async move { overseer_handle.send(FromOverseer::Signal(OverseerSignal::StartWork(relay_parent))).await; overseer_handle.send(FromOverseer::Signal(OverseerSignal::StopWork(relay_parent))).await; - }); - testing_logger::validate(|captured_logs| assert_eq!(captured_logs.len(), 0)); + let errs: Vec<_> = err_rx.collect().await; + assert_eq!(errs.len(), 0); + }); } #[test] @@ -844,13 +949,13 @@ mod tests { let relay_parent: Hash = [0; 32].into(); let run_args = HashMap::new(); - test_harness(run_args, |mut overseer_handle| async move { + test_harness(run_args, |mut overseer_handle, err_rx| async move { overseer_handle.send(FromOverseer::Signal(OverseerSignal::StopWork(relay_parent))).await; - }); - testing_logger::validate(|captured_logs| { - assert_eq!(captured_logs.len(), 1); - assert_eq!(captured_logs[0].level, log::Level::Error); + let errs: Vec<_> = err_rx.collect().await; + assert_eq!(errs.len(), 1); + assert_eq!(errs[0].0, Some(relay_parent)); + assert_eq!(errs[0].1, JobsError::Utility(util::Error::JobNotFound(relay_parent))); }); } } From 66e7d50b3f8a21d3e3ce7206dc6cfd9d32aecda5 Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Fri, 17 Jul 2020 14:26:13 +0200 Subject: [PATCH 08/15] fix issue which caused test to hang on osx --- node/subsystem/src/util.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/node/subsystem/src/util.rs b/node/subsystem/src/util.rs index f52fe14e87c1..33ec6b69de41 100644 --- a/node/subsystem/src/util.rs +++ b/node/subsystem/src/util.rs @@ -924,6 +924,7 @@ mod tests { overseer_handle.recv().await, AllMessages::Test(msg) if msg == test_message ); + std::mem::drop(overseer_handle); let errs: Vec<_> = err_rx.collect().await; assert_eq!(errs.len(), 0); From a9b49af663993c58f3852c9cd854bd3a06f327d5 Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Fri, 17 Jul 2020 14:55:52 +0200 Subject: [PATCH 09/15] only require that job errors are PartialEq when testing also fix polkadot-node-core-backing tests --- node/core/backing/Cargo.toml | 1 + node/core/backing/src/lib.rs | 2 +- node/subsystem/src/util.rs | 33 ++++++++++++++++++++------------- 3 files changed, 22 insertions(+), 14 deletions(-) diff --git a/node/core/backing/Cargo.toml b/node/core/backing/Cargo.toml index 5a354482ce3c..2e2e0f0e15b0 100644 --- a/node/core/backing/Cargo.toml +++ b/node/core/backing/Cargo.toml @@ -23,3 +23,4 @@ bitvec = { version = "0.17.4", default-features = false, features = ["alloc"] } sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "master" } futures = { version = "0.3.5", features = ["thread-pool"] } assert_matches = "1.3.0" +polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem", features = [ "test-helpers" ] } diff --git a/node/core/backing/src/lib.rs b/node/core/backing/src/lib.rs index 6237fa295acb..5e2db1ef70fe 100644 --- a/node/core/backing/src/lib.rs +++ b/node/core/backing/src/lib.rs @@ -748,7 +748,7 @@ where /// Run this subsystem pub async fn run(ctx: Context, keystore: KeyStorePtr, spawner: Spawner) { - >::run(ctx, keystore, spawner).await + >::run(ctx, keystore, spawner, None).await } } diff --git a/node/subsystem/src/util.rs b/node/subsystem/src/util.rs index 33ec6b69de41..4d075d243823 100644 --- a/node/subsystem/src/util.rs +++ b/node/subsystem/src/util.rs @@ -337,8 +337,15 @@ pub trait JobTrait: Unpin { type ToJob: 'static + ToJobTrait + Send; /// Message type from the job. Typically a subset of AllMessages. type FromJob: 'static + Into + Send; + // the test suite here requires that this error type implements PartialEq, but that's + // not a necessary or even an entirely reasonable ask to make of all potential + // job error types. Therefore, only impose that trait bound when testing. /// Job runtime error. + #[cfg(test)] type Error: 'static + std::fmt::Debug + PartialEq + Send; + /// Job runtime error. + #[cfg(not(test))] + type Error: 'static + std::fmt::Debug + Send; /// Extra arguments this job needs to run properly. /// /// If no extra information is needed, it is perfectly acceptable to set it to `()`. @@ -374,12 +381,12 @@ pub trait JobTrait: Unpin { /// /// Wraps the utility error type and the job-specific error #[derive(Debug, derive_more::From, PartialEq)] -pub enum JobsError { +pub enum JobsError { /// utility error #[from] Utility(Error), /// internal job error - Job(JobError), + Job(Job::Error), } /// Jobs manager for a subsystem @@ -396,10 +403,10 @@ pub struct Jobs { #[pin] outgoing_msgs: StreamUnordered>, job: std::marker::PhantomData, - errors: Option, JobsError)>>, + errors: Option, JobsError)>>, } -impl Jobs { +impl Jobs { /// Create a new Jobs manager which handles spawning appropriate jobs. pub fn new(spawner: Spawner) -> Self { Self { @@ -417,7 +424,7 @@ impl Jobs { /// the error is forwarded onto the provided channel. /// /// Errors if the error channel already exists. - pub fn fwd_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { + pub fn fwd_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { if self.errors.is_some() { return Err(Error::AlreadyForwarding) } self.errors = Some(tx); Ok(()) @@ -540,14 +547,14 @@ pub struct JobManager { run_args: Job::RunArgs, context: std::marker::PhantomData, job: std::marker::PhantomData, - errors: Option, JobsError)>>, + errors: Option, JobsError)>>, } impl JobManager where Spawner: Spawn + Clone + Send + Unpin, Context: SubsystemContext, - Job: JobTrait, + Job: 'static + JobTrait, Job::RunArgs: Clone, Job::ToJob: TryFrom + TryFrom<::Message> + Sync, { @@ -568,7 +575,7 @@ where /// the error is forwarded onto the provided channel. /// /// Errors if the error channel already exists. - pub fn fwd_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { + pub fn fwd_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { if self.errors.is_some() { return Err(Error::AlreadyForwarding) } self.errors = Some(tx); Ok(()) @@ -585,7 +592,7 @@ where /// /// If `err_tx` is not `None`, errors are forwarded onto that channel as they occur. /// Otherwise, most are logged and then discarded. - pub async fn run(mut ctx: Context, run_args: Job::RunArgs, spawner: Spawner, mut err_tx: Option, JobsError)>>) { + pub async fn run(mut ctx: Context, run_args: Job::RunArgs, spawner: Spawner, mut err_tx: Option, JobsError)>>) { let mut jobs = Jobs::new(spawner.clone()); if let Some(ref err_tx) = err_tx { jobs.fwd_errors(err_tx.clone()).expect("we never call this twice in this context; qed"); @@ -601,7 +608,7 @@ where } // if we have a channel on which to forward errors, do so - async fn fwd_err(hash: Option, err: JobsError, err_tx: &mut Option, JobsError)>>) { + async fn fwd_err(hash: Option, err: JobsError, err_tx: &mut Option, JobsError)>>) { if let Some(err_tx) = err_tx { // if we can't send on the error transmission channel, we can't do anything about it let _ = err_tx.send((hash, err)).await; @@ -613,7 +620,7 @@ where incoming: SubsystemResult>, jobs: &mut Jobs, run_args: &Job::RunArgs, - err_tx: &mut Option, JobsError)>> + err_tx: &mut Option, JobsError)>> ) -> bool { use crate::FromOverseer::{Communication, Signal}; use crate::OverseerSignal::{Conclude, StartWork, StopWork}; @@ -687,7 +694,7 @@ where } // handle an outgoing message. return true if we should break afterwards. - async fn handle_outgoing(outgoing: Option, ctx: &mut Context, err_tx: &mut Option, JobsError)>>) -> bool { + async fn handle_outgoing(outgoing: Option, ctx: &mut Context, err_tx: &mut Option, JobsError)>>) -> bool { match outgoing { Some(msg) => { if let Err(e) = ctx.send_message(msg.into()).await { @@ -705,7 +712,7 @@ where Spawner: Spawn + Send + Clone + Unpin + 'static, Context: SubsystemContext, ::Message: Into, - Job: JobTrait + Send, + Job: 'static + JobTrait + Send, Job::RunArgs: Clone + Sync, Job::ToJob: TryFrom + Sync, { From 812b35a515f33c94beca737f75b4a129b592c7a8 Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Fri, 17 Jul 2020 15:19:59 +0200 Subject: [PATCH 10/15] get rid of any notion of partialeq --- node/subsystem/src/util.rs | 56 +++++++++++--------------------------- 1 file changed, 16 insertions(+), 40 deletions(-) diff --git a/node/subsystem/src/util.rs b/node/subsystem/src/util.rs index 4d075d243823..20eb14a97c12 100644 --- a/node/subsystem/src/util.rs +++ b/node/subsystem/src/util.rs @@ -83,26 +83,6 @@ pub enum Error { AlreadyForwarding, } -impl PartialEq for Error { - fn eq(&self, other: &Self) -> bool { - match (self, other) { - // spawn errors are not comparable - (Self::Spawn(_), _) | (_, Self::Spawn(_)) => false, - - (Self::Oneshot(left), Self::Oneshot(right)) => left == right, - (Self::Mpsc(left), Self::Mpsc(right)) => left == right, - (Self::Subsystem(left), Self::Subsystem(right)) => left == right, - (Self::Infallible(left), Self::Infallible(right)) => left == right, - (Self::SenderConversion(left), Self::SenderConversion(right)) => left == right, - (Self::JobNotFound(left), Self::JobNotFound(right)) => left == right, - (Self::NotAValidator, Self::NotAValidator) | (Self::AlreadyForwarding, Self::AlreadyForwarding) => true, - - // if the variants don't match, the errors are unequal - _ => false - } - } -} - /// Request some data from the `RuntimeApi`. pub async fn request_from_runtime( parent: Hash, @@ -337,14 +317,7 @@ pub trait JobTrait: Unpin { type ToJob: 'static + ToJobTrait + Send; /// Message type from the job. Typically a subset of AllMessages. type FromJob: 'static + Into + Send; - // the test suite here requires that this error type implements PartialEq, but that's - // not a necessary or even an entirely reasonable ask to make of all potential - // job error types. Therefore, only impose that trait bound when testing. /// Job runtime error. - #[cfg(test)] - type Error: 'static + std::fmt::Debug + PartialEq + Send; - /// Job runtime error. - #[cfg(not(test))] type Error: 'static + std::fmt::Debug + Send; /// Extra arguments this job needs to run properly. /// @@ -380,13 +353,13 @@ pub trait JobTrait: Unpin { /// Error which can be returned by the jobs manager /// /// Wraps the utility error type and the job-specific error -#[derive(Debug, derive_more::From, PartialEq)] -pub enum JobsError { +#[derive(Debug, derive_more::From)] +pub enum JobsError { /// utility error #[from] Utility(Error), /// internal job error - Job(Job::Error), + Job(JobError), } /// Jobs manager for a subsystem @@ -403,7 +376,7 @@ pub struct Jobs { #[pin] outgoing_msgs: StreamUnordered>, job: std::marker::PhantomData, - errors: Option, JobsError)>>, + errors: Option, JobsError)>>, } impl Jobs { @@ -424,7 +397,7 @@ impl Jobs { /// the error is forwarded onto the provided channel. /// /// Errors if the error channel already exists. - pub fn fwd_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { + pub fn fwd_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { if self.errors.is_some() { return Err(Error::AlreadyForwarding) } self.errors = Some(tx); Ok(()) @@ -547,7 +520,7 @@ pub struct JobManager { run_args: Job::RunArgs, context: std::marker::PhantomData, job: std::marker::PhantomData, - errors: Option, JobsError)>>, + errors: Option, JobsError)>>, } impl JobManager @@ -575,7 +548,7 @@ where /// the error is forwarded onto the provided channel. /// /// Errors if the error channel already exists. - pub fn fwd_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { + pub fn fwd_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { if self.errors.is_some() { return Err(Error::AlreadyForwarding) } self.errors = Some(tx); Ok(()) @@ -592,7 +565,7 @@ where /// /// If `err_tx` is not `None`, errors are forwarded onto that channel as they occur. /// Otherwise, most are logged and then discarded. - pub async fn run(mut ctx: Context, run_args: Job::RunArgs, spawner: Spawner, mut err_tx: Option, JobsError)>>) { + pub async fn run(mut ctx: Context, run_args: Job::RunArgs, spawner: Spawner, mut err_tx: Option, JobsError)>>) { let mut jobs = Jobs::new(spawner.clone()); if let Some(ref err_tx) = err_tx { jobs.fwd_errors(err_tx.clone()).expect("we never call this twice in this context; qed"); @@ -608,7 +581,7 @@ where } // if we have a channel on which to forward errors, do so - async fn fwd_err(hash: Option, err: JobsError, err_tx: &mut Option, JobsError)>>) { + async fn fwd_err(hash: Option, err: JobsError, err_tx: &mut Option, JobsError)>>) { if let Some(err_tx) = err_tx { // if we can't send on the error transmission channel, we can't do anything about it let _ = err_tx.send((hash, err)).await; @@ -620,7 +593,7 @@ where incoming: SubsystemResult>, jobs: &mut Jobs, run_args: &Job::RunArgs, - err_tx: &mut Option, JobsError)>> + err_tx: &mut Option, JobsError)>> ) -> bool { use crate::FromOverseer::{Communication, Signal}; use crate::OverseerSignal::{Conclude, StartWork, StopWork}; @@ -694,7 +667,7 @@ where } // handle an outgoing message. return true if we should break afterwards. - async fn handle_outgoing(outgoing: Option, ctx: &mut Context, err_tx: &mut Option, JobsError)>>) -> bool { + async fn handle_outgoing(outgoing: Option, ctx: &mut Context, err_tx: &mut Option, JobsError)>>) -> bool { match outgoing { Some(msg) => { if let Err(e) = ctx.send_message(msg.into()).await { @@ -834,7 +807,7 @@ mod tests { // Error will mostly be a wrapper to make the try operator more convenient; // deriving From implementations for most variants is recommended. // It must implement Debug for logging. - #[derive(Debug, derive_more::From, PartialEq)] + #[derive(Debug, derive_more::From)] enum Error { #[from] Sending(mpsc::SendError) @@ -963,7 +936,10 @@ mod tests { let errs: Vec<_> = err_rx.collect().await; assert_eq!(errs.len(), 1); assert_eq!(errs[0].0, Some(relay_parent)); - assert_eq!(errs[0].1, JobsError::Utility(util::Error::JobNotFound(relay_parent))); + assert_matches!( + errs[0].1, + JobsError::Utility(util::Error::JobNotFound(match_relay_parent)) if relay_parent == match_relay_parent + ); }); } } From 7e758b53a7c592f7860362d9c40d4753a4dd4f8b Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Mon, 20 Jul 2020 08:54:12 +0200 Subject: [PATCH 11/15] rethink testing Combine tests of starting and stopping job: leaving a test executor with a job running was pretty clearly the cause of the sometimes-hang. Also, add a timeout so tests _can't_ hang anymore; they just fail after a while. --- node/subsystem/src/util.rs | 29 ++++++++++++----------------- 1 file changed, 12 insertions(+), 17 deletions(-) diff --git a/node/subsystem/src/util.rs b/node/subsystem/src/util.rs index 20eb14a97c12..d3104139cdc0 100644 --- a/node/subsystem/src/util.rs +++ b/node/subsystem/src/util.rs @@ -719,17 +719,18 @@ mod tests { use futures::{ channel::mpsc, executor::{self, ThreadPool}, - future, Future, FutureExt, stream::{self, StreamExt}, SinkExt, }; + use futures_timer::Delay; use polkadot_primitives::v1::Hash; use std::{ collections::HashMap, convert::TryFrom, pin::Pin, + time::Duration, }; // basic usage: in a nutshell, when you want to define a subsystem, just focus on what its jobs do; @@ -884,15 +885,23 @@ mod tests { let subsystem = FakeCandidateSelectionSubsystem::run(context, run_args, pool, Some(err_tx)); let test_future = test(overseer_handle, err_rx); + let timeout = Delay::new(Duration::from_secs(2)); futures::pin_mut!(test_future); futures::pin_mut!(subsystem); + futures::pin_mut!(timeout); - executor::block_on(future::select(test_future, subsystem)); + executor::block_on(async move { + futures::select! { + _ = test_future.fuse() => (), + _ = subsystem.fuse() => (), + _ = timeout.fuse() => panic!("test timed out instead of completing"), + } + }); } #[test] - fn starting_job_works() { + fn starting_and_stopping_job_works() { let relay_parent: Hash = [0; 32].into(); let mut run_args = HashMap::new(); let test_message = format!("greetings from {}", relay_parent); @@ -904,20 +913,6 @@ mod tests { overseer_handle.recv().await, AllMessages::Test(msg) if msg == test_message ); - std::mem::drop(overseer_handle); - - let errs: Vec<_> = err_rx.collect().await; - assert_eq!(errs.len(), 0); - }); - } - - #[test] - fn stopping_running_job_works() { - let relay_parent: Hash = [0; 32].into(); - let run_args = HashMap::new(); - - test_harness(run_args, |mut overseer_handle, err_rx| async move { - overseer_handle.send(FromOverseer::Signal(OverseerSignal::StartWork(relay_parent))).await; overseer_handle.send(FromOverseer::Signal(OverseerSignal::StopWork(relay_parent))).await; let errs: Vec<_> = err_rx.collect().await; From 6672e2a9dfb789e35de43eb758b191016ff18e0e Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Mon, 20 Jul 2020 08:55:43 +0200 Subject: [PATCH 12/15] rename fwd_errors -> forward_errors --- node/subsystem/src/util.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/node/subsystem/src/util.rs b/node/subsystem/src/util.rs index d3104139cdc0..ab6e0ed88859 100644 --- a/node/subsystem/src/util.rs +++ b/node/subsystem/src/util.rs @@ -397,7 +397,7 @@ impl Jobs { /// the error is forwarded onto the provided channel. /// /// Errors if the error channel already exists. - pub fn fwd_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { + pub fn forward_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { if self.errors.is_some() { return Err(Error::AlreadyForwarding) } self.errors = Some(tx); Ok(()) @@ -548,7 +548,7 @@ where /// the error is forwarded onto the provided channel. /// /// Errors if the error channel already exists. - pub fn fwd_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { + pub fn forward_errors(&mut self, tx: mpsc::Sender<(Option, JobsError)>) -> Result<(), Error> { if self.errors.is_some() { return Err(Error::AlreadyForwarding) } self.errors = Some(tx); Ok(()) @@ -568,7 +568,7 @@ where pub async fn run(mut ctx: Context, run_args: Job::RunArgs, spawner: Spawner, mut err_tx: Option, JobsError)>>) { let mut jobs = Jobs::new(spawner.clone()); if let Some(ref err_tx) = err_tx { - jobs.fwd_errors(err_tx.clone()).expect("we never call this twice in this context; qed"); + jobs.forward_errors(err_tx.clone()).expect("we never call this twice in this context; qed"); } loop { From 6b056bf2e9e3860689061da18e260c3f93d852f9 Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Mon, 20 Jul 2020 09:01:25 +0200 Subject: [PATCH 13/15] warn on error propagation failure --- node/subsystem/src/util.rs | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/node/subsystem/src/util.rs b/node/subsystem/src/util.rs index ab6e0ed88859..04c05794d7d7 100644 --- a/node/subsystem/src/util.rs +++ b/node/subsystem/src/util.rs @@ -424,7 +424,10 @@ impl Jobs { if let Some(mut err_tx) = err_tx { // if we can't send the notification of error on the error channel, then // there's no point trying to propagate this error onto the channel too - let _ = err_tx.send((Some(parent_hash), JobsError::Job(e))).await; + // all we can do is warn that error propagatio has failed + if let Err(e) = err_tx.send((Some(parent_hash), JobsError::Job(e))).await { + log::warn!("failed to forward error: {:?}", e); + } } } }); @@ -583,8 +586,11 @@ where // if we have a channel on which to forward errors, do so async fn fwd_err(hash: Option, err: JobsError, err_tx: &mut Option, JobsError)>>) { if let Some(err_tx) = err_tx { - // if we can't send on the error transmission channel, we can't do anything about it - let _ = err_tx.send((hash, err)).await; + // if we can't send on the error transmission channel, we can't do anything useful about it + // still, we can at least log the failure + if let Err(e) = err_tx.send((hash, err)).await { + log::warn!("failed to forward error: {:?}", e); + } } } From bcd0a6d2411629b3cd16bfa6ebecef84630eabf4 Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Mon, 20 Jul 2020 09:52:21 +0200 Subject: [PATCH 14/15] fix unused import leftover from merge --- node/network/pov-distribution/Cargo.toml | 1 + node/network/pov-distribution/src/lib.rs | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/node/network/pov-distribution/Cargo.toml b/node/network/pov-distribution/Cargo.toml index 845f6389daab..672e697c268f 100644 --- a/node/network/pov-distribution/Cargo.toml +++ b/node/network/pov-distribution/Cargo.toml @@ -20,3 +20,4 @@ polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsys parking_lot = "0.10.0" assert_matches = "1.3.0" sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" } +polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem", features = [ "test-helpers" ] } diff --git a/node/network/pov-distribution/src/lib.rs b/node/network/pov-distribution/src/lib.rs index 9883450df398..aa37cfdc3fa9 100644 --- a/node/network/pov-distribution/src/lib.rs +++ b/node/network/pov-distribution/src/lib.rs @@ -551,7 +551,7 @@ async fn run( #[cfg(test)] mod tests { use super::*; - use futures::executor::{self, ThreadPool}; + use futures::executor; use polkadot_primitives::v1::BlockData; use assert_matches::assert_matches; From 5b464b0b4d90e9728faffcfc6d6fcda36d3d97ea Mon Sep 17 00:00:00 2001 From: Peter Goodspeed-Niklaus Date: Mon, 20 Jul 2020 13:20:26 +0200 Subject: [PATCH 15/15] derive eq for subsystemerror --- node/subsystem/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/node/subsystem/src/lib.rs b/node/subsystem/src/lib.rs index ce076577e922..430a8418d90b 100644 --- a/node/subsystem/src/lib.rs +++ b/node/subsystem/src/lib.rs @@ -73,7 +73,7 @@ pub enum FromOverseer { /// * Subsystems dying when they are not expected to /// * Subsystems not dying when they are told to die /// * etc. -#[derive(Debug, PartialEq)] +#[derive(Debug, PartialEq, Eq)] pub struct SubsystemError; impl From for SubsystemError {