async fn worker(
    mut receiver: mpsc::UnboundedReceiver<ClassBasedCommand>,
    resources_per_second: u32,
    max_stored_resource: u32,
) {
    let mut active_validators = HashSet::new();
    let mut upcoming_validators = HashSet::new();

    let mut resources_available: i64 = 0;
    let mut last_refill: Instant = Instant::now();

    // Whether or not we emitted a warning that the limiter is currently implicitly disabled.
    let mut logged_uninitialized = false;

    while let Some(msg) = receiver.recv().await {
        match msg {
            ClassBasedCommand::UpdateValidators {
                active_validators: new_active_validators,
                upcoming_validators: new_upcoming_validators,
            } => {
                active_validators = new_active_validators;
                upcoming_validators = new_upcoming_validators;
                debug!(
                    ?active_validators,
                    ?upcoming_validators,
                    "resource classes updated"
                );
            }
            ClassBasedCommand::RequestResource {
                amount,
                id,
                responder,
            } => {
                if active_validators.is_empty() && upcoming_validators.is_empty() {
                    // It is likely that we have not been initialized, thus no node is getting the
                    // reserved resources. In this case, do not limit at all.
                    if !logged_uninitialized {
                        logged_uninitialized = true;
                        info!("empty set of validators, not limiting resources at all");
                    }
                    continue;
                }

                let peer_class = if let Some(ref validator_id) = id.validator_id {
                    if active_validators.contains(validator_id) {
                        PeerClass::ActiveValidator
                    } else if upcoming_validators.contains(validator_id) {
                        PeerClass::UpcomingValidator
                    } else {
                        PeerClass::Bulk
                    }
                } else {
                    PeerClass::Bulk
                };

                match peer_class {
                    PeerClass::ActiveValidator | PeerClass::UpcomingValidator => {
                        // No limit imposed on validators.
                    }
                    PeerClass::Bulk => {
                        while resources_available < 0 {
                            // Determine time delta since last refill.
                            let now = Instant::now();
                            let elapsed = now - last_refill;
                            last_refill = now;

                            // Add appropriate amount of resources, capped at `max_stored_bytes`.
                            resources_available +=
                                ((elapsed.as_nanos() * resources_per_second as u128)
                                    / 1_000_000_000) as i64;
                            resources_available =
                                resources_available.min(max_stored_resource as i64);

                            // If we do not have enough resources available, sleep until we do.
                            if resources_available < 0 {
                                let estimated_time_remaining = Duration::from_millis(
                                    (-resources_available) as u64 * 1000
                                        / resources_per_second as u64,
                                );

                                tokio::time::sleep(estimated_time_remaining).await;
                            }
                        }

                        // Subtract requested amount.
                        resources_available -= amount as i64;
                    }
                }

                if responder.send(()).is_err() {
                    debug!("resource requester disappeared before we could answer.")
                }
            }
            ClassBasedCommand::Shutdown => {
                // Shutdown the channel, processing only the remaining messages.
                receiver.close();
            }
        }
    }
    debug!("class based worker exiting");
}
/// Background worker for the limiter.
///
/// Will permit any amount of current/future validator resource requests, while restricting
/// non-validators to `resources_per_second`, although with unlimited one-time burst. The latter
/// guarantees that any size request can be satisfied (e.g. a message larger than the burst limit).
///
/// Stores up to `max_stored_resources` during idle times to smooth this process.
Background worker for the limiter.
Will permit any amount of current/future validator resource requests, while restricting
non-validators to `resources_per_second`, although with unlimited one-time burst. The latter
guarantees that any size request can be satisfied .

Stores up to `max_stored_resources` during idle times to smooth this process.
Background worker for the limiter . Will permit any amount of current / future validator resource requests while restricting non - validators to ` resources_per_second ` although with unlimited one - time burst . The latter guarantees that any size request can be satisfied . Stores up to ` max_stored_resources ` during idle times to smooth this process .
async fn worker(
    mut receiver: mpsc::UnboundedReceiver<ClassBasedCommand>,
    resources_per_second: u32,
    max_stored_resource: u32,
) {
    let mut active_validators = HashSet::new();
    let mut upcoming_validators = HashSet::new();
    let mut resources_available: i64 = 0;
    let mut last_refill: Instant = Instant::now();
    // Whether or not we emitted a warning that the limiter is currently implicitly disabled.
    let mut logged_uninitialized = false;
    while let Some(msg) = receiver.recv().await {
        match msg {
            ClassBasedCommand::UpdateValidators {
                active_validators: new_active_validators,
                upcoming_validators: new_upcoming_validators,
            } => {
                active_validators = new_active_validators;
                upcoming_validators = new_upcoming_validators;
                debug!(
                    ?active_validators,
                    ?upcoming_validators,
                    "resource classes updated"
                );
            }
            ClassBasedCommand::RequestResource {
                amount,
                id,
                responder,
            } => {
                if active_validators.is_empty() && upcoming_validators.is_empty() {
                    // It is likely that we have not been initialized, thus no node is getting the
                    // reserved resources. In this case, do not limit at all.
                    if !logged_uninitialized {
                        logged_uninitialized = true;
                        info!("empty set of validators, not limiting resources at all");
                    }
                    continue;
                }
                let peer_class = if let Some(ref validator_id) = id.validator_id {
                    if active_validators.contains(validator_id) {
                        PeerClass::ActiveValidator
                    } else if upcoming_validators.contains(validator_id) {
                        PeerClass::UpcomingValidator
                    } else {
                        PeerClass::Bulk
                    }
                } else {
                    PeerClass::Bulk
                };
                match peer_class {
                    PeerClass::ActiveValidator | PeerClass::UpcomingValidator => {
                        // No limit imposed on validators.
                    }
                    PeerClass::Bulk => {
                        while resources_available < 0 {
                            // Determine time delta since last refill.
                            let now = Instant::now();
                            let elapsed = now - last_refill;
                            last_refill = now;
                            // Add appropriate amount of resources, capped at `max_stored_bytes`.
                            resources_available +=
                                ((elapsed.as_nanos() * resources_per_second as u128)
                                    / 1_000_000_000) as i64;
                            resources_available =
                                resources_available.min(max_stored_resource as i64);
                            // If we do not have enough resources available, sleep until we do.
                            if resources_available < 0 {
                                let estimated_time_remaining = Duration::from_millis(
                                    (-resources_available) as u64 * 1000
                                        / resources_per_second as u64,
                                );
                                tokio::time::sleep(estimated_time_remaining).await;
                            }
                        }
                        // Subtract requested amount.
                        resources_available -= amount as i64;
                    }
                }
                if responder.send(()).is_err() {
                    debug!("resource requester disappeared before we could answer.")
                }
            }
            ClassBasedCommand::Shutdown => {
                // Shutdown the channel, processing only the remaining messages.
                receiver.close();
            }
        }
    }
    debug!("class based worker exiting");
}
async fn worker ( mut receiver : mpsc :: UnboundedReceiver < ClassBasedCommand > , resources_per_second : u32 , max_stored_resource : u32 , ) { let mut active_validators = HashSet :: new ( ) ; let mut upcoming_validators = HashSet :: new ( ) ; let mut resources_available : i64 = 0 ; let mut last_refill : Instant = Instant :: now ( ) ; // Whether or not we emitted a warning that the limiter is currently implicitly disabled. let mut logged_uninitialized = false ; while let Some ( msg ) = receiver . recv ( ) . await { match msg { ClassBasedCommand :: UpdateValidators { active_validators : new_active_validators , upcoming_validators : new_upcoming_validators , } => { active_validators = new_active_validators ; upcoming_validators = new_upcoming_validators ; debug ! ( active_validators upcoming_validators " " ) ; } ClassBasedCommand :: RequestResource { amount , id , responder , } => { if active_validators . is_empty ( ) && upcoming_validators . is_empty ( ) { // It is likely that we have not been initialized, thus no node is getting the // reserved resources. In this case, do not limit at all. if ! logged_uninitialized { logged_uninitialized = true ; info ! ( " " ) ; } continue ; } let peer_class = if let Some ( ref validator_id ) = id . validator_id { if active_validators . contains ( validator_id ) { PeerClass :: ActiveValidator } else if upcoming_validators . contains ( validator_id ) { PeerClass :: UpcomingValidator } else { PeerClass :: Bulk } } else { PeerClass :: Bulk } ; match peer_class { PeerClass :: ActiveValidator | PeerClass :: UpcomingValidator => { // No limit imposed on validators. } PeerClass :: Bulk => { while resources_available < 0 { // Determine time delta since last refill. let now = Instant :: now ( ) ; let elapsed = now - last_refill ; last_refill = now ; // Add appropriate amount of resources, capped at `max_stored_bytes`. resources_available += ( ( elapsed . as_nanos ( ) * resources_per_second as u128 ) / 1_000_000_000 ) as i64 ; resources_available = resources_available . min ( max_stored_resource as i64 ) ; // If we do not have enough resources available, sleep until we do. if resources_available < 0 { let estimated_time_remaining = Duration :: from_millis ( ( - resources_available ) as u64 * 1000 / resources_per_second as u64 , ) ; tokio :: time :: sleep ( estimated_time_remaining ) . await ; } } // Subtract requested amount. resources_available -= amount as i64 ; } } if responder . send ( ( ) ) . is_err ( ) { debug ! ( " " ) } } ClassBasedCommand :: Shutdown => { // Shutdown the channel, processing only the remaining messages. receiver . close ( ) ; } } } debug ! ( " " ) ; }
Background worker for the limiter.
Background worker for the limiter .
receiver
mpsc::UnboundedReceiver<ClassBasedCommand>
resources_per_second
u32
max_stored_resource
u32
receiver
mpsc::UnboundedReceiver<ClassBasedCommand>
resources_per_second
u32
max_stored_resource
u32