You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by ko...@apache.org on 2025/10/20 16:15:58 UTC
(opendal) branch refactor-memcached created (now 345e56658)
This is an automated email from the ASF dual-hosted git repository.
koushiro pushed a change to branch refactor-memcached
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 345e56658 refactor: migrate memcached service from adapter::kv to impl Access directly
This branch includes the following new commits:
new 345e56658 refactor: migrate memcached service from adapter::kv to impl Access directly
The 1 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.
(opendal) 01/01: refactor: migrate memcached service from adapter::kv to impl Access directly
Posted by ko...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
koushiro pushed a commit to branch refactor-memcached
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 345e56658639ea1b6c68011f7eb431ae9533ff73
Author: koushiro <ko...@gmail.com>
AuthorDate: Tue Oct 21 00:15:45 2025 +0800
refactor: migrate memcached service from adapter::kv to impl Access directly
---
core/src/services/memcached/backend.rs | 182 +++++++++------------
core/src/services/memcached/core.rs | 132 +++++++++++++++
core/src/services/memcached/{mod.rs => deleter.rs} | 28 +++-
core/src/services/memcached/docs.md | 3 +-
core/src/services/memcached/mod.rs | 5 +-
core/src/services/memcached/writer.rs | 59 +++++++
6 files changed, 292 insertions(+), 117 deletions(-)
diff --git a/core/src/services/memcached/backend.rs b/core/src/services/memcached/backend.rs
index 5cdde5a0d..a2562357d 100644
--- a/core/src/services/memcached/backend.rs
+++ b/core/src/services/memcached/backend.rs
@@ -15,14 +15,14 @@
// specific language governing permissions and limitations
// under the License.
+use std::sync::Arc;
use std::time::Duration;
-use bb8::RunError;
-use tokio::net::TcpStream;
use tokio::sync::OnceCell;
-use super::binary;
-use crate::raw::adapters::kv;
+use super::core::*;
+use super::deleter::MemcachedDeleter;
+use super::writer::MemcachedWriter;
use crate::raw::*;
use crate::services::MemcachedConfig;
use crate::*;
@@ -138,11 +138,11 @@ impl Builder for MemcachedBuilder {
);
let conn = OnceCell::new();
- Ok(MemcachedBackend::new(Adapter {
+ Ok(MemcachedBackend::new(MemcachedCore {
+ conn,
endpoint,
username: self.config.username.clone(),
password: self.config.password.clone(),
- conn,
default_ttl: self.config.default_ttl,
})
.with_normalized_root(root))
@@ -150,128 +150,92 @@ impl Builder for MemcachedBuilder {
}
/// Backend for memcached services.
-pub type MemcachedBackend = kv::Backend<Adapter>;
-
#[derive(Clone, Debug)]
-pub struct Adapter {
- endpoint: String,
- username: Option<String>,
- password: Option<String>,
- default_ttl: Option<Duration>,
- conn: OnceCell<bb8::Pool<MemcacheConnectionManager>>,
+pub struct MemcachedBackend {
+ core: Arc<MemcachedCore>,
+ root: String,
+ info: Arc<AccessorInfo>,
}
-impl Adapter {
- async fn conn(&self) -> Result<bb8::PooledConnection<'_, MemcacheConnectionManager>> {
- let pool = self
- .conn
- .get_or_try_init(|| async {
- let mgr = MemcacheConnectionManager::new(
- &self.endpoint,
- self.username.clone(),
- self.password.clone(),
- );
-
- bb8::Pool::builder().build(mgr).await.map_err(|err| {
- Error::new(ErrorKind::ConfigInvalid, "connect to memecached failed")
- .set_source(err)
- })
- })
- .await?;
+impl MemcachedBackend {
+ pub fn new(core: MemcachedCore) -> Self {
+ let info = AccessorInfo::default();
+ info.set_scheme(Scheme::Memcached.into_static());
+ info.set_name("memcached");
+ info.set_root("/");
+ info.set_native_capability(Capability {
+ read: true,
+ stat: true,
+ write: true,
+ write_can_empty: true,
+ delete: true,
+ shared: true,
+ ..Default::default()
+ });
- pool.get().await.map_err(|err| match err {
- RunError::TimedOut => {
- Error::new(ErrorKind::Unexpected, "get connection from pool failed").set_temporary()
- }
- RunError::User(err) => err,
- })
- }
-}
-
-impl kv::Adapter for Adapter {
- type Scanner = ();
-
- fn info(&self) -> kv::Info {
- kv::Info::new(
- Scheme::Memcached,
- "memcached",
- Capability {
- read: true,
- write: true,
- shared: true,
-
- ..Default::default()
- },
- )
- }
-
- async fn get(&self, key: &str) -> Result<Option<Buffer>> {
- let mut conn = self.conn().await?;
- let result = conn.get(&percent_encode_path(key)).await?;
- Ok(result.map(Buffer::from))
+ Self {
+ core: Arc::new(core),
+ root: "/".to_string(),
+ info: Arc::new(info),
+ }
}
- async fn set(&self, key: &str, value: Buffer) -> Result<()> {
- let mut conn = self.conn().await?;
-
- conn.set(
- &percent_encode_path(key),
- &value.to_vec(),
- // Set expiration to 0 if ttl not set.
- self.default_ttl
- .map(|v| v.as_secs() as u32)
- .unwrap_or_default(),
- )
- .await
+ fn with_normalized_root(mut self, root: String) -> Self {
+ self.info.set_root(&root);
+ self.root = root;
+ self
}
+}
- async fn delete(&self, key: &str) -> Result<()> {
- let mut conn = self.conn().await?;
+impl Access for MemcachedBackend {
+ type Reader = Buffer;
+ type Writer = MemcachedWriter;
+ type Lister = ();
+ type Deleter = oio::OneShotDeleter<MemcachedDeleter>;
- conn.delete(&percent_encode_path(key)).await
+ fn info(&self) -> Arc<AccessorInfo> {
+ self.info.clone()
}
-}
-/// A `bb8::ManageConnection` for `memcache_async::ascii::Protocol`.
-#[derive(Clone, Debug)]
-struct MemcacheConnectionManager {
- address: String,
- username: Option<String>,
- password: Option<String>,
-}
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let p = build_abs_path(&self.root, path);
-impl MemcacheConnectionManager {
- fn new(address: &str, username: Option<String>, password: Option<String>) -> Self {
- Self {
- address: address.to_string(),
- username,
- password,
+ if p == build_abs_path(&self.root, "") {
+ Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+ } else {
+ let bs = self.core.get(&p).await?;
+ match bs {
+ Some(bs) => Ok(RpStat::new(
+ Metadata::new(EntryMode::FILE).with_content_length(bs.len() as u64),
+ )),
+ None => Err(Error::new(ErrorKind::NotFound, "kv not found in memcached")),
+ }
}
}
-}
-
-impl bb8::ManageConnection for MemcacheConnectionManager {
- type Connection = binary::Connection;
- type Error = Error;
- /// TODO: Implement unix stream support.
- async fn connect(&self) -> Result<Self::Connection, Self::Error> {
- let conn = TcpStream::connect(&self.address)
- .await
- .map_err(new_std_io_error)?;
- let mut conn = binary::Connection::new(conn);
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let p = build_abs_path(&self.root, path);
+ let bs = match self.core.get(&p).await? {
+ Some(bs) => bs,
+ None => return Err(Error::new(ErrorKind::NotFound, "kv not found in memcached")),
+ };
+ Ok((RpRead::new(), bs.slice(args.range().to_range_as_usize())))
+ }
- if let (Some(username), Some(password)) = (self.username.as_ref(), self.password.as_ref()) {
- conn.auth(username, password).await?;
- }
- Ok(conn)
+ async fn write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ let p = build_abs_path(&self.root, path);
+ Ok((RpWrite::new(), MemcachedWriter::new(self.core.clone(), p)))
}
- async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
- conn.version().await.map(|_| ())
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ Ok((
+ RpDelete::default(),
+ oio::OneShotDeleter::new(MemcachedDeleter::new(self.core.clone(), self.root.clone())),
+ ))
}
- fn has_broken(&self, _: &mut Self::Connection) -> bool {
- false
+ async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
+ let _ = build_abs_path(&self.root, path);
+ Ok((RpList::default(), ()))
}
}
diff --git a/core/src/services/memcached/core.rs b/core/src/services/memcached/core.rs
new file mode 100644
index 000000000..82c21584d
--- /dev/null
+++ b/core/src/services/memcached/core.rs
@@ -0,0 +1,132 @@
+// 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::time::Duration;
+
+use bb8::RunError;
+use tokio::net::TcpStream;
+use tokio::sync::OnceCell;
+
+use super::binary;
+use crate::raw::*;
+use crate::*;
+
+/// A `bb8::ManageConnection` for `memcache_async::ascii::Protocol`.
+#[derive(Clone, Debug)]
+pub struct MemcacheConnectionManager {
+ address: String,
+ username: Option<String>,
+ password: Option<String>,
+}
+
+impl MemcacheConnectionManager {
+ fn new(address: &str, username: Option<String>, password: Option<String>) -> Self {
+ Self {
+ address: address.to_string(),
+ username,
+ password,
+ }
+ }
+}
+
+impl bb8::ManageConnection for MemcacheConnectionManager {
+ type Connection = binary::Connection;
+ type Error = Error;
+
+ /// TODO: Implement unix stream support.
+ async fn connect(&self) -> Result<Self::Connection, Self::Error> {
+ let conn = TcpStream::connect(&self.address)
+ .await
+ .map_err(new_std_io_error)?;
+ let mut conn = binary::Connection::new(conn);
+
+ if let (Some(username), Some(password)) = (self.username.as_ref(), self.password.as_ref()) {
+ conn.auth(username, password).await?;
+ }
+ Ok(conn)
+ }
+
+ async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
+ conn.version().await.map(|_| ())
+ }
+
+ fn has_broken(&self, _: &mut Self::Connection) -> bool {
+ false
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct MemcachedCore {
+ pub conn: OnceCell<bb8::Pool<MemcacheConnectionManager>>,
+ pub endpoint: String,
+ pub username: Option<String>,
+ pub password: Option<String>,
+ pub default_ttl: Option<Duration>,
+}
+
+impl MemcachedCore {
+ async fn conn(&self) -> Result<bb8::PooledConnection<'_, MemcacheConnectionManager>> {
+ let pool = self
+ .conn
+ .get_or_try_init(|| async {
+ let mgr = MemcacheConnectionManager::new(
+ &self.endpoint,
+ self.username.clone(),
+ self.password.clone(),
+ );
+
+ bb8::Pool::builder().build(mgr).await.map_err(|err| {
+ Error::new(ErrorKind::ConfigInvalid, "connect to memecached failed")
+ .set_source(err)
+ })
+ })
+ .await?;
+
+ pool.get().await.map_err(|err| match err {
+ RunError::TimedOut => {
+ Error::new(ErrorKind::Unexpected, "get connection from pool failed").set_temporary()
+ }
+ RunError::User(err) => err,
+ })
+ }
+
+ pub async fn get(&self, key: &str) -> Result<Option<Buffer>> {
+ let mut conn = self.conn().await?;
+ let result = conn.get(&percent_encode_path(key)).await?;
+ Ok(result.map(Buffer::from))
+ }
+
+ pub async fn set(&self, key: &str, value: Buffer) -> Result<()> {
+ let mut conn = self.conn().await?;
+
+ conn.set(
+ &percent_encode_path(key),
+ &value.to_vec(),
+ // Set expiration to 0 if ttl not set.
+ self.default_ttl
+ .map(|v| v.as_secs() as u32)
+ .unwrap_or_default(),
+ )
+ .await
+ }
+
+ pub async fn delete(&self, key: &str) -> Result<()> {
+ let mut conn = self.conn().await?;
+
+ conn.delete(&percent_encode_path(key)).await
+ }
+}
diff --git a/core/src/services/memcached/mod.rs b/core/src/services/memcached/deleter.rs
similarity index 59%
copy from core/src/services/memcached/mod.rs
copy to core/src/services/memcached/deleter.rs
index 4cb6779c7..3bb5e0c35 100644
--- a/core/src/services/memcached/mod.rs
+++ b/core/src/services/memcached/deleter.rs
@@ -15,10 +15,28 @@
// specific language governing permissions and limitations
// under the License.
-mod binary;
+use std::sync::Arc;
-mod backend;
-pub use backend::MemcachedBuilder as Memcached;
+use super::core::MemcachedCore;
+use crate::raw::oio;
+use crate::raw::*;
+use crate::*;
-mod config;
-pub use config::MemcachedConfig;
+pub struct MemcachedDeleter {
+ core: Arc<MemcachedCore>,
+ root: String,
+}
+
+impl MemcachedDeleter {
+ pub fn new(core: Arc<MemcachedCore>, root: String) -> Self {
+ Self { core, root }
+ }
+}
+
+impl oio::OneShotDelete for MemcachedDeleter {
+ async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+ let p = build_abs_path(&self.root, &path);
+ self.core.delete(&p).await?;
+ Ok(())
+ }
+}
diff --git a/core/src/services/memcached/docs.md b/core/src/services/memcached/docs.md
index 40de9df81..b03dba24f 100644
--- a/core/src/services/memcached/docs.md
+++ b/core/src/services/memcached/docs.md
@@ -2,16 +2,15 @@
This service can be used to:
+- [ ] create_dir
- [x] stat
- [x] read
- [x] write
-- [ ] create_dir
- [x] delete
- [ ] copy
- [ ] rename
- [ ] ~~list~~
- [ ] ~~presign~~
-- [ ] blocking
## Configuration
diff --git a/core/src/services/memcached/mod.rs b/core/src/services/memcached/mod.rs
index 4cb6779c7..bd29318fb 100644
--- a/core/src/services/memcached/mod.rs
+++ b/core/src/services/memcached/mod.rs
@@ -15,9 +15,12 @@
// specific language governing permissions and limitations
// under the License.
+mod backend;
mod binary;
+mod core;
+mod deleter;
+mod writer;
-mod backend;
pub use backend::MemcachedBuilder as Memcached;
mod config;
diff --git a/core/src/services/memcached/writer.rs b/core/src/services/memcached/writer.rs
new file mode 100644
index 000000000..bb2e843b3
--- /dev/null
+++ b/core/src/services/memcached/writer.rs
@@ -0,0 +1,59 @@
+// 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::*;
+
+pub struct MemcachedWriter {
+ core: Arc<MemcachedCore>,
+ path: String,
+ buffer: oio::QueueBuf,
+}
+
+impl MemcachedWriter {
+ pub fn new(core: Arc<MemcachedCore>, path: String) -> Self {
+ Self {
+ core,
+ path,
+ buffer: oio::QueueBuf::new(),
+ }
+ }
+}
+
+impl oio::Write for MemcachedWriter {
+ async fn write(&mut self, bs: Buffer) -> Result<()> {
+ self.buffer.push(bs);
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<Metadata> {
+ let buf = self.buffer.clone().collect();
+ let length = buf.len() as u64;
+ self.core.set(&self.path, buf).await?;
+
+ let meta = Metadata::new(EntryMode::from_path(&self.path)).with_content_length(length);
+ Ok(meta)
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ self.buffer.clear();
+ Ok(())
+ }
+}