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 7 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