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 2024/04/24 12:47:33 UTC
(opendal) branch write-chunk created (now 8ccc09cb39)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch write-chunk
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 8ccc09cb39 chore: Align with chunk instead of confusing buffer
This branch includes the following new commits:
new 8ccc09cb39 chore: Align with chunk instead of confusing buffer
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: chore: Align with chunk instead of confusing buffer
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch write-chunk
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 8ccc09cb3963b1435593a283a525682ae36e1362
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed Apr 24 20:46:42 2024 +0800
chore: Align with chunk instead of confusing buffer
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/benches/oio/write.rs | 4 +--
core/fuzz/fuzz_writer.rs | 4 +--
core/src/layers/complete.rs | 17 +++++-------
.../write/{exact_buf_write.rs => chunked_write.rs} | 30 ++++++++++------------
core/src/raw/oio/write/mod.rs | 4 +--
core/src/raw/ops.rs | 23 +++++++++--------
core/src/types/operator/operator.rs | 22 ++++++++--------
core/src/types/operator/operator_functions.rs | 4 +--
core/src/types/operator/operator_futures.rs | 20 +++++++--------
core/tests/behavior/async_write.rs | 8 +++---
10 files changed, 66 insertions(+), 70 deletions(-)
diff --git a/core/benches/oio/write.rs b/core/benches/oio/write.rs
index 633d604727..13f31560d7 100644
--- a/core/benches/oio/write.rs
+++ b/core/benches/oio/write.rs
@@ -18,7 +18,7 @@
use bytes::Buf;
use criterion::Criterion;
use once_cell::sync::Lazy;
-use opendal::raw::oio::ExactBufWriter;
+use opendal::raw::oio::ChunkedWriter;
use opendal::raw::oio::Write;
use rand::thread_rng;
use size::Size;
@@ -44,7 +44,7 @@ pub fn bench_exact_buf_write(c: &mut Criterion) {
group.throughput(criterion::Throughput::Bytes(size.bytes() as u64));
group.bench_with_input(size.to_string(), &content, |b, content| {
b.to_async(&*TOKIO).iter(|| async {
- let mut w = ExactBufWriter::new(BlackHoleWriter, 256 * 1024);
+ let mut w = ChunkedWriter::new(BlackHoleWriter, 256 * 1024);
let mut bs = content.clone();
while !bs.is_empty() {
diff --git a/core/fuzz/fuzz_writer.rs b/core/fuzz/fuzz_writer.rs
index 6528240734..cfd2c50bb0 100644
--- a/core/fuzz/fuzz_writer.rs
+++ b/core/fuzz/fuzz_writer.rs
@@ -81,9 +81,9 @@ async fn fuzz_writer(op: Operator, input: FuzzInput) -> Result<()> {
let mut writer = op.writer_with(&path);
if let Some(buffer) = input.buffer {
- writer = writer.buffer(buffer);
+ writer = writer.chunk(buffer);
} else if let Some(min_size) = op.info().full_capability().write_multi_min_size {
- writer = writer.buffer(min_size);
+ writer = writer.chunk(min_size);
}
if let Some(concurrent) = input.concurrent {
writer = writer.concurrent(concurrent);
diff --git a/core/src/layers/complete.rs b/core/src/layers/complete.rs
index 40814d2dd2..c5d38ffe0a 100644
--- a/core/src/layers/complete.rs
+++ b/core/src/layers/complete.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::cmp;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::sync::Arc;
@@ -376,8 +375,7 @@ impl<A: Accessor> LayeredAccessor for CompleteAccessor<A> {
type Inner = A;
type Reader = CompleteReader<A::Reader>;
type BlockingReader = CompleteReader<A::BlockingReader>;
- type Writer =
- TwoWays<CompleteWriter<A::Writer>, oio::ExactBufWriter<CompleteWriter<A::Writer>>>;
+ type Writer = TwoWays<CompleteWriter<A::Writer>, oio::ChunkedWriter<CompleteWriter<A::Writer>>>;
type BlockingWriter = CompleteWriter<A::BlockingWriter>;
type Lister = CompleteLister<A, A::Lister>;
type BlockingLister = CompleteLister<A, A::BlockingLister>;
@@ -426,16 +424,16 @@ impl<A: Accessor> LayeredAccessor for CompleteAccessor<A> {
}
// Calculate buffer size.
- let buffer_size = args.buffer().map(|mut size| {
+ let chunk_size = args.chunk().map(|mut size| {
if let Some(v) = capability.write_multi_max_size {
- size = cmp::min(v, size);
+ size = size.min(v);
}
if let Some(v) = capability.write_multi_min_size {
- size = cmp::max(v, size);
+ size = size.max(v);
}
if let Some(v) = capability.write_multi_align_size {
// Make sure size >= size first.
- size = cmp::max(v, size);
+ size = size.max(v);
size -= size % v;
}
@@ -445,9 +443,9 @@ impl<A: Accessor> LayeredAccessor for CompleteAccessor<A> {
let (rp, w) = self.inner.write(path, args.clone()).await?;
let w = CompleteWriter::new(w);
- let w = match buffer_size {
+ let w = match chunk_size {
None => TwoWays::One(w),
- Some(size) => TwoWays::Two(oio::ExactBufWriter::new(w, size)),
+ Some(size) => TwoWays::Two(oio::ChunkedWriter::new(w, size)),
};
Ok((rp, w))
@@ -628,7 +626,6 @@ impl<W> CompleteWriter<W> {
impl<W> Drop for CompleteWriter<W> {
fn drop(&mut self) {
if self.inner.is_some() {
- // Do we need to panic here?
log::warn!("writer has not been closed or aborted, must be a bug")
}
}
diff --git a/core/src/raw/oio/write/exact_buf_write.rs b/core/src/raw/oio/write/chunked_write.rs
similarity index 82%
rename from core/src/raw/oio/write/exact_buf_write.rs
rename to core/src/raw/oio/write/chunked_write.rs
index a5650b58a3..8eadaa3dd2 100644
--- a/core/src/raw/oio/write/exact_buf_write.rs
+++ b/core/src/raw/oio/write/chunked_write.rs
@@ -18,40 +18,38 @@
use crate::raw::*;
use crate::*;
-/// ExactBufWriter is used to implement [`oio::Write`] based on exact buffer strategy: flush the
-/// underlying storage when the buffered size is exactly the same as the buffer size.
+/// ChunkedWriter is used to implement [`oio::Write`] based on chunk:
+/// flush the underlying storage at the `chunk`` size.
///
-/// ExactBufWriter makes sure that the size of the data written to the underlying storage is exactly
-/// `buffer_size` bytes. It's useful when the underlying storage requires the size to be written.
-///
-/// For example, R2 requires all parts must be the same size except the last part.
-pub struct ExactBufWriter<W: oio::Write> {
+/// ChunkedWriter makes sure that the size of the data written to the
+/// underlying storage is exactly `chunk` bytes.
+pub struct ChunkedWriter<W: oio::Write> {
inner: W,
/// The size for buffer, we will flush the underlying storage at the size of this buffer.
- buffer_size: usize,
+ chunk_size: usize,
buffer: oio::QueueBuf,
}
-impl<W: oio::Write> ExactBufWriter<W> {
+impl<W: oio::Write> ChunkedWriter<W> {
/// Create a new exact buf writer.
- pub fn new(inner: W, buffer_size: usize) -> Self {
+ pub fn new(inner: W, chunk_size: usize) -> Self {
Self {
inner,
- buffer_size,
+ chunk_size,
buffer: oio::QueueBuf::new(),
}
}
}
-impl<W: oio::Write> oio::Write for ExactBufWriter<W> {
+impl<W: oio::Write> oio::Write for ChunkedWriter<W> {
async fn write(&mut self, mut bs: Buffer) -> Result<usize> {
- if self.buffer.len() >= self.buffer_size {
+ if self.buffer.len() >= self.chunk_size {
let written = self.inner.write(self.buffer.clone().collect()).await?;
self.buffer.advance(written);
}
- let remaining = self.buffer_size - self.buffer.len();
+ let remaining = self.chunk_size - self.buffer.len();
bs.truncate(remaining);
let n = bs.len();
self.buffer.push(bs);
@@ -126,7 +124,7 @@ mod tests {
let mut expected = vec![0; 5];
rng.fill_bytes(&mut expected);
- let mut w = ExactBufWriter::new(MockWriter { buf: vec![] }, 10);
+ let mut w = ChunkedWriter::new(MockWriter { buf: vec![] }, 10);
let mut bs = Bytes::from(expected.clone());
while !bs.is_empty() {
@@ -156,7 +154,7 @@ mod tests {
let mut expected = vec![];
let buffer_size = rng.gen_range(1..10);
- let mut writer = ExactBufWriter::new(MockWriter { buf: vec![] }, buffer_size);
+ let mut writer = ChunkedWriter::new(MockWriter { buf: vec![] }, buffer_size);
debug!("test_fuzz_exact_buf_writer: buffer size: {buffer_size}");
for _ in 0..1000 {
diff --git a/core/src/raw/oio/write/mod.rs b/core/src/raw/oio/write/mod.rs
index 66ffa2f4c0..d8e17434bf 100644
--- a/core/src/raw/oio/write/mod.rs
+++ b/core/src/raw/oio/write/mod.rs
@@ -35,8 +35,8 @@ mod one_shot_write;
pub use one_shot_write::OneShotWrite;
pub use one_shot_write::OneShotWriter;
-mod exact_buf_write;
-pub use exact_buf_write::ExactBufWriter;
+mod chunked_write;
+pub use chunked_write::ChunkedWriter;
mod range_write;
pub use range_write::RangeWrite;
diff --git a/core/src/raw/ops.rs b/core/src/raw/ops.rs
index 1c2e75c44c..f21c2769fb 100644
--- a/core/src/raw/ops.rs
+++ b/core/src/raw/ops.rs
@@ -548,7 +548,7 @@ impl OpStat {
#[derive(Debug, Clone, Default)]
pub struct OpWrite {
append: bool,
- buffer: Option<usize>,
+ chunk: Option<usize>,
concurrent: usize,
content_type: Option<String>,
@@ -583,23 +583,24 @@ impl OpWrite {
self
}
- /// Get the buffer from op.
+ /// Get the chunk from op.
///
- /// The buffer is used by service to decide the buffer size of the underlying writer.
- pub fn buffer(&self) -> Option<usize> {
- self.buffer
+ /// The chunk is used by service to decide the chunk size of the underlying writer.
+ pub fn chunk(&self) -> Option<usize> {
+ self.chunk
}
- /// Set the buffer of op.
+ /// Set the chunk of op.
///
- /// If buffer is set, the data will be buffered by the underlying writer.
+ /// If chunk is set, the data will be chunked by the underlying writer.
///
/// ## NOTE
///
- /// Service could have their own minimum buffer size while perform write operations like
- /// multipart uploads. So the buffer size may be larger than the given buffer size.
- pub fn with_buffer(mut self, buffer: usize) -> Self {
- self.buffer = Some(buffer);
+ /// Service could have their own minimum chunk size while perform write
+ /// operations like multipart uploads. So the chunk size may be larger than
+ /// the given buffer size.
+ pub fn with_chunk(mut self, chunk: usize) -> Self {
+ self.chunk = Some(chunk);
self
}
diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs
index e9ca1bdfba..6804ea3cf4 100644
--- a/core/src/types/operator/operator.rs
+++ b/core/src/types/operator/operator.rs
@@ -680,7 +680,7 @@ impl Operator {
///
/// OpenDAL abstracts the multipart uploads into [`Writer`]. It will automatically
/// handle the multipart uploads for you. You can control the behavior of multipart uploads
- /// by setting `buffer`, `concurrent` via [`Operator::writer_with`]
+ /// by setting `chunk`, `concurrent` via [`Operator::writer_with`]
///
/// # Examples
///
@@ -826,16 +826,16 @@ impl Operator {
/// extra options like `content_type` and `cache_control`, please use [`Operator::write_with`]
/// instead.
///
- /// ## Buffer
+ /// ## Chunk
///
- /// OpenDAL is designed to write files directly without buffering by default, giving users
+ /// OpenDAL is designed to write files directly without chunking by default, giving users
/// control over the exact size of their writes and helping avoid unnecessary costs.
///
/// This is not efficient for cases when users write small chunks of data. Some storage services
/// like `s3` could even return hard errors like `EntityTooSmall`. Besides, cloud storage services
/// will cost more money if we write data in small chunks.
///
- /// Users can use [`Operator::write_with`] to set a good buffer size might improve the performance,
+ /// Users can use [`Operator::write_with`] to set a good chunk size might improve the performance,
///
/// # Examples
///
@@ -885,20 +885,20 @@ impl Operator {
/// # }
/// ```
///
- /// ## `buffer`
+ /// ## `chunk`
///
- /// Set `buffer` for the writer.
+ /// Set `chunk` for the writer.
///
- /// OpenDAL is designed to write files directly without buffering by default, giving users
+ /// OpenDAL is designed to write files directly without chunking by default, giving users
/// control over the exact size of their writes and helping avoid unnecessary costs.
///
/// This is not efficient for cases when users write small chunks of data. Some storage services
/// like `s3` could even return hard errors like `EntityTooSmall`. Besides, cloud storage services
/// will cost more money if we write data in small chunks.
///
- /// Set a good buffer size might improve the performance, reduce the API calls and save money.
+ /// Set a good chunk size might improve the performance, reduce the API calls and save money.
///
- /// The following example will set the writer buffer to 8MiB. Only one API call will be sent at
+ /// The following example will set the writer chunk to 8MiB. Only one API call will be sent at
/// `close` instead.
///
/// ```no_run
@@ -911,7 +911,7 @@ impl Operator {
/// # async fn test(op: Operator) -> Result<()> {
/// let mut w = op
/// .writer_with("path/to/file")
- /// .buffer(8 * 1024 * 1024)
+ /// .chunk(8 * 1024 * 1024)
/// .await?;
/// w.write(vec![0; 4096]).await?;
/// w.write(vec![1; 4096]).await?;
@@ -1078,7 +1078,7 @@ impl Operator {
///
/// OpenDAL abstracts the multipart uploads into [`Writer`]. It will automatically
/// handle the multipart uploads for you. You can control the behavior of multipart uploads
- /// by setting `buffer`, `concurrent` via [`Operator::writer_with`]
+ /// by setting `chunk`, `concurrent` via [`Operator::writer_with`]
///
/// # Options
///
diff --git a/core/src/types/operator/operator_functions.rs b/core/src/types/operator/operator_functions.rs
index 0a7711544a..e59382ebdb 100644
--- a/core/src/types/operator/operator_functions.rs
+++ b/core/src/types/operator/operator_functions.rs
@@ -97,7 +97,7 @@ impl FunctionWrite {
/// Service could have their own minimum buffer size while perform write operations like
/// multipart uploads. So the buffer size may be larger than the given buffer size.
pub fn buffer(mut self, v: usize) -> Self {
- self.0 = self.0.map_args(|(args, bs)| (args.with_buffer(v), bs));
+ self.0 = self.0.map_args(|(args, bs)| (args.with_chunk(v), bs));
self
}
@@ -163,7 +163,7 @@ impl FunctionWriter {
/// Service could have their own minimum buffer size while perform write operations like
/// multipart uploads. So the buffer size may be larger than the given buffer size.
pub fn buffer(mut self, v: usize) -> Self {
- self.0 = self.0.map_args(|args| args.with_buffer(v));
+ self.0 = self.0.map_args(|args| args.with_chunk(v));
self
}
diff --git a/core/src/types/operator/operator_futures.rs b/core/src/types/operator/operator_futures.rs
index 1cdef975b9..7599c94021 100644
--- a/core/src/types/operator/operator_futures.rs
+++ b/core/src/types/operator/operator_futures.rs
@@ -289,13 +289,13 @@ impl<F: Future<Output = Result<()>>> FutureWrite<F> {
///
/// Service could have their own minimum buffer size while perform write operations like
/// multipart uploads. So the buffer size may be larger than the given buffer size.
- pub fn buffer(self, v: usize) -> Self {
- self.map(|(args, bs)| (args.with_buffer(v), bs))
+ pub fn chunk(self, v: usize) -> Self {
+ self.map(|(args, bs)| (args.with_chunk(v), bs))
}
/// Set the maximum concurrent write task amount.
pub fn concurrent(self, v: usize) -> Self {
- self.map(|(args, bs)| (args.with_buffer(v), bs))
+ self.map(|(args, bs)| (args.with_chunk(v), bs))
}
/// Set the content type of option
@@ -331,23 +331,23 @@ impl<F: Future<Output = Result<Writer>>> FutureWriter<F> {
self.map(|args| args.with_append(v))
}
- /// Set the buffer size of op.
+ /// Set the chunk size of op.
///
- /// If buffer size is set, the data will be buffered by the underlying writer.
+ /// If chunk size is set, the data will be chunked by the underlying writer.
///
/// ## NOTE
///
- /// Service could have their own limitation for buffer size. It's possible that buffer size
- /// is not equal to the given buffer size.
+ /// Service could have their own limitation for chunk size. It's possible that chunk size
+ /// is not equal to the given chunk size.
///
/// For example:
///
/// - AWS S3 requires the part size to be in [5MiB, 5GiB].
/// - GCS requires the part size to be aligned with 256 KiB.
///
- /// The services will alter the buffer size to meet their requirements.
- pub fn buffer(self, v: usize) -> Self {
- self.map(|args| args.with_buffer(v))
+ /// The services will alter the chunk size to meet their requirements.
+ pub fn chunk(self, v: usize) -> Self {
+ self.map(|args| args.with_chunk(v))
}
/// Set the maximum concurrent write task amount.
diff --git a/core/tests/behavior/async_write.rs b/core/tests/behavior/async_write.rs
index bb145ab2d5..50462fb356 100644
--- a/core/tests/behavior/async_write.rs
+++ b/core/tests/behavior/async_write.rs
@@ -339,7 +339,7 @@ pub async fn test_writer_sink(op: Operator) -> Result<()> {
let content_b = gen_fixed_bytes(size);
let stream = stream::iter(vec![content_a.clone(), content_b.clone()]).map(Ok);
- let mut w = op.writer_with(&path).buffer(5 * 1024 * 1024).await?;
+ let mut w = op.writer_with(&path).chunk(5 * 1024 * 1024).await?;
w.sink(stream).await?;
w.close().await?;
@@ -377,7 +377,7 @@ pub async fn test_writer_sink_with_concurrent(op: Operator) -> Result<()> {
let mut w = op
.writer_with(&path)
- .buffer(5 * 1024 * 1024)
+ .chunk(5 * 1024 * 1024)
.concurrent(4)
.await?;
w.sink(stream).await?;
@@ -414,7 +414,7 @@ pub async fn test_writer_futures_copy(op: Operator) -> Result<()> {
let mut w = op
.writer_with(&path)
- .buffer(8 * 1024 * 1024)
+ .chunk(8 * 1024 * 1024)
.await?
.into_futures_async_write();
@@ -449,7 +449,7 @@ pub async fn test_writer_futures_copy_with_concurrent(op: Operator) -> Result<()
let mut w = op
.writer_with(&path)
- .buffer(8 * 1024 * 1024)
+ .chunk(8 * 1024 * 1024)
.concurrent(4)
.await?
.into_futures_async_write();