[incubator-opendal] 01/03: Implement buffer cal

[xu...@apache.org]

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

xuanwo pushed a commit to branch apply-buffer
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git

commit f07dfbca783717557cfd4d555b99b938b6c09e4c
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed Sep 13 10:46:54 2023 +0800

    Implement buffer cal
    
    Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
 core/src/layers/complete.rs  | 48 ++++++++++++++++++++++++++++----------------
 core/src/types/capability.rs | 12 +++++++++++
 2 files changed, 43 insertions(+), 17 deletions(-)

diff --git a/core/src/layers/complete.rs b/core/src/layers/complete.rs
index 1abed7bbd..84b587efa 100644
--- a/core/src/layers/complete.rs
+++ b/core/src/layers/complete.rs
@@ -17,11 +17,11 @@
 
 use std::fmt::Debug;
 use std::fmt::Formatter;
-use std::io;
 use std::sync::Arc;
 use std::task::ready;
 use std::task::Context;
 use std::task::Poll;
+use std::{cmp, io};
 
 use async_trait::async_trait;
 use bytes::Bytes;
@@ -95,17 +95,6 @@ use crate::*;
 /// If there is a hint that `ReadStreamable`, we will use existing reader
 /// directly. Otherwise, we will use transform this reader as a stream.
 ///
-/// ### Consume instead of Drop
-///
-/// Normally, if reader is seekable, we need to drop current reader and start
-/// a new read call.
-///
-/// We can consume the data if the seek position is close enough. For
-/// example, users try to seek to `Current(1)`, we can just read the data
-/// can consume it.
-///
-/// In this way, we can reduce the extra cost of dropping reader.
-///
 /// ## List Completion
 ///
 /// There are two styles of list, but not all services support both of
@@ -366,7 +355,10 @@ impl<A: Accessor> LayeredAccessor for CompleteReaderAccessor<A> {
     type Inner = A;
     type Reader = CompleteReader<A, A::Reader>;
     type BlockingReader = CompleteReader<A, A::BlockingReader>;
-    type Writer = CompleteWriter<A::Writer>;
+    type Writer = oio::TwoWaysWriter<
+        CompleteWriter<A::Writer>,
+        oio::ExactBufWriter<CompleteWriter<A::Writer>>,
+    >;
     type BlockingWriter = CompleteWriter<A::BlockingWriter>;
     type Pager = CompletePager<A, A::Pager>;
     type BlockingPager = CompletePager<A, A::BlockingPager>;
@@ -433,10 +425,32 @@ impl<A: Accessor> LayeredAccessor for CompleteReaderAccessor<A> {
             ));
         }
 
-        self.inner
-            .write(path, args)
-            .await
-            .map(|(rp, w)| (rp, CompleteWriter::new(w)))
+        // Calculate buffer size.
+        let buffer_size = args.buffer().map(|mut size| {
+            if let Some(v) = capability.write_multi_max_size {
+                size = cmp::min(v, size);
+            }
+            if let Some(v) = capability.write_multi_min_size {
+                size = cmp::max(v, size);
+            }
+            if let Some(v) = capability.write_multi_align_size {
+                // Make sure size >= size first.
+                size = cmp::max(v, size);
+                size -= size % v;
+            }
+
+            size
+        });
+
+        let (rp, w) = self.inner.write(path, args.clone()).await?;
+        let w = CompleteWriter::new(w);
+
+        let w = match buffer_size {
+            None => oio::TwoWaysWriter::One(w),
+            Some(size) => oio::TwoWaysWriter::Two(oio::ExactBufWriter::new(w, size)),
+        };
+
+        Ok((rp, w))
     }
 
     fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
diff --git a/core/src/types/capability.rs b/core/src/types/capability.rs
index b568b3ee8..abe7eb9c4 100644
--- a/core/src/types/capability.rs
+++ b/core/src/types/capability.rs
@@ -84,6 +84,18 @@ pub struct Capability {
     pub write_with_content_disposition: bool,
     /// If operator supports write with cache control.
     pub write_with_cache_control: bool,
+    /// write_multi_max_size is the max size that services support in write_multi.
+    ///
+    /// For example, AWS S3 supports 5GiB as max in write_multi.
+    pub write_multi_max_size: Option<usize>,
+    /// write_multi_min_size is the min size that services support in write_multi.
+    ///
+    /// For example, AWS S3 requires at least 5MiB in write_multi expect the last one.
+    pub write_multi_min_size: Option<usize>,
+    /// write_multi_align_size is the align size that services required in write_multi.
+    ///
+    /// For example, Google GCS requires align size to 256KiB in write_multi.
+    pub write_multi_align_size: Option<usize>,
 
     /// If operator supports create dir.
     pub create_dir: bool,