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/16 10:47:43 UTC
(opendal) branch main updated: refactor: typed_kv::adapter should use Buffer (#4497)
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 2e9e89b9c9 refactor: typed_kv::adapter should use Buffer (#4497)
2e9e89b9c9 is described below
commit 2e9e89b9c959adb9c7de54e80ea3a1f6ca5a5b95
Author: tison <wa...@gmail.com>
AuthorDate: Tue Apr 16 18:47:31 2024 +0800
refactor: typed_kv::adapter should use Buffer (#4497)
* refactor: typed_kv::adapter should use Buffer
Signed-off-by: tison <wa...@gmail.com>
* impl fn collect(mut self) -> Buffer
Signed-off-by: tison <wa...@gmail.com>
* Revert "impl fn collect(mut self) -> Buffer"
This reverts commit 482ff9700eb918360f7d017d8d0b48de823e85a6.
* Enforce user to call clone outside
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---------
Signed-off-by: tison <wa...@gmail.com>
Signed-off-by: Xuanwo <gi...@xuanwo.io>
Co-authored-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/adapters/typed_kv/api.rs | 7 +++----
core/src/raw/adapters/typed_kv/backend.rs | 26 ++++++++++----------------
core/src/raw/oio/buf/queue_buf.rs | 8 ++++----
core/src/raw/oio/write/exact_buf_write.rs | 4 ++--
4 files changed, 19 insertions(+), 26 deletions(-)
diff --git a/core/src/raw/adapters/typed_kv/api.rs b/core/src/raw/adapters/typed_kv/api.rs
index f1990d5cc7..9b44c67e61 100644
--- a/core/src/raw/adapters/typed_kv/api.rs
+++ b/core/src/raw/adapters/typed_kv/api.rs
@@ -19,15 +19,14 @@ use std::fmt::Debug;
use std::mem::size_of;
use async_trait::async_trait;
-use bytes::Bytes;
use chrono::Utc;
-use crate::EntryMode;
use crate::Error;
use crate::ErrorKind;
use crate::Metadata;
use crate::Result;
use crate::Scheme;
+use crate::{Buffer, EntryMode};
/// Adapter is the typed adapter to underlying kv services.
///
@@ -97,7 +96,7 @@ pub struct Value {
/// Metadata of this value.
pub metadata: Metadata,
/// The corresponding content of this value.
- pub value: Bytes,
+ pub value: Buffer,
}
impl Value {
@@ -107,7 +106,7 @@ impl Value {
metadata: Metadata::new(EntryMode::DIR)
.with_content_length(0)
.with_last_modified(Utc::now()),
- value: Bytes::new(),
+ value: Buffer::new(),
}
}
diff --git a/core/src/raw/adapters/typed_kv/backend.rs b/core/src/raw/adapters/typed_kv/backend.rs
index b9935e2527..500ec205eb 100644
--- a/core/src/raw/adapters/typed_kv/backend.rs
+++ b/core/src/raw/adapters/typed_kv/backend.rs
@@ -19,11 +19,10 @@ use std::sync::Arc;
use std::vec::IntoIter;
use async_trait::async_trait;
-use bytes::{Buf, Bytes};
use super::Adapter;
use super::Value;
-use crate::raw::oio::HierarchyLister;
+use crate::raw::oio::{HierarchyLister, QueueBuf};
use crate::raw::*;
use crate::*;
@@ -56,8 +55,8 @@ where
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<S: Adapter> Accessor for Backend<S> {
- type Reader = Bytes;
- type BlockingReader = Bytes;
+ type Reader = Buffer;
+ type BlockingReader = Buffer;
type Writer = KvWriter<S>;
type BlockingWriter = KvWriter<S>;
type Lister = HierarchyLister<KvLister>;
@@ -238,7 +237,7 @@ pub struct KvWriter<S> {
path: String,
op: OpWrite,
- buf: Option<Vec<u8>>,
+ buf: Option<QueueBuf>,
value: Option<Value>,
}
@@ -259,7 +258,7 @@ impl<S> KvWriter<S> {
}
fn build(&mut self) -> Value {
- let value = self.buf.take().map(Bytes::from).unwrap_or_default();
+ let value = self.buf.take().map(QueueBuf::collect).unwrap_or_default();
let mut metadata = Metadata::new(EntryMode::FILE);
metadata.set_content_length(value.len() as u64);
@@ -280,11 +279,9 @@ impl<S> KvWriter<S> {
impl<S: Adapter> oio::Write for KvWriter<S> {
async fn write(&mut self, bs: Buffer) -> Result<usize> {
- let size = bs.chunk().len();
-
- let mut buf = self.buf.take().unwrap_or_else(|| Vec::with_capacity(size));
- buf.extend_from_slice(bs.chunk());
-
+ let size = bs.len();
+ let mut buf = self.buf.take().unwrap_or_default();
+ buf.push(bs);
self.buf = Some(buf);
Ok(size)
}
@@ -311,12 +308,9 @@ impl<S: Adapter> oio::Write for KvWriter<S> {
impl<S: Adapter> oio::BlockingWrite for KvWriter<S> {
fn write(&mut self, bs: Buffer) -> Result<usize> {
let size = bs.len();
-
- let mut buf = self.buf.take().unwrap_or_else(|| Vec::with_capacity(size));
- buf.extend_from_slice(bs.chunk());
-
+ let mut buf = self.buf.take().unwrap_or_default();
+ buf.push(bs);
self.buf = Some(buf);
-
Ok(size)
}
diff --git a/core/src/raw/oio/buf/queue_buf.rs b/core/src/raw/oio/buf/queue_buf.rs
index 7e81ac4929..102ee8c3d8 100644
--- a/core/src/raw/oio/buf/queue_buf.rs
+++ b/core/src/raw/oio/buf/queue_buf.rs
@@ -29,7 +29,7 @@ use std::collections::VecDeque;
/// - `push`: Push a new buffer in the queue.
/// - `collect`: Collect all buffer in the queue as a new [`Buffer`]
/// - `advance`: Advance the queue by `cnt` bytes.
-#[derive(Default)]
+#[derive(Clone, Default)]
pub struct QueueBuf(VecDeque<Buffer>);
impl QueueBuf {
@@ -68,13 +68,13 @@ impl QueueBuf {
/// most of them should be acceptable since we can expect the item length of buffers are slower
/// than 4k.
#[inline]
- pub fn collect(&self) -> Buffer {
+ pub fn collect(mut self) -> Buffer {
if self.0.is_empty() {
Buffer::new()
} else if self.0.len() == 1 {
- self.0.clone().pop_front().unwrap()
+ self.0.pop_front().unwrap()
} else {
- self.0.clone().into_iter().flatten().collect()
+ self.0.into_iter().flatten().collect()
}
}
diff --git a/core/src/raw/oio/write/exact_buf_write.rs b/core/src/raw/oio/write/exact_buf_write.rs
index 6e6f385975..3934915cc7 100644
--- a/core/src/raw/oio/write/exact_buf_write.rs
+++ b/core/src/raw/oio/write/exact_buf_write.rs
@@ -47,7 +47,7 @@ impl<W: oio::Write> ExactBufWriter<W> {
impl<W: oio::Write> oio::Write for ExactBufWriter<W> {
async fn write(&mut self, mut bs: Buffer) -> Result<usize> {
if self.buffer.len() >= self.buffer_size {
- let written = self.inner.write(self.buffer.collect()).await?;
+ let written = self.inner.write(self.buffer.clone().collect()).await?;
self.buffer.advance(written);
}
@@ -64,7 +64,7 @@ impl<W: oio::Write> oio::Write for ExactBufWriter<W> {
break;
}
- let written = self.inner.write(self.buffer.collect()).await?;
+ let written = self.inner.write(self.buffer.clone().collect()).await?;
self.buffer.advance(written);
}