rate_limiter_aimd 0.1.1

A Rust library for adaptive concurrency control based on network service feedback (forked from vector.dev)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

use crate::registered_event;
use metrics::{Histogram, histogram};
use std::time::Duration;
use tracing;

#[derive(Clone, Copy, Debug)]
pub struct AdaptiveConcurrencyLimitData {
    pub concurrency: u64,
    pub reached_limit: bool,
    pub had_back_pressure: bool,
    pub current_rtt: Option<Duration>,
    pub past_rtt: Duration,
    pub past_rtt_deviation: Duration,
}

registered_event! {
    AdaptiveConcurrencyLimit => {
        // These are histograms, as they may have a number of different
        // values over each reporting interval, and each of those values
        // is valuable for diagnosis.
        limit: Histogram = histogram!("adaptive_concurrency_limit"),
        reached_limit: Histogram = histogram!("adaptive_concurrency_reached_limit"),
        back_pressure: Histogram = histogram!("adaptive_concurrency_back_pressure"),
        past_rtt_mean: Histogram = histogram!("adaptive_concurrency_past_rtt_mean"),
    }

    fn emit(&self, data: AdaptiveConcurrencyLimitData) {
        self.limit.record(data.concurrency as f64);
        let reached_limit_val = data.reached_limit.then_some(1.0).unwrap_or_default();
        self.reached_limit.record(reached_limit_val);
        let back_pressure_val = data.had_back_pressure.then_some(1.0).unwrap_or_default();
        self.back_pressure.record(back_pressure_val);
        self.past_rtt_mean.record(data.past_rtt);

        // ADDED LOGGING FOR THE EXAMPLE
        tracing::info!(
            target: "adaptive_concurrency::stats",
            concurrency_limit = data.concurrency,
            reached_max_limit_this_cycle = data.reached_limit, // Renamed for clarity from "reached_limit"
            had_back_pressure_this_cycle = data.had_back_pressure, // Renamed for clarity
            current_rtt_ms = data.current_rtt.map(|d| d.as_millis()),
            past_rtt_ms = data.past_rtt.as_millis(),
            past_rtt_deviation_ms = data.past_rtt_deviation.as_millis(),
            "Limit Adjusted"
        );
    }
}

registered_event! {
    AdaptiveConcurrencyInFlight => {
        in_flight: Histogram = histogram!("adaptive_concurrency_in_flight"),
    }

    fn emit(&self, in_flight_count: u64) {
        self.in_flight.record(in_flight_count as f64);
        // ADDED LOGGING FOR THE EXAMPLE
        tracing::debug!(target: "adaptive_concurrency::stats", in_flight = in_flight_count, "In-flight Updated");
    }
}

registered_event! {
    AdaptiveConcurrencyObservedRtt => {
        observed_rtt: Histogram = histogram!("adaptive_concurrency_observed_rtt"),
    }

    fn emit(&self, rtt: Duration) {
        self.observed_rtt.record(rtt);
        // Optionally log raw RTTs, can be very verbose
        tracing::trace!(target: "adaptive_concurrency::stats", observed_rtt_ms = rtt.as_millis(), "RTT Observed");
    }
}

registered_event! {
    AdaptiveConcurrencyAveragedRtt => {
        averaged_rtt: Histogram = histogram!("adaptive_concurrency_averaged_rtt"),
    }

    fn emit(&self, rtt: Duration) {
        self.averaged_rtt.record(rtt);
        // Log averaged RTT when it's computed for an adjustment period
        tracing::debug!(target: "adaptive_concurrency::stats", averaged_rtt_ms = rtt.as_millis(), "RTT Averaged for Period");
    }
}