Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add ExponentialBackoff::new_unchecked #1517

Merged
merged 1 commit into from
Feb 23, 2022
Merged

Add ExponentialBackoff::new_unchecked #1517

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion linkerd/app/inbound/src/test_util.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ pub fn default_config() -> Config {
connect: config::ConnectConfig {
keepalive: Keepalive(None),
timeout: Duration::from_secs(1),
backoff: exp_backoff::ExponentialBackoff::new(
backoff: exp_backoff::ExponentialBackoff::try_new(
Duration::from_millis(100),
Duration::from_millis(500),
0.1,
Expand Down
2 changes: 1 addition & 1 deletion linkerd/app/outbound/src/test_util.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ pub(crate) fn default_config() -> Config {
connect: config::ConnectConfig {
keepalive: Keepalive(None),
timeout: Duration::from_secs(1),
backoff: exp_backoff::ExponentialBackoff::new(
backoff: exp_backoff::ExponentialBackoff::try_new(
Duration::from_millis(100),
Duration::from_millis(500),
0.1,
Expand Down
16 changes: 5 additions & 11 deletions linkerd/app/src/env.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -227,19 +227,13 @@ const DEFAULT_METRICS_RETAIN_IDLE: Duration = Duration::from_secs(10 * 60);
const DEFAULT_INBOUND_DISPATCH_TIMEOUT: Duration = Duration::from_secs(1);
const DEFAULT_INBOUND_DETECT_TIMEOUT: Duration = Duration::from_secs(10);
const DEFAULT_INBOUND_CONNECT_TIMEOUT: Duration = Duration::from_millis(300);
const DEFAULT_INBOUND_CONNECT_BACKOFF: ExponentialBackoff = ExponentialBackoff {
min: Duration::from_millis(100),
max: Duration::from_millis(500),
jitter: 0.1,
};
const DEFAULT_INBOUND_CONNECT_BACKOFF: ExponentialBackoff =
ExponentialBackoff::new_unchecked(Duration::from_millis(100), Duration::from_millis(500), 0.1);
const DEFAULT_OUTBOUND_DISPATCH_TIMEOUT: Duration = Duration::from_secs(3);
const DEFAULT_OUTBOUND_DETECT_TIMEOUT: Duration = Duration::from_secs(10);
const DEFAULT_OUTBOUND_CONNECT_TIMEOUT: Duration = Duration::from_secs(1);
const DEFAULT_OUTBOUND_CONNECT_BACKOFF: ExponentialBackoff = ExponentialBackoff {
min: Duration::from_millis(100),
max: Duration::from_millis(500),
jitter: 0.1,
};
const DEFAULT_OUTBOUND_CONNECT_BACKOFF: ExponentialBackoff =
ExponentialBackoff::new_unchecked(Duration::from_millis(100), Duration::from_millis(500), 0.1);
const DEFAULT_RESOLV_CONF: &str = "/etc/resolv.conf";

const DEFAULT_INITIAL_STREAM_WINDOW_SIZE: u32 = 65_535; // Protocol default
Expand Down Expand Up @@ -1066,7 +1060,7 @@ pub fn parse_backoff<S: Strings>(
match (min?, max?, jitter?) {
(None, None, None) => Ok(default),
(Some(min), Some(max), jitter) => {
ExponentialBackoff::new(min, max, jitter.unwrap_or_default()).map_err(|error| {
ExponentialBackoff::try_new(min, max, jitter.unwrap_or_default()).map_err(|error| {
error!(message="Invalid backoff", %error, %min_env, ?min, %max_env, ?max, %jitter_env, ?jitter);
EnvError::InvalidEnvVar
})
Expand Down
72 changes: 38 additions & 34 deletions linkerd/exp-backoff/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,24 +12,22 @@ use rand::{rngs::SmallRng, thread_rng, SeedableRng};
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use thiserror::Error;
use tokio::time;

/// A jittered exponential backoff strategy.
// The raw fields are exposed so this type can be constructed statically.
#[derive(Copy, Clone, Debug, Default)]
pub struct ExponentialBackoff {
/// The minimum amount of time to wait before resuming an operation.
pub min: Duration,
min: time::Duration,

/// The maximum amount of time to wait before resuming an operation.
pub max: Duration,
max: time::Duration,

/// The ratio of the base timeout that may be randomly added to a backoff.
///
/// Must be greater than or equal to 0.0.
pub jitter: f64,
jitter: f64,
}

/// A jittered exponential backoff stream.
Expand All @@ -39,32 +37,28 @@ pub struct ExponentialBackoffStream {
backoff: ExponentialBackoff,
rng: SmallRng,
iterations: u32,
sleep: Pin<Box<time::Sleep>>,
sleeping: bool,
sleep: Pin<Box<time::Sleep>>,
}

#[derive(Clone, Debug, Error)]
#[error("invalid backoff: {0}")]
pub struct InvalidBackoff(&'static str);

impl ExponentialBackoff {
pub fn stream(&self) -> ExponentialBackoffStream {
ExponentialBackoffStream {
backoff: *self,
rng: SmallRng::from_rng(&mut thread_rng()).expect("RNG must be valid"),
iterations: 0,
sleep: Box::pin(time::sleep(Duration::from_secs(0))),
sleeping: false,
}
pub const fn new_unchecked(min: time::Duration, max: time::Duration, jitter: f64) -> Self {
Self { min, max, jitter }
}
}

impl ExponentialBackoff {
pub fn new(min: Duration, max: Duration, jitter: f64) -> Result<Self, InvalidBackoff> {
pub fn try_new(
min: time::Duration,
max: time::Duration,
jitter: f64,
) -> Result<Self, InvalidBackoff> {
if min > max {
return Err(InvalidBackoff("maximum must not be less than minimum"));
}
if max == Duration::from_millis(0) {
if max == time::Duration::from_millis(0) {
return Err(InvalidBackoff("maximum must be non-zero"));
}
if jitter < 0.0 {
Expand All @@ -79,13 +73,23 @@ impl ExponentialBackoff {
Ok(ExponentialBackoff { min, max, jitter })
}

fn base(&self, iterations: u32) -> Duration {
pub fn stream(&self) -> ExponentialBackoffStream {
ExponentialBackoffStream {
backoff: *self,
rng: SmallRng::from_rng(&mut thread_rng()).expect("RNG must be valid"),
iterations: 0,
sleeping: false,
sleep: Box::pin(time::sleep(time::Duration::from_secs(0))),
}
}

fn base(&self, iterations: u32) -> time::Duration {
debug_assert!(
self.min <= self.max,
"maximum backoff must not be less than minimum backoff"
);
debug_assert!(
self.max > Duration::from_millis(0),
self.max > time::Duration::from_millis(0),
"Maximum backoff must be non-zero"
);
self.min
Expand All @@ -96,9 +100,9 @@ impl ExponentialBackoff {

/// Returns a random, uniform duration on `[0, base*self.jitter]` no greater
/// than `self.max`.
fn jitter<R: rand::Rng>(&self, base: Duration, rng: &mut R) -> Duration {
fn jitter<R: rand::Rng>(&self, base: time::Duration, rng: &mut R) -> time::Duration {
if self.jitter == 0.0 {
Duration::default()
time::Duration::default()
} else {
let jitter_factor = rng.gen::<f64>();
debug_assert!(
Expand All @@ -109,7 +113,7 @@ impl ExponentialBackoff {
let secs = (base.as_secs() as f64) * rand_jitter;
let nanos = (base.subsec_nanos() as f64) * rand_jitter;
let remaining = self.max - base;
Duration::new(secs as u64, nanos as u32).min(remaining)
time::Duration::new(secs as u64, nanos as u32).min(remaining)
}
}
}
Expand Down Expand Up @@ -150,9 +154,9 @@ mod tests {

quickcheck! {
fn backoff_base_first(min_ms: u64, max_ms: u64) -> TestResult {
let min = Duration::from_millis(min_ms);
let max = Duration::from_millis(max_ms);
let backoff = match ExponentialBackoff::new(min, max, 0.0) {
let min = time::Duration::from_millis(min_ms);
let max = time::Duration::from_millis(max_ms);
let backoff = match ExponentialBackoff::try_new(min, max, 0.0) {
Err(_) => return TestResult::discard(),
Ok(backoff) => backoff,
};
Expand All @@ -161,9 +165,9 @@ mod tests {
}

fn backoff_base(min_ms: u64, max_ms: u64, iterations: u32) -> TestResult {
let min = Duration::from_millis(min_ms);
let max = Duration::from_millis(max_ms);
let backoff = match ExponentialBackoff::new(min, max, 0.0) {
let min = time::Duration::from_millis(min_ms);
let max = time::Duration::from_millis(max_ms);
let backoff = match ExponentialBackoff::try_new(min, max, 0.0) {
Err(_) => return TestResult::discard(),
Ok(backoff) => backoff,
};
Expand All @@ -172,18 +176,18 @@ mod tests {
}

fn backoff_jitter(base_ms: u64, max_ms: u64, jitter: f64) -> TestResult {
let base = Duration::from_millis(base_ms);
let max = Duration::from_millis(max_ms);
let backoff = match ExponentialBackoff::new(base, max, jitter) {
let base = time::Duration::from_millis(base_ms);
let max = time::Duration::from_millis(max_ms);
let backoff = match ExponentialBackoff::try_new(base, max, jitter) {
Err(_) => return TestResult::discard(),
Ok(backoff) => backoff,
};

let j = backoff.jitter(base, &mut rand::thread_rng());
if jitter == 0.0 || base_ms == 0 || max_ms == base_ms {
TestResult::from_bool(j == Duration::default())
TestResult::from_bool(j == time::Duration::default())
} else {
TestResult::from_bool(j > Duration::default())
TestResult::from_bool(j > time::Duration::default())
}
}
}
Expand Down