You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2024/03/11 07:54:55 UTC
(opendal) 09/24: Fix retry
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch read-async-trait
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 2ee2c2877e5a1dae49a2bfa39532d3896c57250c
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Fri Mar 8 17:38:19 2024 +0800
Fix retry
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/Cargo.lock | 6 +--
core/Cargo.toml | 2 +-
core/src/layers/retry.rs | 113 +++++++++++++++++++++++++++++------------------
3 files changed, 73 insertions(+), 48 deletions(-)
diff --git a/core/Cargo.lock b/core/Cargo.lock
index 2dfab9bbaf..5986d5cae1 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -886,11 +886,11 @@ dependencies = [
[[package]]
name = "backon"
-version = "0.4.1"
+version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0c1a6197b2120bb2185a267f6515038558b019e92b832bb0320e96d66268dcf9"
+checksum = "c491fa80d69c03084223a4e73c378dd9f9a1e612eb54051213f88b2d5249b458"
dependencies = [
- "fastrand 1.9.0",
+ "fastrand 2.0.1",
"futures-core",
"pin-project",
"tokio",
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 451280d71c..b54d8df415 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -235,7 +235,7 @@ async-tls = { version = "0.13.0", optional = true }
# Required dependencies
anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
-backon = "0.4.1"
+backon = "0.4.3"
base64 = "0.21"
bytes = "1.4"
chrono = { version = "0.4.28", default-features = false, features = [
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index 8e103c856e..de87fdf103 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -647,7 +647,7 @@ impl<A: Accessor, I: RetryInterceptor> LayeredAccessor for RetryAccessor<A, I> {
}
pub struct RetryWrapper<R, I> {
- inner: R,
+ inner: Option<R>,
notify: Arc<I>,
path: String,
@@ -659,7 +659,7 @@ pub struct RetryWrapper<R, I> {
impl<R, I> RetryWrapper<R, I> {
fn new(inner: R, notify: Arc<I>, path: &str, backoff: ExponentialBuilder) -> Self {
Self {
- inner,
+ inner: Some(inner),
notify,
path: path.to_string(),
@@ -672,47 +672,72 @@ impl<R, I> RetryWrapper<R, I> {
impl<R: oio::Read, I: RetryInterceptor> oio::Read for RetryWrapper<R, I> {
async fn seek(&mut self, pos: io::SeekFrom) -> Result<u64> {
- todo!()
- // { || self.inner.seek(pos) }
- // .retry(&self.builder)
- // .when(|e| e.is_temporary())
- // .notify(|err, dur| {
- // self.notify.intercept(
- // err,
- // dur,
- // &[
- // ("operation", ReadOperation::Seek.into_static()),
- // ("path", &self.path),
- // ],
- // )
- // })
- // .map(|v| v.map_err(|e| e.set_persistent()))
- // .await
+ use backon::RetryableWithContext;
+
+ let inner = self.inner.take().expect("inner must be valid");
+
+ let (inner, res) = {
+ |mut r: R| async move {
+ let res = r.seek(pos).await;
+
+ (r, res)
+ }
+ }
+ .retry(&self.builder)
+ .context(inner)
+ .when(|e| e.is_temporary())
+ .notify(|err, dur| {
+ self.notify.intercept(
+ err,
+ dur,
+ &[
+ ("operation", ReadOperation::Seek.into_static()),
+ ("path", &self.path),
+ ],
+ )
+ })
+ .map(|(r, res)| (r, res.map_err(|err| err.set_persistent())))
+ .await;
+
+ self.inner = Some(inner);
+ res
}
async fn next_v2(&mut self, size: usize) -> Result<Bytes> {
- todo!()
- // { || self.inner.next_v2(size) }
- // .retry(&self.builder)
- // .when(|e| e.is_temporary())
- // .notify(|err, dur| {
- // self.notify.intercept(
- // err,
- // dur,
- // &[
- // ("operation", ReadOperation::NextV2.into_static()),
- // ("path", &self.path),
- // ],
- // )
- // })
- // .map_err(|e| e.set_persistent())
- // .await
+ use backon::RetryableWithContext;
+
+ let inner = self.inner.take().expect("inner must be valid");
+
+ let (inner, res) = {
+ |mut r: R| async move {
+ let res = r.next_v2(size).await;
+
+ (r, res)
+ }
+ }
+ .retry(&self.builder)
+ .when(|e| e.is_temporary())
+ .notify(|err, dur| {
+ self.notify.intercept(
+ err,
+ dur,
+ &[
+ ("operation", ReadOperation::Next.into_static()),
+ ("path", &self.path),
+ ],
+ )
+ })
+ .map(|(r, res)| (r, res.map_err(|err| err.set_persistent())))
+ .await;
+
+ self.inner = Some(inner);
+ res
}
}
impl<R: oio::BlockingRead, I: RetryInterceptor> oio::BlockingRead for RetryWrapper<R, I> {
fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
- { || self.inner.read(buf) }
+ { || self.inner.as_mut().unwrap().read(buf) }
.retry(&self.builder)
.when(|e| e.is_temporary())
.notify(|err, dur| {
@@ -730,7 +755,7 @@ impl<R: oio::BlockingRead, I: RetryInterceptor> oio::BlockingRead for RetryWrapp
}
fn seek(&mut self, pos: io::SeekFrom) -> Result<u64> {
- { || self.inner.seek(pos) }
+ { || self.inner.as_mut().unwrap().seek(pos) }
.retry(&self.builder)
.when(|e| e.is_temporary())
.notify(|err, dur| {
@@ -748,7 +773,7 @@ impl<R: oio::BlockingRead, I: RetryInterceptor> oio::BlockingRead for RetryWrapp
}
fn next(&mut self) -> Option<Result<Bytes>> {
- { || self.inner.next().transpose() }
+ { || self.inner.as_mut().unwrap().next().transpose() }
.retry(&self.builder)
.when(|e| e.is_temporary())
.notify(|err, dur| {
@@ -774,7 +799,7 @@ impl<R: oio::Write, I: RetryInterceptor> oio::Write for RetryWrapper<R, I> {
self.sleep = None;
}
- match ready!(self.inner.poll_write(cx, bs)) {
+ match ready!(self.inner.as_mut().unwrap().poll_write(cx, bs)) {
Ok(v) => {
self.current_backoff = None;
Poll::Ready(Ok(v))
@@ -820,7 +845,7 @@ impl<R: oio::Write, I: RetryInterceptor> oio::Write for RetryWrapper<R, I> {
self.sleep = None;
}
- match ready!(self.inner.poll_abort(cx)) {
+ match ready!(self.inner.as_mut().unwrap().poll_abort(cx)) {
Ok(v) => {
self.current_backoff = None;
Poll::Ready(Ok(v))
@@ -866,7 +891,7 @@ impl<R: oio::Write, I: RetryInterceptor> oio::Write for RetryWrapper<R, I> {
self.sleep = None;
}
- match ready!(self.inner.poll_close(cx)) {
+ match ready!(self.inner.as_mut().unwrap().poll_close(cx)) {
Ok(v) => {
self.current_backoff = None;
Poll::Ready(Ok(v))
@@ -909,7 +934,7 @@ impl<R: oio::Write, I: RetryInterceptor> oio::Write for RetryWrapper<R, I> {
impl<R: oio::BlockingWrite, I: RetryInterceptor> oio::BlockingWrite for RetryWrapper<R, I> {
fn write(&mut self, bs: &dyn oio::WriteBuf) -> Result<usize> {
- { || self.inner.write(bs) }
+ { || self.inner.as_mut().unwrap().write(bs) }
.retry(&self.builder)
.when(|e| e.is_temporary())
.notify(|err, dur| {
@@ -927,7 +952,7 @@ impl<R: oio::BlockingWrite, I: RetryInterceptor> oio::BlockingWrite for RetryWra
}
fn close(&mut self) -> Result<()> {
- { || self.inner.close() }
+ { || self.inner.as_mut().unwrap().close() }
.retry(&self.builder)
.when(|e| e.is_temporary())
.notify(|err, dur| {
@@ -954,7 +979,7 @@ impl<P: oio::List, I: RetryInterceptor> oio::List for RetryWrapper<P, I> {
self.sleep = None;
}
- match ready!(self.inner.poll_next(cx)) {
+ match ready!(self.inner.as_mut().unwrap().poll_next(cx)) {
Ok(v) => {
self.current_backoff = None;
Poll::Ready(Ok(v))
@@ -997,7 +1022,7 @@ impl<P: oio::List, I: RetryInterceptor> oio::List for RetryWrapper<P, I> {
impl<P: oio::BlockingList, I: RetryInterceptor> oio::BlockingList for RetryWrapper<P, I> {
fn next(&mut self) -> Result<Option<oio::Entry>> {
- { || self.inner.next() }
+ { || self.inner.as_mut().unwrap().next() }
.retry(&self.builder)
.when(|e| e.is_temporary())
.notify(|err, dur| {