cargo +nightly fmt (#3540)

* cargo +nightly fmt

* add cargo-fmt check to ci

* update ci

* fmt

* fmt

* skip macro

* ignore bridges
This commit is contained in:
Shawn Tabrizi
2021-08-02 12:47:33 +02:00
committed by GitHub
parent 30e3012270
commit ff5d56fb76
350 changed files with 20617 additions and 21266 deletions
@@ -19,7 +19,7 @@ use std::collections::HashSet;
use proc_macro2::TokenStream;
use quote::quote;
use syn::{Error, GenericParam, Ident, Result, Type, parse2};
use syn::{parse2, Error, GenericParam, Ident, Result, Type};
#[proc_macro_derive(AllSubsystemsGen)]
pub fn subsystems_gen(item: proc_macro::TokenStream) -> proc_macro::TokenStream {
@@ -40,53 +40,68 @@ fn impl_subsystems_gen(item: TokenStream) -> Result<proc_macro2::TokenStream> {
}
let mut orig_generics = ds.generics;
// remove default types
orig_generics.params = orig_generics.params.into_iter().map(|mut generic| {
match generic {
GenericParam::Type(ref mut param) => {
param.eq_token = None;
param.default = None;
orig_generics.params = orig_generics
.params
.into_iter()
.map(|mut generic| {
match generic {
GenericParam::Type(ref mut param) => {
param.eq_token = None;
param.default = None;
},
_ => {},
}
_ => {}
}
generic
}).collect();
generic
})
.collect();
// prepare a hashmap of generic type to member that uses it
let generic_types = orig_generics.params.iter().filter_map(|generic| {
if let GenericParam::Type(param) = generic {
Some(param.ident.clone())
} else {
None
}
}).collect::<HashSet<Ident>>();
let generic_types = orig_generics
.params
.iter()
.filter_map(|generic| {
if let GenericParam::Type(param) = generic {
Some(param.ident.clone())
} else {
None
}
})
.collect::<HashSet<Ident>>();
let strukt_ty = ds.ident;
if generic_types.is_empty() {
return Err(Error::new(strukt_ty.span(), "struct must have at least one generic parameter."))
return Err(Error::new(
strukt_ty.span(),
"struct must have at least one generic parameter.",
))
}
// collect all fields that exist, and all fields that are replaceable
let mut replacable_items = Vec::<NameTyTup>::with_capacity(64);
let mut all_fields = replacable_items.clone();
let mut duplicate_generic_detection = HashSet::<Ident>::with_capacity(64);
for field in named.named {
let field_ident = field.ident.clone().ok_or_else(|| Error::new(span, "Member field must have a name."))?;
let field_ident = field
.ident
.clone()
.ok_or_else(|| Error::new(span, "Member field must have a name."))?;
let ty = field.ty.clone();
let ntt = NameTyTup { field: field_ident, ty };
replacable_items.push(ntt.clone());
// assure every generic is used exactly once
let ty_ident = match field.ty {
Type::Path(path) => path.path.get_ident().cloned().ok_or_else(|| {
Error::new(proc_macro2::Span::call_site(), "Expected an identifier, but got a path.")
Error::new(
proc_macro2::Span::call_site(),
"Expected an identifier, but got a path.",
)
}),
_ => return Err(Error::new(proc_macro2::Span::call_site(), "Must be path."))
_ => return Err(Error::new(proc_macro2::Span::call_site(), "Must be path.")),
}?;
if generic_types.contains(&ty_ident) {
@@ -98,34 +113,43 @@ fn impl_subsystems_gen(item: TokenStream) -> Result<proc_macro2::TokenStream> {
all_fields.push(ntt);
}
let msg = "Generated by #[derive(AllSubsystemsGen)] derive proc-macro.";
let mut additive = TokenStream::new();
// generate an impl of `fn replace_#name`
for NameTyTup { field: replacable_item, ty: replacable_item_ty } in replacable_items {
let keeper = all_fields.iter().filter(|ntt| ntt.field != replacable_item).map(|ntt| ntt.field.clone());
let keeper = all_fields
.iter()
.filter(|ntt| ntt.field != replacable_item)
.map(|ntt| ntt.field.clone());
let strukt_ty = strukt_ty.clone();
let fname = Ident::new(&format!("replace_{}", replacable_item), span);
// adjust the generics such that the appropriate member type is replaced
let mut modified_generics = orig_generics.clone();
modified_generics.params = modified_generics.params.into_iter().map(|mut generic| {
match generic {
GenericParam::Type(ref mut param) => {
param.eq_token = None;
param.default = None;
if match &replacable_item_ty {
Type::Path(path) =>
path.path.get_ident().filter(|&ident| ident == &param.ident).is_some(),
_ => false
} {
param.ident = Ident::new("NEW", span);
}
modified_generics.params = modified_generics
.params
.into_iter()
.map(|mut generic| {
match generic {
GenericParam::Type(ref mut param) => {
param.eq_token = None;
param.default = None;
if match &replacable_item_ty {
Type::Path(path) => path
.path
.get_ident()
.filter(|&ident| ident == &param.ident)
.is_some(),
_ => false,
} {
param.ident = Ident::new("NEW", span);
}
},
_ => {},
}
_ => {}
}
generic
}).collect();
generic
})
.collect();
additive.extend(quote! {
impl #orig_generics #strukt_ty #orig_generics {
@@ -143,11 +167,13 @@ fn impl_subsystems_gen(item: TokenStream) -> Result<proc_macro2::TokenStream> {
}
Ok(additive)
}
syn::Fields::Unit => Err(Error::new(span, "Must be a struct with named fields. Not an unit struct.")),
syn::Fields::Unnamed(_) => {
Err(Error::new(span, "Must be a struct with named fields. Not an unnamed fields struct."))
}
},
syn::Fields::Unit =>
Err(Error::new(span, "Must be a struct with named fields. Not an unit struct.")),
syn::Fields::Unnamed(_) => Err(Error::new(
span,
"Must be a struct with named fields. Not an unnamed fields struct.",
)),
}
}
@@ -18,39 +18,28 @@
//! * Spawning subsystems and subsystem child jobs
//! * Establishing message passing
use std::time::Duration;
use futures::{
channel::oneshot,
pending, pin_mut, select, stream,
FutureExt, StreamExt,
};
use futures::{channel::oneshot, pending, pin_mut, select, stream, FutureExt, StreamExt};
use futures_timer::Delay;
use std::time::Duration;
use polkadot_node_primitives::{PoV, BlockData};
use polkadot_primitives::v1::Hash;
use polkadot_node_primitives::{BlockData, PoV};
use polkadot_node_subsystem_types::messages::{
CandidateBackingMessage, CandidateValidationMessage,
};
use polkadot_overseer::{
self as overseer,
AllMessages,
AllSubsystems,
HeadSupportsParachains,
Overseer,
OverseerSignal,
SubsystemError,
gen::{
FromOverseer,
SpawnedSubsystem,
},
};
use polkadot_node_subsystem_types::messages::{
CandidateValidationMessage, CandidateBackingMessage,
gen::{FromOverseer, SpawnedSubsystem},
AllMessages, AllSubsystems, HeadSupportsParachains, Overseer, OverseerSignal, SubsystemError,
};
use polkadot_primitives::v1::Hash;
struct AlwaysSupportsParachains;
impl HeadSupportsParachains for AlwaysSupportsParachains {
fn head_supports_parachains(&self, _head: &Hash) -> bool { true }
fn head_supports_parachains(&self, _head: &Hash) -> bool {
true
}
}
////////
struct Subsystem1;
@@ -58,7 +47,11 @@ struct Subsystem1;
impl Subsystem1 {
async fn run<Ctx>(mut ctx: Ctx) -> ()
where
Ctx: overseer::SubsystemContext<Message=CandidateBackingMessage,AllMessages=AllMessages,Signal=OverseerSignal>,
Ctx: overseer::SubsystemContext<
Message = CandidateBackingMessage,
AllMessages = AllMessages,
Signal = OverseerSignal,
>,
{
'louy: loop {
match ctx.try_recv().await {
@@ -66,13 +59,13 @@ impl Subsystem1 {
if let FromOverseer::Communication { msg } = msg {
tracing::info!("msg {:?}", msg);
}
continue 'louy;
}
continue 'louy
},
Ok(None) => (),
Err(_) => {
tracing::info!("exiting");
break 'louy;
}
break 'louy
},
}
Delay::new(Duration::from_secs(1)).await;
@@ -80,21 +73,23 @@ impl Subsystem1 {
let msg = CandidateValidationMessage::ValidateFromChainState(
Default::default(),
PoV {
block_data: BlockData(Vec::new()),
}.into(),
PoV { block_data: BlockData(Vec::new()) }.into(),
tx,
);
ctx.send_message(<Ctx as overseer::SubsystemContext>::AllMessages::from(msg)).await;
ctx.send_message(<Ctx as overseer::SubsystemContext>::AllMessages::from(msg))
.await;
}
()
}
}
impl<Context> overseer::Subsystem<Context,SubsystemError> for Subsystem1
impl<Context> overseer::Subsystem<Context, SubsystemError> for Subsystem1
where
Context: overseer::SubsystemContext<Message=CandidateBackingMessage,AllMessages=AllMessages,Signal=OverseerSignal>,
Context: overseer::SubsystemContext<
Message = CandidateBackingMessage,
AllMessages = AllMessages,
Signal = OverseerSignal,
>,
{
fn start(self, ctx: Context) -> SpawnedSubsystem<SubsystemError> {
let future = Box::pin(async move {
@@ -102,10 +97,7 @@ where
Ok(())
});
SpawnedSubsystem {
name: "subsystem-1",
future,
}
SpawnedSubsystem { name: "subsystem-1", future }
}
}
@@ -116,7 +108,11 @@ struct Subsystem2;
impl Subsystem2 {
async fn run<Ctx>(mut ctx: Ctx)
where
Ctx: overseer::SubsystemContext<Message=CandidateValidationMessage,AllMessages=AllMessages,Signal=OverseerSignal>,
Ctx: overseer::SubsystemContext<
Message = CandidateValidationMessage,
AllMessages = AllMessages,
Signal = OverseerSignal,
>,
{
ctx.spawn(
"subsystem-2-job",
@@ -126,27 +122,34 @@ impl Subsystem2 {
Delay::new(Duration::from_secs(1)).await;
}
}),
).unwrap();
)
.unwrap();
loop {
match ctx.try_recv().await {
Ok(Some(msg)) => {
tracing::info!("Subsystem2 received message {:?}", msg);
continue;
}
Ok(None) => { pending!(); }
continue
},
Ok(None) => {
pending!();
},
Err(_) => {
tracing::info!("exiting");
return;
return
},
}
}
}
}
impl<Context> overseer::Subsystem<Context,SubsystemError> for Subsystem2
impl<Context> overseer::Subsystem<Context, SubsystemError> for Subsystem2
where
Context: overseer::SubsystemContext<Message=CandidateValidationMessage,AllMessages=AllMessages,Signal=OverseerSignal>,
Context: overseer::SubsystemContext<
Message = CandidateValidationMessage,
AllMessages = AllMessages,
Signal = OverseerSignal,
>,
{
fn start(self, ctx: Context) -> SpawnedSubsystem<SubsystemError> {
let future = Box::pin(async move {
@@ -154,10 +157,7 @@ where
Ok(())
});
SpawnedSubsystem {
name: "subsystem-2",
future,
}
SpawnedSubsystem { name: "subsystem-2", future }
}
}
@@ -171,16 +171,10 @@ fn main() {
let all_subsystems = AllSubsystems::<()>::dummy()
.replace_candidate_validation(Subsystem2)
.replace_candidate_backing(Subsystem1)
;
.replace_candidate_backing(Subsystem1);
let (overseer, _handle) = Overseer::new(
vec![],
all_subsystems,
None,
AlwaysSupportsParachains,
spawner,
).unwrap();
let (overseer, _handle) =
Overseer::new(vec![], all_subsystems, None, AlwaysSupportsParachains, spawner).unwrap();
let overseer_fut = overseer.run().fuse();
let timer_stream = timer_stream;
@@ -1,15 +1,14 @@
//! A dummy to be used with cargo expand
use polkadot_overseer_gen::*;
use polkadot_node_network_protocol::WrongVariant;
use polkadot_overseer_gen::*;
/// Concrete subsystem implementation for `MsgStrukt` msg type.
#[derive(Default)]
pub struct AwesomeSubSys;
impl ::polkadot_overseer_gen::Subsystem<XxxSubsystemContext<MsgStrukt>, Yikes> for AwesomeSubSys {
fn start(self, _ctx: XxxSubsystemContext<MsgStrukt>) -> SpawnedSubsystem < Yikes > {
impl ::polkadot_overseer_gen::Subsystem<XxxSubsystemContext<MsgStrukt>, Yikes> for AwesomeSubSys {
fn start(self, _ctx: XxxSubsystemContext<MsgStrukt>) -> SpawnedSubsystem<Yikes> {
unimplemented!("starting yay!")
}
}
@@ -18,22 +17,19 @@ impl ::polkadot_overseer_gen::Subsystem<XxxSubsystemContext<MsgStrukt>, Yikes> f
pub struct GoblinTower;
impl ::polkadot_overseer_gen::Subsystem<XxxSubsystemContext<Plinko>, Yikes> for GoblinTower {
fn start(self, _ctx: XxxSubsystemContext<Plinko>) -> SpawnedSubsystem < Yikes > {
fn start(self, _ctx: XxxSubsystemContext<Plinko>) -> SpawnedSubsystem<Yikes> {
unimplemented!("welcum")
}
}
/// A signal sent by the overseer.
#[derive(Debug, Clone)]
pub struct SigSigSig;
/// The external event.
#[derive(Debug, Clone)]
pub struct EvX;
impl EvX {
pub fn focus<'a, T>(&'a self) -> Result<EvX, ()> {
unimplemented!("dispatch")
@@ -75,7 +71,6 @@ impl From<NetworkMsg> for MsgStrukt {
}
}
#[derive(Debug, Clone, Copy)]
pub enum NetworkMsg {
A,
@@ -83,18 +78,15 @@ pub enum NetworkMsg {
C,
}
impl NetworkMsg {
fn focus(&self) -> Result<Self, WrongVariant> {
Ok(match self {
Self::B => return Err(WrongVariant),
Self::A | Self::C => self.clone()
Self::A | Self::C => self.clone(),
})
}
}
#[overlord(signal=SigSigSig, event=EvX, error=Yikes, network=NetworkMsg, gen=AllMessages)]
struct Xxx {
#[subsystem(MsgStrukt)]
@@ -109,7 +101,7 @@ struct Xxx {
#[derive(Debug, Clone)]
struct DummySpawner;
impl SpawnNamed for DummySpawner{
impl SpawnNamed for DummySpawner {
fn spawn_blocking(&self, name: &'static str, _future: futures::future::BoxFuture<'static, ()>) {
unimplemented!("spawn blocking {}", name)
}
@@ -80,10 +80,8 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
let consumes = &info.consumes();
let subsyste_ctx_name = Ident::new(
&(overseer_name.to_string() + "SubsystemContext"),
overseer_name.span()
);
let subsyste_ctx_name =
Ident::new(&(overseer_name.to_string() + "SubsystemContext"), overseer_name.span());
let builder_where_clause = quote! {
where
@@ -15,7 +15,7 @@
// along with Polkadot. If not, see <http://www.gnu.org/licenses/>.
use super::*;
use proc_macro2::{TokenStream, Ident};
use proc_macro2::{Ident, TokenStream};
use quote::quote;
use syn::Path;
@@ -15,8 +15,7 @@
// along with Polkadot. If not, see <http://www.gnu.org/licenses/>.
use quote::quote;
use syn::Result;
use syn::spanned::Spanned;
use syn::{spanned::Spanned, Result};
use super::*;
@@ -30,21 +29,24 @@ pub(crate) fn impl_message_wrapper_enum(info: &OverseerInfo) -> Result<proc_macr
let message_wrapper = &info.message_wrapper;
let (outgoing_from_impl, outgoing_decl) = if let Some(outgoing) = outgoing {
let outgoing_variant = outgoing
.get_ident()
.ok_or_else(||{
syn::Error::new(outgoing.span(), "Missing identifier to use as enum variant for outgoing.")
})?;
(quote! {
impl ::std::convert::From< #outgoing > for #message_wrapper {
fn from(message: #outgoing) -> Self {
#message_wrapper :: #outgoing_variant ( message )
let outgoing_variant = outgoing.get_ident().ok_or_else(|| {
syn::Error::new(
outgoing.span(),
"Missing identifier to use as enum variant for outgoing.",
)
})?;
(
quote! {
impl ::std::convert::From< #outgoing > for #message_wrapper {
fn from(message: #outgoing) -> Self {
#message_wrapper :: #outgoing_variant ( message )
}
}
}
},
quote! {
#outgoing_variant ( #outgoing ) ,
})
},
quote! {
#outgoing_variant ( #outgoing ) ,
},
)
} else {
(TokenStream::new(), TokenStream::new())
};
@@ -23,8 +23,10 @@ use super::*;
/// which acts as the gateway to constructing the overseer.
pub(crate) fn impl_misc(info: &OverseerInfo) -> proc_macro2::TokenStream {
let overseer_name = info.overseer_name.clone();
let subsystem_sender_name = Ident::new(&(overseer_name.to_string() + "SubsystemSender"), overseer_name.span());
let subsystem_ctx_name = Ident::new(&(overseer_name.to_string() + "SubsystemContext"), overseer_name.span());
let subsystem_sender_name =
Ident::new(&(overseer_name.to_string() + "SubsystemSender"), overseer_name.span());
let subsystem_ctx_name =
Ident::new(&(overseer_name.to_string() + "SubsystemContext"), overseer_name.span());
let consumes = &info.consumes();
let signal = &info.extern_signal_ty;
let wrapper_message = &info.message_wrapper;
@@ -52,7 +52,8 @@ pub(crate) fn impl_overseer_struct(info: &OverseerInfo) -> proc_macro2::TokenStr
let message_channel_capacity = info.message_channel_capacity;
let signal_channel_capacity = info.signal_channel_capacity;
let log_target = syn::LitStr::new(overseer_name.to_string().to_lowercase().as_str(), overseer_name.span());
let log_target =
syn::LitStr::new(overseer_name.to_string().to_lowercase().as_str(), overseer_name.span());
let ts = quote! {
const STOP_DELAY: ::std::time::Duration = ::std::time::Duration::from_secs(1);
@@ -16,18 +16,18 @@
#![deny(unused_crate_dependencies)]
use proc_macro2::{Span, Ident, TokenStream};
use syn::{parse2, Result};
use proc_macro2::{Ident, Span, TokenStream};
use quote::{quote, ToTokens};
use syn::{parse2, Result};
mod impl_builder;
mod impl_channels_out;
mod impl_dispatch;
mod impl_message_wrapper;
mod impl_misc;
mod impl_overseer;
mod parse_attr;
mod parse_struct;
mod impl_channels_out;
mod impl_dispatch;
mod impl_message_wrapper;
use impl_builder::*;
use impl_channels_out::*;
@@ -42,28 +42,36 @@ use parse_struct::*;
mod tests;
#[proc_macro_attribute]
pub fn overlord(attr: proc_macro::TokenStream, item: proc_macro::TokenStream) -> proc_macro::TokenStream {
pub fn overlord(
attr: proc_macro::TokenStream,
item: proc_macro::TokenStream,
) -> proc_macro::TokenStream {
let attr: TokenStream = attr.into();
let item: TokenStream = item.into();
impl_overseer_gen(attr, item).unwrap_or_else(|err| err.to_compile_error()).into()
impl_overseer_gen(attr, item)
.unwrap_or_else(|err| err.to_compile_error())
.into()
}
pub(crate) fn impl_overseer_gen(attr: TokenStream, orig: TokenStream) -> Result<proc_macro2::TokenStream> {
pub(crate) fn impl_overseer_gen(
attr: TokenStream,
orig: TokenStream,
) -> Result<proc_macro2::TokenStream> {
let args: AttrArgs = parse2(attr)?;
let message_wrapper = args.message_wrapper;
let of: OverseerGuts = parse2(orig)?;
let support_crate_name = if cfg!(test) {
quote!{crate}
quote! {crate}
} else {
use proc_macro_crate::{crate_name, FoundCrate};
let crate_name = crate_name("polkadot-overseer-gen")
.expect("Support crate polkadot-overseer-gen is present in `Cargo.toml`. qed");
match crate_name {
FoundCrate::Itself => quote!{crate},
FoundCrate::Name(name) => Ident::new(&name, Span::call_site()).to_token_stream(),
}
FoundCrate::Itself => quote! {crate},
FoundCrate::Name(name) => Ident::new(&name, Span::call_site()).to_token_stream(),
}
};
let info = OverseerInfo {
support_crate_name,
@@ -15,12 +15,14 @@
// along with Polkadot. If not, see <http://www.gnu.org/licenses/>.
use proc_macro2::Span;
use std::collections::{hash_map::RandomState, HashMap};
use syn::parse::{Parse, ParseBuffer};
use syn::punctuated::Punctuated;
use syn::spanned::Spanned;
use syn::{Error, Ident, LitInt, Path, Result, Token};
use quote::{quote, ToTokens};
use std::collections::{hash_map::RandomState, HashMap};
use syn::{
parse::{Parse, ParseBuffer},
punctuated::Punctuated,
spanned::Spanned,
Error, Ident, LitInt, Path, Result, Token,
};
mod kw {
syn::custom_keyword!(event);
@@ -33,62 +35,45 @@ mod kw {
syn::custom_keyword!(message_capacity);
}
#[derive(Clone, Debug)]
enum OverseerAttrItem {
ExternEventType {
tag: kw::event,
eq_token: Token![=],
value: Path
},
ExternNetworkType {
tag: kw::network,
eq_token: Token![=],
value: Path
},
ExternOverseerSignalType {
tag: kw::signal,
eq_token: Token![=],
value: Path
},
ExternErrorType {
tag: kw::error,
eq_token: Token![=],
value: Path
},
OutgoingType {
tag: kw::outgoing,
eq_token: Token![=],
value: Path
},
MessageWrapperName {
tag: kw::gen,
eq_token: Token![=],
value: Ident
},
SignalChannelCapacity {
tag: kw::signal_capacity,
eq_token: Token![=],
value: usize
},
MessageChannelCapacity {
tag: kw::message_capacity,
eq_token: Token![=],
value: usize
},
ExternEventType { tag: kw::event, eq_token: Token![=], value: Path },
ExternNetworkType { tag: kw::network, eq_token: Token![=], value: Path },
ExternOverseerSignalType { tag: kw::signal, eq_token: Token![=], value: Path },
ExternErrorType { tag: kw::error, eq_token: Token![=], value: Path },
OutgoingType { tag: kw::outgoing, eq_token: Token![=], value: Path },
MessageWrapperName { tag: kw::gen, eq_token: Token![=], value: Ident },
SignalChannelCapacity { tag: kw::signal_capacity, eq_token: Token![=], value: usize },
MessageChannelCapacity { tag: kw::message_capacity, eq_token: Token![=], value: usize },
}
impl ToTokens for OverseerAttrItem {
fn to_tokens(&self, tokens: &mut proc_macro2::TokenStream) {
let ts = match self {
Self::ExternEventType { tag, eq_token, value } => { quote!{ #tag #eq_token, #value } }
Self::ExternNetworkType { tag, eq_token, value } => { quote!{ #tag #eq_token, #value } }
Self::ExternOverseerSignalType { tag, eq_token, value } => { quote!{ #tag #eq_token, #value } }
Self::ExternErrorType { tag, eq_token, value } => { quote!{ #tag #eq_token, #value } }
Self::OutgoingType { tag, eq_token, value } => { quote!{ #tag #eq_token, #value } }
Self::MessageWrapperName { tag, eq_token, value } => { quote!{ #tag #eq_token, #value } }
Self::SignalChannelCapacity { tag, eq_token, value } => { quote!{ #tag #eq_token, #value } }
Self::MessageChannelCapacity { tag, eq_token, value } => { quote!{ #tag #eq_token, #value } }
Self::ExternEventType { tag, eq_token, value } => {
quote! { #tag #eq_token, #value }
},
Self::ExternNetworkType { tag, eq_token, value } => {
quote! { #tag #eq_token, #value }
},
Self::ExternOverseerSignalType { tag, eq_token, value } => {
quote! { #tag #eq_token, #value }
},
Self::ExternErrorType { tag, eq_token, value } => {
quote! { #tag #eq_token, #value }
},
Self::OutgoingType { tag, eq_token, value } => {
quote! { #tag #eq_token, #value }
},
Self::MessageWrapperName { tag, eq_token, value } => {
quote! { #tag #eq_token, #value }
},
Self::SignalChannelCapacity { tag, eq_token, value } => {
quote! { #tag #eq_token, #value }
},
Self::MessageChannelCapacity { tag, eq_token, value } => {
quote! { #tag #eq_token, #value }
},
};
tokens.extend(ts.into_iter());
}
@@ -137,7 +122,7 @@ impl Parse for OverseerAttrItem {
Ok(OverseerAttrItem::SignalChannelCapacity {
tag: input.parse::<kw::signal_capacity>()?,
eq_token: input.parse()?,
value: input.parse::<LitInt>()?.base10_parse::<usize>()?
value: input.parse::<LitInt>()?.base10_parse::<usize>()?,
})
} else if lookahead.peek(kw::message_capacity) {
Ok(OverseerAttrItem::MessageChannelCapacity {
@@ -169,42 +154,47 @@ pub(crate) struct AttrArgs {
macro_rules! extract_variant {
($unique:expr, $variant:ident ; default = $fallback:expr) => {
extract_variant!($unique, $variant)
.unwrap_or_else(|| { $fallback })
extract_variant!($unique, $variant).unwrap_or_else(|| $fallback)
};
($unique:expr, $variant:ident ; err = $err:expr) => {
extract_variant!($unique, $variant)
.ok_or_else(|| {
Error::new(Span::call_site(), $err)
})
extract_variant!($unique, $variant).ok_or_else(|| Error::new(Span::call_site(), $err))
};
($unique:expr, $variant:ident) => {
$unique.values()
.find_map(|item| {
if let OverseerAttrItem:: $variant { value, ..} = item {
Some(value.clone())
} else {
None
}
})
$unique.values().find_map(|item| {
if let OverseerAttrItem::$variant { value, .. } = item {
Some(value.clone())
} else {
None
}
})
};
}
impl Parse for AttrArgs {
fn parse(input: &ParseBuffer) -> Result<Self> {
let items: Punctuated<OverseerAttrItem, Token![,]> = input.parse_terminated(OverseerAttrItem::parse)?;
let items: Punctuated<OverseerAttrItem, Token![,]> =
input.parse_terminated(OverseerAttrItem::parse)?;
let mut unique = HashMap::<std::mem::Discriminant<OverseerAttrItem>, OverseerAttrItem, RandomState>::default();
let mut unique = HashMap::<
std::mem::Discriminant<OverseerAttrItem>,
OverseerAttrItem,
RandomState,
>::default();
for item in items {
if let Some(first) = unique.insert(std::mem::discriminant(&item), item.clone()) {
let mut e = Error::new(item.span(), format!("Duplicate definition of overseer generation type found"));
let mut e = Error::new(
item.span(),
format!("Duplicate definition of overseer generation type found"),
);
e.combine(Error::new(first.span(), "previously defined here."));
return Err(e);
return Err(e)
}
}
let signal_channel_capacity = extract_variant!(unique, SignalChannelCapacity; default = 64_usize);
let message_channel_capacity = extract_variant!(unique, MessageChannelCapacity; default = 1024_usize);
let signal_channel_capacity =
extract_variant!(unique, SignalChannelCapacity; default = 64_usize);
let message_channel_capacity =
extract_variant!(unique, MessageChannelCapacity; default = 1024_usize);
let error = extract_variant!(unique, ExternErrorType; err = "Must declare the overseer error type via `error=..`.")?;
let event = extract_variant!(unique, ExternEventType; err = "Must declare the overseer event type via `event=..`.")?;
@@ -15,13 +15,13 @@
// along with Polkadot. If not, see <http://www.gnu.org/licenses/>.
use proc_macro2::{Span, TokenStream};
use std::collections::{hash_map::RandomState, HashSet, HashMap};
use syn::punctuated::Punctuated;
use syn::spanned::Spanned;
use syn::parse::{Parse, ParseStream};
use std::collections::{hash_map::RandomState, HashMap, HashSet};
use syn::{
Attribute, Field, FieldsNamed, Ident, Token, Type, AttrStyle, Path,
Error, GenericParam, ItemStruct, Result, Visibility
parse::{Parse, ParseStream},
punctuated::Punctuated,
spanned::Spanned,
AttrStyle, Attribute, Error, Field, FieldsNamed, GenericParam, Ident, ItemStruct, Path, Result,
Token, Type, Visibility,
};
use quote::{quote, ToTokens};
@@ -32,7 +32,6 @@ mod kw {
syn::custom_keyword!(blocking);
}
#[derive(Clone, Debug)]
enum SubSysAttrItem {
/// The subsystem is still a work in progress
@@ -64,15 +63,20 @@ impl Parse for SubSysAttrItem {
impl ToTokens for SubSysAttrItem {
fn to_tokens(&self, tokens: &mut proc_macro2::TokenStream) {
let ts = match self {
Self::Wip(wip) => { quote!{ #wip } }
Self::Blocking(blocking) => { quote!{ #blocking } }
Self::NoDispatch(no_dispatch) => { quote!{ #no_dispatch } }
Self::Wip(wip) => {
quote! { #wip }
},
Self::Blocking(blocking) => {
quote! { #blocking }
},
Self::NoDispatch(no_dispatch) => {
quote! { #no_dispatch }
},
};
tokens.extend(ts.into_iter());
}
}
/// A field of the struct annotated with
/// `#[subsystem(no_dispatch, , A | B | C)]`
#[derive(Clone, Debug)]
@@ -106,28 +110,22 @@ fn try_type_to_path(ty: Type, span: Span) -> Result<Path> {
macro_rules! extract_variant {
($unique:expr, $variant:ident ; default = $fallback:expr) => {
extract_variant!($unique, $variant)
.unwrap_or_else(|| { $fallback })
extract_variant!($unique, $variant).unwrap_or_else(|| $fallback)
};
($unique:expr, $variant:ident ; err = $err:expr) => {
extract_variant!($unique, $variant)
.ok_or_else(|| {
Error::new(Span::call_site(), $err)
})
extract_variant!($unique, $variant).ok_or_else(|| Error::new(Span::call_site(), $err))
};
($unique:expr, $variant:ident) => {
$unique.values()
.find_map(|item| {
if let SubSysAttrItem:: $variant ( _ ) = item {
Some(true)
} else {
None
}
})
$unique.values().find_map(|item| {
if let SubSysAttrItem::$variant(_) = item {
Some(true)
} else {
None
}
})
};
}
pub(crate) struct SubSystemTags {
#[allow(dead_code)]
pub(crate) attrs: Vec<Attribute>,
@@ -158,12 +156,19 @@ impl Parse for SubSystemTags {
let consumes = content.parse::<Path>()?;
let mut unique = HashMap::<std::mem::Discriminant<SubSysAttrItem>, SubSysAttrItem, RandomState>::default();
let mut unique = HashMap::<
std::mem::Discriminant<SubSysAttrItem>,
SubSysAttrItem,
RandomState,
>::default();
for item in items {
if let Some(first) = unique.insert(std::mem::discriminant(&item), item.clone()) {
let mut e = Error::new(item.span(), format!("Duplicate definition of subsystem attribute found"));
let mut e = Error::new(
item.span(),
format!("Duplicate definition of subsystem attribute found"),
);
e.combine(Error::new(first.span(), "previously defined here."));
return Err(e);
return Err(e)
}
}
@@ -228,10 +233,7 @@ impl OverseerInfo {
}
pub(crate) fn variant_names(&self) -> Vec<Ident> {
self.subsystems
.iter()
.map(|ssf| ssf.generic.clone())
.collect::<Vec<_>>()
self.subsystems.iter().map(|ssf| ssf.generic.clone()).collect::<Vec<_>>()
}
pub(crate) fn variant_names_without_wip(&self) -> Vec<Ident> {
@@ -272,13 +274,8 @@ impl OverseerInfo {
self.baggage
.iter()
.map(|bag| {
let BaggageField {
vis,
field_ty,
field_name,
..
} = bag;
quote!{ #vis #field_name: #field_ty }
let BaggageField { vis, field_ty, field_name, .. } = bag;
quote! { #vis #field_name: #field_ty }
})
.collect::<Vec<TokenStream>>()
}
@@ -330,7 +327,11 @@ pub(crate) struct OverseerGuts {
}
impl OverseerGuts {
pub(crate) fn parse_fields(name: Ident, baggage_generics: HashSet<Ident>, fields: FieldsNamed) -> Result<Self> {
pub(crate) fn parse_fields(
name: Ident,
baggage_generics: HashSet<Ident>,
fields: FieldsNamed,
) -> Result<Self> {
let n = fields.named.len();
let mut subsystems = Vec::with_capacity(n);
let mut baggage = Vec::with_capacity(n);
@@ -340,14 +341,16 @@ impl OverseerGuts {
// for the builder pattern besides other places.
let mut unique_subsystem_idents = HashSet::<Ident>::new();
for Field { attrs, vis, ident, ty, .. } in fields.named.into_iter() {
let mut consumes = attrs.iter().filter(|attr| attr.style == AttrStyle::Outer).filter_map(|attr| {
let span = attr.path.span();
attr.path.get_ident().filter(|ident| *ident == "subsystem").map(move |_ident| {
let attr_tokens = attr.tokens.clone();
(attr_tokens, span)
})
});
let ident = ident.ok_or_else(|| Error::new(ty.span(), "Missing identifier for member. BUG"))?;
let mut consumes =
attrs.iter().filter(|attr| attr.style == AttrStyle::Outer).filter_map(|attr| {
let span = attr.path.span();
attr.path.get_ident().filter(|ident| *ident == "subsystem").map(move |_ident| {
let attr_tokens = attr.tokens.clone();
(attr_tokens, span)
})
});
let ident =
ident.ok_or_else(|| Error::new(ty.span(), "Missing identifier for member. BUG"))?;
if let Some((attr_tokens, span)) = consumes.next() {
if let Some((_attr_tokens2, span2)) = consumes.next() {
@@ -355,7 +358,7 @@ impl OverseerGuts {
let mut err = Error::new(span, "The first subsystem annotation is at");
err.combine(Error::new(span2, "but another here for the same field."));
err
});
})
}
let mut consumes_paths = Vec::with_capacity(attrs.len());
let attr_tokens = attr_tokens.clone();
@@ -363,9 +366,17 @@ impl OverseerGuts {
consumes_paths.push(variant.consumes);
let field_ty = try_type_to_path(ty, span)?;
let generic = field_ty.get_ident().ok_or_else(|| Error::new(field_ty.span(), "Must be an identifier, not a path."))?.clone();
let generic = field_ty
.get_ident()
.ok_or_else(|| {
Error::new(field_ty.span(), "Must be an identifier, not a path.")
})?
.clone();
if let Some(previous) = unique_subsystem_idents.get(&generic) {
let mut e = Error::new(generic.span(), format!("Duplicate subsystem names `{}`", generic));
let mut e = Error::new(
generic.span(),
format!("Duplicate subsystem names `{}`", generic),
);
e.combine(Error::new(previous.span(), "previously defined here."));
return Err(e)
}
@@ -381,7 +392,10 @@ impl OverseerGuts {
});
} else {
let field_ty = try_type_to_path(ty, ident.span())?;
let generic = field_ty.get_ident().map(|ident| baggage_generics.contains(ident)).unwrap_or_default();
let generic = field_ty
.get_ident()
.map(|ident| baggage_generics.contains(ident))
.unwrap_or_default();
baggage.push(BaggageField { field_name: ident, generic, field_ty, vis });
}
}
@@ -411,21 +425,23 @@ impl Parse for OverseerGuts {
baggage_generic_idents.insert(param.ident.clone());
param.eq_token = None;
param.default = None;
}
_ => {}
},
_ => {},
}
generic
})
.collect();
Self::parse_fields(name, baggage_generic_idents, named)
}
syn::Fields::Unit => {
Err(Error::new(ds.fields.span(), "Must be a struct with named fields. Not an unit struct."))
}
syn::Fields::Unnamed(unnamed) => {
Err(Error::new(unnamed.span(), "Must be a struct with named fields. Not an unnamed fields struct."))
}
},
syn::Fields::Unit => Err(Error::new(
ds.fields.span(),
"Must be a struct with named fields. Not an unit struct.",
)),
syn::Fields::Unnamed(unnamed) => Err(Error::new(
unnamed.span(),
"Must be a struct with named fields. Not an unnamed fields struct.",
)),
}
}
}
+49 -62
View File
@@ -62,39 +62,34 @@
pub use polkadot_overseer_gen_proc_macro::overlord;
#[doc(hidden)]
pub use tracing;
#[doc(hidden)]
pub use metered;
#[doc(hidden)]
pub use polkadot_node_primitives::SpawnNamed;
#[doc(hidden)]
pub use tracing;
#[doc(hidden)]
pub use async_trait::async_trait;
#[doc(hidden)]
pub use futures::{
self,
select,
StreamExt,
FutureExt,
poll,
future::{
Fuse, Future, BoxFuture
},
stream::{
self, select, FuturesUnordered,
},
task::{
Poll, Context,
},
channel::{mpsc, oneshot},
future::{BoxFuture, Fuse, Future},
poll, select,
stream::{self, select, FuturesUnordered},
task::{Context, Poll},
FutureExt, StreamExt,
};
#[doc(hidden)]
pub use std::pin::Pin;
#[doc(hidden)]
pub use async_trait::async_trait;
use std::sync::{
atomic::{self, AtomicUsize},
Arc,
};
#[doc(hidden)]
pub use std::time::Duration;
use std::sync::{Arc, atomic::{self, AtomicUsize}};
#[doc(hidden)]
pub use futures_timer::Delay;
@@ -103,7 +98,6 @@ pub use polkadot_node_network_protocol::WrongVariant;
use std::fmt;
#[cfg(test)]
mod tests;
@@ -134,15 +128,12 @@ pub enum ToOverseer {
impl fmt::Debug for ToOverseer {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::SpawnJob{ name, .. } => writeln!(f, "SpawnJob{{ {}, ..}}", name),
Self::SpawnBlockingJob{ name, .. } => writeln!(f, "SpawnBlockingJob{{ {}, ..}}", name),
Self::SpawnJob { name, .. } => writeln!(f, "SpawnJob{{ {}, ..}}", name),
Self::SpawnBlockingJob { name, .. } => writeln!(f, "SpawnBlockingJob{{ {}, ..}}", name),
}
}
}
/// A helper trait to map a subsystem to smth. else.
pub trait MapSubsystem<T> {
/// The output type of the mapping.
@@ -152,7 +143,10 @@ pub trait MapSubsystem<T> {
fn map_subsystem(&self, sub: T) -> Self::Output;
}
impl<F, T, U> MapSubsystem<T> for F where F: Fn(T) -> U {
impl<F, T, U> MapSubsystem<T> for F
where
F: Fn(T) -> U,
{
type Output = U;
fn map_subsystem(&self, sub: T) -> U {
@@ -178,10 +172,7 @@ pub struct MessagePacket<T> {
/// Create a packet from its parts.
pub fn make_packet<T>(signals_received: usize, message: T) -> MessagePacket<T> {
MessagePacket {
signals_received,
message,
}
MessagePacket { signals_received, message }
}
/// Incoming messages from both the bounded and unbounded channel.
@@ -190,7 +181,6 @@ pub type SubsystemIncomingMessages<M> = self::stream::Select<
self::metered::UnboundedMeteredReceiver<MessagePacket<M>>,
>;
/// Watermark to track the received signals.
#[derive(Debug, Default, Clone)]
pub struct SignalsReceived(Arc<AtomicUsize>);
@@ -208,8 +198,6 @@ impl SignalsReceived {
}
}
/// A trait to support the origin annotation
/// such that errors across subsystems can be easier tracked.
pub trait AnnotateErrorOrigin: 'static + Send + Sync + std::error::Error {
@@ -227,12 +215,8 @@ pub trait AnnotateErrorOrigin: 'static + Send + Sync + std::error::Error {
///
/// In essence it's just a new type wrapping a `BoxFuture`.
pub struct SpawnedSubsystem<E>
where
E: std::error::Error
+ Send
+ Sync
+ 'static
+ From<self::OverseerError>,
where
E: std::error::Error + Send + Sync + 'static + From<self::OverseerError>,
{
/// Name of the subsystem being spawned.
pub name: &'static str,
@@ -274,7 +258,8 @@ pub enum OverseerError {
/// An additional annotation tag for the origin of `source`.
origin: &'static str,
/// The wrapped error. Marked as source for tracking the error chain.
#[source] source: Box<dyn 'static + std::error::Error + Send + Sync>
#[source]
source: Box<dyn 'static + std::error::Error + Send + Sync>,
},
}
@@ -303,7 +288,6 @@ impl SubsystemMeters {
}
}
/// Set of readouts of the `Meter`s of a subsystem.
pub struct SubsystemMeterReadouts {
#[allow(missing_docs)]
@@ -377,7 +361,7 @@ pub trait SubsystemContext: Send + 'static {
/// The sender type as provided by `sender()` and underlying.
type Sender: SubsystemSender<Self::AllMessages> + Send + 'static;
/// The error type.
type Error: ::std::error::Error + ::std::convert::From< OverseerError > + Sync + Send + 'static;
type Error: ::std::error::Error + ::std::convert::From<OverseerError> + Sync + Send + 'static;
/// Try to asynchronously receive a message.
///
@@ -392,7 +376,7 @@ pub trait SubsystemContext: Send + 'static {
fn spawn(
&mut self,
name: &'static str,
s: ::std::pin::Pin<Box<dyn crate::Future<Output = ()> + Send>>
s: ::std::pin::Pin<Box<dyn crate::Future<Output = ()> + Send>>,
) -> Result<(), Self::Error>;
/// Spawn a blocking child task on the executor's dedicated thread pool.
@@ -404,22 +388,24 @@ pub trait SubsystemContext: Send + 'static {
/// Send a direct message to some other `Subsystem`, routed based on message type.
async fn send_message<X>(&mut self, msg: X)
where
Self::AllMessages: From<X>,
X: Send,
where
Self::AllMessages: From<X>,
X: Send,
{
self.sender().send_message(<Self::AllMessages>::from(msg)).await
}
/// Send multiple direct messages to other `Subsystem`s, routed based on message type.
async fn send_messages<X, T>(&mut self, msgs: T)
where
T: IntoIterator<Item = X> + Send,
T::IntoIter: Send,
Self::AllMessages: From<X>,
X: Send,
where
T: IntoIterator<Item = X> + Send,
T::IntoIter: Send,
Self::AllMessages: From<X>,
X: Send,
{
self.sender().send_messages(msgs.into_iter().map(|x| <Self::AllMessages>::from(x))).await
self.sender()
.send_messages(msgs.into_iter().map(|x| <Self::AllMessages>::from(x)))
.await
}
/// Send a message using the unbounded connection.
@@ -449,10 +435,9 @@ where
E: std::error::Error + Send + Sync + 'static + From<self::OverseerError>,
{
/// Start this `Subsystem` and return `SpawnedSubsystem`.
fn start(self, ctx: Ctx) -> SpawnedSubsystem < E >;
fn start(self, ctx: Ctx) -> SpawnedSubsystem<E>;
}
/// Sender end of a channel to interface with a subsystem.
#[async_trait::async_trait]
pub trait SubsystemSender<Message>: Send + Clone + 'static {
@@ -461,7 +446,9 @@ pub trait SubsystemSender<Message>: Send + Clone + 'static {
/// Send multiple direct messages to other `Subsystem`s, routed based on message type.
async fn send_messages<T>(&mut self, msgs: T)
where T: IntoIterator<Item = Message> + Send, T::IntoIter: Send;
where
T: IntoIterator<Item = Message> + Send,
T::IntoIter: Send;
/// Send a message onto the unbounded queue of some other `Subsystem`, routed based on message
/// type.
@@ -488,27 +475,27 @@ pub trait TimeoutExt: Future {
where
Self: Sized,
{
Timeout {
future: self,
delay: Delay::new(duration),
}
Timeout { future: self, delay: Delay::new(duration) }
}
}
impl<F> TimeoutExt for F where F: Future{}
impl<F> TimeoutExt for F where F: Future {}
impl<F> Future for Timeout<F> where F: Future {
impl<F> Future for Timeout<F>
where
F: Future,
{
type Output = Option<F::Output>;
fn poll(self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Self::Output> {
let this = self.project();
if this.delay.poll(ctx).is_ready() {
return Poll::Ready(None);
return Poll::Ready(None)
}
if let Poll::Ready(output) = this.future.poll(ctx) {
return Poll::Ready(Some(output));
return Poll::Ready(Some(output))
}
Poll::Pending
@@ -1,4 +1,3 @@
// The generated code requires quite a bit of surrounding code to work.
// Please refer to [the examples](examples/dummy.rs) and
// [the minimal usage example](../examples/minimal-example.rs).
+126 -117
View File
@@ -59,46 +59,36 @@
// yielding false positives
#![warn(missing_docs)]
use std::fmt::{self, Debug};
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
use std::collections::{hash_map, HashMap};
use std::iter::FromIterator;
use futures::channel::oneshot;
use futures::{
select,
future::BoxFuture,
Future, FutureExt, StreamExt,
use std::{
collections::{hash_map, HashMap},
fmt::{self, Debug},
iter::FromIterator,
pin::Pin,
sync::Arc,
time::Duration,
};
use futures::{channel::oneshot, future::BoxFuture, select, Future, FutureExt, StreamExt};
use lru::LruCache;
use parking_lot::RwLock;
use polkadot_primitives::v1::{Block, BlockId,BlockNumber, Hash, ParachainHost};
use client::{BlockImportNotification, BlockchainEvents, FinalityNotification};
use polkadot_primitives::v1::{Block, BlockId, BlockNumber, Hash, ParachainHost};
use sp_api::{ApiExt, ProvideRuntimeApi};
use polkadot_node_network_protocol::{
v1 as protocol_v1,
};
use polkadot_node_network_protocol::v1 as protocol_v1;
use polkadot_node_subsystem_types::messages::{
CandidateValidationMessage, CandidateBackingMessage,
ChainApiMessage, StatementDistributionMessage,
AvailabilityDistributionMessage, BitfieldSigningMessage, BitfieldDistributionMessage,
ProvisionerMessage, RuntimeApiMessage,
AvailabilityStoreMessage, NetworkBridgeMessage, CollationGenerationMessage,
CollatorProtocolMessage, AvailabilityRecoveryMessage, ApprovalDistributionMessage,
ApprovalVotingMessage, GossipSupportMessage,
NetworkBridgeEvent,
DisputeParticipationMessage, DisputeCoordinatorMessage, ChainSelectionMessage,
DisputeDistributionMessage,
ApprovalDistributionMessage, ApprovalVotingMessage, AvailabilityDistributionMessage,
AvailabilityRecoveryMessage, AvailabilityStoreMessage, BitfieldDistributionMessage,
BitfieldSigningMessage, CandidateBackingMessage, CandidateValidationMessage, ChainApiMessage,
ChainSelectionMessage, CollationGenerationMessage, CollatorProtocolMessage,
DisputeCoordinatorMessage, DisputeDistributionMessage, DisputeParticipationMessage,
GossipSupportMessage, NetworkBridgeEvent, NetworkBridgeMessage, ProvisionerMessage,
RuntimeApiMessage, StatementDistributionMessage,
};
pub use polkadot_node_subsystem_types::{
OverseerSignal,
errors::{SubsystemResult, SubsystemError,},
ActiveLeavesUpdate, ActivatedLeaf, LeafStatus,
jaeger,
errors::{SubsystemError, SubsystemResult},
jaeger, ActivatedLeaf, ActiveLeavesUpdate, LeafStatus, OverseerSignal,
};
// TODO legacy, to be deleted, left for easier integration
@@ -110,30 +100,15 @@ mod metrics;
use self::metrics::Metrics;
use polkadot_node_metrics::{
metrics::{
prometheus,
Metrics as MetricsTrait
},
metrics::{prometheus, Metrics as MetricsTrait},
Metronome,
};
pub use polkadot_overseer_gen::{
TimeoutExt,
SpawnNamed,
Subsystem,
SubsystemMeterReadouts,
SubsystemMeters,
SubsystemIncomingMessages,
SubsystemInstance,
SubsystemSender,
SubsystemContext,
overlord,
MessagePacket,
SignalsReceived,
FromOverseer,
ToOverseer,
MapSubsystem,
};
pub use polkadot_overseer_gen as gen;
pub use polkadot_overseer_gen::{
overlord, FromOverseer, MapSubsystem, MessagePacket, SignalsReceived, SpawnNamed, Subsystem,
SubsystemContext, SubsystemIncomingMessages, SubsystemInstance, SubsystemMeterReadouts,
SubsystemMeters, SubsystemSender, TimeoutExt, ToOverseer,
};
/// Store 2 days worth of blocks, not accounting for forks,
/// in the LRU cache. Assumes a 6-second block time.
@@ -142,14 +117,14 @@ const KNOWN_LEAVES_CACHE_SIZE: usize = 2 * 24 * 3600 / 6;
#[cfg(test)]
mod tests;
/// Whether a header supports parachain consensus or not.
pub trait HeadSupportsParachains {
/// Return true if the given header supports parachain consensus. Otherwise, false.
fn head_supports_parachains(&self, head: &Hash) -> bool;
}
impl<Client> HeadSupportsParachains for Arc<Client> where
impl<Client> HeadSupportsParachains for Arc<Client>
where
Client: ProvideRuntimeApi<Block>,
Client::Api: ParachainHost<Block>,
{
@@ -159,7 +134,6 @@ impl<Client> HeadSupportsParachains for Arc<Client> where
}
}
/// A handle used to communicate with the [`Overseer`].
///
/// [`Overseer`]: struct.Overseer.html
@@ -205,11 +179,16 @@ impl Handle {
/// Note that due the fact the overseer doesn't store the whole active-leaves set, only deltas,
/// the response channel may never return if the hash was deactivated before this call.
/// In this case, it's the caller's responsibility to ensure a timeout is set.
pub async fn wait_for_activation(&mut self, hash: Hash, response_channel: oneshot::Sender<SubsystemResult<()>>) {
pub async fn wait_for_activation(
&mut self,
hash: Hash,
response_channel: oneshot::Sender<SubsystemResult<()>>,
) {
self.send_and_log_error(Event::ExternalRequest(ExternalRequest::WaitForActivation {
hash,
response_channel
})).await;
hash,
response_channel,
}))
.await;
}
/// Tell `Overseer` to shutdown.
@@ -293,21 +272,13 @@ pub struct BlockInfo {
impl From<BlockImportNotification<Block>> for BlockInfo {
fn from(n: BlockImportNotification<Block>) -> Self {
BlockInfo {
hash: n.hash,
parent_hash: n.header.parent_hash,
number: n.header.number,
}
BlockInfo { hash: n.hash, parent_hash: n.header.parent_hash, number: n.header.number }
}
}
impl From<FinalityNotification<Block>> for BlockInfo {
fn from(n: FinalityNotification<Block>) -> Self {
BlockInfo {
hash: n.hash,
parent_hash: n.header.parent_hash,
number: n.header.number,
}
BlockInfo { hash: n.hash, parent_hash: n.header.parent_hash, number: n.header.number }
}
}
@@ -345,10 +316,7 @@ pub enum ExternalRequest {
/// Glues together the [`Overseer`] and `BlockchainEvents` by forwarding
/// import and finality notifications into the [`OverseerHandle`].
pub async fn forward_events<P: BlockchainEvents<Block>>(
client: Arc<P>,
mut handle: Handle,
) {
pub async fn forward_events<P: BlockchainEvents<Block>>(client: Arc<P>, mut handle: Handle) {
let mut finality = client.finality_notification_stream();
let mut imports = client.import_notification_stream();
@@ -594,9 +562,53 @@ where
/// # });
/// # }
/// ```
pub fn new<CV, CB, SD, AD, AR, BS, BD, P, RA, AS, NB, CA, CG, CP, ApD, ApV, GS, DC, DP, DD, CS>(
pub fn new<
CV,
CB,
SD,
AD,
AR,
BS,
BD,
P,
RA,
AS,
NB,
CA,
CG,
CP,
ApD,
ApV,
GS,
DC,
DP,
DD,
CS,
>(
leaves: impl IntoIterator<Item = BlockInfo>,
all_subsystems: AllSubsystems<CV, CB, SD, AD, AR, BS, BD, P, RA, AS, NB, CA, CG, CP, ApD, ApV, GS, DC, DP, DD, CS>,
all_subsystems: AllSubsystems<
CV,
CB,
SD,
AD,
AR,
BS,
BD,
P,
RA,
AS,
NB,
CA,
CG,
CP,
ApD,
ApV,
GS,
DC,
DP,
DD,
CS,
>,
prometheus_registry: Option<&prometheus::Registry>,
supports_parachains: SupportsParachains,
s: S,
@@ -604,8 +616,10 @@ where
where
CV: Subsystem<OverseerSubsystemContext<CandidateValidationMessage>, SubsystemError> + Send,
CB: Subsystem<OverseerSubsystemContext<CandidateBackingMessage>, SubsystemError> + Send,
SD: Subsystem<OverseerSubsystemContext<StatementDistributionMessage>, SubsystemError> + Send,
AD: Subsystem<OverseerSubsystemContext<AvailabilityDistributionMessage>, SubsystemError> + Send,
SD: Subsystem<OverseerSubsystemContext<StatementDistributionMessage>, SubsystemError>
+ Send,
AD: Subsystem<OverseerSubsystemContext<AvailabilityDistributionMessage>, SubsystemError>
+ Send,
AR: Subsystem<OverseerSubsystemContext<AvailabilityRecoveryMessage>, SubsystemError> + Send,
BS: Subsystem<OverseerSubsystemContext<BitfieldSigningMessage>, SubsystemError> + Send,
BD: Subsystem<OverseerSubsystemContext<BitfieldDistributionMessage>, SubsystemError> + Send,
@@ -616,7 +630,8 @@ where
CA: Subsystem<OverseerSubsystemContext<ChainApiMessage>, SubsystemError> + Send,
CG: Subsystem<OverseerSubsystemContext<CollationGenerationMessage>, SubsystemError> + Send,
CP: Subsystem<OverseerSubsystemContext<CollatorProtocolMessage>, SubsystemError> + Send,
ApD: Subsystem<OverseerSubsystemContext<ApprovalDistributionMessage>, SubsystemError> + Send,
ApD:
Subsystem<OverseerSubsystemContext<ApprovalDistributionMessage>, SubsystemError> + Send,
ApV: Subsystem<OverseerSubsystemContext<ApprovalVotingMessage>, SubsystemError> + Send,
GS: Subsystem<OverseerSubsystemContext<GossipSupportMessage>, SubsystemError> + Send,
DC: Subsystem<OverseerSubsystemContext<DisputeCoordinatorMessage>, SubsystemError> + Send,
@@ -650,7 +665,9 @@ where
.dispute_distribution(all_subsystems.dispute_distribution)
.chain_selection(all_subsystems.chain_selection)
.leaves(Vec::from_iter(
leaves.into_iter().map(|BlockInfo { hash, parent_hash: _, number }| (hash, number))
leaves
.into_iter()
.map(|BlockInfo { hash, parent_hash: _, number }| (hash, number)),
))
.known_leaves(LruCache::new(KNOWN_LEAVES_CACHE_SIZE))
.active_leaves(Default::default())
@@ -669,33 +686,29 @@ where
type Output = Option<(&'static str, SubsystemMeters)>;
fn map_subsystem(&self, subsystem: &'a OverseenSubsystem<T>) -> Self::Output {
subsystem.instance.as_ref().map(|instance| {
(
instance.name,
instance.meters.clone(),
)
})
subsystem
.instance
.as_ref()
.map(|instance| (instance.name, instance.meters.clone()))
}
}
let subsystem_meters = overseer.map_subsystems(ExtractNameAndMeters);
let metronome_metrics = metrics.clone();
let metronome = Metronome::new(std::time::Duration::from_millis(950))
.for_each(move |_| {
let metronome =
Metronome::new(std::time::Duration::from_millis(950)).for_each(move |_| {
// We combine the amount of messages from subsystems to the overseer
// as well as the amount of messages from external sources to the overseer
// into one `to_overseer` value.
metronome_metrics.channel_fill_level_snapshot(
subsystem_meters.iter()
subsystem_meters
.iter()
.cloned()
.filter_map(|x| x)
.map(|(name, ref meters)| (name, meters.read()))
.map(|(name, ref meters)| (name, meters.read())),
);
async move {
()
}
async move { () }
});
overseer.spawner().spawn("metrics_metronome", Box::pin(metronome));
}
@@ -705,10 +718,7 @@ where
/// Stop the overseer.
async fn stop(mut self) {
let _ = self.wait_terminate(
OverseerSignal::Conclude,
Duration::from_secs(1_u64)
).await;
let _ = self.wait_terminate(OverseerSignal::Conclude, Duration::from_secs(1_u64)).await;
}
/// Run the `Overseer`.
@@ -717,12 +727,8 @@ where
for (hash, number) in std::mem::take(&mut self.leaves) {
let _ = self.active_leaves.insert(hash, number);
if let Some((span, status)) = self.on_head_activated(&hash, None) {
let update = ActiveLeavesUpdate::start_work(ActivatedLeaf {
hash,
number,
status,
span,
});
let update =
ActiveLeavesUpdate::start_work(ActivatedLeaf { hash, number, status, span });
self.broadcast_signal(OverseerSignal::ActiveLeaves(update)).await?;
}
}
@@ -778,8 +784,8 @@ where
hash_map::Entry::Vacant(entry) => entry.insert(block.number),
hash_map::Entry::Occupied(entry) => {
debug_assert_eq!(*entry.get(), block.number);
return Ok(());
}
return Ok(())
},
};
let mut update = match self.on_head_activated(&block.hash, Some(block.parent_hash)) {
@@ -787,7 +793,7 @@ where
hash: block.hash,
number: block.number,
status,
span
span,
}),
None => ActiveLeavesUpdate::default(),
};
@@ -822,7 +828,8 @@ where
self.on_head_deactivated(deactivated)
}
self.broadcast_signal(OverseerSignal::BlockFinalized(block.hash, block.number)).await?;
self.broadcast_signal(OverseerSignal::BlockFinalized(block.hash, block.number))
.await?;
// If there are no leaves being deactivated, we don't need to send an update.
//
@@ -836,11 +843,13 @@ where
/// Handles a header activation. If the header's state doesn't support the parachains API,
/// this returns `None`.
fn on_head_activated(&mut self, hash: &Hash, parent_hash: Option<Hash>)
-> Option<(Arc<jaeger::Span>, LeafStatus)>
{
fn on_head_activated(
&mut self,
hash: &Hash,
parent_hash: Option<Hash>,
) -> Option<(Arc<jaeger::Span>, LeafStatus)> {
if !self.supports_parachains.head_supports_parachains(hash) {
return None;
return None
}
self.metrics.on_head_activated();
@@ -894,9 +903,12 @@ where
// it's fine if the listener is no longer interested
let _ = response_channel.send(Ok(()));
} else {
self.activation_external_listeners.entry(hash).or_default().push(response_channel);
self.activation_external_listeners
.entry(hash)
.or_default()
.push(response_channel);
}
}
},
}
}
@@ -909,15 +921,12 @@ where
}
}
// Additional `From` implementations, in order to deal with incoming network messages.
// Kept out of the proc macro, for sake of simplicity reduce the need to make even
// more types to the proc macro logic.
use polkadot_node_network_protocol::{
request_response::{request::IncomingRequest, v1 as req_res_v1},
use polkadot_node_network_protocol::request_response::{
request::IncomingRequest, v1 as req_res_v1,
};
impl From<IncomingRequest<req_res_v1::PoVFetchingRequest>> for AllMessages {
+38 -39
View File
@@ -33,7 +33,6 @@ struct MetricsInner {
signals_received: prometheus::GaugeVec<prometheus::U64>,
}
/// A shareable metrics type for usage with the overseer.
#[derive(Default, Clone)]
pub struct Metrics(Option<MetricsInner>);
@@ -59,30 +58,42 @@ impl Metrics {
pub(crate) fn channel_fill_level_snapshot(
&self,
collection: impl IntoIterator<Item=(&'static str, SubsystemMeterReadouts)>,
collection: impl IntoIterator<Item = (&'static str, SubsystemMeterReadouts)>,
) {
if let Some(metrics) = &self.0 {
collection.into_iter().for_each(
|(name, readouts): (_, SubsystemMeterReadouts)| {
metrics.to_subsystem_bounded_sent.with_label_values(&[name])
.set(readouts.bounded.sent as u64);
collection
.into_iter()
.for_each(|(name, readouts): (_, SubsystemMeterReadouts)| {
metrics
.to_subsystem_bounded_sent
.with_label_values(&[name])
.set(readouts.bounded.sent as u64);
metrics.to_subsystem_bounded_received.with_label_values(&[name])
.set(readouts.bounded.received as u64);
metrics
.to_subsystem_bounded_received
.with_label_values(&[name])
.set(readouts.bounded.received as u64);
metrics.to_subsystem_unbounded_sent.with_label_values(&[name])
.set(readouts.unbounded.sent as u64);
metrics
.to_subsystem_unbounded_sent
.with_label_values(&[name])
.set(readouts.unbounded.sent as u64);
metrics.to_subsystem_unbounded_received.with_label_values(&[name])
.set(readouts.unbounded.received as u64);
metrics
.to_subsystem_unbounded_received
.with_label_values(&[name])
.set(readouts.unbounded.received as u64);
metrics.signals_sent.with_label_values(&[name])
.set(readouts.signals.sent as u64);
metrics
.signals_sent
.with_label_values(&[name])
.set(readouts.signals.sent as u64);
metrics.signals_received.with_label_values(&[name])
.set(readouts.signals.received as u64);
}
);
metrics
.signals_received
.with_label_values(&[name])
.set(readouts.signals.received as u64);
});
}
}
}
@@ -93,21 +104,21 @@ impl metrics::Metrics for Metrics {
activated_heads_total: prometheus::register(
prometheus::Counter::new(
"parachain_activated_heads_total",
"Number of activated heads."
"Number of activated heads.",
)?,
registry,
)?,
deactivated_heads_total: prometheus::register(
prometheus::Counter::new(
"parachain_deactivated_heads_total",
"Number of deactivated heads."
"Number of deactivated heads.",
)?,
registry,
)?,
messages_relayed_total: prometheus::register(
prometheus::Counter::new(
"parachain_messages_relayed_total",
"Number of messages relayed by Overseer."
"Number of messages relayed by Overseer.",
)?,
registry,
)?,
@@ -117,9 +128,7 @@ impl metrics::Metrics for Metrics {
"parachain_subsystem_bounded_sent",
"Number of elements sent to subsystems' bounded queues",
),
&[
"subsystem_name",
],
&["subsystem_name"],
)?,
registry,
)?,
@@ -129,9 +138,7 @@ impl metrics::Metrics for Metrics {
"parachain_subsystem_bounded_received",
"Number of elements received by subsystems' bounded queues",
),
&[
"subsystem_name",
],
&["subsystem_name"],
)?,
registry,
)?,
@@ -141,9 +148,7 @@ impl metrics::Metrics for Metrics {
"parachain_subsystem_unbounded_sent",
"Number of elements sent to subsystems' unbounded queues",
),
&[
"subsystem_name",
],
&["subsystem_name"],
)?,
registry,
)?,
@@ -153,9 +158,7 @@ impl metrics::Metrics for Metrics {
"parachain_subsystem_unbounded_received",
"Number of elements received by subsystems' unbounded queues",
),
&[
"subsystem_name",
],
&["subsystem_name"],
)?,
registry,
)?,
@@ -165,9 +168,7 @@ impl metrics::Metrics for Metrics {
"parachain_overseer_signals_sent",
"Number of signals sent by overseer to subsystems",
),
&[
"subsystem_name",
],
&["subsystem_name"],
)?,
registry,
)?,
@@ -177,9 +178,7 @@ impl metrics::Metrics for Metrics {
"parachain_overseer_signals_received",
"Number of signals received by subsystems from overseer",
),
&[
"subsystem_name",
],
&["subsystem_name"],
)?,
registry,
)?,
+156 -62
View File
@@ -19,16 +19,12 @@
//! In the future, everything should be set up using the generated
//! overseer builder pattern instead.
use crate::{AllMessages, OverseerSignal};
use polkadot_node_subsystem_types::errors::SubsystemError;
use polkadot_overseer_gen::{
MapSubsystem, SubsystemContext,
Subsystem,
SpawnedSubsystem,
FromOverseer,
};
use polkadot_overseer_all_subsystems_gen::AllSubsystemsGen;
use crate::OverseerSignal;
use crate::AllMessages;
use polkadot_overseer_gen::{
FromOverseer, MapSubsystem, SpawnedSubsystem, Subsystem, SubsystemContext,
};
/// A dummy subsystem that implements [`Subsystem`] for all
/// types of messages. Used for tests or as a placeholder.
@@ -37,7 +33,11 @@ pub struct DummySubsystem;
impl<Context> Subsystem<Context, SubsystemError> for DummySubsystem
where
Context: SubsystemContext<Signal=OverseerSignal, Error=SubsystemError, AllMessages=AllMessages>,
Context: SubsystemContext<
Signal = OverseerSignal,
Error = SubsystemError,
AllMessages = AllMessages,
>,
{
fn start(self, mut ctx: Context) -> SpawnedSubsystem<SubsystemError> {
let future = Box::pin(async move {
@@ -51,20 +51,16 @@ where
"Discarding a message sent from overseer {:?}",
overseer_msg
);
continue;
}
continue
},
}
}
});
SpawnedSubsystem {
name: "dummy-subsystem",
future,
}
SpawnedSubsystem { name: "dummy-subsystem", future }
}
}
/// 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
@@ -75,9 +71,27 @@ where
/// subsystems are implemented and the rest can be mocked with the [`DummySubsystem`].
#[derive(Debug, Clone, AllSubsystemsGen)]
pub struct AllSubsystems<
CV = (), CB = (), SD = (), AD = (), AR = (), BS = (), BD = (), P = (),
RA = (), AS = (), NB = (), CA = (), CG = (), CP = (), ApD = (), ApV = (),
GS = (), DC = (), DP = (), DD = (), CS = (),
CV = (),
CB = (),
SD = (),
AD = (),
AR = (),
BS = (),
BD = (),
P = (),
RA = (),
AS = (),
NB = (),
CA = (),
CG = (),
CP = (),
ApD = (),
ApV = (),
GS = (),
DC = (),
DP = (),
DD = (),
CS = (),
> {
/// A candidate validation subsystem.
pub candidate_validation: CV,
@@ -187,7 +201,31 @@ impl<CV, CB, SD, AD, AR, BS, BD, P, RA, AS, NB, CA, CG, CP, ApD, ApV, GS, DC, DP
}
/// Reference every individual subsystem.
pub fn as_ref(&self) -> AllSubsystems<&'_ CV, &'_ CB, &'_ SD, &'_ AD, &'_ AR, &'_ BS, &'_ BD, &'_ P, &'_ RA, &'_ AS, &'_ NB, &'_ CA, &'_ CG, &'_ CP, &'_ ApD, &'_ ApV, &'_ GS, &'_ DC, &'_ DP, &'_ DD, &'_ CS> {
pub fn as_ref(
&self,
) -> AllSubsystems<
&'_ CV,
&'_ CB,
&'_ SD,
&'_ AD,
&'_ AR,
&'_ BS,
&'_ BD,
&'_ P,
&'_ RA,
&'_ AS,
&'_ NB,
&'_ CA,
&'_ CG,
&'_ CP,
&'_ ApD,
&'_ ApV,
&'_ GS,
&'_ DC,
&'_ DP,
&'_ DD,
&'_ CS,
> {
AllSubsystems {
candidate_validation: &self.candidate_validation,
candidate_backing: &self.candidate_backing,
@@ -214,30 +252,32 @@ impl<CV, CB, SD, AD, AR, BS, BD, P, RA, AS, NB, CA, CG, CP, ApD, ApV, GS, DC, DP
}
/// Map each subsystem.
pub fn map_subsystems<Mapper>(self, mapper: Mapper)
-> AllSubsystems<
<Mapper as MapSubsystem<CV>>::Output,
<Mapper as MapSubsystem<CB>>::Output,
<Mapper as MapSubsystem<SD>>::Output,
<Mapper as MapSubsystem<AD>>::Output,
<Mapper as MapSubsystem<AR>>::Output,
<Mapper as MapSubsystem<BS>>::Output,
<Mapper as MapSubsystem<BD>>::Output,
<Mapper as MapSubsystem<P>>::Output,
<Mapper as MapSubsystem<RA>>::Output,
<Mapper as MapSubsystem<AS>>::Output,
<Mapper as MapSubsystem<NB>>::Output,
<Mapper as MapSubsystem<CA>>::Output,
<Mapper as MapSubsystem<CG>>::Output,
<Mapper as MapSubsystem<CP>>::Output,
<Mapper as MapSubsystem<ApD>>::Output,
<Mapper as MapSubsystem<ApV>>::Output,
<Mapper as MapSubsystem<GS>>::Output,
<Mapper as MapSubsystem<DC>>::Output,
<Mapper as MapSubsystem<DP>>::Output,
<Mapper as MapSubsystem<DD>>::Output,
<Mapper as MapSubsystem<CS>>::Output,
>
pub fn map_subsystems<Mapper>(
self,
mapper: Mapper,
) -> AllSubsystems<
<Mapper as MapSubsystem<CV>>::Output,
<Mapper as MapSubsystem<CB>>::Output,
<Mapper as MapSubsystem<SD>>::Output,
<Mapper as MapSubsystem<AD>>::Output,
<Mapper as MapSubsystem<AR>>::Output,
<Mapper as MapSubsystem<BS>>::Output,
<Mapper as MapSubsystem<BD>>::Output,
<Mapper as MapSubsystem<P>>::Output,
<Mapper as MapSubsystem<RA>>::Output,
<Mapper as MapSubsystem<AS>>::Output,
<Mapper as MapSubsystem<NB>>::Output,
<Mapper as MapSubsystem<CA>>::Output,
<Mapper as MapSubsystem<CG>>::Output,
<Mapper as MapSubsystem<CP>>::Output,
<Mapper as MapSubsystem<ApD>>::Output,
<Mapper as MapSubsystem<ApV>>::Output,
<Mapper as MapSubsystem<GS>>::Output,
<Mapper as MapSubsystem<DC>>::Output,
<Mapper as MapSubsystem<DP>>::Output,
<Mapper as MapSubsystem<DD>>::Output,
<Mapper as MapSubsystem<CS>>::Output,
>
where
Mapper: MapSubsystem<CV>,
Mapper: MapSubsystem<CB>,
@@ -262,27 +302,81 @@ impl<CV, CB, SD, AD, AR, BS, BD, P, RA, AS, NB, CA, CG, CP, ApD, ApV, GS, DC, DP
Mapper: MapSubsystem<CS>,
{
AllSubsystems {
candidate_validation: <Mapper as MapSubsystem<CV>>::map_subsystem(&mapper, self.candidate_validation),
candidate_backing: <Mapper as MapSubsystem<CB>>::map_subsystem(&mapper, self.candidate_backing),
statement_distribution: <Mapper as MapSubsystem<SD>>::map_subsystem(&mapper, self.statement_distribution),
availability_distribution: <Mapper as MapSubsystem<AD>>::map_subsystem(&mapper, self.availability_distribution),
availability_recovery: <Mapper as MapSubsystem<AR>>::map_subsystem(&mapper, self.availability_recovery),
bitfield_signing: <Mapper as MapSubsystem<BS>>::map_subsystem(&mapper, self.bitfield_signing),
bitfield_distribution: <Mapper as MapSubsystem<BD>>::map_subsystem(&mapper, self.bitfield_distribution),
candidate_validation: <Mapper as MapSubsystem<CV>>::map_subsystem(
&mapper,
self.candidate_validation,
),
candidate_backing: <Mapper as MapSubsystem<CB>>::map_subsystem(
&mapper,
self.candidate_backing,
),
statement_distribution: <Mapper as MapSubsystem<SD>>::map_subsystem(
&mapper,
self.statement_distribution,
),
availability_distribution: <Mapper as MapSubsystem<AD>>::map_subsystem(
&mapper,
self.availability_distribution,
),
availability_recovery: <Mapper as MapSubsystem<AR>>::map_subsystem(
&mapper,
self.availability_recovery,
),
bitfield_signing: <Mapper as MapSubsystem<BS>>::map_subsystem(
&mapper,
self.bitfield_signing,
),
bitfield_distribution: <Mapper as MapSubsystem<BD>>::map_subsystem(
&mapper,
self.bitfield_distribution,
),
provisioner: <Mapper as MapSubsystem<P>>::map_subsystem(&mapper, self.provisioner),
runtime_api: <Mapper as MapSubsystem<RA>>::map_subsystem(&mapper, self.runtime_api),
availability_store: <Mapper as MapSubsystem<AS>>::map_subsystem(&mapper, self.availability_store),
network_bridge: <Mapper as MapSubsystem<NB>>::map_subsystem(&mapper, self.network_bridge),
availability_store: <Mapper as MapSubsystem<AS>>::map_subsystem(
&mapper,
self.availability_store,
),
network_bridge: <Mapper as MapSubsystem<NB>>::map_subsystem(
&mapper,
self.network_bridge,
),
chain_api: <Mapper as MapSubsystem<CA>>::map_subsystem(&mapper, self.chain_api),
collation_generation: <Mapper as MapSubsystem<CG>>::map_subsystem(&mapper, self.collation_generation),
collator_protocol: <Mapper as MapSubsystem<CP>>::map_subsystem(&mapper, self.collator_protocol),
approval_distribution: <Mapper as MapSubsystem<ApD>>::map_subsystem(&mapper, self.approval_distribution),
approval_voting: <Mapper as MapSubsystem<ApV>>::map_subsystem(&mapper, self.approval_voting),
gossip_support: <Mapper as MapSubsystem<GS>>::map_subsystem(&mapper, self.gossip_support),
dispute_coordinator: <Mapper as MapSubsystem<DC>>::map_subsystem(&mapper, self.dispute_coordinator),
dispute_participation: <Mapper as MapSubsystem<DP>>::map_subsystem(&mapper, self.dispute_participation),
dispute_distribution: <Mapper as MapSubsystem<DD>>::map_subsystem(&mapper, self.dispute_distribution),
chain_selection: <Mapper as MapSubsystem<CS>>::map_subsystem(&mapper, self.chain_selection),
collation_generation: <Mapper as MapSubsystem<CG>>::map_subsystem(
&mapper,
self.collation_generation,
),
collator_protocol: <Mapper as MapSubsystem<CP>>::map_subsystem(
&mapper,
self.collator_protocol,
),
approval_distribution: <Mapper as MapSubsystem<ApD>>::map_subsystem(
&mapper,
self.approval_distribution,
),
approval_voting: <Mapper as MapSubsystem<ApV>>::map_subsystem(
&mapper,
self.approval_voting,
),
gossip_support: <Mapper as MapSubsystem<GS>>::map_subsystem(
&mapper,
self.gossip_support,
),
dispute_coordinator: <Mapper as MapSubsystem<DC>>::map_subsystem(
&mapper,
self.dispute_coordinator,
),
dispute_participation: <Mapper as MapSubsystem<DP>>::map_subsystem(
&mapper,
self.dispute_participation,
),
dispute_distribution: <Mapper as MapSubsystem<DD>>::map_subsystem(
&mapper,
self.dispute_distribution,
),
chain_selection: <Mapper as MapSubsystem<CS>>::map_subsystem(
&mapper,
self.chain_selection,
),
}
}
}
+162 -186
View File
@@ -14,47 +14,37 @@
// You should have received a copy of the GNU General Public License
// along with Polkadot. If not, see <http://www.gnu.org/licenses/>.
use std::sync::atomic;
use std::collections::HashMap;
use std::task::{Poll};
use futures::{executor, pin_mut, select, FutureExt, pending, poll, stream};
use futures::channel::mpsc;
use futures::{channel::mpsc, executor, pending, pin_mut, poll, select, stream, FutureExt};
use std::{collections::HashMap, sync::atomic, task::Poll};
use polkadot_primitives::v1::{CollatorPair, CandidateHash};
use polkadot_node_primitives::{CollationResult, CollationGenerationConfig, PoV, BlockData};
use polkadot_node_network_protocol::{PeerId, UnifiedReputationChange};
use polkadot_node_primitives::{BlockData, CollationGenerationConfig, CollationResult, PoV};
use polkadot_node_subsystem_types::{
ActivatedLeaf, LeafStatus,
messages::{
RuntimeApiRequest,
NetworkBridgeEvent,
},
jaeger,
messages::{NetworkBridgeEvent, RuntimeApiRequest},
ActivatedLeaf, LeafStatus,
};
use polkadot_primitives::v1::{CandidateHash, CollatorPair};
use crate::{
self as overseer,
Overseer,
HeadSupportsParachains,
gen::Delay,
};
use crate::{self as overseer, gen::Delay, HeadSupportsParachains, Overseer};
use metered_channel as metered;
use sp_core::crypto::Pair as _;
use assert_matches::assert_matches;
use sp_core::crypto::Pair as _;
use super::*;
type SpawnedSubsystem = crate::gen::SpawnedSubsystem<SubsystemError>;
struct TestSubsystem1(metered::MeteredSender<usize>);
impl<C> overseer::Subsystem<C, SubsystemError> for TestSubsystem1
where
C: overseer::SubsystemContext<Message=CandidateValidationMessage,Signal=OverseerSignal,AllMessages=AllMessages>,
C: overseer::SubsystemContext<
Message = CandidateValidationMessage,
Signal = OverseerSignal,
AllMessages = AllMessages,
>,
{
fn start(self, mut ctx: C) -> SpawnedSubsystem {
let mut sender = self.0;
@@ -67,8 +57,8 @@ where
Ok(FromOverseer::Communication { .. }) => {
let _ = sender.send(i).await;
i += 1;
continue;
}
continue
},
Ok(FromOverseer::Signal(OverseerSignal::Conclude)) => return Ok(()),
Err(_) => return Ok(()),
_ => (),
@@ -83,7 +73,11 @@ struct TestSubsystem2(metered::MeteredSender<usize>);
impl<C> overseer::Subsystem<C, SubsystemError> for TestSubsystem2
where
C: overseer::SubsystemContext<Message=CandidateBackingMessage,Signal=OverseerSignal,AllMessages=AllMessages>,
C: overseer::SubsystemContext<
Message = CandidateBackingMessage,
Signal = OverseerSignal,
AllMessages = AllMessages,
>,
{
fn start(self, mut ctx: C) -> SpawnedSubsystem {
let sender = self.0.clone();
@@ -95,25 +89,18 @@ where
loop {
if c < 10 {
let (tx, _) = oneshot::channel();
ctx.send_message(
CandidateValidationMessage::ValidateFromChainState(
Default::default(),
PoV {
block_data: BlockData(Vec::new()),
}.into(),
tx,
)
).await;
ctx.send_message(CandidateValidationMessage::ValidateFromChainState(
Default::default(),
PoV { block_data: BlockData(Vec::new()) }.into(),
tx,
))
.await;
c += 1;
continue;
continue
}
match ctx.try_recv().await {
Ok(Some(FromOverseer::Signal(OverseerSignal::Conclude))) => {
break;
}
Ok(Some(_)) => {
continue;
}
Ok(Some(FromOverseer::Signal(OverseerSignal::Conclude))) => break,
Ok(Some(_)) => continue,
Err(_) => return Ok(()),
_ => (),
}
@@ -130,7 +117,11 @@ struct ReturnOnStart;
impl<C> overseer::Subsystem<C, SubsystemError> for ReturnOnStart
where
C: overseer::SubsystemContext<Message=CandidateBackingMessage,Signal=OverseerSignal,AllMessages=AllMessages>,
C: overseer::SubsystemContext<
Message = CandidateBackingMessage,
Signal = OverseerSignal,
AllMessages = AllMessages,
>,
{
fn start(self, mut _ctx: C) -> SpawnedSubsystem {
SpawnedSubsystem {
@@ -167,13 +158,8 @@ fn overseer_works() {
.replace_candidate_validation(TestSubsystem1(s1_tx))
.replace_candidate_backing(TestSubsystem2(s2_tx));
let (overseer, handle) = Overseer::new(
vec![],
all_subsystems,
None,
MockSupportsParachains,
spawner,
).unwrap();
let (overseer, handle) =
Overseer::new(vec![], all_subsystems, None, MockSupportsParachains, spawner).unwrap();
let mut handle = Handle::Connected(handle);
let overseer_fut = overseer.run().fuse();
@@ -220,21 +206,12 @@ fn overseer_metrics_work() {
let second_block_hash = [2; 32].into();
let third_block_hash = [3; 32].into();
let first_block = BlockInfo {
hash: first_block_hash,
parent_hash: [0; 32].into(),
number: 1,
};
let second_block = BlockInfo {
hash: second_block_hash,
parent_hash: first_block_hash,
number: 2,
};
let third_block = BlockInfo {
hash: third_block_hash,
parent_hash: second_block_hash,
number: 3,
};
let first_block =
BlockInfo { hash: first_block_hash, parent_hash: [0; 32].into(), number: 1 };
let second_block =
BlockInfo { hash: second_block_hash, parent_hash: first_block_hash, number: 2 };
let third_block =
BlockInfo { hash: third_block_hash, parent_hash: second_block_hash, number: 3 };
let all_subsystems = AllSubsystems::<()>::dummy();
let registry = prometheus::Registry::new();
@@ -244,7 +221,8 @@ fn overseer_metrics_work() {
Some(&registry),
MockSupportsParachains,
spawner,
).unwrap();
)
.unwrap();
let mut handle = Handle::Connected(handle);
let overseer_fut = overseer.run().fuse();
@@ -252,7 +230,9 @@ fn overseer_metrics_work() {
handle.block_imported(second_block).await;
handle.block_imported(third_block).await;
handle.send_msg_anon(AllMessages::CandidateValidation(test_candidate_validation_msg())).await;
handle
.send_msg_anon(AllMessages::CandidateValidation(test_candidate_validation_msg()))
.await;
handle.stop().await;
select! {
@@ -291,15 +271,9 @@ fn overseer_ends_on_subsystem_exit() {
let spawner = sp_core::testing::TaskExecutor::new();
executor::block_on(async move {
let all_subsystems = AllSubsystems::<()>::dummy()
.replace_candidate_backing(ReturnOnStart);
let (overseer, _handle) = Overseer::new(
vec![],
all_subsystems,
None,
MockSupportsParachains,
spawner,
).unwrap();
let all_subsystems = AllSubsystems::<()>::dummy().replace_candidate_backing(ReturnOnStart);
let (overseer, _handle) =
Overseer::new(vec![], all_subsystems, None, MockSupportsParachains, spawner).unwrap();
overseer.run().await.unwrap();
})
@@ -309,7 +283,11 @@ struct TestSubsystem5(metered::MeteredSender<OverseerSignal>);
impl<C> overseer::Subsystem<C, SubsystemError> for TestSubsystem5
where
C: overseer::SubsystemContext<Message=CandidateValidationMessage,Signal=OverseerSignal,AllMessages=AllMessages>,
C: overseer::SubsystemContext<
Message = CandidateValidationMessage,
Signal = OverseerSignal,
AllMessages = AllMessages,
>,
{
fn start(self, mut ctx: C) -> SpawnedSubsystem {
let mut sender = self.0.clone();
@@ -322,7 +300,7 @@ where
Ok(Some(FromOverseer::Signal(OverseerSignal::Conclude))) => break,
Ok(Some(FromOverseer::Signal(s))) => {
sender.send(s).await.unwrap();
continue;
continue
},
Ok(Some(_)) => continue,
Err(_) => break,
@@ -341,7 +319,11 @@ struct TestSubsystem6(metered::MeteredSender<OverseerSignal>);
impl<C> Subsystem<C, SubsystemError> for TestSubsystem6
where
C: overseer::SubsystemContext<Message=CandidateBackingMessage,Signal=OverseerSignal,AllMessages=AllMessages>,
C: overseer::SubsystemContext<
Message = CandidateBackingMessage,
Signal = OverseerSignal,
AllMessages = AllMessages,
>,
{
fn start(self, mut ctx: C) -> SpawnedSubsystem {
let mut sender = self.0.clone();
@@ -354,7 +336,7 @@ where
Ok(Some(FromOverseer::Signal(OverseerSignal::Conclude))) => break,
Ok(Some(FromOverseer::Signal(s))) => {
sender.send(s).await.unwrap();
continue;
continue
},
Ok(Some(_)) => continue,
Err(_) => break,
@@ -380,34 +362,21 @@ fn overseer_start_stop_works() {
let second_block_hash = [2; 32].into();
let third_block_hash = [3; 32].into();
let first_block = BlockInfo {
hash: first_block_hash,
parent_hash: [0; 32].into(),
number: 1,
};
let second_block = BlockInfo {
hash: second_block_hash,
parent_hash: first_block_hash,
number: 2,
};
let third_block = BlockInfo {
hash: third_block_hash,
parent_hash: second_block_hash,
number: 3,
};
let first_block =
BlockInfo { hash: first_block_hash, parent_hash: [0; 32].into(), number: 1 };
let second_block =
BlockInfo { hash: second_block_hash, parent_hash: first_block_hash, number: 2 };
let third_block =
BlockInfo { hash: third_block_hash, parent_hash: second_block_hash, number: 3 };
let (tx_5, mut rx_5) = metered::channel(64);
let (tx_6, mut rx_6) = metered::channel(64);
let all_subsystems = AllSubsystems::<()>::dummy()
.replace_candidate_validation(TestSubsystem5(tx_5))
.replace_candidate_backing(TestSubsystem6(tx_6));
let (overseer, handle) = Overseer::new(
vec![first_block],
all_subsystems,
None,
MockSupportsParachains,
spawner,
).unwrap();
let (overseer, handle) =
Overseer::new(vec![first_block], all_subsystems, None, MockSupportsParachains, spawner)
.unwrap();
let mut handle = Handle::Connected(handle);
let overseer_fut = overseer.run().fuse();
@@ -466,8 +435,9 @@ fn overseer_start_stop_works() {
}
if ss5_results.len() == expected_heartbeats.len() &&
ss6_results.len() == expected_heartbeats.len() {
handle.stop().await;
ss6_results.len() == expected_heartbeats.len()
{
handle.stop().await;
}
}
@@ -487,21 +457,12 @@ fn overseer_finalize_works() {
let second_block_hash = [2; 32].into();
let third_block_hash = [3; 32].into();
let first_block = BlockInfo {
hash: first_block_hash,
parent_hash: [0; 32].into(),
number: 1,
};
let second_block = BlockInfo {
hash: second_block_hash,
parent_hash: [42; 32].into(),
number: 2,
};
let third_block = BlockInfo {
hash: third_block_hash,
parent_hash: second_block_hash,
number: 3,
};
let first_block =
BlockInfo { hash: first_block_hash, parent_hash: [0; 32].into(), number: 1 };
let second_block =
BlockInfo { hash: second_block_hash, parent_hash: [42; 32].into(), number: 2 };
let third_block =
BlockInfo { hash: third_block_hash, parent_hash: second_block_hash, number: 3 };
let (tx_5, mut rx_5) = metered::channel(64);
let (tx_6, mut rx_6) = metered::channel(64);
@@ -517,7 +478,8 @@ fn overseer_finalize_works() {
None,
MockSupportsParachains,
spawner,
).unwrap();
)
.unwrap();
let mut handle = Handle::Connected(handle);
let overseer_fut = overseer.run().fuse();
@@ -568,7 +530,9 @@ fn overseer_finalize_works() {
complete => break,
}
if ss5_results.len() == expected_heartbeats.len() && ss6_results.len() == expected_heartbeats.len() {
if ss5_results.len() == expected_heartbeats.len() &&
ss6_results.len() == expected_heartbeats.len()
{
handle.stop().await;
}
}
@@ -590,30 +554,20 @@ fn do_not_send_empty_leaves_update_on_block_finalization() {
let spawner = sp_core::testing::TaskExecutor::new();
executor::block_on(async move {
let imported_block = BlockInfo {
hash: Hash::random(),
parent_hash: Hash::random(),
number: 1,
};
let imported_block =
BlockInfo { hash: Hash::random(), parent_hash: Hash::random(), number: 1 };
let finalized_block = BlockInfo {
hash: Hash::random(),
parent_hash: Hash::random(),
number: 1,
};
let finalized_block =
BlockInfo { hash: Hash::random(), parent_hash: Hash::random(), number: 1 };
let (tx_5, mut rx_5) = metered::channel(64);
let all_subsystems = AllSubsystems::<()>::dummy()
.replace_candidate_backing(TestSubsystem6(tx_5));
let all_subsystems =
AllSubsystems::<()>::dummy().replace_candidate_backing(TestSubsystem6(tx_5));
let (overseer, handle) = Overseer::new(
Vec::new(),
all_subsystems,
None,
MockSupportsParachains,
spawner,
).unwrap();
let (overseer, handle) =
Overseer::new(Vec::new(), all_subsystems, None, MockSupportsParachains, spawner)
.unwrap();
let mut handle = Handle::Connected(handle);
let overseer_fut = overseer.run().fuse();
@@ -673,17 +627,13 @@ impl CounterSubsystem {
signals_received: Arc<atomic::AtomicUsize>,
msgs_received: Arc<atomic::AtomicUsize>,
) -> Self {
Self {
stop_signals_received,
signals_received,
msgs_received,
}
Self { stop_signals_received, signals_received, msgs_received }
}
}
impl<C, M> Subsystem<C, SubsystemError> for CounterSubsystem
where
C: overseer::SubsystemContext<Message=M,Signal=OverseerSignal,AllMessages=AllMessages>,
C: overseer::SubsystemContext<Message = M, Signal = OverseerSignal, AllMessages = AllMessages>,
M: Send,
{
fn start(self, mut ctx: C) -> SpawnedSubsystem {
@@ -694,15 +644,15 @@ where
match ctx.try_recv().await {
Ok(Some(FromOverseer::Signal(OverseerSignal::Conclude))) => {
self.stop_signals_received.fetch_add(1, atomic::Ordering::SeqCst);
break;
break
},
Ok(Some(FromOverseer::Signal(_))) => {
self.signals_received.fetch_add(1, atomic::Ordering::SeqCst);
continue;
continue
},
Ok(Some(FromOverseer::Communication { .. })) => {
self.msgs_received.fetch_add(1, atomic::Ordering::SeqCst);
continue;
continue
},
Err(_) => (),
_ => (),
@@ -872,47 +822,74 @@ fn overseer_all_subsystems_receive_signals_and_messages() {
dispute_distribution: subsystem.clone(),
chain_selection: subsystem.clone(),
};
let (overseer, handle) = Overseer::new(
vec![],
all_subsystems,
None,
MockSupportsParachains,
spawner,
).unwrap();
let (overseer, handle) =
Overseer::new(vec![], all_subsystems, None, MockSupportsParachains, spawner).unwrap();
let mut handle = Handle::Connected(handle);
let overseer_fut = overseer.run().fuse();
pin_mut!(overseer_fut);
// send a signal to each subsystem
handle.block_imported(BlockInfo {
hash: Default::default(),
parent_hash: Default::default(),
number: Default::default(),
}).await;
handle
.block_imported(BlockInfo {
hash: Default::default(),
parent_hash: Default::default(),
number: Default::default(),
})
.await;
// send a msg to each subsystem
// except for BitfieldSigning and GossipSupport as the messages are not instantiable
handle.send_msg_anon(AllMessages::CandidateValidation(test_candidate_validation_msg())).await;
handle.send_msg_anon(AllMessages::CandidateBacking(test_candidate_backing_msg())).await;
handle.send_msg_anon(AllMessages::CollationGeneration(test_collator_generation_msg())).await;
handle.send_msg_anon(AllMessages::CollatorProtocol(test_collator_protocol_msg())).await;
handle.send_msg_anon(AllMessages::StatementDistribution(test_statement_distribution_msg())).await;
handle.send_msg_anon(AllMessages::AvailabilityRecovery(test_availability_recovery_msg())).await;
handle
.send_msg_anon(AllMessages::CandidateValidation(test_candidate_validation_msg()))
.await;
handle
.send_msg_anon(AllMessages::CandidateBacking(test_candidate_backing_msg()))
.await;
handle
.send_msg_anon(AllMessages::CollationGeneration(test_collator_generation_msg()))
.await;
handle
.send_msg_anon(AllMessages::CollatorProtocol(test_collator_protocol_msg()))
.await;
handle
.send_msg_anon(AllMessages::StatementDistribution(test_statement_distribution_msg()))
.await;
handle
.send_msg_anon(AllMessages::AvailabilityRecovery(test_availability_recovery_msg()))
.await;
// handle.send_msg_anon(AllMessages::BitfieldSigning(test_bitfield_signing_msg())).await;
// handle.send_msg_anon(AllMessages::GossipSupport(test_bitfield_signing_msg())).await;
handle.send_msg_anon(AllMessages::BitfieldDistribution(test_bitfield_distribution_msg())).await;
handle
.send_msg_anon(AllMessages::BitfieldDistribution(test_bitfield_distribution_msg()))
.await;
handle.send_msg_anon(AllMessages::Provisioner(test_provisioner_msg())).await;
handle.send_msg_anon(AllMessages::RuntimeApi(test_runtime_api_msg())).await;
handle.send_msg_anon(AllMessages::AvailabilityStore(test_availability_store_msg())).await;
handle.send_msg_anon(AllMessages::NetworkBridge(test_network_bridge_msg())).await;
handle
.send_msg_anon(AllMessages::AvailabilityStore(test_availability_store_msg()))
.await;
handle
.send_msg_anon(AllMessages::NetworkBridge(test_network_bridge_msg()))
.await;
handle.send_msg_anon(AllMessages::ChainApi(test_chain_api_msg())).await;
handle.send_msg_anon(AllMessages::ApprovalDistribution(test_approval_distribution_msg())).await;
handle.send_msg_anon(AllMessages::ApprovalVoting(test_approval_voting_msg())).await;
handle.send_msg_anon(AllMessages::DisputeCoordinator(test_dispute_coordinator_msg())).await;
handle.send_msg_anon(AllMessages::DisputeParticipation(test_dispute_participation_msg())).await;
handle.send_msg_anon(AllMessages::DisputeDistribution(test_dispute_distribution_msg())).await;
handle.send_msg_anon(AllMessages::ChainSelection(test_chain_selection_msg())).await;
handle
.send_msg_anon(AllMessages::ApprovalDistribution(test_approval_distribution_msg()))
.await;
handle
.send_msg_anon(AllMessages::ApprovalVoting(test_approval_voting_msg()))
.await;
handle
.send_msg_anon(AllMessages::DisputeCoordinator(test_dispute_coordinator_msg()))
.await;
handle
.send_msg_anon(AllMessages::DisputeParticipation(test_dispute_participation_msg()))
.await;
handle
.send_msg_anon(AllMessages::DisputeDistribution(test_dispute_distribution_msg()))
.await;
handle
.send_msg_anon(AllMessages::ChainSelection(test_chain_selection_msg()))
.await;
// Wait until all subsystems have received. Otherwise the messages might race against
// the conclude signal.
@@ -927,7 +904,7 @@ fn overseer_all_subsystems_receive_signals_and_messages() {
} else {
break
}
}
},
Some(_) => panic!("exited too early"),
}
}
@@ -1055,17 +1032,16 @@ fn context_holds_onto_message_until_enough_signals_received() {
assert_matches!(ctx.recv().await.unwrap(), FromOverseer::Signal(OverseerSignal::Conclude));
assert_eq!(ctx.signals_received.load(), 1);
bounded_tx.send(MessagePacket {
signals_received: 2,
message: (),
}).await.unwrap();
unbounded_tx.unbounded_send(MessagePacket {
signals_received: 2,
message: (),
}).unwrap();
bounded_tx
.send(MessagePacket { signals_received: 2, message: () })
.await
.unwrap();
unbounded_tx
.unbounded_send(MessagePacket { signals_received: 2, message: () })
.unwrap();
match poll!(ctx.recv()) {
Poll::Pending => {}
Poll::Pending => {},
Poll::Ready(_) => panic!("ready too early"),
};