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 2023/04/02 17:44:57 UTC
[incubator-opendal] branch polish-retry created (now d5e0ccaa)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch polish-retry
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
at d5e0ccaa refactor(core): Simplify the usage of BatchOperation and BatchResults
This branch includes the following new commits:
new d5e0ccaa refactor(core): Simplify the usage of BatchOperation and BatchResults
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-opendal] 01/01: refactor(core): Simplify the usage of BatchOperation and BatchResults
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch polish-retry
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
commit d5e0ccaae6feb6ef5278c40eb75c4bb3256bec73
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Apr 3 01:44:18 2023 +0800
refactor(core): Simplify the usage of BatchOperation and BatchResults
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/error_context.rs | 7 +--
core/src/layers/logging.rs | 6 +-
core/src/layers/retry.rs | 107 ++++++++++++++++++++----------------
core/src/raw/rps.rs | 54 +++---------------
core/src/services/azblob/backend.rs | 97 ++++++++++++++++----------------
core/src/services/oss/backend.rs | 104 +++++++++++++++++------------------
core/src/services/s3/backend.rs | 69 +++++++++++------------
core/src/types/operator/operator.rs | 20 +++----
core/src/types/ops.rs | 40 +++++---------
9 files changed, 231 insertions(+), 273 deletions(-)
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index a5d58cfc..2546c5c4 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -214,9 +214,8 @@ impl<A: Accessor> LayeredAccessor for ErrorContextAccessor<A> {
self.inner
.batch(args)
.map_ok(|v| {
- let BatchedResults::Delete(res) = v.into_results();
-
- let res = res
+ let res = v
+ .into_results()
.into_iter()
.map(|(path, res)| {
let res = res.map_err(|err| {
@@ -228,7 +227,7 @@ impl<A: Accessor> LayeredAccessor for ErrorContextAccessor<A> {
})
.collect();
- RpBatch::new(BatchedResults::Delete(res))
+ RpBatch::new(res)
})
.map_err(|err| {
err.with_operation(Operation::Batch)
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index 69712e6e..923ace20 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -532,7 +532,7 @@ impl<A: Accessor> LayeredAccessor for LoggingAccessor<A> {
}
async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- let (op, count) = (args.operation().operation(), args.operation().len());
+ let (op, count) = (args.operation()[0].1.operation(), args.operation().len());
debug!(
target: LOGGING_TARGET,
@@ -550,8 +550,8 @@ impl<A: Accessor> LayeredAccessor for LoggingAccessor<A> {
self.scheme,
Operation::Batch,
v.results().len(),
- v.results().len_ok(),
- v.results().len_err(),
+ v.results().iter().filter(|(_, v)|v.is_ok()).count(),
+ v.results().iter().filter(|(_, v)|v.is_err()).count(),
);
v
})
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index 3e251a0a..359d59ce 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -291,7 +291,13 @@ impl<A: Accessor> LayeredAccessor for RetryAccessor<A> {
{ || self.inner.batch(args.clone()).map(Err::<(), _>) }
.retry(&self.builder)
.when(|res| match res {
- Ok(rp) => rp.results().errors().any(|e| e.is_temporary()),
+ Ok(rp) => rp.results().iter().any(|(_, v)| {
+ if let Err(e) = v {
+ e.is_temporary()
+ } else {
+ false
+ }
+ }),
Err(err) => err.is_temporary(),
})
.notify(|res, dur| match res {
@@ -299,7 +305,14 @@ impl<A: Accessor> LayeredAccessor for RetryAccessor<A> {
target: "opendal::service",
"operation={} -> retry after {}s: error={:?}",
Operation::Batch, dur.as_secs_f64(),
- rp.results().errors().filter(|e| e.is_temporary()).collect::<Vec<_>>()),
+ rp.results().iter().filter_map(|(_, v)| {
+ if let Err(e) = v {
+ if e.is_temporary() {
+ return Some(e);
+ }
+ }
+ None
+ }).collect::<Vec<_>>()),
Err(err) => warn!(
target: "opendal::service",
"operation={} -> retry after {}s: error={:?}",
@@ -842,54 +855,56 @@ mod tests {
let mut attempt = self.attempt.lock().unwrap();
*attempt += 1;
- use BatchOperations::*;
- match op.into_operation() {
- Delete(v) => match *attempt {
- 1 => Err(
- Error::new(ErrorKind::Unexpected, "retryable_error from reader")
- .set_temporary(),
- ),
- 2 => Ok(RpBatch::new(BatchedResults::Delete(
- v.into_iter()
- .map(|(s, _)| {
- (
- s,
- Err(Error::new(
+ match *attempt {
+ 1 => Err(
+ Error::new(ErrorKind::Unexpected, "retryable_error from reader")
+ .set_temporary(),
+ ),
+ 2 => Ok(RpBatch::new(
+ op.into_operation()
+ .into_iter()
+ .map(|(s, _)| {
+ (
+ s,
+ Err(Error::new(
+ ErrorKind::Unexpected,
+ "retryable_error from reader",
+ )
+ .set_temporary()),
+ )
+ })
+ .collect(),
+ )),
+ 3 => Ok(RpBatch::new(
+ op.into_operation()
+ .into_iter()
+ .enumerate()
+ .map(|(i, (s, _))| {
+ (
+ s,
+ match i {
+ 0 => Err(Error::new(
ErrorKind::Unexpected,
"retryable_error from reader",
)
.set_temporary()),
- )
- })
- .collect(),
- ))),
- 3 => Ok(RpBatch::new(BatchedResults::Delete(
- v.into_iter()
- .enumerate()
- .map(|(i, (s, _))| {
- (
- s,
- match i {
- 0 => Err(Error::new(
- ErrorKind::Unexpected,
- "retryable_error from reader",
- )
- .set_temporary()),
- _ => Ok(RpDelete {}),
- },
- )
- })
- .collect(),
- ))),
- 4 => Err(
- Error::new(ErrorKind::Unexpected, "retryable_error from reader")
- .set_temporary(),
- ),
- 5 => Ok(RpBatch::new(BatchedResults::Delete(
- v.into_iter().map(|(s, _)| (s, Ok(RpDelete {}))).collect(),
- ))),
- _ => unreachable!(),
- },
+ _ => Ok(RpDelete {}.into()),
+ },
+ )
+ })
+ .collect(),
+ )),
+ 4 => Err(
+ Error::new(ErrorKind::Unexpected, "retryable_error from reader")
+ .set_temporary(),
+ ),
+ 5 => Ok(RpBatch::new(
+ op.into_operation()
+ .into_iter()
+ .map(|(s, _)| (s, Ok(RpDelete {}.into())))
+ .collect(),
+ )),
+ _ => unreachable!(),
}
}
}
diff --git a/core/src/raw/rps.rs b/core/src/raw/rps.rs
index 05954afc..37fa37f3 100644
--- a/core/src/raw/rps.rs
+++ b/core/src/raw/rps.rs
@@ -132,71 +132,35 @@ impl RpRead {
/// Reply for `batch` operation.
pub struct RpBatch {
- results: BatchedResults,
+ results: Vec<(String, Result<BatchedReply>)>,
}
impl RpBatch {
/// Create a new RpBatch.
- pub fn new(results: BatchedResults) -> Self {
+ pub fn new(results: Vec<(String, Result<BatchedReply>)>) -> Self {
Self { results }
}
/// Get the results from RpBatch.
- pub fn results(&self) -> &BatchedResults {
+ pub fn results(&self) -> &[(String, Result<BatchedReply>)] {
&self.results
}
/// Consume RpBatch to get the batched results.
- pub fn into_results(self) -> BatchedResults {
+ pub fn into_results(self) -> Vec<(String, Result<BatchedReply>)> {
self.results
}
}
/// Batch results of `batch` operations.
-pub enum BatchedResults {
+pub enum BatchedReply {
/// results of delete batch operation
- Delete(Vec<(String, Result<RpDelete>)>),
+ Delete(RpDelete),
}
-impl BatchedResults {
- /// Return the length of given results.
- pub fn len(&self) -> usize {
- use BatchedResults::*;
- match self {
- Delete(v) => v.len(),
- }
- }
-
- /// Return if given results is empty.
- pub fn is_empty(&self) -> bool {
- use BatchedResults::*;
- match self {
- Delete(v) => v.is_empty(),
- }
- }
-
- /// Return the length of ok results.
- pub fn len_ok(&self) -> usize {
- use BatchedResults::*;
- match self {
- Delete(v) => v.iter().filter(|v| v.1.is_ok()).count(),
- }
- }
-
- /// Return the length of error results.
- pub fn len_err(&self) -> usize {
- use BatchedResults::*;
- match self {
- Delete(v) => v.iter().filter(|v| v.1.is_err()).count(),
- }
- }
-
- /// Return an iterator over error results.
- pub fn errors(&self) -> Box<dyn Iterator<Item = &Error> + '_> {
- use BatchedResults::*;
- match self {
- Delete(v) => Box::new(v.iter().filter_map(|v| v.1.as_ref().err())),
- }
+impl From<RpDelete> for BatchedReply {
+ fn from(rp: RpDelete) -> Self {
+ Self::Delete(rp)
}
}
diff --git a/core/src/services/azblob/backend.rs b/core/src/services/azblob/backend.rs
index c194be5d..78a421cb 100644
--- a/core/src/services/azblob/backend.rs
+++ b/core/src/services/azblob/backend.rs
@@ -598,59 +598,58 @@ impl Accessor for AzblobBackend {
async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
let ops = args.into_operation();
- match ops {
- BatchOperations::Delete(ops) => {
- let paths = ops.into_iter().map(|(p, _)| p).collect::<Vec<_>>();
- if paths.len() > AZBLOB_BATCH_LIMIT {
- return Err(Error::new(
- ErrorKind::Unsupported,
- "batch delete limit exceeded",
- ));
- }
- // construct and complete batch request
- let resp = self.azblob_batch_delete(&paths).await?;
+ let paths = ops.into_iter().map(|(p, _)| p).collect::<Vec<_>>();
+ if paths.len() > AZBLOB_BATCH_LIMIT {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ "batch delete limit exceeded",
+ ));
+ }
+ // construct and complete batch request
+ let resp = self.azblob_batch_delete(&paths).await?;
- // check response status
- if resp.status() != StatusCode::ACCEPTED {
- return Err(parse_error(resp).await?);
- }
+ // check response status
+ if resp.status() != StatusCode::ACCEPTED {
+ return Err(parse_error(resp).await?);
+ }
- // get boundary from response header
- let content_type = resp.headers().get(CONTENT_TYPE).ok_or_else(|| {
- Error::new(
- ErrorKind::Unexpected,
- "response data should have CONTENT_TYPE header",
- )
- })?;
- let content_type = content_type
- .to_str()
- .map(|ty| ty.to_string())
- .map_err(|e| {
- Error::new(
- ErrorKind::Unexpected,
- &format!("get invalid CONTENT_TYPE header in response: {:?}", e),
- )
- })?;
- let splits = content_type.split("boundary=").collect::<Vec<&str>>();
- let boundary = splits.get(1).to_owned().ok_or_else(|| {
- Error::new(
- ErrorKind::Unexpected,
- "No boundary message provided in CONTENT_TYPE",
- )
- })?;
+ // get boundary from response header
+ let content_type = resp.headers().get(CONTENT_TYPE).ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "response data should have CONTENT_TYPE header",
+ )
+ })?;
+ let content_type = content_type
+ .to_str()
+ .map(|ty| ty.to_string())
+ .map_err(|e| {
+ Error::new(
+ ErrorKind::Unexpected,
+ &format!("get invalid CONTENT_TYPE header in response: {:?}", e),
+ )
+ })?;
+ let splits = content_type.split("boundary=").collect::<Vec<&str>>();
+ let boundary = splits.get(1).to_owned().ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "No boundary message provided in CONTENT_TYPE",
+ )
+ })?;
- let body = resp.into_body().bytes().await?;
- let body = String::from_utf8(body.to_vec()).map_err(|e| {
- Error::new(
- ErrorKind::Unexpected,
- &format!("get invalid batch response {e:?}"),
- )
- })?;
+ let body = resp.into_body().bytes().await?;
+ let body = String::from_utf8(body.to_vec()).map_err(|e| {
+ Error::new(
+ ErrorKind::Unexpected,
+ &format!("get invalid batch response {e:?}"),
+ )
+ })?;
- let results = parse_batch_delete_response(boundary, body, paths)?;
- Ok(RpBatch::new(BatchedResults::Delete(results)))
- }
- }
+ let results = parse_batch_delete_response(boundary, body, paths)?
+ .into_iter()
+ .map(|(path, rp)| (path, rp.map(|v| v.into())))
+ .collect();
+ Ok(RpBatch::new(results))
}
}
diff --git a/core/src/services/oss/backend.rs b/core/src/services/oss/backend.rs
index a61d7cdb..e8f59e75 100644
--- a/core/src/services/oss/backend.rs
+++ b/core/src/services/oss/backend.rs
@@ -544,61 +544,57 @@ impl Accessor for OssBackend {
async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
let ops = args.into_operation();
- match ops {
- BatchOperations::Delete(ops) => {
- // Sadly, OSS will not return failed keys, so we will build
- // a set to calculate the failed keys.
- let mut keys = HashSet::new();
-
- let ops_len = ops.len();
- if ops_len > 1000 {
- return Err(Error::new(
- ErrorKind::Unsupported,
- "oss services only allow delete up to 1000 keys at once",
- )
- .with_context("length", ops_len.to_string()));
- }
-
- let paths = ops
- .into_iter()
- .map(|(p, _)| {
- keys.insert(p.clone());
- p
- })
- .collect();
-
- let resp = self.oss_delete_objects(paths).await?;
-
- let status = resp.status();
-
- if let StatusCode::OK = status {
- let bs = resp.into_body().bytes().await?;
-
- let result: DeleteObjectsResult = quick_xml::de::from_reader(bs.reader())
- .map_err(new_xml_deserialize_error)?;
-
- let mut batched_result = Vec::with_capacity(ops_len);
- for i in result.deleted {
- let path = build_rel_path(&self.root, &i.key);
- keys.remove(&path);
- batched_result.push((path, Ok(RpDelete::default())));
- }
- // TODO: we should handle those errors with code.
- for i in keys {
- batched_result.push((
- i,
- Err(Error::new(
- ErrorKind::Unexpected,
- "oss delete this key failed for reason we don't know",
- )),
- ));
- }
-
- Ok(RpBatch::new(BatchedResults::Delete(batched_result)))
- } else {
- Err(parse_error(resp).await?)
- }
+ // Sadly, OSS will not return failed keys, so we will build
+ // a set to calculate the failed keys.
+ let mut keys = HashSet::new();
+
+ let ops_len = ops.len();
+ if ops_len > 1000 {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ "oss services only allow delete up to 1000 keys at once",
+ )
+ .with_context("length", ops_len.to_string()));
+ }
+
+ let paths = ops
+ .into_iter()
+ .map(|(p, _)| {
+ keys.insert(p.clone());
+ p
+ })
+ .collect();
+
+ let resp = self.oss_delete_objects(paths).await?;
+
+ let status = resp.status();
+
+ if let StatusCode::OK = status {
+ let bs = resp.into_body().bytes().await?;
+
+ let result: DeleteObjectsResult =
+ quick_xml::de::from_reader(bs.reader()).map_err(new_xml_deserialize_error)?;
+
+ let mut batched_result = Vec::with_capacity(ops_len);
+ for i in result.deleted {
+ let path = build_rel_path(&self.root, &i.key);
+ keys.remove(&path);
+ batched_result.push((path, Ok(RpDelete::default().into())));
}
+ // TODO: we should handle those errors with code.
+ for i in keys {
+ batched_result.push((
+ i,
+ Err(Error::new(
+ ErrorKind::Unexpected,
+ "oss delete this key failed for reason we don't know",
+ )),
+ ));
+ }
+
+ Ok(RpBatch::new(batched_result))
+ } else {
+ Err(parse_error(resp).await?)
}
}
}
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index 47098eb0..e1f9d7b8 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -1280,49 +1280,44 @@ impl Accessor for S3Backend {
async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
let ops = args.into_operation();
- match ops {
- BatchOperations::Delete(ops) => {
- if ops.len() > 1000 {
- return Err(Error::new(
- ErrorKind::Unsupported,
- "s3 services only allow delete up to 1000 keys at once",
- )
- .with_context("length", ops.len().to_string()));
- }
+ if ops.len() > 1000 {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ "s3 services only allow delete up to 1000 keys at once",
+ )
+ .with_context("length", ops.len().to_string()));
+ }
- let paths = ops.into_iter().map(|(p, _)| p).collect();
+ let paths = ops.into_iter().map(|(p, _)| p).collect();
- let resp = self.s3_delete_objects(paths).await?;
+ let resp = self.s3_delete_objects(paths).await?;
- let status = resp.status();
+ let status = resp.status();
- if let StatusCode::OK = status {
- let bs = resp.into_body().bytes().await?;
+ if let StatusCode::OK = status {
+ let bs = resp.into_body().bytes().await?;
- let result: DeleteObjectsResult = quick_xml::de::from_reader(bs.reader())
- .map_err(new_xml_deserialize_error)?;
-
- let mut batched_result =
- Vec::with_capacity(result.deleted.len() + result.error.len());
- for i in result.deleted {
- let path = build_rel_path(&self.root, &i.key);
- batched_result.push((path, Ok(RpDelete::default())));
- }
- // TODO: we should handle those errors with code.
- for i in result.error {
- let path = build_rel_path(&self.root, &i.key);
-
- batched_result.push((
- path,
- Err(Error::new(ErrorKind::Unexpected, &format!("{i:?}"))),
- ));
- }
-
- Ok(RpBatch::new(BatchedResults::Delete(batched_result)))
- } else {
- Err(parse_error(resp).await?)
- }
+ let result: DeleteObjectsResult =
+ quick_xml::de::from_reader(bs.reader()).map_err(new_xml_deserialize_error)?;
+
+ let mut batched_result = Vec::with_capacity(result.deleted.len() + result.error.len());
+ for i in result.deleted {
+ let path = build_rel_path(&self.root, &i.key);
+ batched_result.push((path, Ok(RpDelete::default().into())));
}
+ // TODO: we should handle those errors with code.
+ for i in result.error {
+ let path = build_rel_path(&self.root, &i.key);
+
+ batched_result.push((
+ path,
+ Err(Error::new(ErrorKind::Unexpected, &format!("{i:?}"))),
+ ));
+ }
+
+ Ok(RpBatch::new(batched_result))
+ } else {
+ Err(parse_error(resp).await?)
}
}
}
diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs
index 0231dc88..791a0403 100644
--- a/core/src/types/operator/operator.rs
+++ b/core/src/types/operator/operator.rs
@@ -732,15 +732,16 @@ impl Operator {
/// ```
pub async fn remove_via(&self, input: impl Stream<Item = String> + Unpin) -> Result<()> {
if self.info().can_batch() {
- let mut input = input.map(|v| (v, OpDelete::default())).chunks(self.limit());
+ let mut input = input
+ .map(|v| (v, OpDelete::default().into()))
+ .chunks(self.limit());
while let Some(batches) = input.next().await {
let results = self
.inner()
- .batch(OpBatch::new(BatchOperations::Delete(batches)))
- .await?;
-
- let BatchedResults::Delete(results) = results.into_results();
+ .batch(OpBatch::new(batches))
+ .await?
+ .into_results();
// TODO: return error here directly seems not a good idea?
for (_, result) in results {
@@ -805,15 +806,14 @@ impl Operator {
let batches = batches
.map_err(|err| err.1)?
.into_iter()
- .map(|v| (v.path().to_string(), OpDelete::default()))
+ .map(|v| (v.path().to_string(), OpDelete::default().into()))
.collect();
let results = self
.inner()
- .batch(OpBatch::new(BatchOperations::Delete(batches)))
- .await?;
-
- let BatchedResults::Delete(results) = results.into_results();
+ .batch(OpBatch::new(batches))
+ .await?
+ .into_results();
// TODO: return error here directly seems not a good idea?
for (_, result) in results {
diff --git a/core/src/types/ops.rs b/core/src/types/ops.rs
index 0612ac34..70addb8e 100644
--- a/core/src/types/ops.rs
+++ b/core/src/types/ops.rs
@@ -172,22 +172,22 @@ impl From<OpWrite> for PresignOperation {
/// Args for `batch` operation.
#[derive(Debug, Clone)]
pub struct OpBatch {
- ops: BatchOperations,
+ ops: Vec<(String, BatchOperation)>,
}
impl OpBatch {
/// Create a new batch options.
- pub fn new(ops: BatchOperations) -> Self {
+ pub fn new(ops: Vec<(String, BatchOperation)>) -> Self {
Self { ops }
}
/// Get operation from op.
- pub fn operation(&self) -> &BatchOperations {
+ pub fn operation(&self) -> &[(String, BatchOperation)] {
&self.ops
}
/// Consume OpBatch into BatchOperation
- pub fn into_operation(self) -> BatchOperations {
+ pub fn into_operation(self) -> Vec<(String, BatchOperation)> {
self.ops
}
}
@@ -195,35 +195,25 @@ impl OpBatch {
/// Batch operation used for batch.
#[derive(Debug, Clone)]
#[non_exhaustive]
-pub enum BatchOperations {
- /// Batch delete operations.
- Delete(Vec<(String, OpDelete)>),
+pub enum BatchOperation {
+ /// Batch delete operation.
+ Delete(OpDelete),
}
-impl BatchOperations {
+impl From<OpDelete> for BatchOperation {
+ fn from(op: OpDelete) -> Self {
+ Self::Delete(op)
+ }
+}
+
+impl BatchOperation {
/// Return the operation of this batch.
pub fn operation(&self) -> Operation {
- use BatchOperations::*;
+ use BatchOperation::*;
match self {
Delete(_) => Operation::Delete,
}
}
-
- /// Return the length of given operations.
- pub fn len(&self) -> usize {
- use BatchOperations::*;
- match self {
- Delete(v) => v.len(),
- }
- }
-
- /// Return if given operations is empty.
- pub fn is_empty(&self) -> bool {
- use BatchOperations::*;
- match self {
- Delete(v) => v.is_empty(),
- }
- }
}
/// Args for `read` operation.