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:34 UTC
(opendal) 01/01: feat: Add prefetch support for concurrent read
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();