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-unsubscr...@opendal.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
