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 2023/05/07 11:40:57 UTC
[incubator-opendal] branch add-typed-kv created (now f998ffcc)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch add-typed-kv
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
at f998ffcc Add bench for moka
This branch includes the following new commits:
new 05b4443c feat: Add typed kv adapter and migrate moka to it
new f998ffcc Add bench for moka
The 2 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.
[incubator-opendal] 02/02: Add bench for moka
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch add-typed-kv
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
commit f998ffcc053e2ffa36ce4d3bc84cd56593b29b0f
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Sun May 7 19:40:44 2023 +0800
Add bench for moka
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/benches/ops/utils.rs | 2 ++
1 file changed, 2 insertions(+)
diff --git a/core/benches/ops/utils.rs b/core/benches/ops/utils.rs
index 15dc1915..7c29aa2f 100644
--- a/core/benches/ops/utils.rs
+++ b/core/benches/ops/utils.rs
@@ -45,6 +45,8 @@ pub fn services() -> Vec<(&'static str, Option<Operator>)> {
("fs", service::<services::Fs>()),
("s3", service::<services::S3>()),
("memory", service::<services::Memory>()),
+ #[cfg(feature = "services-moka")]
+ ("moka", service::<services::Moka>()),
]
}
[incubator-opendal] 01/02: feat: Add typed kv adapter and migrate moka to it
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch add-typed-kv
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
commit 05b4443c4e5abdb6e2fcb2e73d015f967654ac35
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Sun May 7 19:29:38 2023 +0800
feat: Add typed kv adapter and migrate moka to it
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/adapters/kv/mod.rs | 6 -
core/src/raw/adapters/mod.rs | 1 +
core/src/raw/adapters/typed_kv/api.rs | 89 ++++++++
core/src/raw/adapters/typed_kv/backend.rs | 282 ++++++++++++++++++++++++++
core/src/raw/adapters/{kv => typed_kv}/mod.rs | 10 +-
core/src/raw/oio/cursor.rs | 21 ++
core/src/services/moka/backend.rs | 34 ++--
7 files changed, 409 insertions(+), 34 deletions(-)
diff --git a/core/src/raw/adapters/kv/mod.rs b/core/src/raw/adapters/kv/mod.rs
index d24179ca..facb6efe 100644
--- a/core/src/raw/adapters/kv/mod.rs
+++ b/core/src/raw/adapters/kv/mod.rs
@@ -18,12 +18,6 @@
//! Providing Key Value Adapter for OpenDAL.
//!
//! Any services that implement `Adapter` can be used an OpenDAL Service.
-//!
-//! # Notes
-//!
-//! This adapter creates a new storage format which is not stable.
-//!
-//! Any service that built upon this adapter should not be persisted.
mod api;
pub use api::Adapter;
diff --git a/core/src/raw/adapters/mod.rs b/core/src/raw/adapters/mod.rs
index d69a595a..9864797c 100644
--- a/core/src/raw/adapters/mod.rs
+++ b/core/src/raw/adapters/mod.rs
@@ -46,3 +46,4 @@
//! - [`kv::Adapter`]: Adapter for Key Value Services like in-memory map, `redis`.
pub mod kv;
+pub mod typed_kv;
diff --git a/core/src/raw/adapters/typed_kv/api.rs b/core/src/raw/adapters/typed_kv/api.rs
new file mode 100644
index 00000000..1f2008c4
--- /dev/null
+++ b/core/src/raw/adapters/typed_kv/api.rs
@@ -0,0 +1,89 @@
+// 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::{fmt::Debug, mem::size_of};
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use chrono::Utc;
+
+use crate::*;
+
+/// Adapter is the typed adapter to underlying kv services.
+///
+/// By implement this trait, any kv service can work as an OpenDAL Service.
+///
+/// # Notes
+///
+/// `typed_kv::Adapter` is the typed version of `kv::Adapter`. It's more
+/// efficient if the uderlying kv service can store data with its type. For
+/// example, we can store `Bytes` along with it's metadata so that we don't
+/// need to serialize/deserialize it when we get it from the service.
+///
+/// Ideally, we should use `typed_kv::Adapter` instead of `kv::Adapter` for
+/// in-memory rust libs like moka and dashmap.
+#[async_trait]
+pub trait Adapter: Send + Sync + Debug + Unpin + 'static {
+ /// Get the scheme and name of current adapter.
+ fn metadata(&self) -> (Scheme, String);
+
+ /// Get a value from adapter.
+ async fn get(&self, path: &str) -> Result<Option<Value>>;
+
+ /// Get a value from adapter.
+ fn blocking_get(&self, path: &str) -> Result<Option<Value>>;
+
+ /// Set a value into adapter.
+ async fn set(&self, path: &str, value: Value) -> Result<()>;
+
+ /// Set a value into adapter.
+ fn blocking_set(&self, path: &str, value: Value) -> Result<()>;
+
+ /// Delete a value from adapter.
+ async fn delete(&self, path: &str) -> Result<()>;
+
+ /// Delete a value from adapter.
+ fn blocking_delete(&self, path: &str) -> Result<()>;
+}
+
+/// Value is the typed value stored in adapter.
+///
+/// It's cheap to clone so that users can read data without extra copy.
+#[derive(Debug, Clone)]
+pub struct Value {
+ /// Metadata of this value.
+ pub metadata: Metadata,
+ /// The correbonding content of this value.
+ pub value: Bytes,
+}
+
+impl Value {
+ /// Create a new dir of value.
+ pub fn new_dir() -> Self {
+ Self {
+ metadata: Metadata::new(EntryMode::DIR)
+ .with_content_length(0)
+ .with_last_modified(Utc::now()),
+ value: Bytes::new(),
+ }
+ }
+
+ /// Size returns the in-memory size of Value.
+ pub fn size(&self) -> usize {
+ size_of::<Metadata>() + self.value.len()
+ }
+}
diff --git a/core/src/raw/adapters/typed_kv/backend.rs b/core/src/raw/adapters/typed_kv/backend.rs
new file mode 100644
index 00000000..cf7b963f
--- /dev/null
+++ b/core/src/raw/adapters/typed_kv/backend.rs
@@ -0,0 +1,282 @@
+// 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::sync::Arc;
+
+use async_trait::async_trait;
+use bytes::Bytes;
+
+use super::Adapter;
+use super::Value;
+use crate::ops::*;
+use crate::raw::oio::VectorCursor;
+use crate::raw::*;
+use crate::*;
+
+/// The typed kv backend which implmenets Accessor for for typed kv adapter.
+#[derive(Debug, Clone)]
+pub struct Backend<S: Adapter> {
+ kv: Arc<S>,
+ root: String,
+}
+
+impl<S> Backend<S>
+where
+ S: Adapter,
+{
+ /// Create a new kv backend.
+ pub fn new(kv: S) -> Self {
+ Self {
+ kv: Arc::new(kv),
+ root: "/".to_string(),
+ }
+ }
+
+ /// Configure root within this backend.
+ pub fn with_root(mut self, root: &str) -> Self {
+ self.root = normalize_root(root);
+ self
+ }
+}
+
+#[async_trait]
+impl<S: Adapter> Accessor for Backend<S> {
+ type Reader = oio::Cursor;
+ type BlockingReader = oio::Cursor;
+ type Writer = KvWriter<S>;
+ type BlockingWriter = KvWriter<S>;
+ type Pager = ();
+ type BlockingPager = ();
+
+ fn info(&self) -> AccessorInfo {
+ let (scheme, name) = self.kv.metadata();
+
+ let mut am = AccessorInfo::default();
+ am.set_scheme(scheme);
+ am.set_name(&name);
+ am.set_root(&self.root);
+
+ let cap = am.capability_mut();
+ cap.read = true;
+ cap.read_can_seek = true;
+ cap.read_can_next = true;
+ cap.read_with_range = true;
+ cap.stat = true;
+
+ cap.write = true;
+ cap.write_with_cache_control = true;
+ cap.write_with_content_disposition = true;
+ cap.write_with_content_type = true;
+ cap.write_without_content_length = true;
+ cap.create_dir = true;
+ cap.delete = true;
+
+ am
+ }
+
+ async fn create_dir(&self, path: &str, _: OpCreate) -> Result<RpCreate> {
+ let p = build_abs_path(&self.root, path);
+ self.kv.set(&p, Value::new_dir()).await?;
+ Ok(RpCreate::default())
+ }
+
+ fn blocking_create_dir(&self, path: &str, _: OpCreate) -> Result<RpCreate> {
+ let p = build_abs_path(&self.root, path);
+ self.kv.blocking_set(&p, Value::new_dir())?;
+
+ Ok(RpCreate::default())
+ }
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let p = build_abs_path(&self.root, path);
+
+ let bs = match self.kv.get(&p).await? {
+ // TODO: we can reuse the metadata in value to build content range.
+ Some(bs) => bs.value,
+ None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
+ };
+
+ let bs = self.apply_range(bs, args.range());
+
+ let length = bs.len();
+ Ok((RpRead::new(length as u64), oio::Cursor::from(bs)))
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ let p = build_abs_path(&self.root, path);
+
+ let bs = match self.kv.blocking_get(&p)? {
+ // TODO: we can reuse the metadata in value to build content range.
+ Some(bs) => bs.value,
+ None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
+ };
+
+ let bs = self.apply_range(bs, args.range());
+ Ok((RpRead::new(bs.len() as u64), oio::Cursor::from(bs)))
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ let p = build_abs_path(&self.root, path);
+
+ Ok((RpWrite::new(), KvWriter::new(self.kv.clone(), p, args)))
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ let p = build_abs_path(&self.root, path);
+
+ Ok((RpWrite::new(), KvWriter::new(self.kv.clone(), p, args)))
+ }
+
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let p = build_abs_path(&self.root, path);
+
+ if p.is_empty() || p.ends_with('/') {
+ Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+ } else {
+ let bs = self.kv.get(&p).await?;
+ match bs {
+ Some(bs) => Ok(RpStat::new(bs.metadata)),
+ None => Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
+ }
+ }
+ }
+
+ fn blocking_stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let p = build_abs_path(&self.root, path);
+
+ if p.is_empty() || p.ends_with('/') {
+ Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+ } else {
+ let bs = self.kv.blocking_get(&p)?;
+ match bs {
+ Some(bs) => Ok(RpStat::new(bs.metadata)),
+ None => Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
+ }
+ }
+ }
+
+ async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ let p = build_abs_path(&self.root, path);
+
+ self.kv.delete(&p).await?;
+ Ok(RpDelete::default())
+ }
+
+ fn blocking_delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ let p = build_abs_path(&self.root, path);
+
+ self.kv.blocking_delete(&p)?;
+ Ok(RpDelete::default())
+ }
+}
+
+impl<S> Backend<S>
+where
+ S: Adapter,
+{
+ fn apply_range(&self, mut bs: Bytes, br: BytesRange) -> Bytes {
+ match (br.offset(), br.size()) {
+ (Some(offset), Some(size)) => {
+ let mut bs = bs.split_off(offset as usize);
+ if (size as usize) < bs.len() {
+ let _ = bs.split_off(size as usize);
+ }
+ bs
+ }
+ (Some(offset), None) => bs.split_off(offset as usize),
+ (None, Some(size)) => bs.split_off(bs.len() - size as usize),
+ (None, None) => bs,
+ }
+ }
+}
+
+pub struct KvWriter<S> {
+ kv: Arc<S>,
+ path: String,
+
+ op: OpWrite,
+ buf: VectorCursor,
+}
+
+impl<S> KvWriter<S> {
+ fn new(kv: Arc<S>, path: String, op: OpWrite) -> Self {
+ KvWriter {
+ kv,
+ path,
+ op,
+ buf: VectorCursor::new(),
+ }
+ }
+
+ fn build(&self) -> Value {
+ let mut metadata = Metadata::new(EntryMode::FILE);
+ if let Some(v) = self.op.cache_control() {
+ metadata.set_cache_control(v);
+ }
+ if let Some(v) = self.op.content_disposition() {
+ metadata.set_content_disposition(v);
+ }
+ if let Some(v) = self.op.content_type() {
+ metadata.set_content_type(v);
+ }
+ if let Some(v) = self.op.content_length() {
+ metadata.set_content_length(v);
+ } else {
+ metadata.set_content_length(self.buf.len() as u64);
+ }
+
+ Value {
+ metadata,
+ value: self.buf.peak_all(),
+ }
+ }
+}
+
+#[async_trait]
+impl<S: Adapter> oio::Write for KvWriter<S> {
+ // TODO: we need to support append in the future.
+ async fn write(&mut self, bs: Bytes) -> Result<()> {
+ self.buf.push(bs);
+
+ Ok(())
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ self.buf.clear();
+
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<()> {
+ self.kv.set(&self.path, self.build()).await?;
+ Ok(())
+ }
+}
+
+impl<S: Adapter> oio::BlockingWrite for KvWriter<S> {
+ fn write(&mut self, bs: Bytes) -> Result<()> {
+ self.buf.push(bs);
+
+ Ok(())
+ }
+
+ fn close(&mut self) -> Result<()> {
+ self.kv.blocking_set(&self.path, self.build())?;
+
+ Ok(())
+ }
+}
diff --git a/core/src/raw/adapters/kv/mod.rs b/core/src/raw/adapters/typed_kv/mod.rs
similarity index 80%
copy from core/src/raw/adapters/kv/mod.rs
copy to core/src/raw/adapters/typed_kv/mod.rs
index d24179ca..12a4e599 100644
--- a/core/src/raw/adapters/kv/mod.rs
+++ b/core/src/raw/adapters/typed_kv/mod.rs
@@ -15,19 +15,13 @@
// specific language governing permissions and limitations
// under the License.
-//! Providing Key Value Adapter for OpenDAL.
+//! Providing Typed Key Value Adapter for OpenDAL.
//!
//! Any services that implement `Adapter` can be used an OpenDAL Service.
-//!
-//! # Notes
-//!
-//! This adapter creates a new storage format which is not stable.
-//!
-//! Any service that built upon this adapter should not be persisted.
mod api;
pub use api::Adapter;
-pub use api::Metadata;
+pub use api::Value;
mod backend;
pub use backend::Backend;
diff --git a/core/src/raw/oio/cursor.rs b/core/src/raw/oio/cursor.rs
index 1e34c9b9..7ce2247d 100644
--- a/core/src/raw/oio/cursor.rs
+++ b/core/src/raw/oio/cursor.rs
@@ -47,6 +47,12 @@ impl Cursor {
}
}
+impl From<Bytes> for Cursor {
+ fn from(v: Bytes) -> Self {
+ Cursor { inner: v, pos: 0 }
+ }
+}
+
impl From<Vec<u8>> for Cursor {
fn from(v: Vec<u8>) -> Self {
Cursor {
@@ -259,6 +265,21 @@ impl VectorCursor {
bs.freeze()
}
+ /// peak all will read and copy all bytes from current cursor
+ /// without change it's content.
+ pub fn peak_all(&self) -> Bytes {
+ // Avoid data copy if we only have one bytes.
+ if self.inner.len() == 1 {
+ return self.inner[0].clone();
+ }
+
+ let mut bs = BytesMut::with_capacity(self.len());
+ for b in &self.inner {
+ bs.extend_from_slice(b);
+ }
+ bs.freeze()
+ }
+
/// Take will consume n bytes from current cursor.
///
/// # Panics
diff --git a/core/src/services/moka/backend.rs b/core/src/services/moka/backend.rs
index f33b698c..f2dd7eee 100644
--- a/core/src/services/moka/backend.rs
+++ b/core/src/services/moka/backend.rs
@@ -24,7 +24,7 @@ use log::debug;
use moka::sync::CacheBuilder;
use moka::sync::SegmentedCache;
-use crate::raw::adapters::kv;
+use crate::raw::adapters::typed_kv;
use crate::*;
/// [moka](https://github.com/moka-rs/moka) backend support.
@@ -151,11 +151,11 @@ impl Builder for MokaBuilder {
fn build(&mut self) -> Result<Self::Accessor> {
debug!("backend build started: {:?}", &self);
- let mut builder: CacheBuilder<String, Vec<u8>, _> =
+ let mut builder: CacheBuilder<String, typed_kv::Value, _> =
SegmentedCache::builder(self.num_segments.unwrap_or(1))
.thread_pool_enabled(self.thread_pool_enabled.unwrap_or(false));
// Use entries' bytes as capacity weigher.
- builder = builder.weigher(|k, v| (k.len() + v.len()) as u32);
+ builder = builder.weigher(|k, v| (k.len() + v.size()) as u32);
if let Some(v) = &self.name {
builder = builder.name(v);
}
@@ -177,11 +177,11 @@ impl Builder for MokaBuilder {
}
/// Backend is used to serve `Accessor` support in moka.
-pub type MokaBackend = kv::Backend<Adapter>;
+pub type MokaBackend = typed_kv::Backend<Adapter>;
#[derive(Clone)]
pub struct Adapter {
- inner: SegmentedCache<String, Vec<u8>>,
+ inner: SegmentedCache<String, typed_kv::Value>,
}
impl Debug for Adapter {
@@ -194,37 +194,31 @@ impl Debug for Adapter {
}
#[async_trait]
-impl kv::Adapter for Adapter {
- fn metadata(&self) -> kv::Metadata {
- kv::Metadata::new(
+impl typed_kv::Adapter for Adapter {
+ fn metadata(&self) -> (Scheme, String) {
+ (
Scheme::Moka,
- self.inner.name().unwrap_or("moka"),
- Capability {
- read: true,
- write: true,
-
- ..Default::default()
- },
+ self.inner.name().unwrap_or("moka").to_string(),
)
}
- async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ async fn get(&self, path: &str) -> Result<Option<typed_kv::Value>> {
self.blocking_get(path)
}
- fn blocking_get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ fn blocking_get(&self, path: &str) -> Result<Option<typed_kv::Value>> {
match self.inner.get(path) {
None => Ok(None),
Some(bs) => Ok(Some(bs)),
}
}
- async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
+ async fn set(&self, path: &str, value: typed_kv::Value) -> Result<()> {
self.blocking_set(path, value)
}
- fn blocking_set(&self, path: &str, value: &[u8]) -> Result<()> {
- self.inner.insert(path.to_string(), value.to_vec());
+ fn blocking_set(&self, path: &str, value: typed_kv::Value) -> Result<()> {
+ self.inner.insert(path.to_string(), value);
Ok(())
}