Repository navigation
cache: preserve and count lookup read errors #2967
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from 1 commit
978bee5
dc29ed6
804433b
47010f9
cba732f
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -75,18 +75,12 @@ impl Storage for IpcStorage { | |
| GetPathResult::Found(path) => { | ||
| let file = std::fs::File::open(&path) | ||
| .with_context(|| format!("IpcStorage::get: open {}", path.display()))?; | ||
| match CacheRead::from(file) { | ||
| Ok(entry) => Ok(Cache::Hit(entry)), | ||
| Err(_) => Ok(Cache::Miss), | ||
| } | ||
| Ok(Cache::Hit(CacheRead::from(file)?)) | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. maybe add a
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. yes, good idea - cba732f |
||
| } | ||
| GetPathResult::Miss => Ok(Cache::Miss), | ||
| // Backend doesn't support paths (S3, Redis, …); fall back to bytes over IPC. | ||
| GetPathResult::Unsupported => match self.get_raw(key).await? { | ||
| Some(bytes) => match CacheRead::from(Cursor::new(bytes)) { | ||
| Ok(entry) => Ok(Cache::Hit(entry)), | ||
| Err(_) => Ok(Cache::Miss), | ||
| }, | ||
| Some(bytes) => Ok(Cache::Hit(CacheRead::from(Cursor::new(bytes))?)), | ||
| None => Ok(Cache::Miss), | ||
| }, | ||
| } | ||
|
|
@@ -115,7 +109,8 @@ impl Storage for IpcStorage { | |
| }) | ||
| .await?; | ||
| match resp { | ||
| Response::StorageGetRaw(opt) => Ok(opt.map(Bytes::from)), | ||
| Response::StorageGetRaw(Ok(opt)) => Ok(opt.map(Bytes::from)), | ||
| Response::StorageGetRaw(Err(e)) => bail!("IpcStorage::get_raw: daemon error: {e}"), | ||
| other => bail!("IpcStorage::get_raw: unexpected response: {other:?}"), | ||
| } | ||
| } | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -16,6 +16,7 @@ use crate::cache::{Cache, CacheWrite, Storage}; | |
| use crate::config::PreprocessorCacheModeConfig; | ||
| use crate::errors::*; | ||
| use async_trait::async_trait; | ||
| use bytes::Bytes; | ||
| use futures::channel::mpsc; | ||
| use std::sync::Arc; | ||
| use std::time::Duration; | ||
|
|
@@ -26,6 +27,8 @@ use tokio::time::sleep; | |
| pub struct MockStorage { | ||
| rx: Arc<Mutex<mpsc::UnboundedReceiver<Result<Cache>>>>, | ||
| tx: mpsc::UnboundedSender<Result<Cache>>, | ||
| raw_rx: Mutex<mpsc::UnboundedReceiver<Result<Option<Bytes>>>>, | ||
| raw_tx: mpsc::UnboundedSender<Result<Option<Bytes>>>, | ||
| delay: Option<Duration>, | ||
| preprocessor_cache_mode: bool, | ||
| } | ||
|
|
@@ -34,10 +37,13 @@ impl MockStorage { | |
| /// Create a new `MockStorage`. if `delay` is `Some`, wait for that amount of time before returning from operations. | ||
| pub(crate) fn new(delay: Option<Duration>, preprocessor_cache_mode: bool) -> MockStorage { | ||
| let (tx, rx) = mpsc::unbounded(); | ||
| let (raw_tx, raw_rx) = mpsc::unbounded(); | ||
| Self { | ||
| tx, | ||
| rx: Arc::new(Mutex::new(rx)), | ||
| delay, | ||
| raw_tx, | ||
| raw_rx: Mutex::new(raw_rx), | ||
| preprocessor_cache_mode, | ||
| } | ||
| } | ||
|
|
@@ -46,6 +52,11 @@ impl MockStorage { | |
| pub(crate) fn next_get(&self, res: Result<Cache>) { | ||
| self.tx.unbounded_send(res).unwrap(); | ||
| } | ||
|
|
||
| /// Queue up `res` to be returned as the next result from `Storage::get_raw`. | ||
| pub(crate) fn next_get_raw(&self, res: Result<Option<Bytes>>) { | ||
| self.raw_tx.unbounded_send(res).unwrap(); | ||
| } | ||
| } | ||
|
|
||
| #[async_trait] | ||
|
|
@@ -58,6 +69,10 @@ impl Storage for MockStorage { | |
|
|
||
| next.expect("MockStorage get called but no get results available") | ||
| } | ||
| async fn get_raw(&self, _key: &str) -> Result<Option<Bytes>> { | ||
| let next = self.raw_rx.lock().await.try_next().unwrap(); | ||
| next.unwrap_or(Ok(None)) | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. agreed - dc29ed6 |
||
| } | ||
| async fn put(&self, _key: &str, _entry: CacheWrite) -> Result<Duration> { | ||
| Ok(if let Some(delay) = self.delay { | ||
| sleep(delay).await; | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
i didn't understand this line, could you please clarify or drop it? :)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Reworded in 804433b let me know what you think