Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions docs/Architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,19 @@ are exchanged with `StorageGetPreprocessorEntry` / `StoragePutPreprocessorEntry`
Because each CLI process accumulates its own statistics, it flushes them to the
daemon with `RecordStats` before exiting.

`StorageGetRaw` preserves backend read errors across IPC; an error is not an
ordinary missing key. Invalid cached archives also produce lookup errors,
whether the CLI reads a local path or receives raw bytes.

In both execution modes, a cache read error followed by a successful local
compilation increments `Cache read errors` as well as the existing `Cache errors`
and `Cache misses` counters. Ordinary misses do not increment the read-error
counter, and overall lookup timeouts are recorded separately as `Cache timeouts`.
With a multilevel cache, a read failure at one level is logged and lookup
continues at the next level. These per-level failures do not increment
`Cache read errors` or `Cache errors`; if all levels miss or fail, the lookup
is counted as a cache miss.

Client-side mode is enabled with the `SCCACHE_CLIENT_SIDE` environment variable
(or the `client_side_mode` config key). It is expected to become the only
supported configuration in the future, with server-side mode eventually
Expand Down
4 changes: 2 additions & 2 deletions src/cache/cache.rs
Original file line number Diff line number Diff line change
Expand Up @@ -161,8 +161,8 @@ pub trait Storage: Send + Sync {
}
/// Return the filesystem path of the cached entry for `key`.
/// Default impl returns [`GetPathResult::Unsupported`].
async fn get_path(&self, _key: &str) -> GetPathResult {
GetPathResult::Unsupported
async fn get_path(&self, _key: &str) -> Result<GetPathResult> {
Ok(GetPathResult::Unsupported)
}

/// Return the preprocessor cache entry for a given preprocessor key,
Expand Down
16 changes: 5 additions & 11 deletions src/cache/disk.rs
Original file line number Diff line number Diff line change
Expand Up @@ -142,24 +142,18 @@ impl Storage for DiskCache {
.await?
}

async fn get_path(&self, key: &str) -> GetPathResult {
async fn get_path(&self, key: &str) -> Result<GetPathResult> {
let rel_path = make_key_path(key);
let lru = self.lru.clone();
self.pool
.spawn_blocking(move || {
match lru
.lock()
.unwrap()
.get_or_init()
.ok()
.and_then(|c| c.get_abs_path(&rel_path))
{
let path = lru.lock().unwrap().get_or_init()?.get_abs_path(&rel_path);
Ok(match path {
Some(p) => GetPathResult::Found(p),
None => GetPathResult::Miss,
}
})
})
.await
.unwrap_or(GetPathResult::Miss)
.await?
}

async fn put(&self, key: &str, entry: CacheWrite) -> Result<Duration> {
Expand Down
30 changes: 14 additions & 16 deletions src/cache/ipc_storage.rs
Original file line number Diff line number Diff line change
Expand Up @@ -71,36 +71,33 @@ impl IpcStorage {
#[async_trait]
impl Storage for IpcStorage {
async fn get(&self, key: &str) -> Result<Cache> {
match self.get_path(key).await {
match self.get_path(key).await? {
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),
}
let entry = CacheRead::from(file)
.with_context(|| format!("IpcStorage::get: read {}", path.display()))?;
Ok(Cache::Hit(entry))
}
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),
},
}
}

async fn get_path(&self, key: &str) -> GetPathResult {
match self
async fn get_path(&self, key: &str) -> Result<GetPathResult> {
let resp = self
.rpc(Request::StorageGetPath {
key: key.to_owned(),
})
.await
{
Ok(Response::StorageGetPath(result)) => result,
_ => GetPathResult::Unsupported,
.await?;
match resp {
Response::StorageGetPath(Ok(result)) => Ok(result),
Response::StorageGetPath(Err(e)) => bail!("IpcStorage::get_path: daemon error: {e}"),
other => bail!("IpcStorage::get_path: unexpected response: {other:?}"),
}
}

Expand All @@ -115,7 +112,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:?}"),
}
}
Expand Down
8 changes: 4 additions & 4 deletions src/protocol.rs
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ pub enum Request {
/// One-shot handshake: client requests cache metadata from the daemon.
StorageHandshake,
/// Fetch the filesystem path of the cached entry for `key`.
/// Returns `None` if the backend does not support direct file access or the key is absent.
/// Reports a missing key, unsupported direct file access, or a lookup error.
StorageGetPath { key: String },
/// Fetch raw (zip) bytes for `key`; returns `None` on a miss.
StorageGetRaw { key: String },
Expand Down Expand Up @@ -57,9 +57,9 @@ pub enum Response {
/// Response for `Request::StorageHandshake`.
StorageHandshake(StorageHandshakeInfo),
/// Response for `Request::StorageGetPath`.
StorageGetPath(GetPathResult),
/// Response for `Request::StorageGetRaw`: zip bytes on hit, `None` on miss.
StorageGetRaw(Option<Vec<u8>>),
StorageGetPath(Result<GetPathResult, String>),
/// Response for `Request::StorageGetRaw`: zip bytes on hit, `None` on miss, or a read error.
StorageGetRaw(Result<Option<Vec<u8>>, String>),
/// Response for `Request::StoragePutRaw`.
StoragePutRaw(Result<(), String>),
/// Response for `Request::StorageGetPreprocessorEntry`.
Expand Down
210 changes: 199 additions & 11 deletions src/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -949,20 +949,22 @@ where
}
Request::StorageGetPath { key } => {
debug!("handle_client: storage_get_path key={}", key);
Ok(Message::WithoutBody(Response::StorageGetPath(
me.storage.get_path(&key).await,
)))
let result = me
.storage
.get_path(&key)
.await
.map_err(|e| format!("{e:#}"));
Ok(Message::WithoutBody(Response::StorageGetPath(result)))
}
Request::StorageGetRaw { key } => {
debug!("handle_client: storage_get_raw key={}", key);
let resp = match me.storage.get_raw(&key).await {
Ok(opt) => Response::StorageGetRaw(opt.map(|b| b.to_vec())),
Err(e) => {
warn!("storage_get_raw error: {e:#}");
Response::StorageGetRaw(None)
}
};
Ok(Message::WithoutBody(resp))
let result = me
.storage
.get_raw(&key)
.await
.map(|opt| opt.map(|b| b.to_vec()))
.map_err(|e| format!("{e:#}"));
Ok(Message::WithoutBody(Response::StorageGetRaw(result)))
}
Request::StoragePutRaw { key, data } => {
debug!("handle_client: storage_put_raw key={}", key);
Expand Down Expand Up @@ -1618,6 +1620,7 @@ where
stats.cache_timeouts += 1;
}
MissType::CacheReadError => {
stats.cache_read_errors += 1;
stats.cache_errors.increment(&kind, &lang);
}
}
Expand Down Expand Up @@ -2893,6 +2896,191 @@ mod tests {
(body, finished)
}

fn connect_ipc_storage(
runtime: &Runtime,
storage: Arc<dyn Storage>,
) -> (
crate::cache::IpcStorage,
tokio::task::JoinHandle<Result<()>>,
) {
let listener = runtime
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let addr = crate::net::SocketAddr::Net(listener.local_addr().unwrap());
let service = MockService::mock_with_storage(storage, runtime.handle().clone());
let server = runtime.spawn(async move {
let (socket, _) = listener.accept().await?;
service.bind(socket).await
});
let connection = crate::client::connect_to_server(&addr).unwrap();
let storage = crate::cache::ipc_storage::IpcStorage::connect(connection).unwrap();
(storage, server)
}

#[test_case::test_case(Err(anyhow!("backend read failed")), true; "backend_error")]
#[test_case::test_case(Ok(Some(bytes::Bytes::from_static(b"not a zip"))), true; "invalid_archive")]
#[test_case::test_case(Ok(None), false; "normal_miss")]
fn ipc_lookup_outcomes_reach_service_stats(
lookup: Result<Option<bytes::Bytes>>,
read_error: bool,
) {
use crate::test::mock_storage::MockStorage;

let (_f, runtime, mut service, compile) = rustc_miss_fixture();
let backend = Arc::new(MockStorage::new(None, false));
backend.next_get_raw(lookup);
let (storage, server) = connect_ipc_storage(&runtime, backend);
service.storage = Arc::new(storage);

let (response, finished, _) = runtime
.block_on(service.compile_direct(compile, &mut Vec::new()))
.unwrap();
assert!(matches!(response, CompileResponse::CompileStarted));
assert_eq!(finished.unwrap().retcode, Some(0));
assert_eq!(service.creator.lock().unwrap().children.len(), 0);
let stats = runtime.block_on(service.take_stats());
assert_eq!(stats.cache_read_errors, u64::from(read_error));
assert_eq!(stats.cache_errors.all(), u64::from(read_error));
assert_eq!(stats.cache_misses.all(), 1);
assert_eq!(stats.compilations, 1);
assert_eq!(stats.requests_executed, 1);
assert_eq!(stats.cache_timeouts, 0);
assert_eq!(stats.cache_hits.all(), 0);
assert_eq!(stats.compile_fails, 0);

drop(service);
runtime.block_on(server).unwrap().unwrap();
}

#[test]
fn ipc_disk_lookup_preserves_hits_misses_and_archive_errors() {
use crate::cache::{Cache, CacheWrite, GetPathResult};

let (_f, runtime, service, _compile) = rustc_miss_fixture();
let (storage, server) = connect_ipc_storage(&runtime, service.storage.clone());
let key = "0123456789abcdef";
assert!(matches!(
runtime.block_on(storage.get(key)).unwrap(),
Cache::Miss
));

let mut entry = CacheWrite::new();
entry.put_stdout(b"cached output").unwrap();
runtime.block_on(service.storage.put(key, entry)).unwrap();
match runtime.block_on(storage.get(key)).unwrap() {
Cache::Hit(mut entry) => assert_eq!(entry.get_stdout(), b"cached output"),
_ => panic!("expected a cache hit"),
}

runtime
.block_on(
service
.storage
.put_raw(key, bytes::Bytes::from_static(b"not a zip")),
)
.unwrap();
let GetPathResult::Found(path) = runtime.block_on(service.storage.get_path(key)).unwrap()
else {
panic!("expected a cached archive path");
};
let error = runtime.block_on(storage.get(key)).err().unwrap();
assert!(format!("{error:#}").contains(path.to_str().unwrap()));
assert!(error.downcast_ref::<zip::result::ZipError>().is_some());

drop(storage);
runtime.block_on(server).unwrap().unwrap();
}

#[test_case::test_case(true; "initialization_error")]
#[test_case::test_case(false; "normal_miss")]
fn ipc_disk_path_lookup_outcomes_reach_service_stats(initialization_error: bool) {
let (f, runtime, mut service, compile) = rustc_miss_fixture();
if initialization_error {
std::fs::write(f.tempdir.path().join("cache"), b"not a directory").unwrap();
}
let (storage, server) = connect_ipc_storage(&runtime, service.storage.clone());
service.storage = Arc::new(storage);

let (response, finished, _) = runtime
.block_on(service.compile_direct(compile, &mut Vec::new()))
.unwrap();
assert!(matches!(response, CompileResponse::CompileStarted));
assert_eq!(finished.unwrap().retcode, Some(0));
assert_eq!(
std::fs::read(f.tempdir.path().join("libdep.rlib")).unwrap(),
b"rlib"
);
let stats = runtime.block_on(service.take_stats());
assert_eq!(stats.cache_read_errors, u64::from(initialization_error));
assert_eq!(stats.cache_errors.all(), u64::from(initialization_error));
assert_eq!(stats.cache_misses.all(), 1);
assert_eq!(stats.cache_hits.all(), 0);
assert_eq!(stats.cache_timeouts, 0);
assert_eq!(stats.compile_fails, 0);
assert_eq!(stats.cache_writes, u64::from(!initialization_error));
assert_eq!(stats.cache_write_errors, u64::from(initialization_error));

drop(service);
runtime.block_on(server).unwrap().unwrap();
}

#[test_case::test_case(true, false; "daemon_read_error")]
#[test_case::test_case(false, false; "daemon_normal_miss")]
#[test_case::test_case(true, true; "client_side_read_error")]
#[test_case::test_case(false, true; "client_side_normal_miss")]
fn cache_lookup_outcomes_preserve_service_stats(read_error: bool, client_side: bool) {
use crate::cache::Cache;
use crate::test::mock_storage::MockStorage;

let (f, runtime, mut service, compile) = rustc_miss_fixture();
let storage = Arc::new(MockStorage::new(None, false));
storage.next_get(if read_error {
Err(anyhow!("cache lookup failed"))
} else {
Ok(Cache::Miss)
});
service.storage = storage;

let (finished, stats) = if client_side {
let (response, finished, _) = runtime
.block_on(service.compile_direct(compile, &mut Vec::new()))
.unwrap();
assert!(matches!(response, CompileResponse::CompileStarted));
(
finished.expect("compile started"),
runtime.block_on(service.take_stats()),
)
} else {
let (_, finished) = compile_body(&runtime, &service, compile);
(
finished,
runtime.block_on(service.get_info()).unwrap().stats,
)
};

assert_eq!(finished.retcode, Some(0));
assert_eq!(
std::fs::read(f.tempdir.path().join("libdep.rlib")).unwrap(),
b"rlib"
);
assert_eq!(service.creator.lock().unwrap().children.len(), 0);
assert_eq!(stats.requests_executed, 1);
assert_eq!(stats.cache_read_errors, u64::from(read_error));
assert_eq!(stats.cache_errors.all(), u64::from(read_error));
assert_eq!(
stats.cache_errors.get("Rust").copied().unwrap_or(0),
u64::from(read_error)
);
assert_eq!(stats.cache_misses.all(), 1);
assert_eq!(stats.cache_misses.get("Rust"), Some(&1));
assert_eq!(stats.compilations, 1);
assert_eq!(stats.cache_hits.all(), 0);
assert_eq!(stats.cache_timeouts, 0);
assert_eq!(stats.compile_fails, 0);
assert_eq!(stats.cache_writes, 1);
assert_eq!(stats.cache_write_errors, 0);
}

#[test]
fn records_non_cacheable_crate_type_detail_in_existing_stats_map() {
let mut stats = ServerStats::default();
Expand Down
Loading
Loading