Unverified Commit 285a35bf authored by Bernhard Schuster's avatar Bernhard Schuster Committed by GitHub
Browse files

feat/overseer: introduce closure init (#3775)

* feat/overseer: introduce closure init

Enables removal of the connected/disconnected overseer state.

* feat/overseer: allow replacement logic to access the original

Allows to re-use init-once types, which would otherwise error.

* feat/overseer: introduce external connector

Preparation for removal of `AllSubsystems`
which is another prerequisite for removing
the connect/disconnect state.

* fix/test: replace needs closure

* fixup

* simplify

* mea culpa

* all-subsystems-gen test
parent 6d1d883f
Pipeline #155499 passed with stages
in 37 minutes and 11 seconds
......@@ -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,
}
......
......@@ -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::*;
......@@ -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)
......
......@@ -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
......@@ -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,
)*
}
}
......
......@@ -12,5 +12,5 @@ fn main() {
a: 0_u16,
b: 1_u16,
};
let _all = all.replace_a(77u8);
let _all = all.replace_a(|_| 77u8);
}
......@@ -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>`
......@@ -13,5 +13,5 @@ fn main() {
a: 0_f32,
b: 1_u16,
};
let _all = all.replace_a(77u8);
let _all = all.replace_a(|_| 77u8);
}
......@@ -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`
......@@ -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>`
......@@ -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);
}
......@@ -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();
......
......@@ -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::*;
......@@ -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("");
......@@ -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 >,
......@@ -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,
......@@ -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
}
)*
......@@ -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
......@@ -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();
......
......@@ -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,
......
......@@ -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();
......@@ -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();
......@@ -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();
......@@ -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(
......@@ -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)
......
......@@ -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,
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment