diff --git a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs index dab0307448..1cee7ea122 100644 --- a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -303,22 +303,29 @@ where self.sending_delay_controller.current_multiplier() ); - // Even just a single used slot is enough to signal backpressure - if used_slots > 0 { + if self + .sending_delay_controller + .is_backpressure_currently_detected(used_slots) + { log::trace!("Backpressure detected"); self.sending_delay_controller.record_backpressure_detected(); } - // If the buffer is running out, slow down the sending rate + // If the buffer is running out, slow down the sending rate by increasing the delay + // multiplier. if self.mix_tx.capacity() == 0 && self.sending_delay_controller.not_increased_delay_recently() { self.sending_delay_controller.increase_delay_multiplier(); } - // Very carefully step up the sending rate in case it seems like we can solidly handle the - // current rate. - if self.sending_delay_controller.is_sending_reliable() { + // If it looks like we are sending reliably, increase the sending rate by decreasing the + // sending delay multiplier. + if !self + .sending_delay_controller + .was_backpressure_detected_recently() + && self.sending_delay_controller.not_decreased_delay_recently() + { self.sending_delay_controller.decrease_delay_multiplier(); } } diff --git a/clients/client-core/src/client/real_messages_control/real_traffic_stream/sending_delay_controller.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream/sending_delay_controller.rs index b2d656f763..69f33fc80b 100644 --- a/clients/client-core/src/client/real_messages_control/real_traffic_stream/sending_delay_controller.rs +++ b/clients/client-core/src/client/real_messages_control/real_traffic_stream/sending_delay_controller.rs @@ -11,11 +11,14 @@ const INCREASE_DELAY_MIN_CHANGE_INTERVAL_SECS: u64 = 1; // The minimum time between decreasing the average delay between packets. We don't want to change // to quickly to keep things somewhat stable. Also there are buffers downstreams meaning we need to // wait a little to see the effect before we decrease further. -const DECREASE_DELAY_MIN_CHANGE_INTERVAL_SECS: u64 = 30; +const DECREASE_DELAY_MIN_CHANGE_INTERVAL_SECS: u64 = 3; +// The queue length that is required for us to register that backpressure occured. If there are +// more than this many packets waiting to be sent, we consider the channel to be under +// backpressure. +const BACKPRESSURE_THRESHOLD: usize = 10; // If we enough time passes without any sign of backpressure in the channel, we can consider -// lowering the average delay. The goal is to keep somewhat stable, rather than maxing out -// bandwidth at all times. -const ACCEPTABLE_TIME_WITHOUT_BACKPRESSURE_SECS: u64 = 30; +// lowering the average delay. +const ACCEPTABLE_TIME_WITHOUT_BACKPRESSURE_SECS: u64 = 1; // The maximum multiplier we apply to the base average Poisson delay. const MAX_DELAY_MULTIPLIER: u32 = 6; // The minium multiplier we apply to the base average Poisson delay. @@ -100,22 +103,27 @@ impl SendingDelayController { } } - pub(crate) fn record_backpressure_detected(&mut self) { - self.time_when_backpressure_detected = get_time_now(); - } - pub(crate) fn not_increased_delay_recently(&self) -> bool { get_time_now() > self.time_when_changed + Duration::from_secs(INCREASE_DELAY_MIN_CHANGE_INTERVAL_SECS) } - pub(crate) fn is_sending_reliable(&self) -> bool { - let now = get_time_now(); - let delay_change_interval = Duration::from_secs(DECREASE_DELAY_MIN_CHANGE_INTERVAL_SECS); - let acceptable_time_without_backpressure = - Duration::from_secs(ACCEPTABLE_TIME_WITHOUT_BACKPRESSURE_SECS); + pub(crate) fn not_decreased_delay_recently(&self) -> bool { + get_time_now() + > self.time_when_changed + Duration::from_secs(DECREASE_DELAY_MIN_CHANGE_INTERVAL_SECS) + } - now > self.time_when_backpressure_detected + acceptable_time_without_backpressure - && now > self.time_when_changed + delay_change_interval + pub(crate) fn is_backpressure_currently_detected(&self, queue_length: usize) -> bool { + queue_length > BACKPRESSURE_THRESHOLD + } + + pub(crate) fn record_backpressure_detected(&mut self) { + self.time_when_backpressure_detected = get_time_now(); + } + + pub(crate) fn was_backpressure_detected_recently(&self) -> bool { + get_time_now() + < self.time_when_backpressure_detected + + Duration::from_secs(ACCEPTABLE_TIME_WITHOUT_BACKPRESSURE_SECS) } }