diff --git a/cli/src/cli.rs b/cli/src/cli.rs index cc449bd844ae..804c4a9f81c6 100644 --- a/cli/src/cli.rs +++ b/cli/src/cli.rs @@ -138,6 +138,12 @@ pub struct RunCmd { /// telemetry, if telemetry is enabled. #[clap(long)] pub no_hardware_benchmarks: bool, + + /// Overseer message capacity override. + /// + /// **Dangerous!** Do not touch unless explicitly adviced to. + #[clap(long)] + pub overseer_channel_capacity_override: Option, } #[allow(missing_docs)] diff --git a/cli/src/command.rs b/cli/src/command.rs index 8869d9a4ead4..19c046e7c2e2 100644 --- a/cli/src/command.rs +++ b/cli/src/command.rs @@ -332,6 +332,7 @@ where None, false, overseer_gen, + cli.run.overseer_channel_capacity_override, hwbench, ) .map(|full| full.task_manager) diff --git a/node/overseer/overseer-gen/proc-macro/src/impl_builder.rs b/node/overseer/overseer-gen/proc-macro/src/impl_builder.rs index a1c9450c4178..6d634cc9f344 100644 --- a/node/overseer/overseer-gen/proc-macro/src/impl_builder.rs +++ b/node/overseer/overseer-gen/proc-macro/src/impl_builder.rs @@ -150,6 +150,9 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { #baggage_name: self. #baggage_name, )* spawner: self.spawner, + + channel_capacity: self.channel_capacity, + signal_capacity: self.signal_capacity, } } /// Specify the the initialization function for a subsystem @@ -171,6 +174,10 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { #baggage_name: self. #baggage_name, )* spawner: self.spawner, + + + channel_capacity: self.channel_capacity, + signal_capacity: self.signal_capacity, } } } @@ -207,6 +214,9 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { #baggage_name: self. #baggage_name, )* spawner: self.spawner, + + channel_capacity: self.channel_capacity, + signal_capacity: self.signal_capacity, } } } @@ -254,6 +264,9 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { #to_keep_baggage_name: self. #to_keep_baggage_name, )* spawner: self.spawner, + + channel_capacity: self.channel_capacity, + signal_capacity: self.signal_capacity, } } } @@ -272,6 +285,9 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { #to_keep_baggage_name: self. #to_keep_baggage_name, )* spawner: self.spawner, + + channel_capacity: self.channel_capacity, + signal_capacity: self.signal_capacity, } } } @@ -359,13 +375,12 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { pub fn handle(&self) -> #handle { self.handle.clone() } - } - impl ::std::default::Default for #connector { - fn default() -> Self { + /// Create a new connector with non-default event channel capacity. + pub fn with_event_capacity(event_channel_capacity: usize) -> Self { let (events_tx, events_rx) = #support_crate ::metered::channel::< #event - >(SIGNAL_CHANNEL_CAPACITY); + >(event_channel_capacity); Self { handle: events_tx, @@ -373,6 +388,12 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { } } } + + impl ::std::default::Default for #connector { + fn default() -> Self { + Self::with_event_capacity(SIGNAL_CHANNEL_CAPACITY) + } + } }); ts.extend(quote!{ @@ -385,6 +406,11 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { #baggage_name: #baggage_passthrough_state_generics, )* spawner: InitStateSpawner, + // user provided runtime overrides, + // if `None`, the `overlord(message_capacity=123,..)` is used + // or the default value. + channel_capacity: Option, + signal_capacity: Option, } }); @@ -406,6 +432,9 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { #field_name: Missing::<#field_type>::default(), )* spawner: Missing::::default(), + + channel_capacity: None, + signal_capacity: None, } } } @@ -419,18 +448,48 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { #spawner_where_clause { /// The `spawner` to use for spawning tasks. - pub fn spawner(self, spawner: S) -> #builder, #( #subsystem_passthrough_state_generics, )* #( #baggage_passthrough_state_generics, )*> + pub fn spawner(self, spawner: S) -> #builder< + Init, + #( #subsystem_passthrough_state_generics, )* + #( #baggage_passthrough_state_generics, )* + > { #builder { #( #field_name: self. #field_name, )* spawner: Init::::Value(spawner), + + channel_capacity: self.channel_capacity, + signal_capacity: self.signal_capacity, } } } }); + // message and signal channel capacity + ts.extend(quote! { + impl + #builder, #( #subsystem_passthrough_state_generics, )* #( #baggage_passthrough_state_generics, )*> + where + #spawner_where_clause, + { + /// Set the interconnecting signal channel capacity. + pub fn signal_channel_capacity(mut self, capacity: usize) -> Self + { + self.signal_capacity = Some(capacity); + self + } + + /// Set the interconnecting message channel capacities. + pub fn message_channel_capacity(mut self, capacity: usize) -> Self + { + self.channel_capacity = Some(capacity); + self + } + } + }); + ts.extend(quote! { /// Type used to represent a builder where all fields are initialized and the overseer could be constructed. pub type #initialized_builder<#initialized_builder_generics> = #builder, #( Init<#field_type>, )*>; @@ -446,7 +505,9 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { /// Complete the construction and create the overseer type. pub fn build(self) -> ::std::result::Result<(#overseer_name, #handle), #error_ty> { - let connector = #connector ::default(); + let connector = #connector ::with_event_capacity( + self.signal_capacity.unwrap_or(SIGNAL_CHANNEL_CAPACITY) + ); self.build_with_connector(connector) } @@ -470,7 +531,9 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { = #support_crate ::metered::channel::< MessagePacket< #consumes > - >(CHANNEL_CAPACITY); + >( + self.channel_capacity.unwrap_or(CHANNEL_CAPACITY) + ); )* #( @@ -510,7 +573,9 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream { let message_rx: SubsystemIncomingMessages< #consumes > = #support_crate ::select( #channel_name_rx, #channel_name_unbounded_rx ); - let (signal_tx, signal_rx) = #support_crate ::metered::channel(SIGNAL_CHANNEL_CAPACITY); + let (signal_tx, signal_rx) = #support_crate ::metered::channel( + self.signal_capacity.unwrap_or(SIGNAL_CHANNEL_CAPACITY) + ); // Generate subsystem name based on overseer field name. let subsystem_string = String::from(stringify!(#subsystem_name)); diff --git a/node/service/src/lib.rs b/node/service/src/lib.rs index 43897cb8c7d8..810b0b61aaf3 100644 --- a/node/service/src/lib.rs +++ b/node/service/src/lib.rs @@ -719,6 +719,7 @@ pub fn new_full( program_path: Option, overseer_enable_anyways: bool, overseer_gen: OverseerGenerator, + overseer_message_channel_capacity_override: Option, hwbench: Option, ) -> Result>>, Error> where @@ -1038,6 +1039,7 @@ where chain_selection_config, dispute_coordinator_config, pvf_checker_enabled, + overseer_message_channel_capacity_override, }, ) .map_err(|e| { @@ -1326,6 +1328,7 @@ pub fn build_full( telemetry_worker_handle: Option, overseer_enable_anyways: bool, overseer_gen: impl OverseerGen, + overseer_message_channel_override: Option, hwbench: Option, ) -> Result, Error> { #[cfg(feature = "rococo-native")] @@ -1343,6 +1346,7 @@ pub fn build_full( None, overseer_enable_anyways, overseer_gen, + overseer_message_channel_override, hwbench, ) .map(|full| full.with_client(Client::Rococo)) @@ -1360,6 +1364,7 @@ pub fn build_full( None, overseer_enable_anyways, overseer_gen, + overseer_message_channel_override, hwbench, ) .map(|full| full.with_client(Client::Kusama)) @@ -1377,6 +1382,7 @@ pub fn build_full( None, overseer_enable_anyways, overseer_gen, + overseer_message_channel_override, hwbench, ) .map(|full| full.with_client(Client::Westend)) @@ -1394,6 +1400,10 @@ pub fn build_full( None, overseer_enable_anyways, overseer_gen, + overseer_message_channel_override.map(|capacity| { + gum::warn!("Channel capacity should _never_ be tampered with on polkadot!"); + capacity + }), hwbench, ) .map(|full| full.with_client(Client::Polkadot)) diff --git a/node/service/src/overseer.rs b/node/service/src/overseer.rs index fd07ddfe825d..aa90abb91fb2 100644 --- a/node/service/src/overseer.rs +++ b/node/service/src/overseer.rs @@ -111,6 +111,8 @@ where pub dispute_coordinator_config: DisputeCoordinatorConfig, /// Enable PVF pre-checking pub pvf_checker_enabled: bool, + /// Overseer channel capacity override. + pub overseer_message_channel_capacity_override: Option, } /// Obtain a prepared `OverseerBuilder`, that is initialized @@ -138,6 +140,7 @@ pub fn prepared_overseer_builder<'a, Spawner, RuntimeClient>( chain_selection_config, dispute_coordinator_config, pvf_checker_enabled, + overseer_message_channel_capacity_override, }: OverseerGenArgs<'a, Spawner, RuntimeClient>, ) -> Result< InitializedOverseerBuilder< @@ -292,7 +295,12 @@ where .known_leaves(LruCache::new(KNOWN_LEAVES_CACHE_SIZE)) .metrics(metrics) .spawner(spawner); - Ok(builder) + + if let Some(capacity) = overseer_message_channel_capacity_override { + Ok(builder.message_channel_capacity(capacity)) + } else { + Ok(builder) + } } /// Trait for the `fn` generating the overseer. diff --git a/node/test/service/src/lib.rs b/node/test/service/src/lib.rs index 4ac3d765c82b..c1a381e10089 100644 --- a/node/test/service/src/lib.rs +++ b/node/test/service/src/lib.rs @@ -98,6 +98,7 @@ pub fn new_full( false, polkadot_service::RealOverseerGen, None, + None, ) } diff --git a/parachain/test-parachains/adder/collator/src/main.rs b/parachain/test-parachains/adder/collator/src/main.rs index 2b3e468d9b42..00e1532fcf68 100644 --- a/parachain/test-parachains/adder/collator/src/main.rs +++ b/parachain/test-parachains/adder/collator/src/main.rs @@ -65,12 +65,13 @@ fn main() -> Result<()> { config, polkadot_service::IsCollator::Yes(collator.collator_key()), None, - true, + false, None, None, false, polkadot_service::RealOverseerGen, None, + None, ) .map_err(|e| e.to_string())?; let mut overseer_handle = full_node diff --git a/parachain/test-parachains/undying/collator/src/main.rs b/parachain/test-parachains/undying/collator/src/main.rs index 5bacf927e4fb..65e97f34f695 100644 --- a/parachain/test-parachains/undying/collator/src/main.rs +++ b/parachain/test-parachains/undying/collator/src/main.rs @@ -65,12 +65,13 @@ fn main() -> Result<()> { config, polkadot_service::IsCollator::Yes(collator.collator_key()), None, - true, + false, None, None, false, polkadot_service::RealOverseerGen, None, + None, ) .map_err(|e| e.to_string())?; let mut overseer_handle = full_node