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/01/09 10:13:50 UTC
(incubator-opendal) branch main updated: fix: Don't retry close if concurrent > 1 to avoid content lost (#3957)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 65973ebd2e fix: Don't retry close if concurrent > 1 to avoid content lost (#3957)
65973ebd2e is described below
commit 65973ebd2e41f3e5c279b9d9fa990c3886162043
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Jan 9 18:13:46 2024 +0800
fix: Don't retry close if concurrent > 1 to avoid content lost (#3957)
* Fix build
Signed-off-by: Xuanwo <gi...@xuanwo.io>
* Fix
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---------
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/oio/write/multipart_upload_write.rs | 10 +++++++++-
core/src/raw/oio/write/range_write.rs | 11 ++++++++++-
2 files changed, 19 insertions(+), 2 deletions(-)
diff --git a/core/src/raw/oio/write/multipart_upload_write.rs b/core/src/raw/oio/write/multipart_upload_write.rs
index 66105b4d4a..3a3d81f891 100644
--- a/core/src/raw/oio/write/multipart_upload_write.rs
+++ b/core/src/raw/oio/write/multipart_upload_write.rs
@@ -133,6 +133,7 @@ pub struct MultipartUploadWriter<W: MultipartUploadWrite> {
w: Arc<W>,
upload_id: Option<Arc<String>>,
+ concurrent: usize,
parts: Vec<MultipartUploadPart>,
cache: Option<oio::ChunkedBytes>,
futures: ConcurrentFutures<WritePartFuture>,
@@ -163,6 +164,7 @@ impl<W: MultipartUploadWrite> MultipartUploadWriter<W> {
w: Arc::new(inner),
upload_id: None,
+ concurrent,
parts: Vec::new(),
cache: None,
futures: ConcurrentFutures::new(1.max(concurrent)),
@@ -275,7 +277,13 @@ where
})));
}
}
- while let Some(part) = ready!(self.futures.poll_next_unpin(cx)) {
+ while let Some(mut part) = ready!(self.futures.poll_next_unpin(cx))
+ {
+ // Don't retry close if concurrent write failed.
+ // TODO: Remove this after <https://github.com/apache/incubator-opendal/issues/3956> addressed.
+ if self.concurrent > 1 {
+ part = part.map_err(|err| err.set_permanent());
+ }
self.parts.push(part?);
}
}
diff --git a/core/src/raw/oio/write/range_write.rs b/core/src/raw/oio/write/range_write.rs
index bbd81606b0..285f4818fe 100644
--- a/core/src/raw/oio/write/range_write.rs
+++ b/core/src/raw/oio/write/range_write.rs
@@ -111,6 +111,7 @@ pub struct RangeWriter<W: RangeWrite> {
next_offset: u64,
buffer: Option<oio::ChunkedBytes>,
futures: ConcurrentFutures<WriteRangeFuture>,
+ concurrent: usize,
w: Arc<W>,
state: State,
@@ -144,6 +145,7 @@ impl<W: RangeWrite> RangeWriter<W> {
buffer: None,
location: None,
next_offset: 0,
+ concurrent,
}
}
@@ -220,7 +222,14 @@ impl<W: RangeWrite> oio::Write for RangeWriter<W> {
match self.location.clone() {
Some(location) => {
if !self.futures.is_empty() {
- while let Some(result) = ready!(self.futures.poll_next_unpin(cx)) {
+ while let Some(mut result) =
+ ready!(self.futures.poll_next_unpin(cx))
+ {
+ // Don't retry close if concurrent write failed.
+ // TODO: Remove this after <https://github.com/apache/incubator-opendal/issues/3956> addressed.
+ if self.concurrent > 1 {
+ result = result.map_err(|err| err.set_permanent());
+ }
result?;
}
}