You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by "jorgehermo9 (via GitHub)" <gi...@apache.org> on 2025/03/28 16:48:32 UTC
[PR] feat: initial cache layer implementation [opendal]
jorgehermo9 opened a new pull request, #5906:
URL: https://github.com/apache/opendal/pull/5906
Closes https://github.com/apache/opendal/issues/5678
I open this as a draft to get some feedback about what I'm doing right now. There are missing tests and documentation, but I would like to get some feedback before working more on this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2040043336
##########
core/src/layers/foyer.rs:
##########
Review Comment:
Note that as I changed `CacheLayer` to `FoyerLayer`, I renamed the file to `foyer.rs` and the feature flag to `layers-foyer`
##########
core/src/layers/foyer.rs:
##########
Review Comment:
Note that as I changed `CacheLayer` to `FoyerLayer`, I renamed the file to `foyer.rs` and the feature flag to `layers-foyer`
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,187 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer {
Review Comment:
Note that as I changed CacheLayer to FoyerLayer, I renamed the file to foyer.rs and the feature flag to layers-foyer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019028218
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ // TODO: lister cache?
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ // TODO: stat cache?
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ // TODO: handle this error
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ // TODO: log that we have a cache hit?
+ return Ok((entry.rp, Box::new(entry.bytes.clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
+ let reader: oio::Reader = Box::new(CacheWrapper::new(
+ reader,
+ rp,
+ Arc::clone(&self.cache),
+ cache_key,
+ ));
+ (rp, reader)
+ })
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ self.inner.list(path, args).await
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.inner.blocking_read(path, args)
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.inner.blocking_write(path, args)
+ }
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
+ self.inner.blocking_list(path, args)
+ }
+}
+
+pub struct CacheWrapper<R> {
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ buffers: Vec<Buffer>,
+}
+
+impl<R> CacheWrapper<R> {
+ fn new(
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ ) -> Self {
+ Self {
+ inner,
+ rp,
+ cache_key,
+ cache,
+ buffers: Vec::new(),
+ }
+ }
+}
+
+impl<R: oio::Read> oio::Read for CacheWrapper<R> {
+ async fn read(&mut self) -> Result<Buffer> {
+ // TODO: ask if we should accumulate the bytes like this
+ // or use self.inner.read_all() instead
+ let buffer = self.inner.read().await?;
+
+ if !buffer.is_empty() {
+ self.buffers.push(buffer.clone());
+ return Ok(buffer);
+ }
+ // TODO: comment that we only insert in cache when we read all the bytes and
+ // `self.inner.read()` returns empty bytes
+ let flattened_buffer: Buffer = self.buffers.drain(..).flatten().collect();
+
+ let cache_value = CacheValue {
+ rp: self.rp,
+ bytes: flattened_buffer.to_bytes(),
Review Comment:
Should we store `Bytes` or `Buffer`? I think the best would be to store in cache *contiguous bytes* with `Bytes` and convert Non-contiguous bytes of `Buffer` to contiguous `Bytes` with this method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044941804
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ // TODO: lister cache?
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ // TODO: stat cache?
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ // TODO: handle this error
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ // TODO: log that we have a cache hit?
+ return Ok((entry.rp, Box::new(entry.bytes.clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
+ let reader: oio::Reader = Box::new(CacheWrapper::new(
+ reader,
+ rp,
+ Arc::clone(&self.cache),
+ cache_key,
+ ));
+ (rp, reader)
+ })
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ self.inner.list(path, args).await
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.inner.blocking_read(path, args)
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.inner.blocking_write(path, args)
+ }
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
+ self.inner.blocking_list(path, args)
+ }
+}
+
+pub struct CacheWrapper<R> {
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ buffers: Vec<Buffer>,
+}
+
+impl<R> CacheWrapper<R> {
+ fn new(
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ ) -> Self {
+ Self {
+ inner,
+ rp,
+ cache_key,
+ cache,
+ buffers: Vec::new(),
+ }
+ }
+}
+
+impl<R: oio::Read> oio::Read for CacheWrapper<R> {
+ async fn read(&mut self) -> Result<Buffer> {
+ // TODO: ask if we should accumulate the bytes like this
+ // or use self.inner.read_all() instead
+ let buffer = self.inner.read().await?;
+
+ if !buffer.is_empty() {
+ self.buffers.push(buffer.clone());
Review Comment:
> Try handling simple files in small chunks; we can start with fixed-size chunks.
Could you explain further on that? I can't understand which handling are you refering to
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045182590
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
+ version: Option<String>,
+}
+
+impl CacheKey {
+ fn new(path: &str, args: &OpRead) -> CacheKey {
+ CacheKey {
+ path: path.to_string(),
+ range: args.range(),
+ if_match: args.if_match().map(ToString::to_string),
+ if_none_match: args.if_none_match().map(ToString::to_string),
+ if_modified_since: args.if_modified_since(),
+ if_unmodified_since: args.if_unmodified_since(),
+ version: args.version().map(ToString::to_string),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct CacheValue(Buffer);
+
+impl Code for CacheValue {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ writer.write_vectored(&self.0.to_io_slice())?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
+ Ok(CacheValue(Buffer::from(buf)))
+ }
+
+ fn estimated_size(&self) -> usize {
+ self.0.len()
+ }
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ type Reader = TwoWays<Buffer, CacheWrapper<A::Reader>>;
+
+ type Writer = A::Writer;
+
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey::new(path, &args);
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
Review Comment:
Should we not propagate the error and instead log it & continue reading from `self.inner.read()`? So cache errors does not affect the whole operation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045917944
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
+ version: Option<String>,
+}
+
+impl CacheKey {
+ fn new(path: &str, args: &OpRead) -> CacheKey {
+ CacheKey {
+ path: path.to_string(),
+ range: args.range(),
+ if_match: args.if_match().map(ToString::to_string),
+ if_none_match: args.if_none_match().map(ToString::to_string),
+ if_modified_since: args.if_modified_since(),
+ if_unmodified_since: args.if_unmodified_since(),
+ version: args.version().map(ToString::to_string),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct CacheValue(Buffer);
+
+impl Code for CacheValue {
Review Comment:
There is no deserialize for `CacheKey`. The cache just try to read the key we input. And `CacheValue` itself is a plain `Buffer` which contains the raw content.
I think we don't need to concern about the back-compatible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045916957
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
+ version: Option<String>,
+}
+
+impl CacheKey {
+ fn new(path: &str, args: &OpRead) -> CacheKey {
+ CacheKey {
+ path: path.to_string(),
+ range: args.range(),
+ if_match: args.if_match().map(ToString::to_string),
+ if_none_match: args.if_none_match().map(ToString::to_string),
+ if_modified_since: args.if_modified_since(),
+ if_unmodified_since: args.if_unmodified_since(),
+ version: args.version().map(ToString::to_string),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct CacheValue(Buffer);
+
+impl Code for CacheValue {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ writer.write_vectored(&self.0.to_io_slice())?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
+ Ok(CacheValue(Buffer::from(buf)))
+ }
+
+ fn estimated_size(&self) -> usize {
+ self.0.len()
+ }
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ type Reader = TwoWays<Buffer, CacheWrapper<A::Reader>>;
+
+ type Writer = A::Writer;
+
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey::new(path, &args);
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
Review Comment:
Makes sense to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020224749
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
Review Comment:
We don't need to do this, the final users that built the `bin` should take care.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019049553
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,187 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
Review Comment:
I think we need to cache both `RpRead` and the bytes from the reader. This would allow us to early return here https://github.com/apache/opendal/blob/22eec12fde9880d1dc88babdf72c78af6a1c64ad/core/src/layers/cache.rs#L120 and to now call `self.inner.read()` to get the `RpRead` https://github.com/apache/opendal/blob/22eec12fde9880d1dc88babdf72c78af6a1c64ad/core/src/layers/cache.rs#L123
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044915017
##########
core/Cargo.toml:
##########
@@ -233,10 +240,11 @@ anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
-bytes = "1.6"
+bytes = { version = "1.6", features = ["serde"] }
chrono = { version = "0.4.28", default-features = false, features = [
"clock",
"std",
+ "serde",
Review Comment:
we can't `impl Serialize for DateTime` because both `Serialize` and `Datetime` are outside of this crate.
What we could do, is to not `derive(Serialize)` for `CacheKey` and manually implement `impl Serialize for CacheKey`, and serializing each one of the fields manually.
I think it is a mistake to do this manual implementation instead of relying in `derive(Serialize,Deserialize)`. I think we should rely on `serde` feature of `chrono` to provide the `DateTime` implementations
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2046861354
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,249 @@
+// 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 crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache};
+use log::warn;
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<String, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(
+ cache: HybridCache<String, CacheValue>,
+ max_value_size: usize,
+ ) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ max_value_size: self.max_value_size,
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct CacheValue(Buffer);
+
+impl Code for CacheValue {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ writer.write_vectored(&self.0.to_io_slice())?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
+ Ok(CacheValue(Buffer::from(buf)))
+ }
+
+ fn estimated_size(&self) -> usize {
+ self.0.len()
+ }
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ cache: Arc<HybridCache<String, CacheValue>>,
+ max_value_size: usize,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ type Reader = TwoWays<Buffer, CacheWrapper<A::Reader>>;
+
+ type Writer = A::Writer;
+
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = build_cache_key(path, &args);
+
+ match self.cache.get(&cache_key).await {
+ Ok(Some(entry)) => return Ok((RpRead::default(), TwoWays::One(entry.0.clone()))),
+ Ok(None) => (),
+ Err(error) => warn!("cache access failed for key {cache_key}: {error}"),
+ };
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
+ let reader = TwoWays::Two(CacheWrapper::new(
+ reader,
+ Arc::clone(&self.cache),
+ cache_key,
+ self.max_value_size,
+ ));
+ (rp, reader)
+ })
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ self.inner.list(path, args).await
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.inner.blocking_read(path, args)
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.inner.blocking_write(path, args)
+ }
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
+ self.inner.blocking_list(path, args)
+ }
+}
+
+pub struct CacheWrapper<R> {
+ inner: R,
+ cache: Arc<HybridCache<String, CacheValue>>,
+ cache_key: String,
+ buffers: Vec<Buffer>,
+ max_value_size: usize,
+ current_value_size: usize,
+}
+
+impl<R> CacheWrapper<R> {
+ fn new(
+ inner: R,
+ cache: Arc<HybridCache<String, CacheValue>>,
+ cache_key: String,
+ max_value_size: usize,
+ ) -> Self {
+ Self {
+ inner,
+ cache_key,
+ cache,
+ buffers: Vec::new(),
+ max_value_size,
+ current_value_size: 0,
+ }
+ }
+}
+
+impl<R: oio::Read> oio::Read for CacheWrapper<R> {
+ async fn read(&mut self) -> Result<Buffer> {
+ let buffer = self.inner.read().await?;
+ self.current_value_size += buffer.len();
+
+ if self.current_value_size > self.max_value_size {
+ return Ok(buffer);
+ }
+
+ if !buffer.is_empty() {
+ self.buffers.push(buffer.clone());
+ return Ok(buffer);
+ }
+ let flattened_buffer: Buffer = self.buffers.drain(..).flatten().collect();
+
+ self.cache
+ .insert(self.cache_key.clone(), CacheValue(flattened_buffer));
+
+ Ok(buffer)
+ }
+}
+
+fn build_cache_key(path: &str, args: &OpRead) -> String {
+ let version = args.version().unwrap_or_default();
+ let range = args.range();
+ format!("{path}-{version}-{range}")
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_build_cache_key_with_default_args() {
Review Comment:
I wonder if it would be better to just have one parametrized test with all these cases instead of 1 test for each case. What is the current approach in the codebase?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019015877
##########
core/Cargo.toml:
##########
@@ -173,6 +175,12 @@ services-obs = [
"reqsign?/reqwest_request",
]
services-onedrive = []
+services-opfs = [
Review Comment:
Autoformatter moved this. Resolve once this comment is seen.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019018134
##########
core/Cargo.toml:
##########
@@ -233,10 +240,11 @@ anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
-bytes = "1.6"
+bytes = { version = "1.6", features = ["serde"] }
chrono = { version = "0.4.28", default-features = false, features = [
"clock",
"std",
+ "serde",
Review Comment:
The same as `Bytes`. `OpRead` contained `chrono::DateTime` and I had to use `OpRead` as the cache key, which needed `DateTime` to implement `Serialize` and `Deserialize`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020225405
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,206 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
Review Comment:
Agreed. Let's focus on reading first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019026131
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
Review Comment:
Should we expose foyer metrics? https://foyer.rs/docs/tutorial/monitor#2-setup-metrics-monitoring
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019049553
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,187 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
Review Comment:
I think we need to cache both `RpRead` and the bytes from the reader. This would allow us to early return here https://github.com/apache/opendal/blob/22eec12fde9880d1dc88babdf72c78af6a1c64ad/core/src/layers/cache.rs#L120 and to now call `self.inner.read()` to get the `RpRead` https://github.com/apache/opendal/blob/22eec12fde9880d1dc88babdf72c78af6a1c64ad/core/src/layers/cache.rs#L123.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019037549
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,206 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
Review Comment:
using `oio::Reader` because we return two different structs that implement `Read`
https://github.com/apache/opendal/blob/22eec12fde9880d1dc88babdf72c78af6a1c64ad/core/src/layers/cache.rs#L120
and
https://github.com/apache/opendal/blob/22eec12fde9880d1dc88babdf72c78af6a1c64ad/core/src/layers/cache.rs#L124
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019041648
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,187 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
Review Comment:
as commented [here](https://github.com/apache/opendal/issues/5857#issuecomment-2746663073) we will only implement cache for `async` operations, since foyer's hybrid cache is not sync-friendly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019046616
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,187 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
Review Comment:
I think we should use both `path` and `OpRead` as the `CacheKey`. Using `OpRead` directly instead of a new struct declared here and mapping from `OpRead` to that struct will ensure that new fields added in the future to `OpRead` will be always taken into account in the `CacheKey`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2782266500
> If you are willing, I can take over this PR and continue with the adaptation.
I would like to continue with it if possible, I have more time this ongoing week to push this further, if everyone is okay with it...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "MrCroxx (via GitHub)" <gi...@apache.org>.
MrCroxx commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2782898126
> > Hi, @MrCroxx, I believe it's better to cut off a release now so that @jorgehermo9 can work on the base without serde directly.
>
> Sure. I'll release a new version in hours.
Done. foyer v0.16.0 is released.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "meteorgan (via GitHub)" <gi...@apache.org>.
meteorgan commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2782689163
> Yes, we could do that in the `Operator` that does the `write` operation, but files in storages can be modified by external sources (or other process in the same machine) and I don't know if it makes sense to just evict the cache key on the writer, other readers will have the same problem and have not easy way to detect those changes. I don't know how to proceed with evictions here. We could take a look to other implementations such as RisingWave's and see what they do regarding to cache invalidation
I completely agree with you, these are indeed tricky scenarios. I think we could invalidate the cache during `write` and `delete` operations, ensuring that as least the same process can always read the most recent data it has written. For the data modified by other processes, the simplest solution might be to evict data based on resident time, which users could configure. However, it would be wise to investigate further and reference to other implementations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044847870
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
Review Comment:
That's because the generic type `S` of `HybridCacheBuilderPhaseStorage` needs the `HashBuilder` bound. If we receive `HybridCache` instead of the builder (https://github.com/apache/opendal/pull/5906#discussion_r2044381889), we can remove this bound
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044904499
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl Code for Buffer {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ let mut reader = self.clone();
Review Comment:
Nice catch. I didn't know that `std::io::copy` wrapped the reader into a buffered reader. Its surprinsing that rust does not provide some utility to just pipe a reader into a writer with no extra allocations.
What do you think of using this instead of what you suggested:
```rust
writer.write_vectored(&self.0.to_io_slice())?;
```
If that doesn't looks good to you, I'll change to your suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044924465
##########
core/src/layers/mod.rs:
##########
@@ -117,6 +117,15 @@ mod dtrace;
#[cfg(all(target_os = "linux", feature = "layers-dtrace"))]
pub use self::dtrace::DtraceLayer;
+#[cfg(feature = "layers-foyer")]
+mod foyer;
+#[cfg(feature = "layers-foyer")]
+pub use self::foyer::CacheKey;
+#[cfg(feature = "layers-foyer")]
+pub use self::foyer::CacheValue;
Review Comment:
I have to expose both `CacheKey` and `CacheValue` because they are part of public API (generic params of the `HybridCache` we receive in `FoyerLayer::new`)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045914533
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
Review Comment:
No, we don't need to include them in cache key generation. The backend will simply return an error when the condition check fails. So basically, we only need to consider `path`, `version`, and `range`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045915930
##########
core/src/layers/mod.rs:
##########
@@ -117,6 +117,15 @@ mod dtrace;
#[cfg(all(target_os = "linux", feature = "layers-dtrace"))]
pub use self::dtrace::DtraceLayer;
+#[cfg(feature = "layers-foyer")]
+mod foyer;
+#[cfg(feature = "layers-foyer")]
+pub use self::foyer::CacheKey;
+#[cfg(feature = "layers-foyer")]
+pub use self::foyer::CacheValue;
Review Comment:
Negative. I don't want to expose anything other than `FoyerLayer`. The `HybridCache` should be `HybridCache<String, Buffer>`, which means we don't need any additional types. It's the user's responsibility to use compatible versions of Foyer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2046849503
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
Review Comment:
Ok.
Just to note, this could lead into situations where, for example, foyer introduces breaking changes in `0.17.0` version.
Opendal uses `0.16.0` and user's binary crate uses `0.17.0`. If opendal does not export its `0.16.0` foyer and the user uses its `0.17.0` foyer, then a `0.17.0` `HybridCache` (with breaking changes) will be passed to Opendal (which expects an `HybridCache` from version 0.16.0`)
This situation could lead to obscure compilation errors. Please, read carefully the example at https://www.lurklurk.org/effective-rust/re-export.html .
> as this would make the external crate part of opendal's public API, which we should avoid.
Unfortunately, we are already exposing `HybridCache` in our public API and the only options I see is to
1. export opendal's foyer dep as public api
2. remove `HybridCache` from our public api. This will complicate things a little bit, and there are *a lot* of things to consider in the case we do not expose the `HybridCache` in the `new` method (i.e. cache metrics, storage options, etc)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044934355
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl Code for Buffer {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ let mut reader = self.clone();
+ std::io::copy(&mut reader, writer).map_err(CodeError::Io)?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
+ Ok(Buffer::from(buf))
+ }
+
+ fn estimated_size(&self) -> usize {
+ self.len()
+ }
+}
+
+impl<A, S> LayeredAccess for CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ return Ok((RpRead::default(), Box::new(entry.deref().clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
Review Comment:
I have doubts about this. If someone tries to read the whole file and uses `BytesRange::from(..)`, we can't know at this point the size of the file and we can't rely purely in `OpRead`, we must check that *while* reading the file
What we could do is to add that check here https://github.com/apache/opendal/blob/53ace144ef51e09d2f715bf4cec2adf320f9f2e5/core/src/layers/foyer.rs#L185
and count how many bytes we have read (and stop buffering them in the `Vec<Buffer>` array if the threshold is exceeded) before inserting the cache value.
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl Code for Buffer {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ let mut reader = self.clone();
+ std::io::copy(&mut reader, writer).map_err(CodeError::Io)?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
+ Ok(Buffer::from(buf))
+ }
+
+ fn estimated_size(&self) -> usize {
+ self.len()
+ }
+}
+
+impl<A, S> LayeredAccess for CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ return Ok((RpRead::default(), Box::new(entry.deref().clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
Review Comment:
I have doubts about this. If someone tries to read the whole file and uses `BytesRange::from(..)`, we can't know at this point the size of the file and we can't rely purely in `OpRead`, we must check that *while* reading the file
What we could do is to add that check here https://github.com/apache/opendal/blob/53ace144ef51e09d2f715bf4cec2adf320f9f2e5/core/src/layers/foyer.rs#L185
and count how many bytes we have read (and stop buffering them in the `Vec<Buffer>` array if the threshold is exceeded) before inserting the cache value.
What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2046685710
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
Review Comment:
I don't think this is a case we can handle at the cache layer. Including `if_modified_since` in the cache key would result in constant cache misses. For example, users are reading data with `op.read_with(path).if_modified_since(now-Duration::from_secs(3600))`. If users are concerned about whether the data has been modified, the best thing the cache layer can do is simply bypass the cache.
In the future, we could add some cache metadata, including `last_modified` and `etag`, so that we can attempt to serve cached content even when users use `if_modified_since`. However, this should be handled within the cache metadata rather than the cache key. This means we would still use the same cache key to fetch the content; the only difference would be whether or not we validate the content.
I want to emphasize that the target users of the cache layer typically work with immutable files, so it is rare for them to update the same file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2046689203
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
Review Comment:
> This means we would still use the same cache key to fetch the content; the only difference would be whether or not we validate the content
That makes a lot of sense now. Sorry, I didn't know in which contexts will this layer be used. I will address all these comments asap!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 closed pull request #5906: feat: Initial foyer layer implementation
URL: https://github.com/apache/opendal/pull/5906
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019035940
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,206 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ // TODO: lister cache?
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ // TODO: stat cache?
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ // TODO: handle this error
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
Review Comment:
should we `log::debug!` that we had a cache hit?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019025059
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
Review Comment:
Rename to `FoyerLayer` or `FoyerCacheLayer`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020176694
##########
core/Cargo.toml:
##########
@@ -399,6 +406,8 @@ prometheus-client = { version = "0.23.1", optional = true }
tracing = { version = "0.1", optional = true }
# for layers-dtrace
probe = { version = "0.5.1", optional = true }
+# for layers-cache
+foyer = { version = "0.14.1", optional = true }
Review Comment:
thanks for noticing! will update the version asap!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020224549
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
Review Comment:
I'm open to have `FoyerLayer` to accept an `HybridCacheBuilder`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "MrCroxx (via GitHub)" <gi...@apache.org>.
MrCroxx commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2774594612
> What I mean is, if a key gets deleted or updated, we should refresh the cache. Otherwise, we'll end up with stale data.
That's true. However, foyer has a lot of room for optimization when handling deletion and updatable data for now. Currently, in almost all user scenarios in Foyer, deleted data is not read, and existing data is not updated. Although Foyer has already made relevant designs and optimizations (e.g., tombstone log to ensure that delete is also append-only), it is not commonly used, so it still needs to be validated and optimized.
Therefore, for scenarios that require access to deleted data or updatable needs, I think Foyer should not be considered for use in the production environment for the time being.
However, based on my current experience, such scenarios are not very common.
Moreover, other similar cache systems also have some corner cases or performance issues in terms of consistency. We can discuss together how to improve it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019044774
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,187 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ return Ok((entry.rp, Box::new(entry.bytes.clone())));
Review Comment:
It is necessary to clone bytes with `entry.bytes.clone()` since foyer's `CacheEntry` only returns references to the `CacheValue` and does not allows to get the owned data
https://docs.rs/foyer/latest/foyer/enum.CacheEntry.html#method.value
https://docs.rs/foyer/latest/foyer/enum.CacheEntry.html#method.deref
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019044774
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,187 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ return Ok((entry.rp, Box::new(entry.bytes.clone())));
Review Comment:
It is necessary to `entry.bytes.clone()` since foyer's `CacheEntry` only returns references to the `CacheValue` and does not allows to get the owned data
https://docs.rs/foyer/latest/foyer/enum.CacheEntry.html#method.value
https://docs.rs/foyer/latest/foyer/enum.CacheEntry.html#method.deref
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020225356
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,206 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
Review Comment:
Please use `opendal::raw::TwoWays`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020224611
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
Review Comment:
Yes, I think `FoyerLayer` is good enough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019024779
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
Review Comment:
I have some doubts about the parameters of this function. Should we receive `HybridCacheBuilder` in the `new` method args? This would allow users to fully customize their cache settings.
Taking `Risingwave` as an example, there are *a lot* of customizable settings https://github.com/risingwavelabs/risingwave/blob/db881a877e4a0d9bb886695d8f78b715d016a1c6/src/storage/src/store_impl.rs#L687
something like:
```
pub async fn new (cache_builder: HybridCacheBuilder){
// ...
}
```
What do you think?
The problem is, I'm hesitant of exposing foyer's structs in opendal's public API, what are your thoughts about this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2781852735
> I don't know how to proceed with evictions here.
It's fine to just ignore evictions if foyer doesn't design for that so far. We just need to make it clear of FoyerLayer's API docs to make sure users know about this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2781859290
Changed PR title to `foyer layer` instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2046653626
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
Review Comment:
I think don't including those in the key is incorrect. I will depict the situation with two examples and the `if_modified_since`
# Situation
we have a file `blob.bin` that has a modication date of `01-01-2025`
# Example 1 - Including `if_modified_since` in the cache key
We first do a read request to `blob.bin` with `if_modified_since = None` -> We read `blob.bin` bytes and everything is OK
Then, we read `blob.bin` again with `if_modified_since= 02-02-2025` -> As we included the `if_modified_since` in the cache key, and it is different from the first call (`None` vs `02-02-2025`), we call the backend again and we get an error as response, because the modification date is earlier than the `if_modified_since` value.
# Example 2 - Not including `if_modified_since` in the cache key
We first do a read request to `blob.bin` with `if_modified_since = None` -> We read `blob.bin` bytes and everything is OK
Then, we read `blob.bin` again with `if_modified_since= 02-02-2025` -> As we did not include `if_modified_since` in the cache key, we have a cache hit here and we return the result from the previous operation -> the `blob.bin` bytes. This would not be correct, because we are returning a file with an earlier modification date than the `if_modified_since` value.
-----
The second example is expected behaviour? The `if_modified_since` parameter is not working as the user would expect in my opinion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045201674
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
+ version: Option<String>,
+}
+
+impl CacheKey {
+ fn new(path: &str, args: &OpRead) -> CacheKey {
+ CacheKey {
+ path: path.to_string(),
+ range: args.range(),
+ if_match: args.if_match().map(ToString::to_string),
+ if_none_match: args.if_none_match().map(ToString::to_string),
+ if_modified_since: args.if_modified_since(),
+ if_unmodified_since: args.if_unmodified_since(),
+ version: args.version().map(ToString::to_string),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct CacheValue(Buffer);
+
+impl Code for CacheValue {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ writer.write_vectored(&self.0.to_io_slice())?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
Review Comment:
Is it okay to read all contents from reader at once? I don't know how to avoid this copy.
Is there any way to wrap the `reader` we receive in the parameters and return it instead of copying to an intermediate buffer? Or to at least lazily read it on-demand and not beforehand
I'm afraid of https://github.com/apache/opendal/pull/5906#discussion_r2020225245
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044849654
##########
core/Cargo.toml:
##########
@@ -406,6 +410,9 @@ prometheus-client = { version = "0.23.1", optional = true }
tracing = { version = "0.1", optional = true }
# for layers-dtrace
probe = { version = "0.5.1", optional = true }
+# for layers-cache
+foyer = { version = "0.16.1", features = ["serde"], optional = true }
+foyer-common = "0.9.5"
Review Comment:
with this change https://github.com/apache/opendal/pull/5906#discussion_r2044847870 we dont need that package anymore
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044924465
##########
core/src/layers/mod.rs:
##########
@@ -117,6 +117,15 @@ mod dtrace;
#[cfg(all(target_os = "linux", feature = "layers-dtrace"))]
pub use self::dtrace::DtraceLayer;
+#[cfg(feature = "layers-foyer")]
+mod foyer;
+#[cfg(feature = "layers-foyer")]
+pub use self::foyer::CacheKey;
+#[cfg(feature = "layers-foyer")]
+pub use self::foyer::CacheValue;
Review Comment:
I have to expose both `CacheKey` and `CacheValue` because they are part of public API (generic params of the `HybridCache` we receive in `FoyerLayer::new`)
Should we expose those two this way, or do `pub mod foyer` instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044952355
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
Review Comment:
As we are exposing public API from other crate, I think we should re-export the foyer crate opendal is using internally.
something like `pub use foyer`. People should use something like `use opendal::exports::foyer::HybridCacheBuilder` instead of their own foyer dependency `use foyer::HybridCacheBuilder` because I think otherwise it could cause dependency versioning pain.
See https://www.lurklurk.org/effective-rust/re-export.html for more details
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044213260
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,206 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
Review Comment:
great utility, thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045921097
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
+ version: Option<String>,
+}
+
+impl CacheKey {
+ fn new(path: &str, args: &OpRead) -> CacheKey {
+ CacheKey {
+ path: path.to_string(),
+ range: args.range(),
+ if_match: args.if_match().map(ToString::to_string),
+ if_none_match: args.if_none_match().map(ToString::to_string),
+ if_modified_since: args.if_modified_since(),
+ if_unmodified_since: args.if_unmodified_since(),
+ version: args.version().map(ToString::to_string),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct CacheValue(Buffer);
+
+impl Code for CacheValue {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ writer.write_vectored(&self.0.to_io_slice())?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
Review Comment:
> Is it okay to read all contents from reader at once? I don't know how to avoid this copy.
That’s why we need to limit the cache’s read size. The current cache layer doesn’t handle chunking for users; it’s up to users to ensure the read size is optimal.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045915002
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
Review Comment:
The same as https://github.com/apache/opendal/pull/5906#discussion_r2045914533
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2809478532
Hi @Xuanwo, I think I addressed everything (left some comments in a few threads). I plan to address documentation & tests once we are confident about how the final implementation looks like
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2782621325
Hi, @MrCroxx, I believe it's better to cut off a release now so that @jorgehermo9 can work on the base without serde directly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "yihong0618 (via GitHub)" <gi...@apache.org>.
yihong0618 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019772258
##########
core/Cargo.toml:
##########
@@ -399,6 +406,8 @@ prometheus-client = { version = "0.23.1", optional = true }
tracing = { version = "0.1", optional = true }
# for layers-dtrace
probe = { version = "0.5.1", optional = true }
+# for layers-cache
+foyer = { version = "0.14.1", optional = true }
Review Comment:
FYI foyer update its version to 0.15.3 now with some improvement and bug fix
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020225301
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,206 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ // TODO: lister cache?
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ // TODO: stat cache?
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ // TODO: handle this error
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
Review Comment:
Better not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020225604
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,187 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
Review Comment:
The cache key could be too large. I suggest we only take required parts as cache key. Most metadata in `OpRead` have not effect on returned data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "meteorgan (via GitHub)" <gi...@apache.org>.
meteorgan commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2765003696
> @meteorgan I think there is no concept of TTL in foyer. You specify a max size in bytes of the cache, and it will be filled until the cache is full. If a new key is going yo be inserted with the cache being full, another entry is evicted (see https://foyer.rs/docs/tutorial/in-memory-cache#22-count-entries-by-weight)
What I mean is, if a key gets deleted or updated, we should refresh the cache. Otherwise, we'll end up with stale data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "MrCroxx (via GitHub)" <gi...@apache.org>.
MrCroxx commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2774598687
Btw, foyer will not require serde as a must after v0.16.x (not released yet, but ready to release anytime.
I plan to do more refactoring in v0.16 to avoid frequently breaking existing user dependencies. However, if it is a blocker for this PR, please feel free to contact me for the release. 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045916670
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
Review Comment:
I don't want to re-export the crate from opendal, as this would make the external crate part of opendal's public API, which we should avoid.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2046659447
##########
core/src/layers/mod.rs:
##########
@@ -117,6 +117,15 @@ mod dtrace;
#[cfg(all(target_os = "linux", feature = "layers-dtrace"))]
pub use self::dtrace::DtraceLayer;
+#[cfg(feature = "layers-foyer")]
+mod foyer;
+#[cfg(feature = "layers-foyer")]
+pub use self::foyer::CacheKey;
+#[cfg(feature = "layers-foyer")]
+pub use self::foyer::CacheValue;
Review Comment:
I think we could do that for the `CacheKey`, but the cache value can't be `Buffer` because of what you suggested here
As `Code` should not be implemented for raw `Buffer`, the cache value should be the `CacheValue` wrapper. We can't have `Buffer` as the cache value because it does not implement `Code` (`CacheValue` does).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2047230531
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
Review Comment:
> Opendal uses `0.16.0` and user's binary crate uses `0.17.0`. If opendal does not export its `0.16.0` foyer and the user uses its `0.17.0` foyer, then a `0.17.0` `HybridCache` (with breaking changes) will be passed to Opendal (which expects an `HybridCache` from version 0.16.0`)
It's expected. Users must use compatible `foyer` that `FoyerLayer` uses. The same rule applies to all other layers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044915017
##########
core/Cargo.toml:
##########
@@ -233,10 +240,11 @@ anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
-bytes = "1.6"
+bytes = { version = "1.6", features = ["serde"] }
chrono = { version = "0.4.28", default-features = false, features = [
"clock",
"std",
+ "serde",
Review Comment:
we can't `impl Serialize for DateTime` because both `Serialize` and `Datetime` are outside of this crate.
What we could do, is to not `derive(Serialize)` for `CacheKey` and manually implement `impl Serialize for CacheKey`, and serializing each one of the fields manually.
I think it is a mistake to do this manual implementation instead of relying in `derive(Serialize,Deserialize)`. I think we should rely on `serde` feature of `chrono` to provide the `DateTime` (de)serialization implementations
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2806421884
Hi @Xuanwo, I addressed all the comments I could, but left doubts on some, could you revisit that?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044919963
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
Review Comment:
Please, note that as we don't have `OpRead` anymore here, I fear that if a new parameter that changes the response is added to `OpRead` we forgot to include it in the `CacheKey` and then the cache will misfunction. I previously embedded `OpRead` here to avoid that type of mistakes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044910995
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
Review Comment:
I think those 4 parameters should be included in the CacheKey, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045192053
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
+ version: Option<String>,
+}
+
+impl CacheKey {
+ fn new(path: &str, args: &OpRead) -> CacheKey {
+ CacheKey {
+ path: path.to_string(),
+ range: args.range(),
+ if_match: args.if_match().map(ToString::to_string),
+ if_none_match: args.if_none_match().map(ToString::to_string),
+ if_modified_since: args.if_modified_since(),
+ if_unmodified_since: args.if_unmodified_since(),
+ version: args.version().map(ToString::to_string),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct CacheValue(Buffer);
+
+impl Code for CacheValue {
Review Comment:
I wonder @MrCroxx how should we take care of `CacheKey` and `CacheValue` evolution throught time. Should we take care of breaking changes such including a new mandatory field in the `CacheKey` struct?
For example, if we add a new mandatory field to `CacheKey`, previous version's keys in disk cache will fail to deserialize? How should we do things here? We must always do back-compatible changes to both `CacheKey` and `CacheValue`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019018134
##########
core/Cargo.toml:
##########
@@ -233,10 +240,11 @@ anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
-bytes = "1.6"
+bytes = { version = "1.6", features = ["serde"] }
chrono = { version = "0.4.28", default-features = false, features = [
"clock",
"std",
+ "serde",
Review Comment:
The same as `Bytes`. `OpRead` contained `chrono::DateTime` and I had to use `OpRead` as the cache key, which needed to `DateTime` to implement `Serialize` and `Deserialize`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2764596434
I think there is no concept of TTL in foyer. You specify a max size in bytes of the cache, and it will be filled until the cache is full. If a new key is going yo be inserted with the cache being full, another entry is evicted (see https://foyer.rs/docs/tutorial/in-memory-cache#22-count-entries-by-weight)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020224488
##########
core/Cargo.toml:
##########
@@ -233,10 +240,11 @@ anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
-bytes = "1.6"
+bytes = { version = "1.6", features = ["serde"] }
chrono = { version = "0.4.28", default-features = false, features = [
"clock",
"std",
+ "serde",
Review Comment:
Please implement serialize in `FoyerLayer` instead. For example, convert `DateTime` into an `i64`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "MrCroxx (via GitHub)" <gi...@apache.org>.
MrCroxx commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2782300656
> > If you are willing, I can take over this PR and continue with the adaptation.
>
> I would like to continue with this if possible, I have more time this ongoing week to push this further, if everyone is okay with it...
No problem. And feel free to ask for any help from my side. 🥰
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2781674674
Hi,
I had been busy these days and couldn't work on this further. I will resume my work as soon as I can, thank everyone for their review & comments!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019030835
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ // TODO: lister cache?
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ // TODO: stat cache?
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ // TODO: handle this error
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ // TODO: log that we have a cache hit?
+ return Ok((entry.rp, Box::new(entry.bytes.clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
+ let reader: oio::Reader = Box::new(CacheWrapper::new(
+ reader,
+ rp,
+ Arc::clone(&self.cache),
+ cache_key,
+ ));
+ (rp, reader)
+ })
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ self.inner.list(path, args).await
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.inner.blocking_read(path, args)
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.inner.blocking_write(path, args)
+ }
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
+ self.inner.blocking_list(path, args)
+ }
+}
+
+pub struct CacheWrapper<R> {
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ buffers: Vec<Buffer>,
+}
+
+impl<R> CacheWrapper<R> {
+ fn new(
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ ) -> Self {
+ Self {
+ inner,
+ rp,
+ cache_key,
+ cache,
+ buffers: Vec::new(),
+ }
+ }
+}
+
+impl<R: oio::Read> oio::Read for CacheWrapper<R> {
+ async fn read(&mut self) -> Result<Buffer> {
+ // TODO: ask if we should accumulate the bytes like this
+ // or use self.inner.read_all() instead
+ let buffer = self.inner.read().await?;
+
+ if !buffer.is_empty() {
+ self.buffers.push(buffer.clone());
Review Comment:
We accumulate the read bytes until `inner.read()` returns empty bytes, then we flush the accumulated bytes to cache. This is an existing pattern in the codebase, inspired from https://github.com/apache/opendal/blob/a0dd08090a9684badee8f335b93578d2458a1570/core/src/raw/oio/read/api.rs#L49
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019032598
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ // TODO: lister cache?
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ // TODO: stat cache?
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ // TODO: handle this error
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ // TODO: log that we have a cache hit?
+ return Ok((entry.rp, Box::new(entry.bytes.clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
+ let reader: oio::Reader = Box::new(CacheWrapper::new(
+ reader,
+ rp,
+ Arc::clone(&self.cache),
+ cache_key,
+ ));
+ (rp, reader)
+ })
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ self.inner.list(path, args).await
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.inner.blocking_read(path, args)
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.inner.blocking_write(path, args)
+ }
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
+ self.inner.blocking_list(path, args)
+ }
+}
+
+pub struct CacheWrapper<R> {
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ buffers: Vec<Buffer>,
+}
+
+impl<R> CacheWrapper<R> {
+ fn new(
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ ) -> Self {
+ Self {
+ inner,
+ rp,
+ cache_key,
+ cache,
+ buffers: Vec::new(),
+ }
+ }
+}
+
+impl<R: oio::Read> oio::Read for CacheWrapper<R> {
+ async fn read(&mut self) -> Result<Buffer> {
+ // TODO: ask if we should accumulate the bytes like this
+ // or use self.inner.read_all() instead
+ let buffer = self.inner.read().await?;
+
+ if !buffer.is_empty() {
+ self.buffers.push(buffer.clone());
+ return Ok(buffer);
+ }
+ // TODO: comment that we only insert in cache when we read all the bytes and
+ // `self.inner.read()` returns empty bytes
+ let flattened_buffer: Buffer = self.buffers.drain(..).flatten().collect();
+
+ let cache_value = CacheValue {
+ rp: self.rp,
+ bytes: flattened_buffer.to_bytes(),
+ };
+ // TODO: clone or store Option<CacheKey> and use here `self.cache_key.take()`?
+ self.cache.insert(self.cache_key.clone(), cache_value);
Review Comment:
Should we `self.cache_key.clone()` or change the `cache_key` to be `Option<CacheKey>` and here just do `self.cache_key.take()` instead? We should only flush to the cache when we read all bytes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019024779
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
Review Comment:
I have some doubts about the parameters of this function. Should we receive `HybridCacheBuilder` instead in the `new` method args? This would allow users to fully customize their cache settings.
Taking `Risingwave` as an example, there are *a lot* of customizable settings https://github.com/risingwavelabs/risingwave/blob/db881a877e4a0d9bb886695d8f78b715d016a1c6/src/storage/src/store_impl.rs#L687
something like:
```
pub async fn new (cache_builder: HybridCacheBuilder){
// ...
}
```
What do you think?
The problem is, I'm hesitant of exposing foyer's structs in opendal's public API, what are your thoughts about this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020195610
##########
core/Cargo.toml:
##########
@@ -399,6 +406,8 @@ prometheus-client = { version = "0.23.1", optional = true }
tracing = { version = "0.1", optional = true }
# for layers-dtrace
probe = { version = "0.5.1", optional = true }
+# for layers-cache
+foyer = { version = "0.14.1", optional = true }
Review Comment:
done in fd3f54bd
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020224791
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ // TODO: lister cache?
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ // TODO: stat cache?
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ // TODO: handle this error
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ // TODO: log that we have a cache hit?
+ return Ok((entry.rp, Box::new(entry.bytes.clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
+ let reader: oio::Reader = Box::new(CacheWrapper::new(
+ reader,
+ rp,
+ Arc::clone(&self.cache),
+ cache_key,
+ ));
+ (rp, reader)
+ })
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ self.inner.list(path, args).await
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.inner.blocking_read(path, args)
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.inner.blocking_write(path, args)
+ }
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
+ self.inner.blocking_list(path, args)
+ }
+}
+
+pub struct CacheWrapper<R> {
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ buffers: Vec<Buffer>,
+}
+
+impl<R> CacheWrapper<R> {
+ fn new(
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ ) -> Self {
+ Self {
+ inner,
+ rp,
+ cache_key,
+ cache,
+ buffers: Vec::new(),
+ }
+ }
+}
+
+impl<R: oio::Read> oio::Read for CacheWrapper<R> {
+ async fn read(&mut self) -> Result<Buffer> {
+ // TODO: ask if we should accumulate the bytes like this
+ // or use self.inner.read_all() instead
+ let buffer = self.inner.read().await?;
+
+ if !buffer.is_empty() {
+ self.buffers.push(buffer.clone());
+ return Ok(buffer);
+ }
+ // TODO: comment that we only insert in cache when we read all the bytes and
+ // `self.inner.read()` returns empty bytes
+ let flattened_buffer: Buffer = self.buffers.drain(..).flatten().collect();
+
+ let cache_value = CacheValue {
+ rp: self.rp,
+ bytes: flattened_buffer.to_bytes(),
Review Comment:
Please always use `Buffer` across opendal.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020224107
##########
core/Cargo.toml:
##########
@@ -233,10 +240,11 @@ anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
-bytes = "1.6"
+bytes = { version = "1.6", features = ["serde"] }
Review Comment:
I hope we don't have to rely on this. Even if we do need to implement it, I hope we can keep it scoped within `FoyerLayer` itself.
I've also opened https://github.com/foyer-rs/foyer/issues/919 as the upstream issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045924535
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl Code for Buffer {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ let mut reader = self.clone();
+ std::io::copy(&mut reader, writer).map_err(CodeError::Io)?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
+ Ok(Buffer::from(buf))
+ }
+
+ fn estimated_size(&self) -> usize {
+ self.len()
+ }
+}
+
+impl<A, S> LayeredAccess for CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ return Ok((RpRead::default(), Box::new(entry.deref().clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
Review Comment:
> I have doubts about this. If someone tries to read the whole file and uses `BytesRange::from(..)`, we can't know at this point the size of the file and we can't rely purely in `OpRead`, we must check that _while_ reading the file
In this case, we should bypass and don't try to cache it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045926000
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl Code for Buffer {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ let mut reader = self.clone();
+ std::io::copy(&mut reader, writer).map_err(CodeError::Io)?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
+ Ok(Buffer::from(buf))
+ }
+
+ fn estimated_size(&self) -> usize {
+ self.len()
+ }
+}
+
+impl<A, S> LayeredAccess for CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ return Ok((RpRead::default(), Box::new(entry.deref().clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
Review Comment:
> and count how many bytes we have read (and stop buffering them in the `Vec<Buffer>` array if the threshold is exceeded) before inserting the cache value.
This seems a good idea. We can skip the cache insert if `max_value_size` has been reached.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2046694461
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
+ version: Option<String>,
+}
+
+impl CacheKey {
+ fn new(path: &str, args: &OpRead) -> CacheKey {
+ CacheKey {
+ path: path.to_string(),
+ range: args.range(),
+ if_match: args.if_match().map(ToString::to_string),
+ if_none_match: args.if_none_match().map(ToString::to_string),
+ if_modified_since: args.if_modified_since(),
+ if_unmodified_since: args.if_unmodified_since(),
+ version: args.version().map(ToString::to_string),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct CacheValue(Buffer);
+
+impl Code for CacheValue {
Review Comment:
> I think deserialization is done for the cache key.
Good point!
cc @MrCroxx, what happens if the cache key changes? Will Foyer silently ignore it and remove the stale cache value, or will we encounter an error?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044915017
##########
core/Cargo.toml:
##########
@@ -233,10 +240,11 @@ anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
-bytes = "1.6"
+bytes = { version = "1.6", features = ["serde"] }
chrono = { version = "0.4.28", default-features = false, features = [
"clock",
"std",
+ "serde",
Review Comment:
we can't `impl Serialize for DateTime` because both `Serialize` and `Datetime` are outside of this crate.
What we could do, is to not `derive(Serialize)` for `CacheKey` and manually implement `impl Serialize for CacheKey`, and serializing each one of the fields manually.
I think it is a mistake to do this manual implementation instead of relying in `derive(Serialize,Deserialize)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044909642
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
Review Comment:
> OpRead is large and most of it's value is not related to caching, we should not include OpRead in cache key directly.
I suggest to use an interal format like format!("{path}-{version}-{range}").
I think it would be better to have this key structured and not use a raw String, what are your thoughts about this? We should rely on `serde` to handle key serialization
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045923283
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ // TODO: lister cache?
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ // TODO: stat cache?
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ // TODO: handle this error
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ // TODO: log that we have a cache hit?
+ return Ok((entry.rp, Box::new(entry.bytes.clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
+ let reader: oio::Reader = Box::new(CacheWrapper::new(
+ reader,
+ rp,
+ Arc::clone(&self.cache),
+ cache_key,
+ ));
+ (rp, reader)
+ })
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ self.inner.list(path, args).await
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.inner.blocking_read(path, args)
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.inner.blocking_write(path, args)
+ }
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
+ self.inner.blocking_list(path, args)
+ }
+}
+
+pub struct CacheWrapper<R> {
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ buffers: Vec<Buffer>,
+}
+
+impl<R> CacheWrapper<R> {
+ fn new(
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ ) -> Self {
+ Self {
+ inner,
+ rp,
+ cache_key,
+ cache,
+ buffers: Vec::new(),
+ }
+ }
+}
+
+impl<R: oio::Read> oio::Read for CacheWrapper<R> {
+ async fn read(&mut self) -> Result<Buffer> {
+ // TODO: ask if we should accumulate the bytes like this
+ // or use self.inner.read_all() instead
+ let buffer = self.inner.read().await?;
+
+ if !buffer.is_empty() {
+ self.buffers.push(buffer.clone());
Review Comment:
I used to think we should handle chunk reading at the cache layer, but now I believe it's better to let users decide or manage it in another layer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045922699
##########
core/Cargo.toml:
##########
@@ -233,10 +240,11 @@ anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
-bytes = "1.6"
+bytes = { version = "1.6", features = ["serde"] }
chrono = { version = "0.4.28", default-features = false, features = [
"clock",
"std",
+ "serde",
Review Comment:
> we can't `impl Serialize for DateTime`
We don't need to.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044469185
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl Code for Buffer {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ let mut reader = self.clone();
+ std::io::copy(&mut reader, writer).map_err(CodeError::Io)?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
+ Ok(Buffer::from(buf))
+ }
+
+ fn estimated_size(&self) -> usize {
+ self.len()
+ }
+}
+
+impl<A, S> LayeredAccess for CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ type Reader = oio::Reader;
Review Comment:
yes, I have that change on local but I'm still working on the previous comments, Didn't forget about them!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2804956417
Thanks for the review @Xuanwo, I was still working on the previous reviews and didn't have all things done, thats why I didnt mentioned you yet. I will work a lot more on this today though :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2044373825
##########
core/Cargo.toml:
##########
@@ -241,10 +243,12 @@ anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
-bytes = "1.6"
+bincode = "1.3.3"
Review Comment:
It's better to avoid introduce `bincode`.
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
Review Comment:
I'm a bit confused why we need expose `HashBuilder`.
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl Code for Buffer {
Review Comment:
Please avoid implementing external traits directly for `Buffer`, as this can lead to abstraction leaks. It's better to create a `CacheValue(Buffer)` and implement `Code` for that instead.
##########
core/Cargo.toml:
##########
@@ -406,6 +410,9 @@ prometheus-client = { version = "0.23.1", optional = true }
tracing = { version = "0.1", optional = true }
# for layers-dtrace
probe = { version = "0.5.1", optional = true }
+# for layers-cache
+foyer = { version = "0.16.1", features = ["serde"], optional = true }
+foyer-common = "0.9.5"
Review Comment:
`foyer-common` should also be `optional`.
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
Review Comment:
`OpRead` is large and most of it's value is not related to caching, we should not include `OpRead` in cache key directly.
I suggest to use an interal format like `format!("{path}-{version}-{range}")`.
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl Code for Buffer {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ let mut reader = self.clone();
+ std::io::copy(&mut reader, writer).map_err(CodeError::Io)?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
Review Comment:
cc @MrCroxx, this can be much better if we can know the size before `decode`.
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl Code for Buffer {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ let mut reader = self.clone();
Review Comment:
Please don't use `copy` to do this which lead to an extra in-memory copy. Instead, please use a simple loop like:
```rust
loop {
let n = writer.write(self.0.chunk())?
self.0.advance(n);
if self.0.is_empty() {
break;
}
}
```
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
Review Comment:
I believe we should accept a `HybridCache<String, Buffer>` directly.
The `HybridCacheBuilderPhaseStorage` API itself is part of builder, and should not be reply on directly.
##########
core/src/raw/rps.rs:
##########
@@ -98,7 +98,7 @@ impl<T: Default> From<PresignedRequest> for Request<T> {
}
/// Reply for `read` operation.
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize)]
Review Comment:
Please don't change this.
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl Code for Buffer {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ let mut reader = self.clone();
+ std::io::copy(&mut reader, writer).map_err(CodeError::Io)?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
+ Ok(Buffer::from(buf))
+ }
+
+ fn estimated_size(&self) -> usize {
+ self.len()
+ }
+}
+
+impl<A, S> LayeredAccess for CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
Review Comment:
Please move `inner` under all types.
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl Code for Buffer {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ let mut reader = self.clone();
+ std::io::copy(&mut reader, writer).map_err(CodeError::Io)?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
+ Ok(Buffer::from(buf))
+ }
+
+ fn estimated_size(&self) -> usize {
+ self.len()
+ }
+}
+
+impl<A, S> LayeredAccess for CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ return Ok((RpRead::default(), Box::new(entry.deref().clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
Review Comment:
Users might be reading very large files. I suggest adding a configuration value called `max_value_size` to control the maximum size we attempt to cache. We can check the range in `OpRead` and only cache it if the range is smaller than `max_value_size`.
##########
core/Cargo.toml:
##########
@@ -233,10 +240,11 @@ anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
-bytes = "1.6"
+bytes = { version = "1.6", features = ["serde"] }
Review Comment:
We can remove this change now.
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,207 @@
+use core::fmt;
+use std::{ops::Deref, sync::Arc};
+
+use crate::raw::*;
+use crate::*;
+use foyer::{Code, CodeError, HybridCache, HybridCacheBuilderPhaseStorage};
+use foyer_common::code::HashBuilder;
+use serde::{Deserialize, Serialize};
+
+pub struct FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl<S> FoyerLayer<S>
+where
+ S: HashBuilder + fmt::Debug,
+{
+ pub async fn new(builder: HybridCacheBuilderPhaseStorage<CacheKey, Buffer, S>) -> Result<Self> {
+ let cache = builder
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A, S> Layer<A> for FoyerLayer<S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type LayeredAccess = CacheAccessor<A, S>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ inner: A,
+ cache: Arc<HybridCache<CacheKey, Buffer, S>>,
+}
+
+impl Code for Buffer {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ let mut reader = self.clone();
+ std::io::copy(&mut reader, writer).map_err(CodeError::Io)?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
+ Ok(Buffer::from(buf))
+ }
+
+ fn estimated_size(&self) -> usize {
+ self.len()
+ }
+}
+
+impl<A, S> LayeredAccess for CacheAccessor<A, S>
+where
+ A: Access,
+ S: HashBuilder + fmt::Debug,
+{
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ type Reader = oio::Reader;
Review Comment:
we can have a `oio::TwoWays<Buffer, A::Reader>`.
##########
core/Cargo.toml:
##########
@@ -233,10 +240,11 @@ anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
-bytes = "1.6"
+bytes = { version = "1.6", features = ["serde"] }
chrono = { version = "0.4.28", default-features = false, features = [
"clock",
"std",
+ "serde",
Review Comment:
The same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2047276118
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
Review Comment:
okay, perfect then
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2046668759
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
+ version: Option<String>,
+}
+
+impl CacheKey {
+ fn new(path: &str, args: &OpRead) -> CacheKey {
+ CacheKey {
+ path: path.to_string(),
+ range: args.range(),
+ if_match: args.if_match().map(ToString::to_string),
+ if_none_match: args.if_none_match().map(ToString::to_string),
+ if_modified_since: args.if_modified_since(),
+ if_unmodified_since: args.if_unmodified_since(),
+ version: args.version().map(ToString::to_string),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct CacheValue(Buffer);
+
+impl Code for CacheValue {
+ fn encode(&self, writer: &mut impl std::io::Write) -> std::result::Result<(), CodeError> {
+ writer.write_vectored(&self.0.to_io_slice())?;
+ Ok(())
+ }
+
+ fn decode(reader: &mut impl std::io::Read) -> std::result::Result<Self, CodeError>
+ where
+ Self: Sized,
+ {
+ let mut buf = Vec::new();
+ reader.read_to_end(&mut buf).map_err(CodeError::Io)?;
Review Comment:
We will handle this situation with this https://github.com/apache/opendal/pull/5906#discussion_r2045926000 right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2046665683
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
+ path: String,
+ range: BytesRange,
+ if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_modified_since: Option<DateTime<Utc>>,
+ if_unmodified_since: Option<DateTime<Utc>>,
+ version: Option<String>,
+}
+
+impl CacheKey {
+ fn new(path: &str, args: &OpRead) -> CacheKey {
+ CacheKey {
+ path: path.to_string(),
+ range: args.range(),
+ if_match: args.if_match().map(ToString::to_string),
+ if_none_match: args.if_none_match().map(ToString::to_string),
+ if_modified_since: args.if_modified_since(),
+ if_unmodified_since: args.if_unmodified_since(),
+ version: args.version().map(ToString::to_string),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct CacheValue(Buffer);
+
+impl Code for CacheValue {
Review Comment:
I think deserialization is done for the cache key.
https://github.com/foyer-rs/foyer/blob/efa47f7c70ed70f0970d2b04ffc8e6422b5c78c9/foyer-storage/src/serde.rs#L178
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2045911831
##########
core/src/layers/foyer.rs:
##########
@@ -0,0 +1,191 @@
+use crate::raw::*;
+use crate::*;
+use chrono::{DateTime, Utc};
+use foyer::{Code, CodeError, HybridCache};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+pub struct FoyerLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl FoyerLayer {
+ pub async fn new(cache: HybridCache<CacheKey, CacheValue>) -> Result<Self> {
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for FoyerLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+pub struct CacheKey {
Review Comment:
Hi, the `CacheKey` is generated solely from the input of each read operation, specifically the `path` and `OpRead`. Therefore, rather than creating a new struct and implementing `Serialize` for it, it’s much simpler to implement a function like `build_cache_key(path, op)`.
Structuring the cache key doesn’t make sense to me, since there’s no need for us to parse it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019031405
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ // TODO: lister cache?
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ // TODO: stat cache?
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ // TODO: handle this error
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ // TODO: log that we have a cache hit?
+ return Ok((entry.rp, Box::new(entry.bytes.clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
+ let reader: oio::Reader = Box::new(CacheWrapper::new(
+ reader,
+ rp,
+ Arc::clone(&self.cache),
+ cache_key,
+ ));
+ (rp, reader)
+ })
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ self.inner.list(path, args).await
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.inner.blocking_read(path, args)
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.inner.blocking_write(path, args)
+ }
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
+ self.inner.blocking_list(path, args)
+ }
+}
+
+pub struct CacheWrapper<R> {
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ buffers: Vec<Buffer>,
+}
+
+impl<R> CacheWrapper<R> {
+ fn new(
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ ) -> Self {
+ Self {
+ inner,
+ rp,
+ cache_key,
+ cache,
+ buffers: Vec::new(),
+ }
+ }
+}
+
+impl<R: oio::Read> oio::Read for CacheWrapper<R> {
+ async fn read(&mut self) -> Result<Buffer> {
+ // TODO: ask if we should accumulate the bytes like this
+ // or use self.inner.read_all() instead
+ let buffer = self.inner.read().await?;
Review Comment:
should we partially read and return with `self.inner.read()` or should we just call `self.inner.read_all()`? I think it is better to do the first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019016770
##########
core/Cargo.toml:
##########
@@ -233,10 +240,11 @@ anyhow = { version = "1.0.30", features = ["std"] }
async-trait = "0.1.68"
backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
-bytes = "1.6"
+bytes = { version = "1.6", features = ["serde"] }
Review Comment:
I need `Bytes` to implement `Serialize` and `Serialize` in order to store it as a cache value
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020225028
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ // TODO: lister cache?
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ // TODO: stat cache?
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ // TODO: handle this error
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ // TODO: log that we have a cache hit?
+ return Ok((entry.rp, Box::new(entry.bytes.clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
+ let reader: oio::Reader = Box::new(CacheWrapper::new(
+ reader,
+ rp,
+ Arc::clone(&self.cache),
+ cache_key,
+ ));
+ (rp, reader)
+ })
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ self.inner.list(path, args).await
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.inner.blocking_read(path, args)
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.inner.blocking_write(path, args)
+ }
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
+ self.inner.blocking_list(path, args)
+ }
+}
+
+pub struct CacheWrapper<R> {
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ buffers: Vec<Buffer>,
+}
+
+impl<R> CacheWrapper<R> {
+ fn new(
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ ) -> Self {
+ Self {
+ inner,
+ rp,
+ cache_key,
+ cache,
+ buffers: Vec::new(),
+ }
+ }
+}
+
+impl<R: oio::Read> oio::Read for CacheWrapper<R> {
+ async fn read(&mut self) -> Result<Buffer> {
+ // TODO: ask if we should accumulate the bytes like this
+ // or use self.inner.read_all() instead
+ let buffer = self.inner.read().await?;
+
+ if !buffer.is_empty() {
+ self.buffers.push(buffer.clone());
+ return Ok(buffer);
+ }
+ // TODO: comment that we only insert in cache when we read all the bytes and
+ // `self.inner.read()` returns empty bytes
+ let flattened_buffer: Buffer = self.buffers.drain(..).flatten().collect();
+
+ let cache_value = CacheValue {
+ rp: self.rp,
+ bytes: flattened_buffer.to_bytes(),
Review Comment:
> I think the best would be to store in cache contiguous bytes with Bytes and convert Non-contiguous bytes of Buffer to contiguous Bytes with this method.
This can lead to non-trivial memory usage and performance bottlenecks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020225245
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,213 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+// TODO: rename to foyer layer and receive a HybridCacheBuilder as the parameter?
+// TODO: expose foyer metrics here?
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl<A: Access> LayeredAccess for CacheAccessor<A> {
+ type Inner = A;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ // TODO: add a comment here that we use `oio::Reader` (i.e Box<dyn ReadDyn>) because
+ // if there is a cache hit, we return Box<Buffer>
+ // but if there isn't, we return a Box<CacheWrapper<..>> so when the reader
+ // is read, the value is inserted into the cache.
+ // This allow to lazy reading and inserting in cache not to be done in the
+ // `CacheAccessor::read` method, but when calling `reader.read()` on the reader
+ // output of `CacheAccessor`
+ type Reader = oio::Reader;
+
+ type Writer = A::Writer;
+
+ // TODO: lister cache?
+ type Lister = A::Lister;
+
+ type Deleter = A::Deleter;
+
+ type BlockingReader = A::BlockingReader;
+
+ type BlockingWriter = A::BlockingWriter;
+
+ type BlockingLister = A::BlockingLister;
+
+ type BlockingDeleter = A::BlockingDeleter;
+
+ // TODO: stat cache?
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let cache_key = CacheKey {
+ path: path.to_string(),
+ args: args.clone(),
+ };
+ if let Some(entry) = self
+ .cache
+ .get(&cache_key)
+ .await
+ // TODO: handle this error
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?
+ {
+ // TODO: log that we have a cache hit?
+ return Ok((entry.rp, Box::new(entry.bytes.clone())));
+ }
+
+ self.inner.read(path, args).await.map(|(rp, reader)| {
+ let reader: oio::Reader = Box::new(CacheWrapper::new(
+ reader,
+ rp,
+ Arc::clone(&self.cache),
+ cache_key,
+ ));
+ (rp, reader)
+ })
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ self.inner.list(path, args).await
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.inner.blocking_read(path, args)
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.inner.blocking_write(path, args)
+ }
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
+ self.inner.blocking_list(path, args)
+ }
+}
+
+pub struct CacheWrapper<R> {
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ buffers: Vec<Buffer>,
+}
+
+impl<R> CacheWrapper<R> {
+ fn new(
+ inner: R,
+ rp: RpRead,
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+ cache_key: CacheKey,
+ ) -> Self {
+ Self {
+ inner,
+ rp,
+ cache_key,
+ cache,
+ buffers: Vec::new(),
+ }
+ }
+}
+
+impl<R: oio::Read> oio::Read for CacheWrapper<R> {
+ async fn read(&mut self) -> Result<Buffer> {
+ // TODO: ask if we should accumulate the bytes like this
+ // or use self.inner.read_all() instead
+ let buffer = self.inner.read().await?;
+
+ if !buffer.is_empty() {
+ self.buffers.push(buffer.clone());
Review Comment:
While implementing the cache layer, please keep in mind that users may read files up to 1 TiB in size. Try handling simple files in small chunks; we can start with fixed-size chunks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2020225713
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,187 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
Review Comment:
At the current of time, we can always return `RpRead::default()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019046616
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,187 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
Review Comment:
I think we should use both `path` and `OpRead` as the `CacheKey`. Using `OpRead` directly instead of a new struct will ensure that new fields added in the future to `OpRead` will be always taken into account in the `CacheKey`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "jorgehermo9 (via GitHub)" <gi...@apache.org>.
jorgehermo9 commented on code in PR #5906:
URL: https://github.com/apache/opendal/pull/5906#discussion_r2019039579
##########
core/src/layers/cache.rs:
##########
@@ -0,0 +1,206 @@
+use std::sync::Arc;
+
+use crate::raw::*;
+use crate::*;
+use bytes::Bytes;
+use foyer::{DirectFsDeviceOptions, Engine, HybridCache, HybridCacheBuilder};
+use serde::{Deserialize, Serialize};
+
+pub struct CacheLayer {
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
+}
+
+impl CacheLayer {
+ pub async fn new(
+ disk_cache_dir: &str,
+ disk_capacity_mb: usize,
+ memory_capacity_mb: usize,
+ ) -> Result<Self> {
+ const MB: usize = 1 << 20;
+
+ let cache = HybridCacheBuilder::new()
+ .with_name("opendal")
+ .memory(memory_capacity_mb * MB)
+ .storage(Engine::Large)
+ .with_device_options(
+ DirectFsDeviceOptions::new(disk_cache_dir).with_capacity(disk_capacity_mb * MB),
+ )
+ .build()
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, e.to_string()))?;
+
+ Ok(Self {
+ cache: Arc::new(cache),
+ })
+ }
+}
+
+impl<A: Access> Layer<A> for CacheLayer {
+ type LayeredAccess = CacheAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ CacheAccessor {
+ inner,
+ cache: Arc::clone(&self.cache),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
+struct CacheKey {
+ path: String,
+ args: OpRead,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CacheValue {
+ rp: RpRead,
+ // TODO: store Buffer or Bytes?
+ bytes: Bytes,
+}
+
+#[derive(Debug)]
+pub struct CacheAccessor<A: Access> {
+ inner: A,
+ // TODO: if cache should be used for other operations (such as stat or list)
+ // maybe we should create different caches for each operation?
+ // So the keys and values does not mix
+ // Although, we could use an enum as the cachekey and as the cache value, but
+ // this way we wouldn't be making invalid states unrepresentable, as given
+ // a Read Cache key, there might be a List cached value associated.
+ cache: Arc<HybridCache<CacheKey, CacheValue>>,
Review Comment:
If we need to cache other operations that are not read, maybe we have to introduce different caches,
for example
`read_cache: Arc<HybridCache<ReadCacheKey,ReadCacheValue>>`
and
`stat_cache: Arc<HybridCache<StatCacheKey,StatCacheValue>>`
etc...
But as of right now, we only cache read operations, we don't need to be so specific...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: initial cache layer implementation [opendal]
Posted by "meteorgan (via GitHub)" <gi...@apache.org>.
meteorgan commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2764556869
Hi, I've got a question: since it's a `cache`, when do we clear out expired data ? I couldn't spot the code handling that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "MrCroxx (via GitHub)" <gi...@apache.org>.
MrCroxx commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2781860308
Hi, folks. Thank you for your contribution and effort in working on this PR.
As the author of foyer, I also hope to see Foyer better integrated with OpenDAL. If you are willing, I can take over this PR and continue with the adaptation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Initial foyer layer implementation [opendal]
Posted by "MrCroxx (via GitHub)" <gi...@apache.org>.
MrCroxx commented on PR #5906:
URL: https://github.com/apache/opendal/pull/5906#issuecomment-2782646078
> Hi, @MrCroxx, I believe it's better to cut off a release now so that @jorgehermo9 can work on the base without serde directly.
Sure. I'll release a new version in hours.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org