This is an automated email from the ASF dual-hosted git repository. meteorgan pushed a commit to branch refactor-dashmap in repository https://gitbox.apache.org/repos/asf/opendal.git
commit c7a525832e66503cc30a3bb8a7ba94733f18d872 Author: meteorgan <[email protected]> AuthorDate: Mon Jun 30 01:31:34 2025 +0800 refactor: Migrate dashmap service to implement Access directly --- core/src/services/dashmap/backend.rs | 179 +++++++++++++++------ core/src/services/dashmap/config.rs | 19 ++- core/src/services/dashmap/core.rs | 61 +++++++ core/src/services/dashmap/{config.rs => delete.rs} | 28 +++- core/src/services/dashmap/docs.md | 28 +++- core/src/services/dashmap/lister.rs | 63 ++++++++ core/src/services/dashmap/mod.rs | 9 ++ core/src/services/dashmap/writer.rs | 87 ++++++++++ 8 files changed, 407 insertions(+), 67 deletions(-) diff --git a/core/src/services/dashmap/backend.rs b/core/src/services/dashmap/backend.rs index 64cc0716e..300900785 100644 --- a/core/src/services/dashmap/backend.rs +++ b/core/src/services/dashmap/backend.rs @@ -17,11 +17,17 @@ use std::fmt::Debug; use std::fmt::Formatter; +use std::sync::Arc; use dashmap::DashMap; - -use crate::raw::adapters::typed_kv; -use crate::raw::Access; +use log::debug; + +use super::core::DashmapCore; +use super::delete::DashmapDeleter; +use super::lister::DashmapLister; +use super::writer::DashmapWriter; +use crate::raw::oio; +use crate::raw::*; use crate::services::DashmapConfig; use crate::*; @@ -39,6 +45,14 @@ pub struct DashmapBuilder { config: DashmapConfig, } +impl Debug for DashmapBuilder { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DashmapBuilder") + .field("config", &self.config) + .finish() + } +} + impl DashmapBuilder { /// Set the root for dashmap. pub fn root(mut self, path: &str) -> Self { @@ -57,73 +71,134 @@ impl Builder for DashmapBuilder { type Config = DashmapConfig; fn build(self) -> Result<impl Access> { - let mut backend = DashmapBackend::new(Adapter { - inner: DashMap::default(), - }); - if let Some(v) = self.config.root { - backend = backend.with_root(&v); - } + debug!("backend build started: {:?}", &self); + + let root = normalize_root( + self.config + .root + .clone() + .unwrap_or_else(|| "/".to_string()) + .as_str(), + ); + + debug!("backend build finished: {:?}", self.config); + + let core = DashmapCore { + cache: DashMap::new(), + }; - Ok(backend) + Ok(DashmapAccessor::new(core, root)) } } -/// Backend is used to serve `Accessor` support in dashmap. -pub type DashmapBackend = typed_kv::Backend<Adapter>; - -#[derive(Clone)] -pub struct Adapter { - inner: DashMap<String, typed_kv::Value>, +#[derive(Debug, Clone)] +pub struct DashmapAccessor { + core: Arc<DashmapCore>, + root: String, + info: Arc<AccessorInfo>, } -impl Debug for Adapter { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - f.debug_struct("DashmapAdapter") - .field("size", &self.inner.len()) - .finish_non_exhaustive() +impl DashmapAccessor { + fn new(core: DashmapCore, root: String) -> Self { + let info = AccessorInfo::default(); + info.set_scheme(Scheme::Dashmap); + info.set_name("dashmap"); + info.set_root(&root); + info.set_native_capability(Capability { + read: true, + + write: true, + write_can_empty: true, + write_with_cache_control: true, + write_with_content_type: true, + write_with_content_disposition: true, + write_with_content_encoding: true, + + delete: true, + stat: true, + list: true, + shared: false, + ..Default::default() + }); + + Self { + core: Arc::new(core), + root, + info: Arc::new(info), + } } } -impl typed_kv::Adapter for Adapter { - fn info(&self) -> typed_kv::Info { - typed_kv::Info::new( - Scheme::Dashmap, - "dashmap", - typed_kv::Capability { - get: true, - set: true, - scan: true, - delete: true, - shared: false, - }, - ) +impl Access for DashmapAccessor { + type Reader = Buffer; + type Writer = DashmapWriter; + type Lister = oio::HierarchyLister<DashmapLister>; + type Deleter = oio::OneShotDeleter<DashmapDeleter>; + + fn info(&self) -> Arc<AccessorInfo> { + self.info.clone() } - async fn get(&self, path: &str) -> Result<Option<typed_kv::Value>> { - match self.inner.get(path) { - None => Ok(None), - Some(bs) => Ok(Some(bs.value().to_owned())), + async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> { + let p = build_abs_path(&self.root, path); + + match self.core.get(&p)? { + Some(value) => { + let metadata = value.metadata; + Ok(RpStat::new(metadata)) + } + None => { + if p.ends_with('/') { + let has_children = self.core.cache.iter().any(|kv| kv.key().starts_with(&p)); + if has_children { + return Ok(RpStat::new(Metadata::new(EntryMode::DIR))); + } + } + Err(Error::new(ErrorKind::NotFound, "key not found in dashmap")) + } } } - async fn set(&self, path: &str, value: typed_kv::Value) -> Result<()> { - self.inner.insert(path.to_string(), value); - - Ok(()) + async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> { + let p = build_abs_path(&self.root, path); + + match self.core.get(&p)? { + Some(value) => { + let buffer = if args.range().is_full() { + value.content + } else { + let range = args.range(); + let start = range.offset() as usize; + let end = match range.size() { + Some(size) => (range.offset() + size) as usize, + None => value.content.len(), + }; + value.content.slice(start..end.min(value.content.len())) + }; + Ok((RpRead::new(), buffer)) + } + None => Err(Error::new(ErrorKind::NotFound, "key not found in dashmap")), + } } - async fn delete(&self, path: &str) -> Result<()> { - self.inner.remove(path); + async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> { + let p = build_abs_path(&self.root, path); + Ok(( + RpWrite::new(), + DashmapWriter::new(self.core.clone(), p, args), + )) + } - Ok(()) + async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> { + Ok(( + RpDelete::default(), + oio::OneShotDeleter::new(DashmapDeleter::new(self.core.clone(), self.root.clone())), + )) } - async fn scan(&self, path: &str) -> Result<Vec<String>> { - let keys = self.inner.iter().map(|kv| kv.key().to_string()); - if path.is_empty() { - Ok(keys.collect()) - } else { - Ok(keys.filter(|k| k.starts_with(path)).collect()) - } + async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> { + let lister = DashmapLister::new(self.core.clone(), self.root.clone(), path.to_string()); + let lister = oio::HierarchyLister::new(lister, path, args.recursive()); + Ok((RpList::default(), lister)) } } diff --git a/core/src/services/dashmap/config.rs b/core/src/services/dashmap/config.rs index bb747508a..7e9833d9c 100644 --- a/core/src/services/dashmap/config.rs +++ b/core/src/services/dashmap/config.rs @@ -15,14 +15,23 @@ // specific language governing permissions and limitations // under the License. -use std::fmt::Debug; - use serde::Deserialize; use serde::Serialize; +use std::fmt::Debug; -/// [dashmap](https://github.com/xacrimon/dashmap) backend support. -#[derive(Default, Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] +/// Config for Dashmap services support. +#[derive(Default, Serialize, Deserialize, Clone, PartialEq, Eq)] +#[serde(default)] +#[non_exhaustive] pub struct DashmapConfig { - /// The root path for dashmap. + /// root path of this backend pub root: Option<String>, } + +impl Debug for DashmapConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DashmapConfig") + .field("root", &self.root) + .finish_non_exhaustive() + } +} diff --git a/core/src/services/dashmap/core.rs b/core/src/services/dashmap/core.rs new file mode 100644 index 000000000..cdb4caf88 --- /dev/null +++ b/core/src/services/dashmap/core.rs @@ -0,0 +1,61 @@ +// 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; +use std::fmt::Formatter; + +use dashmap::DashMap; + +use crate::*; + +/// Value stored in dashmap cache containing both metadata and content +#[derive(Clone)] +pub struct DashmapValue { + /// Stored metadata in dashmap cache. + pub metadata: Metadata, + /// Stored content in dashmap cache. + pub content: Buffer, +} + +#[derive(Clone)] +pub struct DashmapCore { + pub cache: DashMap<String, DashmapValue>, +} + +impl Debug for DashmapCore { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DashmapCore") + .field("size", &self.cache.len()) + .finish() + } +} + +impl DashmapCore { + pub fn get(&self, key: &str) -> Result<Option<DashmapValue>> { + Ok(self.cache.get(key).map(|v| v.value().clone())) + } + + pub fn set(&self, key: &str, value: DashmapValue) -> Result<()> { + self.cache.insert(key.to_string(), value); + Ok(()) + } + + pub fn delete(&self, key: &str) -> Result<()> { + self.cache.remove(key); + Ok(()) + } +} diff --git a/core/src/services/dashmap/config.rs b/core/src/services/dashmap/delete.rs similarity index 60% copy from core/src/services/dashmap/config.rs copy to core/src/services/dashmap/delete.rs index bb747508a..2883fcc51 100644 --- a/core/src/services/dashmap/config.rs +++ b/core/src/services/dashmap/delete.rs @@ -15,14 +15,26 @@ // specific language governing permissions and limitations // under the License. -use std::fmt::Debug; +use std::sync::Arc; -use serde::Deserialize; -use serde::Serialize; +use super::core::DashmapCore; +use crate::raw::{build_abs_path, oio, OpDelete}; +use crate::*; -/// [dashmap](https://github.com/xacrimon/dashmap) backend support. -#[derive(Default, Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] -pub struct DashmapConfig { - /// The root path for dashmap. - pub root: Option<String>, +pub struct DashmapDeleter { + core: Arc<DashmapCore>, + root: String, +} + +impl DashmapDeleter { + pub fn new(core: Arc<DashmapCore>, root: String) -> Self { + Self { core, root } + } +} + +impl oio::OneShotDelete for DashmapDeleter { + async fn delete_once(&self, path: String, _op: OpDelete) -> Result<()> { + let p = build_abs_path(&self.root, &path); + self.core.delete(&p) + } } diff --git a/core/src/services/dashmap/docs.md b/core/src/services/dashmap/docs.md index d2a894ada..4a66723c9 100644 --- a/core/src/services/dashmap/docs.md +++ b/core/src/services/dashmap/docs.md @@ -9,6 +9,30 @@ This service can be used to: - [x] delete - [x] copy - [x] rename -- [ ] list +- [x] list - [ ] presign -- [ ] blocking + +## Configuration + +- `root`: Set the root path for this dashmap instance. + +You can refer to [`DashmapBuilder`]'s docs for more information + +## Example + +### Via Builder + +```rust,no_run +use anyhow::Result; +use opendal::services::Dashmap; +use opendal::Operator; + +#[tokio::main] +async fn main() -> Result<()> { + let mut builder = Dashmap::default() + .root("/"); + + let op: Operator = Operator::new(builder)?.finish(); + Ok(()) +} +``` diff --git a/core/src/services/dashmap/lister.rs b/core/src/services/dashmap/lister.rs new file mode 100644 index 000000000..e276ac3d1 --- /dev/null +++ b/core/src/services/dashmap/lister.rs @@ -0,0 +1,63 @@ +// 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 super::core::DashmapCore; +use crate::raw::oio::Entry; +use crate::raw::{build_abs_path, build_rel_path, oio}; +use crate::*; +use std::sync::Arc; +use std::vec::IntoIter; + +pub struct DashmapLister { + root: String, + path: String, + iter: IntoIter<String>, +} + +impl DashmapLister { + pub fn new(core: Arc<DashmapCore>, root: String, path: String) -> Self { + let entries: Vec<_> = core.cache.iter().map(|item| item.key().clone()).collect(); + let path = build_abs_path(&root, &path); + + Self { + root, + path, + iter: entries.into_iter(), + } + } +} + +impl oio::List for DashmapLister { + async fn next(&mut self) -> Result<Option<Entry>> { + for key in self.iter.by_ref() { + if key.starts_with(&self.path) { + let path = build_rel_path(&self.root, &key); + + // Determine if it's a file or directory based on trailing slash + let mode = if key.ends_with('/') { + EntryMode::DIR + } else { + EntryMode::FILE + }; + let entry = Entry::new(&path, Metadata::new(mode)); + return Ok(Some(entry)); + } + } + + Ok(None) + } +} diff --git a/core/src/services/dashmap/mod.rs b/core/src/services/dashmap/mod.rs index 3b11d9b66..0d128063d 100644 --- a/core/src/services/dashmap/mod.rs +++ b/core/src/services/dashmap/mod.rs @@ -17,6 +17,15 @@ #[cfg(feature = "services-dashmap")] mod backend; +#[cfg(feature = "services-dashmap")] +mod core; +#[cfg(feature = "services-dashmap")] +mod delete; +#[cfg(feature = "services-dashmap")] +mod lister; +#[cfg(feature = "services-dashmap")] +mod writer; + #[cfg(feature = "services-dashmap")] pub use backend::DashmapBuilder as Dashmap; diff --git a/core/src/services/dashmap/writer.rs b/core/src/services/dashmap/writer.rs new file mode 100644 index 000000000..a4a6dfb9b --- /dev/null +++ b/core/src/services/dashmap/writer.rs @@ -0,0 +1,87 @@ +// 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 std::time::SystemTime; + +use super::core::DashmapCore; +use super::core::DashmapValue; +use crate::raw::{oio, OpWrite}; +use crate::*; + +pub struct DashmapWriter { + core: Arc<DashmapCore>, + path: String, + op: OpWrite, + + buf: oio::QueueBuf, +} + +impl DashmapWriter { + pub fn new(core: Arc<DashmapCore>, path: String, op: OpWrite) -> Self { + DashmapWriter { + core, + path, + op, + buf: oio::QueueBuf::new(), + } + } +} + +impl oio::Write for DashmapWriter { + async fn write(&mut self, bs: Buffer) -> Result<()> { + self.buf.push(bs); + Ok(()) + } + + async fn close(&mut self) -> Result<Metadata> { + let content = self.buf.clone().collect(); + + let entry_mode = EntryMode::from_path(&self.path); + let mut meta = Metadata::new(entry_mode); + meta.set_content_length(content.len() as u64); + meta.set_last_modified(SystemTime::now().into()); + + if let Some(v) = self.op.content_type() { + meta.set_content_type(v); + } + if let Some(v) = self.op.content_disposition() { + meta.set_content_disposition(v); + } + if let Some(v) = self.op.cache_control() { + meta.set_cache_control(v); + } + if let Some(v) = self.op.content_encoding() { + meta.set_content_encoding(v); + } + + self.core.set( + &self.path, + DashmapValue { + metadata: meta.clone(), + content, + }, + )?; + + Ok(meta) + } + + async fn abort(&mut self) -> Result<()> { + self.buf.clear(); + Ok(()) + } +}
