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 2025/07/29 15:32:33 UTC
(opendal) branch fix-concurrent-read-backpressure created (now 885583d1c)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch fix-concurrent-read-backpressure
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 885583d1c feat: Add prefetch support for concurrent read
This branch includes the following new commits:
new 885583d1c feat: Add prefetch support for concurrent read
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.
(opendal) 01/01: feat: Add prefetch support for concurrent read
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-read-backpressure
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 885583d1c956367acd4111349d5693b47941bd84
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Jul 29 23:31:26 2025 +0800
feat: Add prefetch support for concurrent read
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/benches/types/tasks.rs | 2 +-
core/src/raw/futures_util.rs | 93 ++++++++++++++++++++++++++++++-
core/src/raw/oio/write/block_write.rs | 2 +-
core/src/raw/oio/write/multipart_write.rs | 2 +-
core/src/raw/oio/write/position_write.rs | 2 +-
core/src/raw/ops.rs | 16 ++++++
core/src/types/options.rs | 14 +++++
core/src/types/read/buffer_stream.rs | 1 +
core/src/types/read/reader.rs | 1 +
9 files changed, 126 insertions(+), 7 deletions(-)
diff --git a/core/benches/types/tasks.rs b/core/benches/types/tasks.rs
index fbedf0d1b..06ef4eaeb 100644
--- a/core/benches/types/tasks.rs
+++ b/core/benches/types/tasks.rs
@@ -41,7 +41,7 @@ fn baseline(b: Bencher) {
#[divan::bench(args = [1, 2, 8, 32])]
fn concurrent(b: Bencher, concurrent: usize) {
b.with_inputs(|| {
- ConcurrentTasks::new(Executor::new(), concurrent, |()| {
+ ConcurrentTasks::new(Executor::new(), concurrent, 8, |()| {
Box::pin(async {
tokio::time::sleep(Duration::from_millis(1)).await;
((), Ok(()))
diff --git a/core/src/raw/futures_util.rs b/core/src/raw/futures_util.rs
index 1c6cd9386..948124242 100644
--- a/core/src/raw/futures_util.rs
+++ b/core/src/raw/futures_util.rs
@@ -134,6 +134,8 @@ pub struct ConcurrentTasks<I, O> {
/// The maximum number of concurrent tasks.
concurrent: usize,
+ /// The maximum number of completed tasks that can be buffered.
+ prefetch: usize,
/// Tracks the number of tasks that have finished execution but have not yet been collected.
/// This count is subtracted from the total concurrency capacity, ensuring that the system
/// always schedules new tasks to maintain the user's desired concurrency level.
@@ -150,12 +152,13 @@ pub struct ConcurrentTasks<I, O> {
}
impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
- /// Create a new concurrent tasks with given executor, concurrent and factory.
+ /// Create a new concurrent tasks with given executor, concurrent, prefetch and factory.
///
/// The factory is a function pointer that shouldn't capture any context.
pub fn new(
executor: Executor,
concurrent: usize,
+ prefetch: usize,
factory: fn(I) -> BoxedStaticFuture<(I, Result<O>)>,
) -> Self {
Self {
@@ -165,6 +168,7 @@ impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
tasks: VecDeque::with_capacity(concurrent),
results: VecDeque::with_capacity(concurrent),
concurrent,
+ prefetch,
completed_but_unretrieved: Arc::default(),
errored: false,
}
@@ -187,7 +191,9 @@ impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
/// Check if there are remaining space to push new tasks.
#[inline]
pub fn has_remaining(&self) -> bool {
- self.tasks.len() < self.concurrent + self.completed_but_unretrieved.load(Ordering::Relaxed)
+ let completed = self.completed_but_unretrieved.load(Ordering::Relaxed);
+ // Allow up to `prefetch` completed tasks to be buffered
+ self.tasks.len() < self.concurrent + completed.min(self.prefetch)
}
/// Chunk if there are remaining results to fetch.
@@ -324,7 +330,7 @@ mod tests {
async fn test_concurrent_tasks() {
let executor = Executor::new();
- let mut tasks = ConcurrentTasks::new(executor, 16, |(i, dur)| {
+ let mut tasks = ConcurrentTasks::new(executor, 16, 8, |(i, dur)| {
Box::pin(async move {
sleep(dur).await;
@@ -362,4 +368,85 @@ mod tests {
assert_eq!(ans, (0..10240).collect::<Vec<_>>())
}
+
+ #[tokio::test]
+ async fn test_prefetch_backpressure() {
+ let executor = Executor::new();
+ let concurrent = 4;
+ let prefetch = 2;
+
+ // Create a slower task to ensure they don't complete immediately
+ let mut tasks = ConcurrentTasks::new(executor, concurrent, prefetch, |i: usize| {
+ Box::pin(async move {
+ sleep(Duration::from_millis(100)).await;
+ (i, Ok(i))
+ })
+ });
+
+ // Initially, we should have space for concurrent tasks
+ assert!(tasks.has_remaining(), "Should have space initially");
+
+ // Submit concurrent tasks
+ for i in 0..concurrent {
+ assert!(tasks.has_remaining(), "Should have space for task {}", i);
+ tasks.execute(i).await.unwrap();
+ }
+
+ // Now we shouldn't have any more space (since no tasks have completed yet)
+ assert!(!tasks.has_remaining(), "Should not have space after submitting concurrent tasks");
+
+ // Wait for some tasks to complete
+ sleep(Duration::from_millis(150)).await;
+
+ // Now we should have space up to prefetch limit
+ for i in concurrent..concurrent + prefetch {
+ assert!(tasks.has_remaining(), "Should have space for prefetch task {}", i);
+ tasks.execute(i).await.unwrap();
+ }
+
+ // Now has_remaining should return false
+ assert!(!tasks.has_remaining(), "Should not have remaining space after filling up prefetch buffer");
+
+ // Retrieve one result
+ let result = tasks.next().await;
+ assert!(result.is_some());
+
+ // Now there should be space for one more task
+ assert!(tasks.has_remaining(), "Should have remaining space after retrieving one result");
+ }
+
+ #[tokio::test]
+ async fn test_prefetch_zero() {
+ let executor = Executor::new();
+ let concurrent = 4;
+ let prefetch = 0; // No prefetching allowed
+
+ let mut tasks = ConcurrentTasks::new(executor, concurrent, prefetch, |i: usize| {
+ Box::pin(async move {
+ sleep(Duration::from_millis(10)).await;
+ (i, Ok(i))
+ })
+ });
+
+ // With prefetch=0, we can only submit up to concurrent tasks
+ for i in 0..concurrent {
+ tasks.execute(i).await.unwrap();
+ }
+
+ // Should not have space for more
+ assert!(!tasks.has_remaining(), "Should not have remaining space with prefetch=0");
+
+ // Retrieve one result
+ let result = tasks.next().await;
+ assert!(result.is_some());
+
+ // Now there should be space for exactly one more task
+ assert!(tasks.has_remaining(), "Should have remaining space after retrieving one result");
+
+ // Execute one more
+ tasks.execute(concurrent).await.unwrap();
+
+ // Should be full again
+ assert!(!tasks.has_remaining(), "Should be full again");
+ }
}
diff --git a/core/src/raw/oio/write/block_write.rs b/core/src/raw/oio/write/block_write.rs
index e65b92d76..31b9dd76d 100644
--- a/core/src/raw/oio/write/block_write.rs
+++ b/core/src/raw/oio/write/block_write.rs
@@ -124,7 +124,7 @@ impl<W: BlockWrite> BlockWriter<W> {
block_ids: Vec::new(),
cache: None,
- tasks: ConcurrentTasks::new(executor, concurrent, |input| {
+ tasks: ConcurrentTasks::new(executor, concurrent, 8192, |input| {
Box::pin(async move {
let fut = input
.w
diff --git a/core/src/raw/oio/write/multipart_write.rs b/core/src/raw/oio/write/multipart_write.rs
index 105ecaa24..32b97774e 100644
--- a/core/src/raw/oio/write/multipart_write.rs
+++ b/core/src/raw/oio/write/multipart_write.rs
@@ -158,7 +158,7 @@ impl<W: MultipartWrite> MultipartWriter<W> {
cache: None,
next_part_number: 0,
- tasks: ConcurrentTasks::new(executor, concurrent, |input| {
+ tasks: ConcurrentTasks::new(executor, concurrent, 8192, |input| {
Box::pin({
async move {
let fut = input.w.write_part(
diff --git a/core/src/raw/oio/write/position_write.rs b/core/src/raw/oio/write/position_write.rs
index fc46732d5..134df2cb7 100644
--- a/core/src/raw/oio/write/position_write.rs
+++ b/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, |input| {
+ tasks: ConcurrentTasks::new(executor, concurrent, 8192, |input| {
Box::pin(async move {
let fut = input.w.write_all_at(input.offset, input.bytes.clone());
match input.executor.timeout() {
diff --git a/core/src/raw/ops.rs b/core/src/raw/ops.rs
index 7a6760ab2..416f2fe71 100644
--- a/core/src/raw/ops.rs
+++ b/core/src/raw/ops.rs
@@ -441,6 +441,8 @@ pub struct OpReader {
chunk: Option<usize>,
/// The gap size of each request.
gap: Option<usize>,
+ /// The maximum number of buffers that can be prefetched.
+ prefetch: usize,
}
impl Default for OpReader {
@@ -449,6 +451,7 @@ impl Default for OpReader {
concurrent: 1,
chunk: None,
gap: None,
+ prefetch: 0,
}
}
}
@@ -491,6 +494,17 @@ impl OpReader {
pub fn gap(&self) -> Option<usize> {
self.gap
}
+
+ /// Set the prefetch of the option
+ pub fn with_prefetch(mut self, prefetch: usize) -> Self {
+ self.prefetch = prefetch;
+ self
+ }
+
+ /// Get prefetch from option
+ pub fn prefetch(&self) -> usize {
+ self.prefetch
+ }
}
impl From<options::ReadOptions> for (OpRead, OpReader) {
@@ -512,6 +526,7 @@ impl From<options::ReadOptions> for (OpRead, OpReader) {
concurrent: value.concurrent.max(1),
chunk: value.chunk,
gap: value.gap,
+ prefetch: 0,
},
)
}
@@ -536,6 +551,7 @@ impl From<options::ReaderOptions> for (OpRead, OpReader) {
concurrent: value.concurrent.max(1),
chunk: value.chunk,
gap: value.gap,
+ prefetch: value.prefetch,
},
)
}
diff --git a/core/src/types/options.rs b/core/src/types/options.rs
index d63be89bb..81863366d 100644
--- a/core/src/types/options.rs
+++ b/core/src/types/options.rs
@@ -231,6 +231,20 @@ pub struct ReaderOptions {
///
/// Refer to [`crate::docs::performance`] for more details.
pub gap: Option<usize>,
+ /// Controls the number of prefetched bytes ranges that can be buffered in memory
+ /// during concurrent reading.
+ ///
+ /// When performing concurrent reads with `Reader`, this option limits how many
+ /// completed-but-not-yet-read chunks can be buffered. Once the number of buffered
+ /// chunks reaches this limit, no new read tasks will be spawned until some of the
+ /// buffered chunks are consumed.
+ ///
+ /// - Default value: 0 (no prefetching, strict back-pressure control)
+ /// - Set to a higher value to allow more aggressive prefetching at the cost of memory
+ ///
+ /// This option helps prevent memory exhaustion when reading large files with high
+ /// concurrency settings.
+ pub prefetch: usize,
}
/// Options for stat operations.
diff --git a/core/src/types/read/buffer_stream.rs b/core/src/types/read/buffer_stream.rs
index 5073426bb..a3ab8a85e 100644
--- a/core/src/types/read/buffer_stream.rs
+++ b/core/src/types/read/buffer_stream.rs
@@ -90,6 +90,7 @@ impl ChunkedReader {
let tasks = ConcurrentTasks::new(
ctx.accessor().info().executor(),
ctx.options().concurrent(),
+ ctx.options().prefetch(),
|mut r: oio::Reader| {
Box::pin(async {
match r.read_all().await {
diff --git a/core/src/types/read/reader.rs b/core/src/types/read/reader.rs
index 15c1cd46b..2672a4acc 100644
--- a/core/src/types/read/reader.rs
+++ b/core/src/types/read/reader.rs
@@ -155,6 +155,7 @@ impl Reader {
let mut tasks = ConcurrentTasks::new(
self.ctx.accessor().info().executor(),
self.ctx.options().concurrent(),
+ self.ctx.options().prefetch(),
|input: FetchInput| {
Box::pin(async move {
let FetchInput { range, reader } = input.clone();