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:02:43 UTC
(incubator-opendal) branch fix-concurrent created (now ff42b8561c)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch fix-concurrent
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
at ff42b8561c Fix build
This branch includes the following new commits:
new ff42b8561c Fix build
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: Fix build
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch fix-concurrent
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
commit ff42b8561c5af886c9a32e3a6d183cfbc307969e
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Jan 9 18:01:59 2024 +0800
Fix build
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/oio/write/block_write.rs | 9 ++++++++-
core/src/raw/oio/write/multipart_upload_write.rs | 9 ++++++++-
core/src/raw/oio/write/range_write.rs | 9 ++++++++-
3 files changed, 24 insertions(+), 3 deletions(-)
diff --git a/core/src/raw/oio/write/block_write.rs b/core/src/raw/oio/write/block_write.rs
index 8665d50c00..f7acd8480b 100644
--- a/core/src/raw/oio/write/block_write.rs
+++ b/core/src/raw/oio/write/block_write.rs
@@ -103,6 +103,7 @@ pub struct BlockWriter<W: BlockWrite> {
state: State,
w: Arc<W>,
+ concurrent: usize,
block_ids: Vec<String>,
cache: Option<oio::ChunkedBytes>,
futures: ConcurrentFutures<WriteBlockFuture>,
@@ -132,6 +133,7 @@ impl<W: BlockWrite> BlockWriter<W> {
w: Arc::new(inner),
block_ids: Vec::new(),
cache: None,
+ concurrent,
futures: ConcurrentFutures::new(1.max(concurrent)),
}
}
@@ -236,7 +238,12 @@ where
}
}
State::Close(fut) => {
- let res = futures::ready!(fut.as_mut().poll(cx));
+ let mut res = futures::ready!(fut.as_mut().poll(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 {
+ res = res.map_err(|err| err.set_permanent());
+ }
self.state = State::Idle;
// We should check res first before clean up cache.
res?;
diff --git a/core/src/raw/oio/write/multipart_upload_write.rs b/core/src/raw/oio/write/multipart_upload_write.rs
index 66105b4d4a..7f3994a3bd 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)),
@@ -300,7 +302,12 @@ where
}
}
State::Close(fut) => {
- let res = futures::ready!(fut.as_mut().poll(cx));
+ let mut res = futures::ready!(fut.as_mut().poll(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 {
+ res = res.map_err(|err| err.set_permanent());
+ }
self.state = State::Idle;
// We should check res first before clean up cache.
res?;
diff --git a/core/src/raw/oio/write/range_write.rs b/core/src/raw/oio/write/range_write.rs
index bbd81606b0..978043cfa7 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,
}
}
@@ -262,7 +264,12 @@ impl<W: RangeWrite> oio::Write for RangeWriter<W> {
unreachable!("RangeWriter must not go into State::Init during poll_close")
}
State::Complete(fut) => {
- let res = ready!(fut.poll_unpin(cx));
+ let mut res = ready!(fut.poll_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 {
+ res = res.map_err(|err| err.set_permanent());
+ }
self.state = State::Idle;
return Poll::Ready(res);
}