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 12:55:25 UTC
[incubator-opendal] branch main updated: feat: Add typed kv adapter and migrate moka to it (#2222)
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/incubator-opendal.git
The following commit(s) were added to refs/heads/main by this push:
new c3c3c9f5 feat: Add typed kv adapter and migrate moka to it (#2222)
c3c3c9f5 is described below
commit c3c3c9f50cc405cee49d2ee2ed5e9c82b7de8a6c
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Sun May 7 20:55:19 2023 +0800
feat: Add typed kv adapter and migrate moka to it (#2222)
* feat: Add typed kv adapter and migrate moka to it
Signed-off-by: Xuanwo <gi...@xuanwo.io>
* Add bench for moka
Signed-off-by: Xuanwo <gi...@xuanwo.io>
* Fix typo
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---------
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/benches/ops/utils.rs | 2 +
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 ++--
8 files changed, 411 insertions(+), 34 deletions(-)
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>()),
]
}
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..e7a85dbe
--- /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 implements 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(())
}