Skip to content

Commit

Permalink
Fix cache for 0471 (#7)
Browse files Browse the repository at this point in the history
* Fix cache for 0471

Signed-off-by: Xuanwo <github@xuanwo.io>

* Make clippy happy

Signed-off-by: Xuanwo <github@xuanwo.io>

---------

Signed-off-by: Xuanwo <github@xuanwo.io>
  • Loading branch information
Xuanwo authored Jun 29, 2024
1 parent c4025fb commit 7867276
Show file tree
Hide file tree
Showing 4 changed files with 150 additions and 124 deletions.
9 changes: 4 additions & 5 deletions src/datanode/src/store.rs
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ mod gcs;
mod oss;
mod s3;

use std::sync::Arc;
use std::time::Duration;
use std::{env, path};

Expand All @@ -28,7 +29,7 @@ use common_telemetry::info;
use object_store::layers::{LruCacheLayer, RetryLayer};
use object_store::services::Fs;
use object_store::util::{join_dir, normalize_dir, with_instrument_layers};
use object_store::{HttpClient, ObjectStore};
use object_store::{HttpClient, ObjectStore, ObjectStoreBuilder};
use snafu::prelude::*;

use crate::config::{ObjectStoreConfig, DEFAULT_OBJECT_STORE_CACHE_SIZE};
Expand Down Expand Up @@ -108,11 +109,9 @@ async fn create_object_store_with_cache(
clean_temp_dir(&atomic_temp_dir)?;
let mut builder = Fs::default();
builder.root(path).atomic_write_dir(&atomic_temp_dir);
let cache_store = ObjectStore::new(builder)
.context(error::InitBackendSnafu)?
.finish();
let cache_store = builder.build().context(error::InitBackendSnafu)?;

let cache_layer = LruCacheLayer::new(cache_store, cache_capacity.0 as usize)
let cache_layer = LruCacheLayer::new(Arc::new(cache_store), cache_capacity.0 as usize)
.await
.context(error::InitBackendSnafu)?;

Expand Down
28 changes: 16 additions & 12 deletions src/object-store/src/layers/lru_cache.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,26 +12,28 @@
// See the License for the specific language governing permissions and
// limitations under the License.

use std::sync::Arc;

use opendal::raw::oio::ReadDyn;
use opendal::raw::{
Access, Layer, LayeredAccess, OpDelete, OpList, OpRead, OpWrite, RpDelete, RpList, RpRead,
RpWrite,
};
use opendal::{Operator, Result};
use opendal::Result;
mod read_cache;
use common_telemetry::info;
use read_cache::ReadCache;

/// An opendal layer with local LRU file cache supporting.
#[derive(Clone)]
pub struct LruCacheLayer {
pub struct LruCacheLayer<C: Access> {
// The read cache
read_cache: ReadCache,
read_cache: ReadCache<C>,
}

impl LruCacheLayer {
impl<C: Access> LruCacheLayer<C> {
/// Create a `[LruCacheLayer]` with local file cache and capacity in bytes.
pub async fn new(file_cache: Operator, capacity: usize) -> Result<Self> {
pub async fn new(file_cache: Arc<C>, capacity: usize) -> Result<Self> {
let read_cache = ReadCache::new(file_cache, capacity);
let (entries, bytes) = read_cache.recover_cache().await?;

Expand All @@ -50,12 +52,12 @@ impl LruCacheLayer {

/// Returns the read cache statistics info `(EntryCount, SizeInBytes)`.
pub async fn read_cache_stat(&self) -> (u64, u64) {
self.read_cache.stat().await
self.read_cache.cache_stat().await
}
}

impl<I: Access> Layer<I> for LruCacheLayer {
type LayeredAccess = LruCacheAccess<I>;
impl<I: Access, C: Access> Layer<I> for LruCacheLayer<C> {
type LayeredAccess = LruCacheAccess<I, C>;

fn layer(&self, inner: I) -> Self::LayeredAccess {
LruCacheAccess {
Expand All @@ -66,12 +68,12 @@ impl<I: Access> Layer<I> for LruCacheLayer {
}

#[derive(Debug)]
pub struct LruCacheAccess<I> {
pub struct LruCacheAccess<I, C> {
inner: I,
read_cache: ReadCache,
read_cache: ReadCache<C>,
}

impl<I: Access> LayeredAccess for LruCacheAccess<I> {
impl<I: Access, C: Access> LayeredAccess for LruCacheAccess<I, C> {
type Inner = I;
type Reader = Box<dyn ReadDyn>;
type BlockingReader = I::BlockingReader;
Expand All @@ -85,7 +87,9 @@ impl<I: Access> LayeredAccess for LruCacheAccess<I> {
}

async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
self.read_cache.read(&self.inner, path, args).await
self.read_cache
.read_from_cache(&self.inner, path, args)
.await
}

async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
Expand Down
Loading

0 comments on commit 7867276

Please sign in to comment.