tikv_utils: make retry returns a `Send` future even the result type is `!Sync`
close tikv/tikv#13811
This PR have modified the `retry_ext` function in the `tikv_utils::stream` package, making the `final_result` not live across await points any more. So even the result type is `!Sync`, `retry` returns a `Send` future now.
@@ -152,24 +152,28 @@ where
})();
let mut retry_wait_dur = Duration::from_secs(1);
-
- let mut final_result = action().await;
- for _ in 1..max_retry_times {
- if let Err(e) = &final_result {
+ let mut retry_time = 0;
+ loop {
+ match action().await {
+ Ok(r) => return Ok(r),
+ Err(e) => {
if let Some(ref mut f) = ext.on_failure {
- f(e);
+ f(&e);
+ }
+ if !e.is_retryable() {
+ return Err(e);
+ }
+ retry_time += 1;
+ if retry_time > max_retry_times {
+ return Err(e);
+ }
}
- if e.is_retryable() {
+ }
+
let backoff = thread_rng().gen_range(0..1000);
sleep(retry_wait_dur + Duration::from_millis(backoff)).await;
retry_wait_dur = MAX_RETRY_DELAY.min(retry_wait_dur * 2);
- final_result = action().await;
- continue;
- }
}
- break;
- }
- final_result
}
// Return an error if the future does not finish by the timeout
@@ -206,3 +210,55 @@ impl RetryError for HttpDispatchError {
true
}
}
+
+#[cfg(test)]
+mod tests {
+ use std::{cell::RefCell, pin::Pin};
+
+ use futures::{Future, FutureExt};
+ use rusoto_core::HttpDispatchError;
+
+ use super::RetryError;
+ use crate::stream::retry;
+
+ #[derive(Debug)]
+ struct TriviallyRetry;
+
+ impl RetryError for TriviallyRetry {
+ fn is_retryable(&self) -> bool {
+ true
+ }
+ }
+
+ fn assert_send<T: Send>(_t: T) {}
+
+ #[test]
+ fn test_retry_is_send_even_return_type_not_sync() {
+ struct BangSync(Option<RefCell<()>>);
+ let fut = retry(|| futures::future::ok::<_, HttpDispatchError>(BangSync(None)));
+ assert_send(fut)
+ }
+
+ fn gen_action_fail_for(
+ n_times: usize,
+ ) -> impl FnMut() -> Pin<Box<dyn Future<Output = Result<(), TriviallyRetry>>>> {
+ let mut n = 0;
+ move || {
+ if n < n_times {
+ n += 1;
+ futures::future::err(TriviallyRetry).boxed()
+ } else {
+ futures::future::ok(()).boxed()
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn test_failure() {
+ fail::cfg("retry_count", "return(2)").unwrap();
+ let r = retry(gen_action_fail_for(3)).await;
+ assert!(r.is_err(), "{:?}", r);
+ let r = retry(gen_action_fail_for(1)).await;
+ assert!(r.is_ok(), "{:?}", r);
+ }
+}
