A0-4318: network rate-limiter for the substrate network

bin/node/src/aleph_cli.rs

@@ -47,9 +47,13 @@ pub struct AlephCli {
     #[clap(long, default_value_t = false)]
     enable_pruning: bool,
 
-    /// Maximum bit-rate per node in bytes per second of the alephbft validator network.
-    #[clap(long, default_value_t = 64 * 1024)]
-    alephbft_bit_rate_per_connection: u64,
+    /// Maximum bit-rate in bits per second of the alephbft validator network.
+    #[clap(long, default_value_t = 768 * 1024)]
+    alephbft_network_bit_rate: u64,
+
+    /// Maximum bit-rate in bits per second of the substrate network.
+    #[clap(long, default_value_t = 5*1024*1024)]
+    substrate_network_bit_rate: u64,
 
     /// Don't spend some extra time to collect more debugging data (e.g. validator network details).
     /// By default collecting is enabled, as the impact on performance is negligible, if any.
@@ -93,8 +97,12 @@ impl AlephCli {
         self.enable_pruning
     }
 
-    pub fn alephbft_bit_rate_per_connection(&self) -> u64 {
-        self.alephbft_bit_rate_per_connection
+    pub fn alephbft_network_bit_rate(&self) -> u64 {
+        self.alephbft_network_bit_rate
+    }
+
+    pub fn substrate_network_bit_rate(&self) -> u64 {
+        self.substrate_network_bit_rate
     }
 
     pub fn no_collection_of_extra_debugging_data(&self) -> bool {

bin/node/src/service.rs

@@ -234,10 +234,8 @@ fn get_proposer_factory(
 
 fn get_rate_limit_config(aleph_config: &AlephCli) -> RateLimiterConfig {
     RateLimiterConfig {
-        alephbft_bit_rate_per_connection: aleph_config
-            .alephbft_bit_rate_per_connection()
-            .try_into()
-            .unwrap_or(usize::MAX),
+        alephbft_network_bit_rate: aleph_config.alephbft_network_bit_rate(),
+        substrate_network_bit_rate: aleph_config.substrate_network_bit_rate(),
     }
 }
 
@@ -296,6 +294,11 @@ pub fn new_authority(
     )?;
 
     let import_queue_handle = BlockImporter::new(service_components.import_queue.service());
+    let rate_limiter_config = get_rate_limit_config(&aleph_config);
+    let network_config = finality_aleph::SubstrateNetworkConfig {
+        substrate_network_bit_rate: rate_limiter_config.substrate_network_bit_rate,
+        network_config: config.network.clone(),
+    };
 
     let BuildNetworkOutput {
         network,
@@ -305,7 +308,7 @@ pub fn new_authority(
         tx_handler_controller,
         system_rpc_tx,
     } = build_network(
-        &config.network,
+        network_config,
         config.protocol_id(),
         service_components.client.clone(),
         major_sync,
@@ -370,8 +373,6 @@ pub fn new_authority(
         .spawn_essential_handle()
         .spawn_blocking("aura", None, aura);
 
-    let rate_limiter_config = get_rate_limit_config(&aleph_config);
-
     let AlephRuntimeVars {
         millisecs_per_block,
         session_period,

clique/src/rate_limiting.rs

@@ -1,46 +1,22 @@
-use rate_limiter::{RateLimiter, SleepingRateLimiter};
-use tokio::io::AsyncRead;
+use rate_limiter::{RateLimitedAsyncRead, RateLimiterImpl, SharingRateLimiter};
 
 use crate::{ConnectionInfo, Data, Dialer, Listener, PeerAddressInfo, Splittable, Splitted};
 
-pub struct RateLimitedAsyncRead<Read> {
-    rate_limiter: RateLimiter,
-    read: Read,
-}
-
-impl<Read> RateLimitedAsyncRead<Read> {
-    pub fn new(read: Read, rate_limiter: RateLimiter) -> Self {
-        Self { rate_limiter, read }
-    }
-}
-
-impl<Read: AsyncRead + Unpin> AsyncRead for RateLimitedAsyncRead<Read> {
-    fn poll_read(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-        buf: &mut tokio::io::ReadBuf<'_>,
-    ) -> std::task::Poll<std::io::Result<()>> {
-        let this = self.get_mut();
-        let read = std::pin::Pin::new(&mut this.read);
-        this.rate_limiter.rate_limit(read, cx, buf)
-    }
-}
-
 impl<Read: ConnectionInfo> ConnectionInfo for RateLimitedAsyncRead<Read> {
     fn peer_address_info(&self) -> PeerAddressInfo {
-        self.read.peer_address_info()
+        self.inner().peer_address_info()
     }
 }
 
 /// Implementation of the [Dialer] trait governing all returned [Dialer::Connection] instances by a rate-limiting wrapper.
 #[derive(Clone)]
 pub struct RateLimitingDialer<D> {
     dialer: D,
-    rate_limiter: SleepingRateLimiter,
+    rate_limiter: SharingRateLimiter,
 }
 
 impl<D> RateLimitingDialer<D> {
-    pub fn new(dialer: D, rate_limiter: SleepingRateLimiter) -> Self {
+    pub fn new(dialer: D, rate_limiter: SharingRateLimiter) -> Self {
         Self {
             dialer,
             rate_limiter,
@@ -66,7 +42,7 @@ where
         let connection = self.dialer.connect(address).await?;
         let (sender, receiver) = connection.split();
         Ok(Splitted(
-            RateLimitedAsyncRead::new(receiver, RateLimiter::new(self.rate_limiter.clone())),
+            RateLimitedAsyncRead::new(receiver, RateLimiterImpl::new(self.rate_limiter.clone())),
             sender,
         ))
     }
@@ -75,11 +51,11 @@ where
 /// Implementation of the [Listener] trait governing all returned [Listener::Connection] instances by a rate-limiting wrapper.
 pub struct RateLimitingListener<L> {
     listener: L,
-    rate_limiter: SleepingRateLimiter,
+    rate_limiter: SharingRateLimiter,
 }
 
 impl<L> RateLimitingListener<L> {
-    pub fn new(listener: L, rate_limiter: SleepingRateLimiter) -> Self {
+    pub fn new(listener: L, rate_limiter: SharingRateLimiter) -> Self {
         Self {
             listener,
             rate_limiter,
@@ -88,7 +64,10 @@ impl<L> RateLimitingListener<L> {
 }
 
 #[async_trait::async_trait]
-impl<L: Listener + Send> Listener for RateLimitingListener<L> {
+impl<L> Listener for RateLimitingListener<L>
+where
+    L: Listener + Send,
+{
     type Connection = Splitted<
         RateLimitedAsyncRead<<L::Connection as Splittable>::Receiver>,
         <L::Connection as Splittable>::Sender,
@@ -99,7 +78,7 @@ impl<L: Listener + Send> Listener for RateLimitingListener<L> {
         let connection = self.listener.accept().await?;
         let (sender, receiver) = connection.split();
         Ok(Splitted(
-            RateLimitedAsyncRead::new(receiver, RateLimiter::new(self.rate_limiter.clone())),
+            RateLimitedAsyncRead::new(receiver, RateLimiterImpl::new(self.rate_limiter.clone())),
             sender,
         ))
     }

docker/docker_entrypoint.sh

@@ -35,6 +35,8 @@ MAX_RUNTIME_INSTANCES=${MAX_RUNTIME_INSTANCES:-8}
 BACKUP_PATH=${BACKUP_PATH:-${BASE_PATH}/backup-stash}
 DATABASE_ENGINE=${DATABASE_ENGINE:-}
 PRUNING_ENABLED=${PRUNING_ENABLED:-false}
+ALEPHBFT_NETWORK_BIT_RATE=${ALEPHBFT_NETWORK_BIT_RATE:-}
+SUBSTRATE_NETWORK_BIT_RATE=${SUBSTRATE_NETWORK_BIT_RATE:-}
 
 if [[ "true" == "$PURGE_BEFORE_START" ]]; then
   echo "Purging chain (${CHAIN}) at path ${BASE_PATH}"
@@ -141,4 +143,12 @@ if [[ -n "${MAX_SUBSCRIPTIONS_PER_CONNECTION:-}" ]]; then
   ARGS+=(--rpc-max-subscriptions-per-connection ${MAX_SUBSCRIPTIONS_PER_CONNECTION})
 fi
 
+if [[ -n "${ALEPHBFT_NETWORK_BIT_RATE}" ]]; then
+    ARGS+=(--alephbft-network-bit-rate ${ALEPHBFT_NETWORK_BIT_RATE})
+fi
+
+if [[ -n "${SUBSTRATE_NETWORK_BIT_RATE}" ]]; then
+    ARGS+=(--substrate-network-bit-rate ${SUBSTRATE_NETWORK_BIT_RATE})
+fi
+
 echo "${CUSTOM_ARGS}" | xargs aleph-node "${ARGS[@]}"

finality-aleph/Cargo.toml

@@ -44,6 +44,7 @@ serde = { workspace = true }
 static_assertions = { workspace = true }
 tiny-bip39 = { workspace = true }
 tokio = { workspace = true, features = ["sync", "macros", "time", "rt-multi-thread"] }
+libp2p = { workspace = true }
 
 substrate-prometheus-endpoint = { workspace = true }
 

finality-aleph/src/lib.rs

@@ -72,7 +72,8 @@ pub use crate::{
     justification::AlephJustification,
     network::{
         address_cache::{ValidatorAddressCache, ValidatorAddressingInfo},
-        build_network, BuildNetworkOutput, ProtocolNetwork, SubstratePeerId,
+        build_network, BuildNetworkOutput, ProtocolNetwork, SubstrateNetworkConfig,
+        SubstratePeerId,
     },
     nodes::run_validator_node,
     session::SessionPeriod,
@@ -255,8 +256,10 @@ type Hasher = abft::HashWrapper<BlakeTwo256>;
 
 #[derive(Clone)]
 pub struct RateLimiterConfig {
-    /// Maximum bit-rate per node in bytes per second of the alephbft validator network.
-    pub alephbft_bit_rate_per_connection: usize,
+    /// Maximum bit-rate in bits per second of the alephbft validator network.
+    pub alephbft_network_bit_rate: u64,
+    /// Maximum bit-rate in bits per second of the substrate network (shared by sync, gossip, etc.).
+    pub substrate_network_bit_rate: u64,
 }
 
 pub struct AlephConfig<C, T> {

finality-aleph/src/network/build/base.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use libp2p::{core::StreamMuxer, PeerId, Transport};
 use sc_client_api::Backend;
 use sc_network::{
     config::{
@@ -8,6 +9,7 @@ use sc_network::{
     },
     error::Error as NetworkError,
     peer_store::PeerStore,
+    transport::NetworkConfig,
     NetworkService, NetworkWorker,
 };
 use sc_network_light::light_client_requests::handler::LightClientRequestHandler;
@@ -69,8 +71,9 @@ type BaseNetworkOutput<B> = (
 );
 
 /// Create a base network with all the protocols already included. Also spawn (almost) all the necessary services.
-pub fn network<B, BE, C>(
+pub fn network<B, BE, C, T, SM>(
     network_config: &NetworkConfiguration,
+    transport_builder: impl FnOnce(NetworkConfig) -> T,
     protocol_id: ProtocolId,
     client: Arc<C>,
     spawn_handle: &SpawnTaskHandle,
@@ -82,6 +85,13 @@ where
     B::Header: Header<Number = BlockNumber>,
     BE: Backend<B>,
     C: ClientForAleph<B, BE>,
+    T: Transport<Output = (PeerId, SM)> + Send + Unpin + 'static,
+    T::Dial: Send,
+    T::ListenerUpgrade: Send,
+    T::Error: Send + Sync,
+    SM: StreamMuxer + Unpin + Send + 'static,
+    SM::Substream: Unpin + Send,
+    SM::Error: Send + Sync,
 {
     let mut full_network_config = FullNetworkConfiguration::new(network_config);
     let genesis_hash = client
@@ -135,7 +145,8 @@ where
         block_announce_config: base_protocol_config,
     };
 
-    let network_service = NetworkWorker::new(network_params)?;
+    let network_service =
+        NetworkWorker::new_with_custom_transport(network_params, transport_builder)?;
     let network = network_service.service().clone();
     spawn_handle.spawn_blocking("network-worker", SPAWN_CATEGORY, network_service.run());
     Ok((network, networks, transactions_prototype))

finality-aleph/src/network/build/mod.rs

@@ -1,6 +1,7 @@
 use std::sync::{atomic::AtomicBool, Arc};
 
 use log::error;
+use rate_limiter::SharingRateLimiter;
 use sc_client_api::Backend;
 use sc_network::{
     config::{NetworkConfiguration, ProtocolId},
@@ -28,6 +29,7 @@ mod base;
 mod own_protocols;
 mod rpc;
 mod transactions;
+mod transport;
 
 use base::network as base_network;
 use own_protocols::Networks;
@@ -47,10 +49,17 @@ pub struct NetworkOutput<TP: TransactionPool + 'static> {
     pub system_rpc_tx: TracingUnboundedSender<RpcRequest<TP::Block>>,
 }
 
+pub struct SubstrateNetworkConfig {
+    /// Maximum bit-rate in bits per second of the substrate network (shared by sync, gossip, etc.).
+    pub substrate_network_bit_rate: u64,
+    /// Configuration of the network service.
+    pub network_config: NetworkConfiguration,
+}
+
 /// Start everything necessary to run the inter-node network and return the interfaces for it.
 /// This includes everything in the base network, the base protocol service, and services for handling transactions and RPCs.
 pub fn network<TP, BE, C>(
-    network_config: &NetworkConfiguration,
+    network_config: SubstrateNetworkConfig,
     protocol_id: ProtocolId,
     client: Arc<C>,
     major_sync: Arc<AtomicBool>,
@@ -72,6 +81,11 @@ where
         .expect("Genesis block exists.");
     let (base_protocol_config, events_from_network) =
         setup_base_protocol::<TP::Block>(genesis_hash);
+
+    let network_rate_limit = network_config.substrate_network_bit_rate;
+    let rate_limiter = SharingRateLimiter::new(network_rate_limit.into());
+    let transport_builder = |config| transport::build_transport(rate_limiter, config);
+
     let (
         network,
         Networks {
@@ -80,7 +94,8 @@ where
         },
         transaction_prototype,
     ) = base_network(
-        network_config,
+        &network_config.network_config,
+        transport_builder,
         protocol_id,
         client.clone(),
         spawn_handle,
@@ -91,7 +106,7 @@ where
     let (base_service, syncing_service) = BaseProtocolService::new(
         major_sync,
         genesis_hash,
-        network_config,
+        &network_config.network_config,
         protocol_names,
         network.clone(),
         events_from_network,