You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by ti...@apache.org on 2025/07/26 14:59:21 UTC
(opendal) 01/01: fix: patch futures buffer_by_ordered
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch fixup-buffered-ordered
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 65eafda3a858bd62cd53442aec00bf6b0767360e
Author: tison <wa...@gmail.com>
AuthorDate: Sat Jul 26 22:59:06 2025 +0800
fix: patch futures buffer_by_ordered
Signed-off-by: tison <wa...@gmail.com>
---
bin/oli/Cargo.lock | 1 +
core/Cargo.lock | 1 +
core/Cargo.toml | 1 +
core/src/lib.rs | 5 +-
core/src/patches/buffer_by_ordered.rs | 158 ++++++++++++++++++++++++++++++++++
core/src/patches/mod.rs | 19 ++++
core/src/patches/stream.rs | 34 ++++++++
core/src/types/read/reader.rs | 6 +-
8 files changed, 221 insertions(+), 4 deletions(-)
diff --git a/bin/oli/Cargo.lock b/bin/oli/Cargo.lock
index faf8fa562..cb3f63654 100644
--- a/bin/oli/Cargo.lock
+++ b/bin/oli/Cargo.lock
@@ -1397,6 +1397,7 @@ dependencies = [
"log",
"md-5",
"percent-encoding",
+ "pin-project-lite",
"prost",
"quick-xml 0.38.0",
"reqsign",
diff --git a/core/Cargo.lock b/core/Cargo.lock
index c6edf5426..db1b6b061 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -5419,6 +5419,7 @@ dependencies = [
"ouroboros",
"percent-encoding",
"persy",
+ "pin-project-lite",
"pretty_assertions",
"probe",
"prometheus 0.14.0",
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 50e4c897a..9bd23f2b3 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -255,6 +255,7 @@ http-body = "1"
log = "0.4"
md-5 = "0.10"
percent-encoding = "2"
+pin-project-lite = { version = "0.2.16"}
quick-xml = { version = "0.38", features = ["serialize", "overlapped-lists"] }
reqwest = { version = "0.12.22", features = [
"stream",
diff --git a/core/src/lib.rs b/core/src/lib.rs
index 864dfb452..1ecaf22a2 100644
--- a/core/src/lib.rs
+++ b/core/src/lib.rs
@@ -153,7 +153,10 @@
//! - [Performance Guide][crate::docs::performance]
// Make sure all our public APIs have docs.
-#![warn(missing_docs)]
+#![deny(missing_docs)]
+
+// Private modules, they will not be accessed by users.
+mod patches;
// Private module with public types, they will be accessed via `opendal::Xxxx`
mod types;
diff --git a/core/src/patches/buffer_by_ordered.rs b/core/src/patches/buffer_by_ordered.rs
new file mode 100644
index 000000000..3f6b0da01
--- /dev/null
+++ b/core/src/patches/buffer_by_ordered.rs
@@ -0,0 +1,158 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::VecDeque;
+use std::fmt;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+
+use futures::future::Future;
+use futures::stream::Fuse;
+use futures::stream::FuturesOrdered;
+use futures::stream::Stream;
+use futures::stream::StreamExt;
+
+use pin_project_lite::pin_project;
+
+pin_project! {
+ /// Stream for the [`buffer_by_ordered`] method.
+ ///
+ /// [`buffer_by_ordered`]: crate::StreamExt::buffer_by_ordered
+ #[must_use = "streams do nothing unless polled"]
+ pub struct BufferByOrdered<St, F>
+ where
+ St: Stream<Item = (F, usize)>,
+ F: Future,
+ {
+ #[pin]
+ stream: Fuse<St>,
+ in_progress_queue: FuturesOrdered<SizedFuture<F>>,
+ ready_queue: VecDeque<(F::Output, usize)>,
+ max_size: usize,
+ current_size: usize,
+ }
+}
+
+impl<St, F> fmt::Debug for BufferByOrdered<St, F>
+where
+ St: Stream<Item = (F, usize)> + fmt::Debug,
+ F: Future,
+{
+ fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+ f.debug_struct("BufferByOrdered")
+ .field("stream", &self.stream)
+ .field("in_progress_queue", &self.in_progress_queue)
+ .field("max_size", &self.max_size)
+ .field("current_size", &self.current_size)
+ .finish()
+ }
+}
+
+impl<St, F> BufferByOrdered<St, F>
+where
+ St: Stream<Item = (F, usize)>,
+ F: Future,
+{
+ pub(crate) fn new(stream: St, max_size: usize) -> Self {
+ Self {
+ stream: stream.fuse(),
+ in_progress_queue: FuturesOrdered::new(),
+ ready_queue: VecDeque::new(),
+ max_size,
+ current_size: 0,
+ }
+ }
+}
+
+impl<St, F> Stream for BufferByOrdered<St, F>
+where
+ St: Stream<Item = (F, usize)>,
+ F: Future,
+{
+ type Item = F::Output;
+
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let mut this = self.project();
+
+ // First up, try to spawn off as many futures as possible by filling up
+ // our queue of futures.
+ while this.current_size < this.max_size {
+ match this.stream.as_mut().poll_next(cx) {
+ Poll::Ready(Some((future, size))) => {
+ *this.current_size += size;
+ this.in_progress_queue
+ .push_back(SizedFuture { future, size });
+ }
+ Poll::Ready(None) => break,
+ Poll::Pending => break,
+ }
+ }
+
+ // Try to poll all ready futures in the in_progress_queue.
+ loop {
+ match this.in_progress_queue.poll_next_unpin(cx) {
+ Poll::Ready(Some(output)) => {
+ this.ready_queue.push_back(output);
+ }
+ Poll::Ready(None) => break,
+ Poll::Pending => break,
+ }
+ }
+
+ if let Some((output, size)) = this.ready_queue.pop_front() {
+ // If we have any ready outputs, return the first one.
+ *this.current_size -= size;
+ Poll::Ready(Some(output))
+ } else if this.stream.is_done() && this.in_progress_queue.is_empty() {
+ Poll::Ready(None)
+ } else {
+ Poll::Pending
+ }
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ let queue_len = self.in_progress_queue.len() + self.ready_queue.len();
+ let (lower, upper) = self.stream.size_hint();
+ let lower = lower.saturating_add(queue_len);
+ let upper = match upper {
+ Some(x) => x.checked_add(queue_len),
+ None => None,
+ };
+ (lower, upper)
+ }
+}
+
+pin_project! {
+ struct SizedFuture<F> {
+ #[pin]
+ future: F,
+ size: usize,
+ }
+}
+
+impl<F: Future> Future for SizedFuture<F> {
+ type Output = (F::Output, usize);
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let this = self.project();
+ match this.future.poll(cx) {
+ Poll::Ready(output) => Poll::Ready((output, *this.size)),
+ Poll::Pending => Poll::Pending,
+ }
+ }
+}
diff --git a/core/src/patches/mod.rs b/core/src/patches/mod.rs
new file mode 100644
index 000000000..86f0420ca
--- /dev/null
+++ b/core/src/patches/mod.rs
@@ -0,0 +1,19 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+pub(crate) mod stream;
+pub(crate) mod buffer_by_ordered;
diff --git a/core/src/patches/stream.rs b/core/src/patches/stream.rs
new file mode 100644
index 000000000..55c0c67c4
--- /dev/null
+++ b/core/src/patches/stream.rs
@@ -0,0 +1,34 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::patches::buffer_by_ordered::BufferByOrdered;
+
+use futures::Stream;
+use std::future::Future;
+
+pub trait StreamExt: Stream {
+ fn buffer_by_ordered<F>(self, max_size: usize) -> BufferByOrdered<Self, F>
+ where
+ Self: Sized,
+ Self: Stream<Item = (F, usize)>,
+ F: Future,
+ {
+ BufferByOrdered::new(self, max_size)
+ }
+}
+
+impl<T: ?Sized> StreamExt for T where T: Stream {}
diff --git a/core/src/types/read/reader.rs b/core/src/types/read/reader.rs
index 34e8cc67f..c827e9fd3 100644
--- a/core/src/types/read/reader.rs
+++ b/core/src/types/read/reader.rs
@@ -21,9 +21,9 @@ use std::sync::Arc;
use bytes::BufMut;
use futures::stream;
-use futures::StreamExt;
use futures::TryStreamExt;
+use crate::patches::stream::StreamExt;
use crate::*;
/// Reader is designed to read data from given path in an asynchronous
@@ -147,8 +147,8 @@ impl Reader {
let merged_ranges = self.merge_ranges(ranges.clone());
let merged_bufs: Vec<_> =
- stream::iter(merged_ranges.clone().into_iter().map(|v| self.read(v)))
- .buffered(self.ctx.options().concurrent())
+ stream::iter(merged_ranges.clone().into_iter().map(|v| (self.read(v), 1)))
+ .buffer_by_ordered(self.ctx.options().concurrent())
.try_collect()
.await?;