(opendal) branch main updated: feat(core): Tune buffer operations based on benchmark results (#4468)

[xuanwo]

This is an automated email from the ASF dual-hosted git repository.

xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git


The following commit(s) were added to refs/heads/main by this push:
     new 62fe9b0841 feat(core): Tune buffer operations  based on benchmark 
results (#4468)
62fe9b0841 is described below

commit 62fe9b08413b99895d474764acec8f7dd2dc9937
Author: Xuanwo <git...@xuanwo.io>
AuthorDate: Fri Apr 12 20:36:51 2024 +0800

    feat(core): Tune buffer operations  based on benchmark results (#4468)
    
    * Debug
    
    Signed-off-by: Xuanwo <git...@xuanwo.io>
    
    * Add inline for buffer operations
    
    Signed-off-by: Xuanwo <git...@xuanwo.io>
    
    * Try
    
    Signed-off-by: Xuanwo <git...@xuanwo.io>
    
    * Drop buffer before start new request
    
    Signed-off-by: Xuanwo <git...@xuanwo.io>
    
    * Add cap to 8MiB
    
    Signed-off-by: Xuanwo <git...@xuanwo.io>
    
    ---------
    
    Signed-off-by: Xuanwo <git...@xuanwo.io>
---
 core/src/types/buffer.rs | 11 +++++++++++
 core/src/types/reader.rs |  8 +++++++-
 2 files changed, 18 insertions(+), 1 deletion(-)

diff --git a/core/src/types/buffer.rs b/core/src/types/buffer.rs
index dc9cc6ec2b..33aab95b55 100644
--- a/core/src/types/buffer.rs
+++ b/core/src/types/buffer.rs
@@ -226,42 +226,49 @@ impl Buffer {
 }
 
 impl From<Vec<u8>> for Buffer {
+    #[inline]
     fn from(bs: Vec<u8>) -> Self {
         Self(Inner::Contiguous(bs.into()))
     }
 }
 
 impl From<Bytes> for Buffer {
+    #[inline]
     fn from(bs: Bytes) -> Self {
         Self(Inner::Contiguous(bs))
     }
 }
 
 impl From<String> for Buffer {
+    #[inline]
     fn from(s: String) -> Self {
         Self(Inner::Contiguous(Bytes::from(s)))
     }
 }
 
 impl From<&'static [u8]> for Buffer {
+    #[inline]
     fn from(s: &'static [u8]) -> Self {
         Self(Inner::Contiguous(Bytes::from_static(s)))
     }
 }
 
 impl From<&'static str> for Buffer {
+    #[inline]
     fn from(s: &'static str) -> Self {
         Self(Inner::Contiguous(Bytes::from_static(s.as_bytes())))
     }
 }
 
 impl FromIterator<u8> for Buffer {
+    #[inline]
     fn from_iter<T: IntoIterator<Item = u8>>(iter: T) -> Self {
         Self(Inner::Contiguous(Bytes::from_iter(iter)))
     }
 }
 
 impl From<VecDeque<Bytes>> for Buffer {
+    #[inline]
     fn from(bs: VecDeque<Bytes>) -> Self {
         let size = bs.iter().map(Bytes::len).sum();
         Self(Inner::NonContiguous {
@@ -274,6 +281,7 @@ impl From<VecDeque<Bytes>> for Buffer {
 }
 
 impl From<Vec<Bytes>> for Buffer {
+    #[inline]
     fn from(bs: Vec<Bytes>) -> Self {
         let size = bs.iter().map(Bytes::len).sum();
         Self(Inner::NonContiguous {
@@ -286,6 +294,7 @@ impl From<Vec<Bytes>> for Buffer {
 }
 
 impl From<Arc<[Bytes]>> for Buffer {
+    #[inline]
     fn from(bs: Arc<[Bytes]>) -> Self {
         let size = bs.iter().map(Bytes::len).sum();
         Self(Inner::NonContiguous {
@@ -298,6 +307,7 @@ impl From<Arc<[Bytes]>> for Buffer {
 }
 
 impl FromIterator<Bytes> for Buffer {
+    #[inline]
     fn from_iter<T: IntoIterator<Item = Bytes>>(iter: T) -> Self {
         let mut size = 0;
         let bs = iter.into_iter().inspect(|v| size += v.len());
@@ -340,6 +350,7 @@ impl Buf for Buffer {
         }
     }
 
+    #[inline]
     fn chunks_vectored<'a>(&'a self, dst: &mut [IoSlice<'a>]) -> usize {
         match &self.0 {
             Inner::Contiguous(b) => {
diff --git a/core/src/types/reader.rs b/core/src/types/reader.rs
index 7dc8944ef7..4031ac79a5 100644
--- a/core/src/types/reader.rs
+++ b/core/src/types/reader.rs
@@ -248,7 +248,7 @@ pub mod into_futures_async_read {
                 offset: range.start,
                 size: range.end - range.start,
                 // TODO: should use services preferred io size.
-                cap: 4 * 1024 * 1024,
+                cap: 8 * 1024 * 1024,
 
                 cur: 0,
                 buf: Buffer::new(),
@@ -297,6 +297,12 @@ pub mod into_futures_async_read {
 
         fn consume(mut self: Pin<&mut Self>, amt: usize) {
             self.buf.advance(amt);
+            // Make sure buf has been dropped before starting new request.
+            // Otherwise, we will hold those bytes in memory until next
+            // buffer reaching.
+            if self.buf.is_empty() {
+                self.buf = Buffer::new();
+            }
             self.cur += amt as u64;
         }
     }