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 2026/09/21 10:20:17 UTC
(opendal) 01/01: fix(core): disable prefetch for concurrent writes and copies
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/fix-concurrent-write-prefetch
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 9570420a5df566c83415b9b46393a21dbb89a000
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Sep 21 18:19:58 2026 +0800
fix(core): disable prefetch for concurrent writes and copies
These paths do not collect ConcurrentTasks results while submitting work, so a non-zero prefetch lets failed inputs accumulate and delays write() errors.
Co-authored-by: Cursor <cu...@cursor.com>
---
core/core/src/raw/futures_util.rs | 5 +++
core/core/src/raw/oio/copy/block_copy.rs | 2 +-
core/core/src/raw/oio/copy/multipart_copy.rs | 2 +-
core/core/src/raw/oio/write/block_write.rs | 2 +-
core/core/src/raw/oio/write/multipart_write.rs | 2 +-
core/core/src/raw/oio/write/position_write.rs | 42 +++++++++++++++++++++++++-
6 files changed, 50 insertions(+), 5 deletions(-)
diff --git a/core/core/src/raw/futures_util.rs b/core/core/src/raw/futures_util.rs
index 8e65323170..778330458c 100644
--- a/core/core/src/raw/futures_util.rs
+++ b/core/core/src/raw/futures_util.rs
@@ -157,6 +157,11 @@ pub struct ConcurrentTasks<I, O> {
impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
/// Create a new concurrent tasks with given executor, concurrent, prefetch and factory.
///
+ /// `prefetch` is extra completed tasks allowed in the queue before
+ /// [`Self::execute`] waits to collect the head. Pass `0` when the caller
+ /// does not collect results while still submitting work; otherwise failed
+ /// tasks keep their inputs until collection.
+ ///
/// The factory is a function pointer that shouldn't capture any context.
pub fn new(
executor: Executor,
diff --git a/core/core/src/raw/oio/copy/block_copy.rs b/core/core/src/raw/oio/copy/block_copy.rs
index 60f2b840ea..f6d52ded1e 100644
--- a/core/core/src/raw/oio/copy/block_copy.rs
+++ b/core/core/src/raw/oio/copy/block_copy.rs
@@ -133,7 +133,7 @@ impl<C: BlockCopy> BlockCopier<C> {
completed: false,
metadata: None,
- tasks: ConcurrentTasks::new(executor, concurrent, 8192, |input| {
+ tasks: ConcurrentTasks::new(executor, concurrent, 0, |input| {
Box::pin(async move {
let size = input.range.size().expect("block copy range must be sized");
let fut = input.copier.copy_block(input.block_id, input.range);
diff --git a/core/core/src/raw/oio/copy/multipart_copy.rs b/core/core/src/raw/oio/copy/multipart_copy.rs
index 1b2b5fe468..858b9c8e13 100644
--- a/core/core/src/raw/oio/copy/multipart_copy.rs
+++ b/core/core/src/raw/oio/copy/multipart_copy.rs
@@ -159,7 +159,7 @@ impl<C: MultipartCopy> MultipartCopier<C> {
completed: false,
metadata: None,
- tasks: ConcurrentTasks::new(executor, concurrent, 8192, |input| {
+ tasks: ConcurrentTasks::new(executor, concurrent, 0, |input| {
Box::pin(async move {
let size = input
.range
diff --git a/core/core/src/raw/oio/write/block_write.rs b/core/core/src/raw/oio/write/block_write.rs
index f45c899068..a96da1a737 100644
--- a/core/core/src/raw/oio/write/block_write.rs
+++ b/core/core/src/raw/oio/write/block_write.rs
@@ -125,7 +125,7 @@ impl<W: BlockWrite> BlockWriter<W> {
block_ids: Vec::new(),
cache: None,
- tasks: ConcurrentTasks::new(executor, concurrent, 8192, |input| {
+ tasks: ConcurrentTasks::new(executor, concurrent, 0, |input| {
Box::pin(async move {
let fut = input
.w
diff --git a/core/core/src/raw/oio/write/multipart_write.rs b/core/core/src/raw/oio/write/multipart_write.rs
index 4134245183..5f61168aab 100644
--- a/core/core/src/raw/oio/write/multipart_write.rs
+++ b/core/core/src/raw/oio/write/multipart_write.rs
@@ -186,7 +186,7 @@ impl<W: MultipartWrite> MultipartWriter<W> {
cache: None,
next_part_number: 0,
- tasks: ConcurrentTasks::new(executor, concurrent, 8192, |input| {
+ tasks: ConcurrentTasks::new(executor, concurrent, 0, |input| {
Box::pin({
async move {
let fut = async {
diff --git a/core/core/src/raw/oio/write/position_write.rs b/core/core/src/raw/oio/write/position_write.rs
index bfa4d81a2a..56272f4253 100644
--- a/core/core/src/raw/oio/write/position_write.rs
+++ b/core/core/src/raw/oio/write/position_write.rs
@@ -88,7 +88,7 @@ impl<W: PositionWrite> PositionWriter<W> {
next_offset: 0,
cache: None,
- tasks: ConcurrentTasks::new(executor, concurrent, 8192, |input| {
+ tasks: ConcurrentTasks::new(executor, concurrent, 0, |input| {
Box::pin(async move {
let fut = input.w.write_all_at(input.offset, input.bytes.clone());
match input.executor.timeout() {
@@ -280,4 +280,44 @@ mod tests {
let actual_size = w.w.lock().unwrap().length;
assert_eq!(actual_size, total_size);
}
+
+ struct FailingWrite;
+
+ impl PositionWrite for FailingWrite {
+ async fn write_all_at(&self, _offset: u64, _buf: Buffer) -> Result<()> {
+ Err(Error::new(ErrorKind::Unexpected, "backend write failed"))
+ }
+
+ async fn close(&self, _size: u64) -> Result<Metadata> {
+ Ok(MetadataBuilder::unknown().build())
+ }
+
+ async fn abort(&self) -> Result<()> {
+ Ok(())
+ }
+ }
+
+ #[tokio::test]
+ async fn test_position_writer_surfaces_permanent_error_within_concurrent_window() {
+ let concurrent = 8;
+ let mut w = PositionWriter::new(Executor::default(), FailingWrite, concurrent);
+ let bs = Buffer::from(vec![0u8; 8]);
+
+ let mut first_error_at = None;
+ for i in 1..=concurrent + 4 {
+ match w.write(bs.clone()).await {
+ Ok(()) => continue,
+ Err(_) => {
+ first_error_at = Some(i);
+ break;
+ }
+ }
+ }
+
+ let first_error_at = first_error_at
+ .expect("write must surface the backend error inside the concurrent window");
+ // One write stays in cache; ConcurrentTasks then holds at most `concurrent`
+ // tasks when prefetch is 0. The next write awaits the queue head.
+ assert_eq!(first_error_at, concurrent + 2);
+ }
}