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:44 UTC
(incubator-opendal) 01/01: Fix build
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);
}