zealchen commented on code in PR #1486: URL: https://github.com/apache/incubator-horaedb/pull/1486#discussion_r1503449057
########## src/components/future_ext/src/retry.rs: ########## @@ -20,35 +20,99 @@ use std::time::Duration; use futures::Future; +use rand::prelude::*; -// TODO: add backoff -// https://github.com/apache/arrow-rs/blob/dfb642809e93c2c1b8343692f4e4b3080000f988/object_store/src/client/backoff.rs#L26 pub struct RetryConfig { pub max_retries: usize, - pub interval: Duration, + pub backoff: BackoffConfig, } impl Default for RetryConfig { fn default() -> Self { Self { max_retries: 3, - interval: Duration::from_millis(500), + backoff: Default::default(), } } } +pub struct BackoffConfig { + /// The initial backoff duration + pub init_backoff: Duration, + /// The maximum backoff duration + pub max_backoff: Duration, + /// The base of the exponential to use + pub base: f64, +} + +impl Default for BackoffConfig { + fn default() -> Self { + Self { + init_backoff: Duration::from_millis(100), + max_backoff: Duration::from_millis(500), + base: 3., + } + } +} + +pub struct Backoff { + init_backoff: f64, + next_backoff_secs: f64, + max_backoff_secs: f64, + base: f64, + rng: Option<Box<dyn RngCore + Sync + Send>>, +} + +impl Backoff { + /// Create a new [`Backoff`] from the provided [`BackoffConfig`] + pub fn new(config: &BackoffConfig) -> Self { + Self::new_with_rng(config, None) + } + + /// Creates a new `Backoff` with the optional `rng` + /// + /// Used [`rand::thread_rng()`] if no rng provided + pub fn new_with_rng( + config: &BackoffConfig, + rng: Option<Box<dyn RngCore + Sync + Send>>, + ) -> Self { + let init_backoff = config.init_backoff.as_secs_f64(); + Self { + init_backoff, + next_backoff_secs: init_backoff, + max_backoff_secs: config.max_backoff.as_secs_f64(), + base: config.base, + rng, + } + } + + /// Returns the next backoff duration to wait for + pub fn next(&mut self) -> Duration { + let range = self.init_backoff..(self.next_backoff_secs * self.base); + + let rand_backoff = match self.rng.as_mut() { + Some(rng) => rng.gen_range(range), + None => thread_rng().gen_range(range), + }; + + let next_backoff = self.max_backoff_secs.min(rand_backoff); + Duration::from_secs_f64(std::mem::replace(&mut self.next_backoff_secs, next_backoff)) + } +} + pub async fn retry_async<F, Fut, T, E>(f: F, config: &RetryConfig) -> Fut::Output where F: Fn() -> Fut, Fut: Future<Output = Result<T, E>>, { + let mut backoff = Backoff::new(&config.backoff); for _ in 0..config.max_retries { - let result = f().await; + let result: Result<T, E> = f().await; Review Comment: Not necessary. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
