/// Adaptive pool scaling function
///
/// This allows to spawn new threads to make room for incoming task pressure.
/// Works in the background detached from the pool system and scales up the pool based
/// on the request rate.
///
/// It uses frequency based calculation to define work. Utilizing average processing rate.
Adaptive pool scaling function
This allows to spawn new threads to make room for incoming task pressure.
Works in the background detached from the pool system and scales up the pool based
on the request rate.

It uses frequency based calculation to define work. Utilizing average processing rate.
Adaptive pool scaling function This allows to spawn new threads to make room for incoming task pressure . Works in the background detached from the pool system and scales up the pool based on the request rate . It uses frequency based calculation to define work . Utilizing average processing rate .
fn scale_pool(&'static self) {
        // Fetch current frequency, it does matter that operations are ordered in this approach.
        let current_frequency = self.last_frequency.swap(0, Ordering::SeqCst);
        let mut freq_queue = self.frequencies.lock();
        // Make it safe to start for calculations by adding initial frequency scale
        if freq_queue.len() == 0 {
            freq_queue.push_back(0);
        }
        // Calculate message rate for the given time window
        let frequency = (current_frequency as f64 / SCALER_POLL_INTERVAL as f64) as u64;
        // Calculates current time window's EMA value (including last sample)
        let prev_ema_frequency = Self::calculate_ema(&freq_queue);
        // Add seen frequency data to the frequency histogram.
        freq_queue.push_back(frequency);
        if freq_queue.len() == FREQUENCY_QUEUE_SIZE.saturating_add(1) {
            freq_queue.pop_front();
        }
        // Calculates current time window's EMA value (including last sample)
        let curr_ema_frequency = Self::calculate_ema(&freq_queue);
        // Adapts the thread count of pool
        //
        // Sliding window of frequencies visited by the pool manager.
        // Pool manager creates EMA value for previous window and current window.
        // Compare them to determine scaling amount based on the trends.
        // If current EMA value is bigger, we will scale up.
        if curr_ema_frequency > prev_ema_frequency {
            // "Scale by" amount can be seen as "how much load is coming".
            // "Scale" amount is "how many threads we should spawn".
            let scale_by: f64 = curr_ema_frequency - prev_ema_frequency;
            let scale = num_cpus::get().min(
                ((DEFAULT_LOW_WATERMARK as f64 * scale_by) + DEFAULT_LOW_WATERMARK as f64) as usize,
            );
            trace!("unparking {} threads", scale);
            // It is time to scale the pool!
            self.provision_threads(scale);
        } else if (curr_ema_frequency - prev_ema_frequency).abs() < std::f64::EPSILON
            && current_frequency != 0
        {
            // Throughput is low. Allocate more threads to unblock flow.
            // If we fall to this case, scheduler is congested by longhauling tasks.
            // For unblock the flow we should add up some threads to the pool, but not that many to
            // stagger the program's operation.
            trace!("unparking {} threads", DEFAULT_LOW_WATERMARK);
            self.provision_threads(DEFAULT_LOW_WATERMARK as usize);
        }
    }
fn scale_pool ( & ' static self ) { // Fetch current frequency, it does matter that operations are ordered in this approach. let current_frequency = self . last_frequency . swap ( 0 , Ordering :: SeqCst ) ; let mut freq_queue = self . frequencies . lock ( ) ; // Make it safe to start for calculations by adding initial frequency scale if freq_queue . len ( ) == 0 { freq_queue . push_back ( 0 ) ; } // Calculate message rate for the given time window let frequency = ( current_frequency as f64 / SCALER_POLL_INTERVAL as f64 ) as u64 ; // Calculates current time window's EMA value (including last sample) let prev_ema_frequency = Self :: calculate_ema ( & freq_queue ) ; // Add seen frequency data to the frequency histogram. freq_queue . push_back ( frequency ) ; if freq_queue . len ( ) == FREQUENCY_QUEUE_SIZE . saturating_add ( 1 ) { freq_queue . pop_front ( ) ; } // Calculates current time window's EMA value (including last sample) let curr_ema_frequency = Self :: calculate_ema ( & freq_queue ) ; // Adapts the thread count of pool // // Sliding window of frequencies visited by the pool manager. // Pool manager creates EMA value for previous window and current window. // Compare them to determine scaling amount based on the trends. // If current EMA value is bigger, we will scale up. if curr_ema_frequency > prev_ema_frequency { // "Scale by" amount can be seen as "how much load is coming". // "Scale" amount is "how many threads we should spawn". let scale_by : f64 = curr_ema_frequency - prev_ema_frequency ; let scale = num_cpus :: get ( ) . min ( ( ( DEFAULT_LOW_WATERMARK as f64 * scale_by ) + DEFAULT_LOW_WATERMARK as f64 ) as usize , ) ; trace ! ( " " scale ) ; // It is time to scale the pool! self . provision_threads ( scale ) ; } else if ( curr_ema_frequency - prev_ema_frequency ) . abs ( ) < std :: f64 :: EPSILON && current_frequency != 0 { // Throughput is low. Allocate more threads to unblock flow. // If we fall to this case, scheduler is congested by longhauling tasks. // For unblock the flow we should add up some threads to the pool, but not that many to // stagger the program's operation. trace ! ( " " DEFAULT_LOW_WATERMARK ) ; self . provision_threads ( DEFAULT_LOW_WATERMARK as usize ) ; } }
Adaptive pool scaling function
This allows to spawn new threads to make room for incoming task pressure.
Adaptive pool scaling function This allows to spawn new threads to make room for incoming task pressure .
&'static self