Skip to content
This repository has been archived by the owner on Nov 15, 2023. It is now read-only.

feat/overseer: introduce closure init #3775

Merged
8 commits merged into from
Sep 4, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions node/core/candidate-validation/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -69,8 +69,10 @@ pub struct Config {

/// The candidate validation subsystem.
pub struct CandidateValidationSubsystem {
metrics: Metrics,
pvf_metrics: polkadot_node_core_pvf::Metrics,
#[allow(missing_docs)]
pub metrics: Metrics,
#[allow(missing_docs)]
pub pvf_metrics: polkadot_node_core_pvf::Metrics,
config: Config,
}

Expand Down
21 changes: 11 additions & 10 deletions node/malus/src/variant-a.rs
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,8 @@ use polkadot_cli::{

// Import extra types relevant to the particular
// subsystem.
use polkadot_node_core_candidate_validation::{CandidateValidationSubsystem, Metrics};
use polkadot_node_core_candidate_validation::CandidateValidationSubsystem;
use polkadot_node_subsystem::messages::CandidateValidationMessage;
use polkadot_node_subsystem_util::metrics::Metrics as _;

// Filter wrapping related types.
use malus::*;
Expand Down Expand Up @@ -88,14 +87,16 @@ impl OverseerGen for BehaveMaleficient {
// modify the subsystem(s) as needed:
let all_subsystems = create_default_subsystems(args)?.replace_candidate_validation(
// create the filtered subsystem
FilteredSubsystem::new(
CandidateValidationSubsystem::with_config(
candidate_validation_config,
Metrics::register(registry)?,
polkadot_node_core_pvf::Metrics::register(registry)?,
),
Skippy::default(),
),
|orig: CandidateValidationSubsystem| {
FilteredSubsystem::new(
CandidateValidationSubsystem::with_config(
candidate_validation_config,
orig.metrics,
orig.pvf_metrics,
),
Skippy::default(),
)
},
);

Overseer::new(leaves, all_subsystems, registry, runtime_client, spawner)
Expand Down
25 changes: 21 additions & 4 deletions node/overseer/all-subsystems-gen/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -118,10 +118,11 @@ fn impl_subsystems_gen(item: TokenStream) -> Result<proc_macro2::TokenStream> {

// generate an impl of `fn replace_#name`
for NameTyTup { field: replacable_item, ty: replacable_item_ty } in replacable_items {
let keeper = all_fields
let keeper = &all_fields
.iter()
.filter(|ntt| ntt.field != replacable_item)
.map(|ntt| ntt.field.clone());
.map(|ntt| ntt.field.clone())
.collect::<Vec<_>>();
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
Expand Down Expand Up @@ -154,11 +155,27 @@ fn impl_subsystems_gen(item: TokenStream) -> Result<proc_macro2::TokenStream> {
additive.extend(quote! {
impl #orig_generics #strukt_ty #orig_generics {
#[doc = #msg]
pub fn #fname < NEW > (self, replacement: NEW) -> #strukt_ty #modified_generics {
pub fn #fname < NEW, F > (self, gen_replacement_fn: F) -> #strukt_ty #modified_generics
where
F: FnOnce(#replacable_item_ty) -> NEW,
{
let Self {
// To be replaced field:
#replacable_item,
// Fields to keep:
#(
#keeper,
)*
} = self;

// Some cases require that parts of the original are copied
// over, since they include a one time initialization.
let replacement = gen_replacement_fn(#replacable_item);

#strukt_ty :: #modified_generics {
#replacable_item: replacement,
#(
#keeper: self.#keeper,
#keeper,
)*
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,5 +12,5 @@ fn main() {
a: 0_u16,
b: 1_u16,
};
let _all = all.replace_a(77u8);
let _all = all.replace_a(|_| 77u8);
}
Original file line number Diff line number Diff line change
Expand Up @@ -10,5 +10,5 @@ error[E0599]: no method named `replace_a` found for struct `AllSubsystems<u16>`
5 | struct AllSubsystems<X> {
| ----------------------- method `replace_a` not found for this
...
15 | let _all = all.replace_a(77u8);
15 | let _all = all.replace_a(|_| 77u8);
| ^^^^^^^^^ method not found in `AllSubsystems<u16>`
Original file line number Diff line number Diff line change
Expand Up @@ -13,5 +13,5 @@ fn main() {
a: 0_f32,
b: 1_u16,
};
let _all = all.replace_a(77u8);
let _all = all.replace_a(|_| 77u8);
}
Original file line number Diff line number Diff line change
Expand Up @@ -10,5 +10,5 @@ error[E0599]: no method named `replace_a` found for struct `AllSubsystems` in th
6 | struct AllSubsystems {
| -------------------- method `replace_a` not found for this
...
16 | let _all = all.replace_a(77u8);
16 | let _all = all.replace_a(|_| 77u8);
| ^^^^^^^^^ method not found in `AllSubsystems`
Original file line number Diff line number Diff line change
Expand Up @@ -10,5 +10,5 @@ error[E0599]: no method named `replace_a` found for struct `AllSubsystems<u16>`
6 | struct AllSubsystems<X> {
| ----------------------- method `replace_a` not found for this
...
16 | let _all = all.replace_a(77u8);
16 | let _all = all.replace_a(|_| 77u8);
| ^^^^^^^^^ method not found in `AllSubsystems<u16>`
Original file line number Diff line number Diff line change
Expand Up @@ -13,5 +13,5 @@ fn main() {
a: 0u8,
b: 1u16,
};
let _all: AllSubsystems<_,_> = all.replace_a::<u32>(777_777u32);
let _all: AllSubsystems<_,_> = all.replace_a::<u32,_>(|_| 777_777u32);
}
4 changes: 2 additions & 2 deletions node/overseer/examples/minimal-example.rs
Original file line number Diff line number Diff line change
Expand Up @@ -170,8 +170,8 @@ fn main() {
});

let all_subsystems = AllSubsystems::<()>::dummy()
.replace_candidate_validation(Subsystem2)
.replace_candidate_backing(Subsystem1);
.replace_candidate_validation(|_| Subsystem2)
.replace_candidate_backing(|orig| orig);

let (overseer, _handle) =
Overseer::new(vec![], all_subsystems, None, AlwaysSupportsParachains, spawner).unwrap();
Expand Down
104 changes: 94 additions & 10 deletions node/overseer/overseer-gen/proc-macro/src/impl_builder.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
// You should have received a copy of the GNU General Public License
// along with Polkadot. If not, see <http://www.gnu.org/licenses/>.

use quote::quote;
use quote::{format_ident, quote};
use syn::Ident;

use super::*;
Expand All @@ -27,8 +27,14 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
let overseer_name = info.overseer_name.clone();
let builder = Ident::new(&(overseer_name.to_string() + "Builder"), overseer_name.span());
let handle = Ident::new(&(overseer_name.to_string() + "Handle"), overseer_name.span());
let connector = Ident::new(&(overseer_name.to_string() + "Connector"), overseer_name.span());

let subsystem_name = &info.subsystem_names_without_wip();
let subsystem_name_init_with = &info
.subsystem_names_without_wip()
.iter()
.map(|subsystem_name| format_ident!("{}_with", subsystem_name))
.collect::<Vec<_>>();
let builder_generic_ty = &info.builder_generic_types();

let channel_name = &info.channel_names_without_wip("");
Expand Down Expand Up @@ -106,10 +112,65 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
/// Handle for an overseer.
pub type #handle = #support_crate ::metered::MeteredSender< #event >;

/// External connector.
pub struct #connector {
/// Publicly accessible handle, to be used for setting up
/// components that are _not_ subsystems but access is needed
/// due to other limitations.
///
/// For subsystems, use the `_with` variants of the builder.
handle: #handle,
/// The side consumed by the `spawned` side of the overseer pattern.
consumer: #support_crate ::metered::MeteredReceiver < #event >,
}

impl #connector {
/// Obtain access to the overseer handle.
pub fn as_handle_mut(&mut self) -> &mut #handle {
&mut self.handle
}
/// Obtain access to the overseer handle.
pub fn as_handle(&mut self) -> &#handle {
&self.handle
}
}

impl ::std::default::Default for #connector {
fn default() -> Self {
let (events_tx, events_rx) = #support_crate ::metered::channel::<
#event
>(SIGNAL_CHANNEL_CAPACITY);

Self {
handle: events_tx,
consumer: events_rx,
}
}
}

/// Convenience alias.
type SubsystemInitFn<T> = Box<dyn FnOnce(#handle) -> ::std::result::Result<T, #error_ty> >;

/// Init kind of a field of the overseer.
enum FieldInitMethod<T> {
/// Defer initialization to a point where the `handle` is available.
Fn(SubsystemInitFn<T>),
/// Directly initialize the subsystem with the given subsystem type `T`.
Value(T),
/// Subsystem field does not have value just yet.
Uninitialized
}

impl<T> ::std::default::Default for FieldInitMethod<T> {
fn default() -> Self {
Self::Uninitialized
}
}

#[allow(missing_docs)]
pub struct #builder #builder_generics {
#(
#subsystem_name : ::std::option::Option< #builder_generic_ty >,
#subsystem_name : FieldInitMethod< #builder_generic_ty >,
)*
#(
#baggage_name : ::std::option::Option< #baggage_ty >,
Expand All @@ -129,7 +190,7 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {

Self {
#(
#subsystem_name: None,
#subsystem_name: Default::default(),
)*
#(
#baggage_name: None,
Expand All @@ -152,7 +213,18 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
#(
/// Specify the particular subsystem implementation.
pub fn #subsystem_name (mut self, subsystem: #builder_generic_ty ) -> Self {
self. #subsystem_name = Some( subsystem );
self. #subsystem_name = FieldInitMethod::Value( subsystem );
self
}

/// Specify the particular subsystem by giving a init function.
pub fn #subsystem_name_init_with <'a, F> (mut self, subsystem_init_fn: F ) -> Self
where
F: 'static + FnOnce(#handle) -> ::std::result::Result<#builder_generic_ty, #error_ty>,
{
self. #subsystem_name = FieldInitMethod::Fn(
Box::new(subsystem_init_fn) as SubsystemInitFn<#builder_generic_ty>
);
self
}
)*
Expand All @@ -166,13 +238,20 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
)*

/// Complete the construction and create the overseer type.
pub fn build(mut self) -> ::std::result::Result<(#overseer_name #generics, #handle), #error_ty>
pub fn build(mut self) -> ::std::result::Result<(#overseer_name #generics, #handle), #error_ty> {
let connector = #connector ::default();
self.build_with_connector(connector)
}

/// Complete the construction and create the overseer type based on an existing `connector`.
pub fn build_with_connector(mut self, connector: #connector) -> ::std::result::Result<(#overseer_name #generics, #handle), #error_ty>
{
let (events_tx, events_rx) = #support_crate ::metered::channel::<
#event
>(SIGNAL_CHANNEL_CAPACITY);
let #connector {
handle: events_tx,
consumer: events_rx,
} = connector;

let handle: #handle = events_tx.clone();
let handle = events_tx.clone();

let (to_overseer_tx, to_overseer_rx) = #support_crate ::metered::unbounded::<
ToOverseer
Expand Down Expand Up @@ -212,7 +291,12 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
#(
// TODO generate a builder pattern that ensures this
// TODO https://github.com/paritytech/polkadot/issues/3427
let #subsystem_name = self. #subsystem_name .expect("All subsystem must exist with the builder pattern.");
let #subsystem_name = match self. #subsystem_name {
FieldInitMethod::Fn(func) => func(handle.clone())?,
FieldInitMethod::Value(val) => val,
FieldInitMethod::Uninitialized =>
panic!("All subsystems must exist with the builder pattern."),
};

let unbounded_meter = #channel_name_unbounded_rx.meter().clone();

Expand Down
2 changes: 1 addition & 1 deletion node/overseer/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -543,7 +543,7 @@ where
/// }
/// let spawner = sp_core::testing::TaskExecutor::new();
/// let all_subsystems = AllSubsystems::<()>::dummy()
/// .replace_candidate_validation(ValidationSubsystem);
/// .replace_candidate_validation(|_| ValidationSubsystem);
/// let (overseer, _handle) = Overseer::new(
/// vec![],
/// all_subsystems,
Expand Down
17 changes: 9 additions & 8 deletions node/overseer/src/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -161,8 +161,8 @@ fn overseer_works() {
let mut s2_rx = s2_rx.fuse();

let all_subsystems = AllSubsystems::<()>::dummy()
.replace_candidate_validation(TestSubsystem1(s1_tx))
.replace_candidate_backing(TestSubsystem2(s2_tx));
.replace_candidate_validation(move |_| TestSubsystem1(s1_tx))
.replace_candidate_backing(move |_| TestSubsystem2(s2_tx));

let (overseer, handle) =
Overseer::new(vec![], all_subsystems, None, MockSupportsParachains, spawner).unwrap();
Expand Down Expand Up @@ -278,7 +278,8 @@ 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 all_subsystems =
AllSubsystems::<()>::dummy().replace_candidate_backing(|_| ReturnOnStart);
let (overseer, _handle) =
Overseer::new(vec![], all_subsystems, None, MockSupportsParachains, spawner).unwrap();

Expand Down Expand Up @@ -379,8 +380,8 @@ fn overseer_start_stop_works() {
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));
.replace_candidate_validation(move |_| TestSubsystem5(tx_5))
.replace_candidate_backing(move |_| TestSubsystem6(tx_6));
let (overseer, handle) =
Overseer::new(vec![first_block], all_subsystems, None, MockSupportsParachains, spawner)
.unwrap();
Expand Down Expand Up @@ -475,8 +476,8 @@ fn overseer_finalize_works() {
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));
.replace_candidate_validation(move |_| TestSubsystem5(tx_5))
.replace_candidate_backing(move |_| TestSubsystem6(tx_6));

// start with two forks of different height.
let (overseer, handle) = Overseer::new(
Expand Down Expand Up @@ -570,7 +571,7 @@ fn do_not_send_empty_leaves_update_on_block_finalization() {
let (tx_5, mut rx_5) = metered::channel(64);

let all_subsystems =
AllSubsystems::<()>::dummy().replace_candidate_backing(TestSubsystem6(tx_5));
AllSubsystems::<()>::dummy().replace_candidate_backing(move |_| TestSubsystem6(tx_5));

let (overseer, handle) =
Overseer::new(Vec::new(), all_subsystems, None, MockSupportsParachains, spawner)
Expand Down
2 changes: 1 addition & 1 deletion node/subsystem-test-helpers/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -384,7 +384,7 @@ mod tests {
let spawner = sp_core::testing::TaskExecutor::new();
let (tx, rx) = mpsc::channel(2);
let all_subsystems =
AllSubsystems::<()>::dummy().replace_collator_protocol(ForwardSubsystem(tx));
AllSubsystems::<()>::dummy().replace_collator_protocol(|_| ForwardSubsystem(tx));
let (overseer, handle) = Overseer::new(
Vec::new(),
all_subsystems,
Expand Down