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/03/14 13:22:32 UTC
(opendal) branch cleanup created (now b7322fce80)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch cleanup
in repository https://gitbox.apache.org/repos/asf/opendal.git
at b7322fce80 Remove write buf
This branch includes the following new commits:
new dcc298db0d clean block
new 48eb515fec Refactor exact buf write
new e69809d988 refactor multipart write
new 87b9095a46 Refactor range write
new 8c0e91e189 Cleanup async body
new ba864e8542 remove chunked bytes
new b7322fce80 Remove write buf
The 7 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) 03/07: refactor multipart write
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch cleanup
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit e69809d98808cbf8d0686401fc841566c6d98241
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Mar 14 21:15:58 2024 +0800
refactor multipart write
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/oio/write/multipart_write.rs | 11 +++++------
1 file changed, 5 insertions(+), 6 deletions(-)
diff --git a/core/src/raw/oio/write/multipart_write.rs b/core/src/raw/oio/write/multipart_write.rs
index 10b260210a..2f43393290 100644
--- a/core/src/raw/oio/write/multipart_write.rs
+++ b/core/src/raw/oio/write/multipart_write.rs
@@ -146,7 +146,7 @@ pub struct MultipartPart {
/// WritePartResult is the result returned by [`WritePartFuture`].
///
/// The error part will carries input `(part_number, bytes, err)` so caller can retry them.
-type WritePartResult = std::result::Result<MultipartPart, (usize, oio::ChunkedBytes, Error)>;
+type WritePartResult = std::result::Result<MultipartPart, (usize, Bytes, Error)>;
struct WritePartFuture(BoxedStaticFuture<WritePartResult>);
@@ -172,14 +172,14 @@ impl WritePartFuture {
w: Arc<W>,
upload_id: Arc<String>,
part_number: usize,
- bytes: oio::ChunkedBytes,
+ bytes: Bytes,
) -> Self {
let fut = async move {
w.write_part(
&upload_id,
part_number,
bytes.len() as u64,
- AsyncBody::ChunkedBytes(bytes.clone()),
+ AsyncBody::Bytes(bytes.clone()),
)
.await
.map_err(|err| (part_number, bytes, err))
@@ -196,7 +196,7 @@ pub struct MultipartWriter<W: MultipartWrite> {
upload_id: Option<Arc<String>>,
parts: Vec<MultipartPart>,
- cache: Option<oio::ChunkedBytes>,
+ cache: Option<Bytes>,
futures: ConcurrentFutures<WritePartFuture>,
next_part_number: usize,
}
@@ -220,7 +220,6 @@ impl<W: MultipartWrite> MultipartWriter<W> {
fn fill_cache(&mut self, bs: Bytes) -> usize {
let size = bs.len();
- let bs = oio::ChunkedBytes::from_vec(vec![bs]);
assert!(self.cache.is_none());
self.cache = Some(bs);
size
@@ -289,7 +288,7 @@ where
Some(v) => v,
None => {
let (size, body) = match self.cache.clone() {
- Some(cache) => (cache.len(), AsyncBody::ChunkedBytes(cache)),
+ Some(cache) => (cache.len(), AsyncBody::Bytes(cache)),
None => (0, AsyncBody::Empty),
};
// Call write_once if there is no upload_id.
(opendal) 05/07: Cleanup async body
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch cleanup
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 8c0e91e189723743b36cace50a65eedd1fe2d7ec
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Mar 14 21:18:35 2024 +0800
Cleanup async body
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/http_util/body.rs | 4 ----
core/src/raw/http_util/client.rs | 11 -----------
core/src/raw/http_util/multipart.rs | 1 -
core/src/raw/oio/write/block_write.rs | 2 --
core/src/services/gcs/core.rs | 3 ---
5 files changed, 21 deletions(-)
diff --git a/core/src/raw/http_util/body.rs b/core/src/raw/http_util/body.rs
index 4c39dfa74e..e32dcb46d4 100644
--- a/core/src/raw/http_util/body.rs
+++ b/core/src/raw/http_util/body.rs
@@ -35,10 +35,6 @@ pub enum AsyncBody {
Empty,
/// Body with bytes.
Bytes(Bytes),
- /// Body with chunked bytes.
- ///
- /// This is nearly the same with stream, but we can save an extra box.
- ChunkedBytes(oio::ChunkedBytes),
/// Body with stream.
Stream(oio::Streamer),
}
diff --git a/core/src/raw/http_util/client.rs b/core/src/raw/http_util/client.rs
index 2c5e4bc5c7..e8108e3a6a 100644
--- a/core/src/raw/http_util/client.rs
+++ b/core/src/raw/http_util/client.rs
@@ -118,17 +118,6 @@ impl HttpClient {
req_builder = match body {
AsyncBody::Empty => req_builder.body(reqwest::Body::from("")),
AsyncBody::Bytes(bs) => req_builder.body(reqwest::Body::from(bs)),
- AsyncBody::ChunkedBytes(bs) => {
- #[cfg(not(target_arch = "wasm32"))]
- {
- req_builder.body(reqwest::Body::wrap_stream(bs))
- }
- #[cfg(target_arch = "wasm32")]
- {
- let bs = oio::WriteBuf::bytes(&bs, bs.len());
- req_builder.body(reqwest::Body::from(bs))
- }
- }
AsyncBody::Stream(s) => {
#[cfg(not(target_arch = "wasm32"))]
{
diff --git a/core/src/raw/http_util/multipart.rs b/core/src/raw/http_util/multipart.rs
index 81e6db7147..4d40b01a61 100644
--- a/core/src/raw/http_util/multipart.rs
+++ b/core/src/raw/http_util/multipart.rs
@@ -404,7 +404,6 @@ impl MixedPart {
bs.len() as u64,
Some(Box::new(oio::Cursor::from(bs)) as Streamer),
),
- AsyncBody::ChunkedBytes(bs) => (bs.len() as u64, Some(Box::new(bs) as Streamer)),
AsyncBody::Stream(stream) => {
let len = parts
.headers
diff --git a/core/src/raw/oio/write/block_write.rs b/core/src/raw/oio/write/block_write.rs
index 28e6d88ed9..3b0266ffbd 100644
--- a/core/src/raw/oio/write/block_write.rs
+++ b/core/src/raw/oio/write/block_write.rs
@@ -287,7 +287,6 @@ mod tests {
use super::*;
use crate::raw::oio::StreamExt;
use crate::raw::oio::Write;
- use crate::raw::oio::WriteBuf;
struct TestWrite {
length: u64,
@@ -321,7 +320,6 @@ mod tests {
let bs = match body {
AsyncBody::Empty => Bytes::new(),
AsyncBody::Bytes(bs) => bs,
- AsyncBody::ChunkedBytes(cb) => cb.bytes(cb.remaining()),
AsyncBody::Stream(s) => s.collect().await.unwrap(),
};
diff --git a/core/src/services/gcs/core.rs b/core/src/services/gcs/core.rs
index df112e504b..46cc2c4123 100644
--- a/core/src/services/gcs/core.rs
+++ b/core/src/services/gcs/core.rs
@@ -288,9 +288,6 @@ impl GcsCore {
AsyncBody::Bytes(bytes) => {
media_part = media_part.content(bytes);
}
- AsyncBody::ChunkedBytes(bs) => {
- media_part = media_part.stream(bs.len() as u64, Box::new(bs));
- }
AsyncBody::Stream(stream) => {
media_part = media_part.stream(size.unwrap(), stream);
}
(opendal) 07/07: Remove write buf
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch cleanup
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit b7322fce80e5d3d532ee770c3a4dfe7716ad7106
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Mar 14 21:22:02 2024 +0800
Remove write buf
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/oio/buf/mod.rs | 3 -
core/src/raw/oio/buf/write_buf.rs | 295 ---------------------------
core/src/raw/oio/write/exact_buf_write.rs | 8 +-
core/src/types/operator/blocking_operator.rs | 3 +-
4 files changed, 4 insertions(+), 305 deletions(-)
diff --git a/core/src/raw/oio/buf/mod.rs b/core/src/raw/oio/buf/mod.rs
index 5aec02b8ca..9c4881ec83 100644
--- a/core/src/raw/oio/buf/mod.rs
+++ b/core/src/raw/oio/buf/mod.rs
@@ -15,8 +15,5 @@
// specific language governing permissions and limitations
// under the License.
-mod write_buf;
-pub use write_buf::WriteBuf;
-
mod adaptive;
pub use adaptive::AdaptiveBuf;
diff --git a/core/src/raw/oio/buf/write_buf.rs b/core/src/raw/oio/buf/write_buf.rs
deleted file mode 100644
index f905dd71e4..0000000000
--- a/core/src/raw/oio/buf/write_buf.rs
+++ /dev/null
@@ -1,295 +0,0 @@
-// 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::io::IoSlice;
-
-use bytes::Bytes;
-use bytes::BytesMut;
-
-/// WriteBuf is used in [`oio::Write`] to provide in-memory buffer support.
-pub trait WriteBuf: Send + Sync {
- /// Returns the number of bytes between the current position and the end of the buffer.
- ///
- /// This value is greater than or equal to the length of the slice returned by chunk().
- ///
- /// # Notes
- ///
- /// Implementations of remaining should ensure that the return value does not change unless a
- /// call is made to advance or any other function that is documented to change the Buf's
- /// current position.
- fn remaining(&self) -> usize;
-
- /// Advance the internal cursor of the Buf
- ///
- /// The next call to chunk() will return a slice starting cnt bytes further into the underlying buffer.
- ///
- /// # Panics
- ///
- /// This function may panic if cnt > self.remaining().
- fn advance(&mut self, cnt: usize);
-
- /// Returns a slice starting at the current position and of length between 0 and
- /// Buf::remaining(). Note that this can return shorter slice (this allows non-continuous
- /// internal representation).
- ///
- /// # Notes
- ///
- /// This function should never panic. Once the end of the buffer is reached, i.e.,
- /// Buf::remaining returns 0, calls to chunk() should return an empty slice.
- fn chunk(&self) -> &[u8];
-
- /// Returns a vectored view of the underlying buffer at the current position and of
- /// length between 0 and Buf::remaining(). Note that this can return shorter slice
- /// (this allows non-continuous internal representation).
- ///
- /// # Notes
- ///
- /// This function should never panic.
- fn vectored_chunk(&self) -> Vec<IoSlice>;
-
- /// Returns a bytes starting at the current position and of length between 0 and
- /// Buf::remaining().
- ///
- /// # Notes
- ///
- /// This functions is used to concat a single bytes from underlying chunks.
- /// Use `vectored_bytes` if you want to avoid copy when possible.
- ///
- /// # Panics
- ///
- /// This function will panic if size > self.remaining().
- fn bytes(&self, size: usize) -> Bytes;
-
- /// Returns true if the underlying buffer is optimized for bytes with given size.
- ///
- /// # Notes
- ///
- /// This function is used to avoid copy when possible. Implementers should return `true`
- /// the given `self.bytes(size)` could be done without cost. For example, the underlying
- /// buffer is `Bytes`.
- ///
- /// # Panics
- ///
- /// This function will panic if size > self.remaining().
- fn is_bytes_optimized(&self, size: usize) -> bool {
- let _ = size;
- false
- }
-
- /// Returns a vectored bytes of the underlying buffer at the current position and of
- /// length between 0 and Buf::remaining().
- ///
- /// # Notes for Users
- ///
- /// This functions is used to return a vectored bytes from underlying chunks.
- /// Use `bytes` if you just want to get a continuous bytes.
- ///
- /// # Notes for implementers
- ///
- /// It's better to align the vectored bytes with underlying chunks to avoid copy.
- ///
- /// # Panics
- ///
- /// This function will panic if size > self.remaining().
- fn vectored_bytes(&self, size: usize) -> Vec<Bytes>;
-}
-
-macro_rules! deref_forward_buf {
- () => {
- fn remaining(&self) -> usize {
- (**self).remaining()
- }
-
- fn advance(&mut self, cnt: usize) {
- (**self).advance(cnt)
- }
-
- fn chunk(&self) -> &[u8] {
- (**self).chunk()
- }
-
- fn vectored_chunk(&self) -> Vec<IoSlice> {
- (**self).vectored_chunk()
- }
-
- fn bytes(&self, size: usize) -> Bytes {
- (**self).bytes(size)
- }
-
- fn is_bytes_optimized(&self, size: usize) -> bool {
- (**self).is_bytes_optimized(size)
- }
-
- fn vectored_bytes(&self, size: usize) -> Vec<Bytes> {
- (**self).vectored_bytes(size)
- }
- };
-}
-
-impl<T: WriteBuf + ?Sized> WriteBuf for &mut T {
- deref_forward_buf!();
-}
-
-impl<T: WriteBuf + ?Sized> WriteBuf for Box<T> {
- deref_forward_buf!();
-}
-
-impl WriteBuf for &[u8] {
- #[inline]
- fn remaining(&self) -> usize {
- self.len()
- }
-
- #[inline]
- fn advance(&mut self, cnt: usize) {
- *self = &self[cnt..];
- }
-
- #[inline]
- fn chunk(&self) -> &[u8] {
- self
- }
-
- #[inline]
- fn vectored_chunk(&self) -> Vec<IoSlice> {
- vec![IoSlice::new(self)]
- }
-
- #[inline]
- fn bytes(&self, size: usize) -> Bytes {
- Bytes::copy_from_slice(&self[..size])
- }
-
- #[inline]
- fn vectored_bytes(&self, size: usize) -> Vec<Bytes> {
- vec![self.bytes(size)]
- }
-}
-
-impl<T: AsRef<[u8]> + Send + Sync> WriteBuf for std::io::Cursor<T> {
- fn remaining(&self) -> usize {
- let len = self.get_ref().as_ref().len();
- let pos = self.position();
-
- if pos >= len as u64 {
- return 0;
- }
-
- len - pos as usize
- }
-
- fn advance(&mut self, cnt: usize) {
- let pos = (self.position() as usize)
- .checked_add(cnt)
- .expect("overflow");
-
- assert!(pos <= self.get_ref().as_ref().len());
- self.set_position(pos as u64);
- }
-
- fn chunk(&self) -> &[u8] {
- let len = self.get_ref().as_ref().len();
- let pos = self.position();
-
- if pos >= len as u64 {
- return &[];
- }
-
- &self.get_ref().as_ref()[pos as usize..]
- }
-
- fn vectored_chunk(&self) -> Vec<IoSlice> {
- vec![IoSlice::new(self.chunk())]
- }
-
- fn bytes(&self, size: usize) -> Bytes {
- Bytes::copy_from_slice(&self.chunk()[..size])
- }
-
- fn vectored_bytes(&self, size: usize) -> Vec<Bytes> {
- vec![self.bytes(size)]
- }
-}
-
-impl WriteBuf for Bytes {
- #[inline]
- fn remaining(&self) -> usize {
- self.len()
- }
-
- #[inline]
- fn advance(&mut self, cnt: usize) {
- bytes::Buf::advance(self, cnt)
- }
-
- #[inline]
- fn chunk(&self) -> &[u8] {
- self
- }
-
- #[inline]
- fn vectored_chunk(&self) -> Vec<IoSlice> {
- vec![IoSlice::new(self)]
- }
-
- #[inline]
- fn bytes(&self, size: usize) -> Bytes {
- self.slice(..size)
- }
-
- #[inline]
- fn is_bytes_optimized(&self, _: usize) -> bool {
- true
- }
-
- #[inline]
- fn vectored_bytes(&self, size: usize) -> Vec<Bytes> {
- vec![self.slice(..size)]
- }
-}
-
-impl WriteBuf for BytesMut {
- #[inline]
- fn remaining(&self) -> usize {
- self.len()
- }
-
- #[inline]
- fn advance(&mut self, cnt: usize) {
- bytes::Buf::advance(self, cnt)
- }
-
- #[inline]
- fn chunk(&self) -> &[u8] {
- self
- }
-
- #[inline]
- fn vectored_chunk(&self) -> Vec<IoSlice> {
- vec![IoSlice::new(self)]
- }
-
- #[inline]
- fn bytes(&self, size: usize) -> Bytes {
- Bytes::copy_from_slice(&self[..size])
- }
-
- #[inline]
- fn vectored_bytes(&self, size: usize) -> Vec<Bytes> {
- vec![self.bytes(size)]
- }
-}
diff --git a/core/src/raw/oio/write/exact_buf_write.rs b/core/src/raw/oio/write/exact_buf_write.rs
index f1097ab140..0de3138ff2 100644
--- a/core/src/raw/oio/write/exact_buf_write.rs
+++ b/core/src/raw/oio/write/exact_buf_write.rs
@@ -70,11 +70,9 @@ impl<W: oio::Write> oio::Write for ExactBufWriter<W> {
//
// if buffer is empty and bs is larger than buffer_size, we can directly
// freeze the first buffer_size bytes.
- if self.buffer.is_empty() {
- if bs.len() >= self.buffer_size {
- self.frozen = Some(bs.slice(0..self.buffer_size));
- return Ok(self.buffer_size);
- }
+ if self.buffer.is_empty() && bs.len() >= self.buffer_size {
+ self.frozen = Some(bs.slice(0..self.buffer_size));
+ return Ok(self.buffer_size);
}
let remaining = self.buffer_size - self.buffer.len();
diff --git a/core/src/types/operator/blocking_operator.rs b/core/src/types/operator/blocking_operator.rs
index eff9afb614..c9b2ee7586 100644
--- a/core/src/types/operator/blocking_operator.rs
+++ b/core/src/types/operator/blocking_operator.rs
@@ -15,10 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use bytes::Bytes;
+use bytes::{Buf, Bytes};
use super::operator_functions::*;
-use crate::raw::oio::WriteBuf;
use crate::raw::*;
use crate::*;
(opendal) 04/07: Refactor range write
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch cleanup
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 87b9095a46a6b4fb4a00bfb04609af5e511e0fef
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Mar 14 21:16:57 2024 +0800
Refactor range write
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/oio/write/range_write.rs | 19 ++++++-------------
1 file changed, 6 insertions(+), 13 deletions(-)
diff --git a/core/src/raw/oio/write/range_write.rs b/core/src/raw/oio/write/range_write.rs
index f766943fbb..a8de1366fe 100644
--- a/core/src/raw/oio/write/range_write.rs
+++ b/core/src/raw/oio/write/range_write.rs
@@ -94,7 +94,7 @@ pub trait RangeWrite: Send + Sync + Unpin + 'static {
/// WritePartResult is the result returned by [`WriteRangeFuture`].
///
/// The error part will carries input `(offset, bytes, err)` so caller can retry them.
-type WriteRangeResult = std::result::Result<(), (u64, oio::ChunkedBytes, Error)>;
+type WriteRangeResult = std::result::Result<(), (u64, Bytes, Error)>;
struct WriteRangeFuture(BoxedStaticFuture<WriteRangeResult>);
@@ -116,18 +116,13 @@ impl Future for WriteRangeFuture {
}
impl WriteRangeFuture {
- pub fn new<W: RangeWrite>(
- w: Arc<W>,
- location: Arc<String>,
- offset: u64,
- bytes: oio::ChunkedBytes,
- ) -> Self {
+ pub fn new<W: RangeWrite>(w: Arc<W>, location: Arc<String>, offset: u64, bytes: Bytes) -> Self {
let fut = async move {
w.write_range(
&location,
offset,
bytes.len() as u64,
- AsyncBody::ChunkedBytes(bytes.clone()),
+ AsyncBody::Bytes(bytes.clone()),
)
.await
.map_err(|err| (offset, bytes, err))
@@ -141,8 +136,7 @@ impl WriteRangeFuture {
pub struct RangeWriter<W: RangeWrite> {
location: Option<Arc<String>>,
next_offset: u64,
- /// TODO: Use Bytes directly.
- buffer: Option<oio::ChunkedBytes>,
+ buffer: Option<Bytes>,
futures: ConcurrentFutures<WriteRangeFuture>,
w: Arc<W>,
@@ -163,7 +157,6 @@ impl<W: RangeWrite> RangeWriter<W> {
fn fill_cache(&mut self, bs: Bytes) -> usize {
let size = bs.len();
- let bs = oio::ChunkedBytes::from_vec(vec![bs]);
assert!(self.buffer.is_none());
self.buffer = Some(bs);
size
@@ -219,7 +212,7 @@ impl<W: RangeWrite> oio::Write for RangeWriter<W> {
async fn close(&mut self) -> Result<()> {
let Some(location) = self.location.clone() else {
let (size, body) = match self.buffer.clone() {
- Some(cache) => (cache.len(), AsyncBody::ChunkedBytes(cache)),
+ Some(cache) => (cache.len(), AsyncBody::Bytes(cache)),
None => (0, AsyncBody::Empty),
};
// Call write_once if there is no data in buffer and no location.
@@ -247,7 +240,7 @@ impl<W: RangeWrite> oio::Write for RangeWriter<W> {
&location,
offset,
buffer.len() as u64,
- AsyncBody::ChunkedBytes(buffer),
+ AsyncBody::Bytes(buffer),
)
.await?;
self.buffer = None;
(opendal) 02/07: Refactor exact buf write
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch cleanup
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 48eb515fecfc755be00c18d6f5f734fd371f7197
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Mar 14 21:10:47 2024 +0800
Refactor exact buf write
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/oio/write/exact_buf_write.rs | 65 +++++++++++++++++++++++--------
1 file changed, 49 insertions(+), 16 deletions(-)
diff --git a/core/src/raw/oio/write/exact_buf_write.rs b/core/src/raw/oio/write/exact_buf_write.rs
index cd95055c83..f1097ab140 100644
--- a/core/src/raw/oio/write/exact_buf_write.rs
+++ b/core/src/raw/oio/write/exact_buf_write.rs
@@ -15,10 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use bytes::Bytes;
-use std::cmp::min;
+use bytes::{Buf, BufMut, Bytes};
+use std::mem;
-use crate::raw::oio::WriteBuf;
use crate::raw::*;
use crate::*;
@@ -38,7 +37,8 @@ pub struct ExactBufWriter<W: oio::Write> {
/// The size for buffer, we will flush the underlying storage at the size of this buffer.
buffer_size: usize,
- buffer: oio::ChunkedBytes,
+ buffer: Vec<u8>,
+ frozen: Option<Bytes>,
}
impl<W: oio::Write> ExactBufWriter<W> {
@@ -47,35 +47,68 @@ impl<W: oio::Write> ExactBufWriter<W> {
Self {
inner,
buffer_size,
- buffer: oio::ChunkedBytes::default(),
+ buffer: Vec::new(),
+ frozen: None,
}
}
}
impl<W: oio::Write> oio::Write for ExactBufWriter<W> {
async fn write(&mut self, bs: Bytes) -> Result<usize> {
- if self.buffer.len() >= self.buffer_size {
- let bs = self.buffer.bytes(self.buffer.remaining());
- let written = self.inner.write(bs).await?;
- self.buffer.advance(written);
+ if let Some(bs) = self.frozen.as_mut() {
+ let written = self.inner.write(bs.clone()).await?;
+ bs.advance(written);
+ if bs.is_empty() {
+ self.frozen = None;
+ } else {
+ // Return remaining bytes back to buffer
+ self.buffer = bs.to_vec();
+ }
+ }
+
+ // Quick Path
+ //
+ // if buffer is empty and bs is larger than buffer_size, we can directly
+ // freeze the first buffer_size bytes.
+ if self.buffer.is_empty() {
+ if bs.len() >= self.buffer_size {
+ self.frozen = Some(bs.slice(0..self.buffer_size));
+ return Ok(self.buffer_size);
+ }
}
- let remaining = min(self.buffer_size - self.buffer.len(), bs.len());
- self.buffer.push(bs.slice(0..remaining));
- Ok(remaining)
+ let remaining = self.buffer_size - self.buffer.len();
+ if bs.len() >= remaining {
+ self.buffer.put_slice(&bs[0..remaining]);
+ self.frozen = Some(Bytes::from(mem::take(&mut self.buffer)));
+ Ok(remaining)
+ } else {
+ self.buffer.put_slice(&bs);
+ Ok(bs.len())
+ }
}
async fn close(&mut self) -> Result<()> {
- while !self.buffer.is_empty() {
- let bs = self.buffer.bytes(self.buffer.remaining());
- let n = self.inner.write(bs).await?;
- self.buffer.advance(n);
+ loop {
+ if let Some(bs) = self.frozen.as_mut() {
+ let written = self.inner.write(bs.clone()).await?;
+ bs.advance(written);
+ if bs.is_empty() {
+ self.frozen = None;
+ }
+ } else if !self.buffer.is_empty() {
+ // freeze the remaining buffer
+ self.frozen = Some(Bytes::from(mem::take(&mut self.buffer)));
+ } else {
+ break;
+ }
}
self.inner.close().await
}
async fn abort(&mut self) -> Result<()> {
+ self.frozen = None;
self.buffer.clear();
self.inner.abort().await
}
(opendal) 06/07: remove chunked bytes
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch cleanup
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit ba864e854246e76018b5cb88d4abc94adae8defb
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Mar 14 21:19:41 2024 +0800
remove chunked bytes
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/oio/buf/chunked_bytes.rs | 514 ----------------------------------
core/src/raw/oio/buf/mod.rs | 3 -
2 files changed, 517 deletions(-)
diff --git a/core/src/raw/oio/buf/chunked_bytes.rs b/core/src/raw/oio/buf/chunked_bytes.rs
deleted file mode 100644
index 95ca55a219..0000000000
--- a/core/src/raw/oio/buf/chunked_bytes.rs
+++ /dev/null
@@ -1,514 +0,0 @@
-// 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::cmp::min;
-use std::collections::VecDeque;
-use std::io::IoSlice;
-use std::pin::Pin;
-use std::task::Context;
-use std::task::Poll;
-
-use bytes::Bytes;
-use bytes::BytesMut;
-use futures::Stream;
-
-use crate::raw::*;
-use crate::*;
-
-// TODO: 64KiB is picked based on experiences, should be configurable
-const DEFAULT_CHUNK_SIZE: usize = 64 * 1024;
-
-/// ChunkedBytes is used represents a non-contiguous bytes in memory.
-#[derive(Clone)]
-pub struct ChunkedBytes {
- chunk_size: usize,
-
- frozen: VecDeque<Bytes>,
- active: BytesMut,
- size: usize,
-}
-
-impl Default for ChunkedBytes {
- fn default() -> Self {
- Self::new()
- }
-}
-
-impl ChunkedBytes {
- /// Create a new chunked bytes.
- pub fn new() -> Self {
- Self {
- frozen: VecDeque::new(),
- active: BytesMut::new(),
- size: 0,
-
- chunk_size: DEFAULT_CHUNK_SIZE,
- }
- }
-
- /// Create a new chunked cursor with given chunk size.
- pub fn with_chunk_size(chunk_size: usize) -> Self {
- Self {
- frozen: VecDeque::new(),
- active: BytesMut::new(),
- size: 0,
-
- chunk_size,
- }
- }
-
- /// Build a chunked bytes from a vector of bytes.
- ///
- /// This function is guaranteed to run in O(1) time and to not re-allocate the Vec’s buffer
- /// or allocate any additional memory.
- ///
- /// Reference: <https://doc.rust-lang.org/stable/std/collections/struct.VecDeque.html#impl-From%3CVec%3CT,+A%3E%3E-for-VecDeque%3CT,+A%3E>
- pub fn from_vec(bs: Vec<Bytes>) -> Self {
- Self {
- size: bs.iter().map(|v| v.len()).sum(),
- frozen: bs.into(),
- active: BytesMut::new(),
-
- chunk_size: DEFAULT_CHUNK_SIZE,
- }
- }
-
- /// Returns `true` if current cursor is empty.
- pub fn is_empty(&self) -> bool {
- self.size == 0
- }
-
- /// Return current bytes size of cursor.
- pub fn len(&self) -> usize {
- self.size
- }
-
- /// Clear the entire cursor.
- pub fn clear(&mut self) {
- self.size = 0;
- self.frozen.clear();
- self.active.clear();
- }
-
- /// Push a new bytes into ChunkedBytes.
- pub fn push(&mut self, mut bs: Bytes) {
- self.size += bs.len();
-
- // Optimization: if active is empty, we can push to frozen directly if possible.
- if self.active.is_empty() {
- let aligned_size = bs.len() - bs.len() % self.chunk_size;
- if aligned_size > 0 {
- self.frozen.push_back(bs.split_to(aligned_size));
- }
- if !bs.is_empty() {
- self.active.extend_from_slice(&bs);
- }
- return;
- }
-
- // Try to fill bytes into active first.
- let remaining = self.chunk_size.saturating_sub(self.active.len());
- if remaining > 0 {
- let len = min(remaining, bs.len());
- self.active.extend_from_slice(&bs.split_to(len));
- }
-
- // If active is full, freeze it and push it into frozen.
- if self.active.len() == self.chunk_size {
- self.frozen.push_back(self.active.split().freeze());
- }
-
- // Split remaining bytes into chunks.
- let aligned_size = bs.len() - bs.len() % self.chunk_size;
- if aligned_size > 0 {
- self.frozen.push_back(bs.split_to(aligned_size));
- }
-
- // Append to active if there are remaining bytes.
- if !bs.is_empty() {
- self.active.extend_from_slice(&bs);
- }
- }
-
- /// Push a new &[u8] into ChunkedBytes.
- pub fn extend_from_slice(&mut self, bs: &[u8]) {
- self.size += bs.len();
-
- let mut remaining = bs;
-
- while !remaining.is_empty() {
- let available = self.chunk_size.saturating_sub(self.active.len());
-
- // available == 0 means self.active.len() >= CHUNK_SIZE
- if available == 0 {
- self.frozen.push_back(self.active.split().freeze());
- self.active.reserve(self.chunk_size);
- continue;
- }
-
- let size = min(remaining.len(), available);
- self.active.extend_from_slice(&remaining[0..size]);
-
- remaining = &remaining[size..];
- }
- }
-
- /// Pull data from [`oio::WriteBuf`] into ChunkedBytes.
- pub fn extend_from_write_buf(&mut self, size: usize, buf: &dyn oio::WriteBuf) -> usize {
- let to_write = min(buf.chunk().len(), size);
-
- if buf.is_bytes_optimized(to_write) && to_write > self.chunk_size {
- // If the chunk is optimized, we can just push it directly.
- self.push(buf.bytes(to_write));
- } else {
- // Otherwise, we should copy it into the buffer.
- self.extend_from_slice(&buf.chunk()[..to_write]);
- }
-
- to_write
- }
-}
-
-impl oio::WriteBuf for ChunkedBytes {
- fn remaining(&self) -> usize {
- self.size
- }
-
- fn advance(&mut self, mut cnt: usize) {
- debug_assert!(
- cnt <= self.size,
- "cnt size {} is larger than bytes size {}",
- cnt,
- self.size
- );
-
- self.size -= cnt;
-
- while cnt > 0 {
- if let Some(front) = self.frozen.front_mut() {
- if front.len() <= cnt {
- cnt -= front.len();
- self.frozen.pop_front(); // Remove the entire chunk.
- } else {
- front.advance(cnt); // Split and keep the remaining part.
- break;
- }
- } else {
- // Here, cnt must be <= self.active.len() due to the checks above
- self.active.advance(cnt); // Remove cnt bytes from the active buffer.
- break;
- }
- }
- }
-
- fn chunk(&self) -> &[u8] {
- match self.frozen.front() {
- Some(v) => v,
- None => &self.active,
- }
- }
-
- fn vectored_chunk(&self) -> Vec<IoSlice> {
- let it = self.frozen.iter().map(|v| IoSlice::new(v));
-
- if !self.active.is_empty() {
- it.chain([IoSlice::new(&self.active)]).collect()
- } else {
- it.collect()
- }
- }
-
- fn bytes(&self, size: usize) -> Bytes {
- debug_assert!(
- size <= self.size,
- "input size {} is larger than bytes size {}",
- size,
- self.size
- );
-
- if size == 0 {
- return Bytes::new();
- }
-
- if let Some(bs) = self.frozen.front() {
- if size <= bs.len() {
- return bs.slice(..size);
- }
- }
-
- let mut remaining = size;
- let mut result = BytesMut::with_capacity(size);
-
- // First, go through the frozen buffer.
- for chunk in &self.frozen {
- let to_copy = min(remaining, chunk.len());
- result.extend_from_slice(&chunk[0..to_copy]);
- remaining -= to_copy;
-
- if remaining == 0 {
- break;
- }
- }
-
- // Then, get from the active buffer if necessary.
- if remaining > 0 {
- result.extend_from_slice(&self.active[0..remaining]);
- }
-
- result.freeze()
- }
-
- fn is_bytes_optimized(&self, size: usize) -> bool {
- if let Some(bs) = self.frozen.front() {
- return size <= bs.len();
- }
-
- false
- }
-
- fn vectored_bytes(&self, size: usize) -> Vec<Bytes> {
- debug_assert!(
- size <= self.size,
- "input size {} is larger than bytes size {}",
- size,
- self.size
- );
-
- let mut remaining = size;
- let mut buf = vec![];
- for bs in self.frozen.iter() {
- if remaining == 0 {
- break;
- }
-
- let to_take = min(remaining, bs.len());
-
- if to_take == bs.len() {
- buf.push(bs.clone()); // Clone is shallow; no data copy occurs.
- } else {
- buf.push(bs.slice(0..to_take));
- }
-
- remaining -= to_take;
- }
-
- if remaining > 0 {
- buf.push(Bytes::copy_from_slice(&self.active[0..remaining]));
- }
-
- buf
- }
-}
-
-impl oio::Stream for ChunkedBytes {
- fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- match self.frozen.pop_front() {
- Some(bs) => {
- self.size -= bs.len();
- Poll::Ready(Some(Ok(bs)))
- }
- None if !self.active.is_empty() => {
- self.size -= self.active.len();
- Poll::Ready(Some(Ok(self.active.split().freeze())))
- }
- None => Poll::Ready(None),
- }
- }
-}
-
-impl Stream for ChunkedBytes {
- type Item = Result<Bytes>;
-
- fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
- match self.frozen.pop_front() {
- Some(bs) => {
- self.size -= bs.len();
- Poll::Ready(Some(Ok(bs)))
- }
- None if !self.active.is_empty() => {
- self.size -= self.active.len();
- Poll::Ready(Some(Ok(self.active.split().freeze())))
- }
- None => Poll::Ready(None),
- }
- }
-}
-
-#[cfg(test)]
-mod tests {
- use log::debug;
- use pretty_assertions::assert_eq;
- use rand::thread_rng;
- use rand::Rng;
- use rand::RngCore;
- use sha2::Digest;
- use sha2::Sha256;
-
- use super::*;
- use crate::raw::oio::WriteBuf;
-
- #[test]
- fn test_chunked_bytes_write_buf() -> Result<()> {
- let mut c = ChunkedBytes::with_chunk_size(5);
-
- c.push(Bytes::from("hello"));
- assert_eq!(c.len(), 5);
- assert!(!c.is_empty());
-
- c.push(Bytes::from("world"));
- assert_eq!(c.len(), 10);
- assert!(!c.is_empty());
-
- // Test chunk
- let bs = c.chunk();
- assert_eq!(bs, "hello".as_bytes());
- assert_eq!(c.len(), 10);
- assert!(!c.is_empty());
-
- // The second chunk should return the same content.
- let bs = c.chunk();
- assert_eq!(bs, "hello".as_bytes());
- assert_eq!(c.remaining(), 10);
- assert!(!c.is_empty());
-
- // Test vectored chunk
- let bs = c.vectored_chunk();
- assert_eq!(
- bs.iter().map(|v| v.as_ref()).collect::<Vec<_>>(),
- vec!["hello".as_bytes(), "world".as_bytes()]
- );
- assert_eq!(c.remaining(), 10);
- assert!(!c.is_empty());
-
- // Test bytes
- let bs = c.bytes(4);
- assert_eq!(bs, Bytes::from("hell"));
- assert_eq!(c.remaining(), 10);
- assert!(!c.is_empty());
-
- // Test bytes again
- let bs = c.bytes(6);
- assert_eq!(bs, Bytes::from("hellow"));
- assert_eq!(c.remaining(), 10);
- assert!(!c.is_empty());
-
- // Test vectored bytes
- let bs = c.vectored_bytes(4);
- assert_eq!(bs, vec![Bytes::from("hell")]);
- assert_eq!(c.remaining(), 10);
- assert!(!c.is_empty());
-
- // Test vectored bytes again
- let bs = c.vectored_bytes(6);
- assert_eq!(bs, vec![Bytes::from("hello"), Bytes::from("w")]);
- assert_eq!(c.remaining(), 10);
- assert!(!c.is_empty());
-
- // Test Advance.
- c.advance(4);
-
- // Test chunk
- let bs = c.chunk();
- assert_eq!(bs, "o".as_bytes());
- assert_eq!(c.len(), 6);
- assert!(!c.is_empty());
-
- c.clear();
- assert_eq!(c.len(), 0);
- assert!(c.is_empty());
-
- Ok(())
- }
-
- #[test]
- fn test_fuzz_chunked_bytes_push() -> Result<()> {
- let _ = tracing_subscriber::fmt()
- .pretty()
- .with_test_writer()
- .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
- .try_init();
-
- let mut rng = thread_rng();
-
- let chunk_size = rng.gen_range(1..10);
- let mut cb = ChunkedBytes::with_chunk_size(chunk_size);
- debug!("test_fuzz_chunked_bytes_push: chunk size: {chunk_size}");
-
- let mut expected = BytesMut::new();
- for _ in 0..1000 {
- let size = rng.gen_range(1..20);
- debug!("test_fuzz_chunked_bytes_push: write size: {size}");
-
- let mut content = vec![0; size];
- rng.fill_bytes(&mut content);
-
- expected.extend_from_slice(&content);
- cb.push(Bytes::from(content.clone()));
-
- let cnt = rng.gen_range(0..expected.len());
- expected.advance(cnt);
- cb.advance(cnt);
-
- assert_eq!(expected.len(), cb.len());
- assert_eq!(
- format!("{:x}", Sha256::digest(&expected)),
- format!("{:x}", Sha256::digest(&cb.bytes(cb.len())))
- );
- }
-
- Ok(())
- }
-
- #[test]
- fn test_fuzz_chunked_bytes_extend_from_slice() -> Result<()> {
- let _ = tracing_subscriber::fmt()
- .pretty()
- .with_test_writer()
- .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
- .try_init();
-
- let mut rng = thread_rng();
-
- let chunk_size = rng.gen_range(1..10);
- let mut cb = ChunkedBytes::with_chunk_size(chunk_size);
- debug!("test_fuzz_chunked_bytes_extend_from_slice: chunk size: {chunk_size}");
-
- let mut expected = BytesMut::new();
- for _ in 0..1000 {
- let size = rng.gen_range(1..20);
- debug!("test_fuzz_chunked_bytes_extend_from_slice: write size: {size}");
-
- let mut content = vec![0; size];
- rng.fill_bytes(&mut content);
-
- expected.extend_from_slice(&content);
- cb.extend_from_slice(&content);
-
- let cnt = rng.gen_range(0..expected.len());
- expected.advance(cnt);
- cb.advance(cnt);
-
- assert_eq!(expected.len(), cb.len());
- assert_eq!(
- format!("{:x}", Sha256::digest(&expected)),
- format!("{:x}", Sha256::digest(&cb.bytes(cb.len())))
- );
- }
-
- Ok(())
- }
-}
diff --git a/core/src/raw/oio/buf/mod.rs b/core/src/raw/oio/buf/mod.rs
index abc8bf3286..5aec02b8ca 100644
--- a/core/src/raw/oio/buf/mod.rs
+++ b/core/src/raw/oio/buf/mod.rs
@@ -15,9 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-mod chunked_bytes;
-pub use chunked_bytes::ChunkedBytes;
-
mod write_buf;
pub use write_buf::WriteBuf;
(opendal) 01/07: clean block
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch cleanup
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit dcc298db0d602f84b7aa5e3d68c0d3c085f03b37
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Mar 14 20:22:38 2024 +0800
clean block
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/oio/write/block_write.rs | 16 +++++-----------
1 file changed, 5 insertions(+), 11 deletions(-)
diff --git a/core/src/raw/oio/write/block_write.rs b/core/src/raw/oio/write/block_write.rs
index 2e836afe75..28e6d88ed9 100644
--- a/core/src/raw/oio/write/block_write.rs
+++ b/core/src/raw/oio/write/block_write.rs
@@ -111,7 +111,7 @@ pub trait BlockWrite: Send + Sync + Unpin + 'static {
/// WriteBlockResult is the result returned by [`WriteBlockFuture`].
///
/// The error part will carries input `(block_id, bytes, err)` so caller can retry them.
-type WriteBlockResult = std::result::Result<Uuid, (Uuid, oio::ChunkedBytes, Error)>;
+type WriteBlockResult = std::result::Result<Uuid, (Uuid, Bytes, Error)>;
struct WriteBlockFuture(BoxedStaticFuture<WriteBlockResult>);
@@ -133,12 +133,12 @@ impl Future for WriteBlockFuture {
}
impl WriteBlockFuture {
- pub fn new<W: BlockWrite>(w: Arc<W>, block_id: Uuid, bytes: oio::ChunkedBytes) -> Self {
+ pub fn new<W: BlockWrite>(w: Arc<W>, block_id: Uuid, bytes: Bytes) -> Self {
let fut = async move {
w.write_block(
block_id,
bytes.len() as u64,
- AsyncBody::ChunkedBytes(bytes.clone()),
+ AsyncBody::Bytes(bytes.clone()),
)
.await
// Return bytes while we got an error to allow retry.
@@ -157,15 +157,10 @@ pub struct BlockWriter<W: BlockWrite> {
w: Arc<W>,
block_ids: Vec<Uuid>,
- /// TODO: use Bytes directly.
- cache: Option<oio::ChunkedBytes>,
+ cache: Option<Bytes>,
futures: ConcurrentFutures<WriteBlockFuture>,
}
-/// # Safety
-///
-/// wasm32 is a special target that we only have one event-loop for this state.
-
impl<W: BlockWrite> BlockWriter<W> {
/// Create a new BlockWriter.
pub fn new(inner: W, concurrent: usize) -> Self {
@@ -179,7 +174,6 @@ impl<W: BlockWrite> BlockWriter<W> {
fn fill_cache(&mut self, bs: Bytes) -> usize {
let size = bs.len();
- let bs = oio::ChunkedBytes::from_vec(vec![bs]);
assert!(self.cache.is_none());
self.cache = Some(bs);
size
@@ -231,7 +225,7 @@ where
// No write block has been sent.
if self.futures.is_empty() && self.block_ids.is_empty() {
let (size, body) = match self.cache.clone() {
- Some(cache) => (cache.len(), AsyncBody::ChunkedBytes(cache)),
+ Some(cache) => (cache.len(), AsyncBody::Bytes(cache)),
None => (0, AsyncBody::Empty),
};
self.w.write_once(size as u64, body).await?;