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 2025/06/17 05:11:14 UTC
(opendal) 01/01: refactor: Migrate memory service to implment Access directly
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch refactor-memoery
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 2390969b2ca497a7dba162af0ce1f33cde7e79dd
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Jun 17 13:10:56 2025 +0800
refactor: Migrate memory service to implment Access directly
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/services/memory/backend.rs | 157 ++++++++++++++++---------
core/src/services/memory/core.rs | 80 +++++++++++++
core/src/services/memory/{mod.rs => delete.rs} | 30 ++++-
core/src/services/memory/lister.rs | 56 +++++++++
core/src/services/memory/mod.rs | 9 ++
core/src/services/memory/writer.rs | 85 +++++++++++++
6 files changed, 355 insertions(+), 62 deletions(-)
diff --git a/core/src/services/memory/backend.rs b/core/src/services/memory/backend.rs
index ba464ac02..7dc981c4b 100644
--- a/core/src/services/memory/backend.rs
+++ b/core/src/services/memory/backend.rs
@@ -15,13 +15,15 @@
// specific language governing permissions and limitations
// under the License.
-use std::collections::BTreeMap;
use std::fmt::Debug;
use std::sync::Arc;
-use std::sync::Mutex;
-use crate::raw::adapters::typed_kv;
-use crate::raw::Access;
+use super::core::*;
+use super::delete::MemoryDeleter;
+use super::lister::MemoryLister;
+use super::writer::MemoryWriter;
+use crate::raw::oio;
+use crate::raw::*;
use crate::services::MemoryConfig;
use crate::*;
@@ -52,77 +54,120 @@ impl Builder for MemoryBuilder {
type Config = MemoryConfig;
fn build(self) -> Result<impl Access> {
- let adapter = Adapter {
- inner: Arc::new(Mutex::new(BTreeMap::default())),
- };
-
- Ok(MemoryBackend::new(adapter).with_root(self.config.root.as_deref().unwrap_or_default()))
+ let root = normalize_root(
+ self.config
+ .root
+ .as_deref()
+ .unwrap_or("/"),
+ );
+
+ let core = MemoryCore::new();
+ Ok(MemoryAccessor::new(core).with_normalized_root(root))
}
}
-/// Backend is used to serve `Accessor` support in memory.
-pub type MemoryBackend = typed_kv::Backend<Adapter>;
-
-#[derive(Clone)]
-pub struct Adapter {
- inner: Arc<Mutex<BTreeMap<String, typed_kv::Value>>>,
+/// MemoryAccessor implements Access trait directly
+#[derive(Debug, Clone)]
+pub struct MemoryAccessor {
+ core: Arc<MemoryCore>,
+ root: String,
+ info: Arc<AccessorInfo>,
}
-impl Debug for Adapter {
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- f.debug_struct("MemoryBackend").finish_non_exhaustive()
+impl MemoryAccessor {
+ fn new(core: MemoryCore) -> Self {
+ let info = AccessorInfo::default();
+ info.set_scheme(Scheme::Memory);
+ info.set_name(&format!("{:p}", Arc::as_ptr(&core.data)));
+ info.set_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,
+ stat_has_cache_control: true,
+ stat_has_content_type: true,
+ stat_has_content_disposition: true,
+ stat_has_content_encoding: true,
+ stat_has_content_length: true,
+ list: true,
+ list_with_recursive: true,
+ shared: false,
+ ..Default::default()
+ });
+
+ Self {
+ core: Arc::new(core),
+ root: "/".to_string(),
+ info: Arc::new(info),
+ }
+ }
+
+ fn with_normalized_root(mut self, root: String) -> Self {
+ self.info.set_root(&root);
+ self.root = root;
+ self
}
}
-impl typed_kv::Adapter for Adapter {
- fn info(&self) -> typed_kv::Info {
- typed_kv::Info::new(
- Scheme::Memory,
- &format!("{:p}", Arc::as_ptr(&self.inner)),
- typed_kv::Capability {
- get: true,
- set: true,
- delete: true,
- scan: true,
- shared: false,
- },
- )
+impl Access for MemoryAccessor {
+ type Reader = Buffer;
+ type Writer = MemoryWriter;
+ type Lister = oio::HierarchyLister<MemoryLister>;
+ type Deleter = oio::OneShotDeleter<MemoryDeleter>;
+
+ fn info(&self) -> Arc<AccessorInfo> {
+ self.info.clone()
}
- async fn get(&self, path: &str) -> Result<Option<typed_kv::Value>> {
- match self.inner.lock().unwrap().get(path) {
- None => Ok(None),
- Some(bs) => Ok(Some(bs.to_owned())),
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let p = build_abs_path(&self.root, path);
+
+ if p == build_abs_path(&self.root, "") {
+ Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+ } else {
+ match self.core.get(&p)? {
+ Some(value) => Ok(RpStat::new(value.metadata)),
+ None => Err(Error::new(ErrorKind::NotFound, "memory doesn't have this path")),
+ }
}
}
- async fn set(&self, path: &str, value: typed_kv::Value) -> Result<()> {
- self.inner.lock().unwrap().insert(path.to_string(), value);
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let p = build_abs_path(&self.root, path);
+
+ let value = match self.core.get(&p)? {
+ Some(value) => value,
+ None => return Err(Error::new(ErrorKind::NotFound, "memory doesn't have this path")),
+ };
- Ok(())
+ Ok((RpRead::new(), value.content.slice(args.range().to_range_as_usize())))
}
- async fn delete(&self, path: &str) -> Result<()> {
- self.inner.lock().unwrap().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(), MemoryWriter::new(self.core.clone(), p, args)))
+ }
- Ok(())
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ Ok((
+ RpDelete::default(),
+ oio::OneShotDeleter::new(MemoryDeleter::new(self.core.clone(), self.root.clone())),
+ ))
}
- async fn scan(&self, path: &str) -> Result<Vec<String>> {
- let inner = self.inner.lock().unwrap();
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ let p = build_abs_path(&self.root, path);
+ let keys = self.core.scan(&p)?;
+ let lister = MemoryLister::new(&self.root, keys);
+ let lister = oio::HierarchyLister::new(lister, path, args.recursive());
- if path.is_empty() {
- return Ok(inner.keys().cloned().collect());
- }
-
- let mut keys = Vec::new();
- for (key, _) in inner.range(path.to_string()..) {
- if !key.starts_with(path) {
- break;
- }
- keys.push(key.to_string());
- }
- Ok(keys)
+ Ok((RpList::default(), lister))
}
}
@@ -138,4 +183,4 @@ mod tests {
let b2 = MemoryBuilder::default().build().unwrap();
assert_ne!(b1.info().name(), b2.info().name())
}
-}
+}
\ No newline at end of file
diff --git a/core/src/services/memory/core.rs b/core/src/services/memory/core.rs
new file mode 100644
index 000000000..68c42d092
--- /dev/null
+++ b/core/src/services/memory/core.rs
@@ -0,0 +1,80 @@
+// 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::collections::BTreeMap;
+use std::fmt::Debug;
+use std::sync::Arc;
+use std::sync::Mutex;
+
+use crate::*;
+
+/// Value stored in memory containing both metadata and content
+#[derive(Clone)]
+pub struct MemoryValue {
+ pub metadata: Metadata,
+ pub content: Buffer,
+}
+
+#[derive(Clone)]
+pub struct MemoryCore {
+ pub data: Arc<Mutex<BTreeMap<String, MemoryValue>>>,
+}
+
+impl Debug for MemoryCore {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("MemoryCore").finish_non_exhaustive()
+ }
+}
+
+impl MemoryCore {
+ pub fn new() -> Self {
+ Self {
+ data: Arc::new(Mutex::new(BTreeMap::new())),
+ }
+ }
+
+ pub fn get(&self, key: &str) -> Result<Option<MemoryValue>> {
+ Ok(self.data.lock().unwrap().get(key).cloned())
+ }
+
+ pub fn set(&self, key: &str, value: MemoryValue) -> Result<()> {
+ self.data.lock().unwrap().insert(key.to_string(), value);
+ Ok(())
+ }
+
+ pub fn delete(&self, key: &str) -> Result<()> {
+ self.data.lock().unwrap().remove(key);
+ Ok(())
+ }
+
+ pub fn scan(&self, prefix: &str) -> Result<Vec<String>> {
+ let data = self.data.lock().unwrap();
+
+ if prefix.is_empty() {
+ return Ok(data.keys().cloned().collect());
+ }
+
+ let mut keys = Vec::new();
+ for (key, _) in data.range(prefix.to_string()..) {
+ if !key.starts_with(prefix) {
+ break;
+ }
+ keys.push(key.clone());
+ }
+ Ok(keys)
+ }
+}
\ No newline at end of file
diff --git a/core/src/services/memory/mod.rs b/core/src/services/memory/delete.rs
similarity index 61%
copy from core/src/services/memory/mod.rs
copy to core/src/services/memory/delete.rs
index ecf2e8d27..f5befe709 100644
--- a/core/src/services/memory/mod.rs
+++ b/core/src/services/memory/delete.rs
@@ -15,10 +15,28 @@
// specific language governing permissions and limitations
// under the License.
-#[cfg(feature = "services-memory")]
-mod backend;
-#[cfg(feature = "services-memory")]
-pub use backend::MemoryBuilder as Memory;
+use std::sync::Arc;
-mod config;
-pub use config::MemoryConfig;
+use super::core::*;
+use crate::raw::oio;
+use crate::raw::*;
+use crate::*;
+
+pub struct MemoryDeleter {
+ core: Arc<MemoryCore>,
+ root: String,
+}
+
+impl MemoryDeleter {
+ pub fn new(core: Arc<MemoryCore>, root: String) -> Self {
+ Self { core, root }
+ }
+}
+
+impl oio::OneShotDelete for MemoryDeleter {
+ async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+ let p = build_abs_path(&self.root, &path);
+ self.core.delete(&p)?;
+ Ok(())
+ }
+}
\ No newline at end of file
diff --git a/core/src/services/memory/lister.rs b/core/src/services/memory/lister.rs
new file mode 100644
index 000000000..985603329
--- /dev/null
+++ b/core/src/services/memory/lister.rs
@@ -0,0 +1,56 @@
+// 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::vec::IntoIter;
+
+use crate::raw::oio;
+use crate::raw::*;
+use crate::*;
+
+pub struct MemoryLister {
+ root: String,
+ keys: IntoIter<String>,
+}
+
+impl MemoryLister {
+ pub fn new(root: &str, keys: Vec<String>) -> Self {
+ Self {
+ root: root.to_string(),
+ keys: keys.into_iter(),
+ }
+ }
+}
+
+impl oio::List for MemoryLister {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ match self.keys.next() {
+ Some(key) => {
+ let mode = if key.ends_with('/') {
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ };
+ let mut path = build_rel_path(&self.root, &key);
+ if path.is_empty() {
+ path = "/".to_string();
+ }
+ Ok(Some(oio::Entry::new(&path, Metadata::new(mode))))
+ }
+ None => Ok(None),
+ }
+ }
+}
\ No newline at end of file
diff --git a/core/src/services/memory/mod.rs b/core/src/services/memory/mod.rs
index ecf2e8d27..cefcd972a 100644
--- a/core/src/services/memory/mod.rs
+++ b/core/src/services/memory/mod.rs
@@ -22,3 +22,12 @@ pub use backend::MemoryBuilder as Memory;
mod config;
pub use config::MemoryConfig;
+
+#[cfg(feature = "services-memory")]
+mod core;
+#[cfg(feature = "services-memory")]
+mod delete;
+#[cfg(feature = "services-memory")]
+mod lister;
+#[cfg(feature = "services-memory")]
+mod writer;
diff --git a/core/src/services/memory/writer.rs b/core/src/services/memory/writer.rs
new file mode 100644
index 000000000..e28440dff
--- /dev/null
+++ b/core/src/services/memory/writer.rs
@@ -0,0 +1,85 @@
+// 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 super::core::*;
+use crate::raw::oio;
+use crate::raw::*;
+use crate::*;
+
+pub struct MemoryWriter {
+ core: Arc<MemoryCore>,
+ path: String,
+ op: OpWrite,
+ buf: Option<oio::QueueBuf>,
+}
+
+impl MemoryWriter {
+ pub fn new(core: Arc<MemoryCore>, path: String, op: OpWrite) -> Self {
+ Self {
+ core,
+ path,
+ op,
+ buf: None,
+ }
+ }
+}
+
+impl oio::Write for MemoryWriter {
+ async fn write(&mut self, bs: Buffer) -> Result<()> {
+ let mut buf = self.buf.take().unwrap_or_default();
+ buf.push(bs);
+ self.buf = Some(buf);
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<Metadata> {
+ let buf = self.buf.take().unwrap_or_default();
+ let content = buf.collect();
+
+ let mut metadata = Metadata::new(EntryMode::FILE);
+ metadata.set_content_length(content.len() as u64);
+
+ 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_encoding() {
+ metadata.set_content_encoding(v);
+ }
+
+ let value = MemoryValue {
+ metadata: metadata.clone(),
+ content,
+ };
+
+ self.core.set(&self.path, value)?;
+
+ Ok(metadata)
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ self.buf = None;
+ Ok(())
+ }
+}
\ No newline at end of file