You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by ho...@apache.org on 2024/04/11 05:25:40 UTC
(opendal) 01/02: feat(raw/oio): block write change to buffer
This is an automated email from the ASF dual-hosted git repository.
hoslo pushed a commit to branch block-write-buffer
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit d8f83f858e0622d1ab1221962f22333964d02417
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Apr 11 11:19:03 2024 +0800
feat(raw/oio): block write change to buffer
---
core/src/raw/oio/write/block_write.rs | 29 ++++++++++++++++++++---------
1 file changed, 20 insertions(+), 9 deletions(-)
diff --git a/core/src/raw/oio/write/block_write.rs b/core/src/raw/oio/write/block_write.rs
index a88b00cb46..1c0e9a0dc3 100644
--- a/core/src/raw/oio/write/block_write.rs
+++ b/core/src/raw/oio/write/block_write.rs
@@ -20,7 +20,6 @@ use std::sync::Arc;
use std::task::Context;
use std::task::Poll;
-use bytes::Bytes;
use futures::Future;
use futures::FutureExt;
use futures::StreamExt;
@@ -68,8 +67,10 @@ pub trait BlockWrite: Send + Sync + Unpin + 'static {
/// - All the data has been written to the buffer and we can perform the upload at once.
#[cfg(not(target_arch = "wasm32"))]
fn write_once(&self, size: u64, body: Buffer) -> impl Future<Output = Result<()>> + Send;
+ fn write_once(&self, size: u64, body: Buffer) -> impl Future<Output = Result<()>> + Send;
#[cfg(target_arch = "wasm32")]
fn write_once(&self, size: u64, body: Buffer) -> impl Future<Output = Result<()>>;
+ fn write_once(&self, size: u64, body: Buffer) -> impl Future<Output = Result<()>>;
/// write_block will write a block of the data and returns the result
/// [`Block`].
@@ -84,6 +85,7 @@ pub trait BlockWrite: Send + Sync + Unpin + 'static {
block_id: Uuid,
size: u64,
body: Buffer,
+ body: Buffer,
) -> impl Future<Output = Result<()>> + Send;
#[cfg(target_arch = "wasm32")]
fn write_block(
@@ -91,6 +93,7 @@ pub trait BlockWrite: Send + Sync + Unpin + 'static {
block_id: Uuid,
size: u64,
body: Buffer,
+ body: Buffer,
) -> impl Future<Output = Result<()>>;
/// complete_block will complete the block upload to build the final
@@ -110,7 +113,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, Bytes, Error)>;
+type WriteBlockResult = std::result::Result<Uuid, (Uuid, Buffer, Error)>;
struct WriteBlockFuture(BoxedStaticFuture<WriteBlockResult>);
@@ -132,9 +135,9 @@ impl Future for WriteBlockFuture {
}
impl WriteBlockFuture {
- pub fn new<W: BlockWrite>(w: Arc<W>, block_id: Uuid, bytes: Bytes) -> Self {
+ pub fn new<W: BlockWrite>(w: Arc<W>, block_id: Uuid, bytes: Buffer) -> Self {
let fut = async move {
- w.write_block(block_id, bytes.len() as u64, Buffer::from(bytes.clone()))
+ w.write_block(block_id, bytes.len() as u64, bytes.clone())
.await
// Return bytes while we got an error to allow retry.
.map_err(|err| (block_id, bytes, err))
@@ -152,7 +155,7 @@ pub struct BlockWriter<W: BlockWrite> {
w: Arc<W>,
block_ids: Vec<Uuid>,
- cache: Option<Bytes>,
+ cache: Option<Buffer>,
futures: ConcurrentFutures<WriteBlockFuture>,
}
@@ -167,7 +170,7 @@ impl<W: BlockWrite> BlockWriter<W> {
}
}
- fn fill_cache(&mut self, bs: Bytes) -> usize {
+ fn fill_cache(&mut self, bs: Buffer) -> usize {
let size = bs.len();
assert!(self.cache.is_none());
self.cache = Some(bs);
@@ -184,7 +187,7 @@ where
if self.futures.has_remaining() {
// Fill cache with the first write.
if self.cache.is_none() {
- let size = self.fill_cache(bs.to_bytes());
+ let size = self.fill_cache(bs);
return Ok(size);
}
@@ -195,7 +198,7 @@ where
cache,
));
- let size = self.fill_cache(bs.to_bytes());
+ let size = self.fill_cache(bs);
return Ok(size);
} else if let Some(res) = self.futures.next().await {
match res {
@@ -220,7 +223,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(), Buffer::from(cache)),
+ Some(cache) => (cache.len(), cache),
None => (0, Buffer::new()),
};
self.w.write_once(size as u64, body).await?;
@@ -285,6 +288,8 @@ mod tests {
length: u64,
bytes: HashMap<Uuid, Buffer>,
content: Option<Buffer>,
+ bytes: HashMap<Uuid, Buffer>,
+ content: Option<Buffer>,
}
impl TestWrite {
@@ -300,10 +305,12 @@ mod tests {
}
impl BlockWrite for Arc<Mutex<TestWrite>> {
+ async fn write_once(&self, _: u64, _: Buffer) -> Result<()> {
async fn write_once(&self, _: u64, _: Buffer) -> Result<()> {
Ok(())
}
+ async fn write_block(&self, block_id: Uuid, size: u64, body: Buffer) -> Result<()> {
async fn write_block(&self, block_id: Uuid, size: u64, body: Buffer) -> Result<()> {
// We will have 50% percent rate for write part to fail.
if thread_rng().gen_bool(5.0 / 10.0) {
@@ -313,6 +320,7 @@ mod tests {
let mut this = self.lock().unwrap();
this.length += size;
this.bytes.insert(block_id, body);
+ this.bytes.insert(block_id, body);
Ok(())
}
@@ -322,8 +330,10 @@ mod tests {
let mut bs = Vec::new();
for id in block_ids {
bs.push(this.bytes[&id].clone());
+ bs.push(this.bytes[&id].clone());
}
this.content = Some(bs.into_iter().flatten().collect());
+ this.content = Some(bs.into_iter().flatten().collect());
Ok(())
}
@@ -372,6 +382,7 @@ mod tests {
assert_eq!(
expected_content,
inner.content.clone().unwrap().to_bytes(),
+ inner.content.clone().unwrap().to_bytes(),
"content must be the same"
);
}