refactor(tvix/castore/blob): use near/far for CombinedBlobService

Align this naming with CachePathInfoService.

Change-Id: Ib9a0d73b8ca57a93e9fc027ae907fc6ed370842a
Reviewed-on: https://cl.tvl.fyi/c/depot/+/12751
Tested-by: BuildkiteCI
Reviewed-by: yuka <yuka@yuka.dev>
This commit is contained in:
Florian Klink 2024-11-09 17:12:55 +00:00 committed by flokli
parent a218f421b2
commit 8400e523ce
2 changed files with 37 additions and 37 deletions

View file

@ -8,17 +8,17 @@ use crate::{B3Digest, Error};
use super::{BlobReader, BlobService, BlobWriter, ChunkedReader}; use super::{BlobReader, BlobService, BlobWriter, ChunkedReader};
/// Combinator for a BlobService, using a "local" and "remote" blobservice. /// Combinator for a BlobService, using a "near" and "far" blobservice.
/// Requests are tried in (and returned from) the local store first, only if /// Requests are tried in (and returned from) the near store first, only if
/// things are not present there, the remote BlobService is queried. /// things are not present there, the far BlobService is queried.
/// In case the local blobservice doesn't have the blob, we ask the remote /// In case the near blobservice doesn't have the blob, we ask the remote
/// blobservice for chunks, and try to read each of these chunks from the local /// blobservice for chunks, and try to read each of these chunks from the near
/// blobservice again, before falling back to the remote one. /// blobservice again, before falling back to the far one.
/// The remote BlobService is never written to. /// The far BlobService is never written to.
pub struct CombinedBlobService<BL, BR> { pub struct CombinedBlobService<BL, BR> {
instance_name: String, instance_name: String,
local: BL, near: BL,
remote: BR, far: BR,
} }
impl<BL, BR> Clone for CombinedBlobService<BL, BR> impl<BL, BR> Clone for CombinedBlobService<BL, BR>
@ -29,8 +29,8 @@ where
fn clone(&self) -> Self { fn clone(&self) -> Self {
Self { Self {
instance_name: self.instance_name.clone(), instance_name: self.instance_name.clone(),
local: self.local.clone(), near: self.near.clone(),
remote: self.remote.clone(), far: self.far.clone(),
} }
} }
} }
@ -43,33 +43,33 @@ where
{ {
#[instrument(skip(self, digest), fields(blob.digest=%digest, instance_name=%self.instance_name))] #[instrument(skip(self, digest), fields(blob.digest=%digest, instance_name=%self.instance_name))]
async fn has(&self, digest: &B3Digest) -> std::io::Result<bool> { async fn has(&self, digest: &B3Digest) -> std::io::Result<bool> {
Ok(self.local.as_ref().has(digest).await? || self.remote.as_ref().has(digest).await?) Ok(self.near.as_ref().has(digest).await? || self.far.as_ref().has(digest).await?)
} }
#[instrument(skip(self, digest), fields(blob.digest=%digest, instance_name=%self.instance_name), err)] #[instrument(skip(self, digest), fields(blob.digest=%digest, instance_name=%self.instance_name), err)]
async fn open_read(&self, digest: &B3Digest) -> std::io::Result<Option<Box<dyn BlobReader>>> { async fn open_read(&self, digest: &B3Digest) -> std::io::Result<Option<Box<dyn BlobReader>>> {
if self.local.as_ref().has(digest).await? { if self.near.as_ref().has(digest).await? {
// local store has the blob, so we can assume it also has all chunks. // near store has the blob, so we can assume it also has all chunks.
self.local.as_ref().open_read(digest).await self.near.as_ref().open_read(digest).await
} else { } else {
// Local store doesn't have the blob. // near store doesn't have the blob.
// Ask the remote one for the list of chunks, // Ask the remote one for the list of chunks,
// and create a chunked reader that uses self.open_read() for // and create a chunked reader that uses self.open_read() for
// individual chunks. There's a chance we already have some chunks // individual chunks. There's a chance we already have some chunks
// locally, meaning we don't need to fetch them all from the remote // in near, meaning we don't need to fetch them all from the far
// BlobService. // BlobService.
match self.remote.as_ref().chunks(digest).await? { match self.far.as_ref().chunks(digest).await? {
// blob doesn't exist on the remote side either, nothing we can do. // blob doesn't exist on the near side either, nothing we can do.
None => Ok(None), None => Ok(None),
Some(remote_chunks) => { Some(remote_chunks) => {
// if there's no more granular chunks, or the remote // if there's no more granular chunks, or the far
// blobservice doesn't support chunks, read the blob from // blobservice doesn't support chunks, read the blob from
// the remote blobservice directly. // the far blobservice directly.
if remote_chunks.is_empty() { if remote_chunks.is_empty() {
return self.remote.as_ref().open_read(digest).await; return self.far.as_ref().open_read(digest).await;
} }
// otherwise, a chunked reader, which will always try the // otherwise, a chunked reader, which will always try the
// local backend first. // near backend first.
let chunked_reader = ChunkedReader::from_chunks( let chunked_reader = ChunkedReader::from_chunks(
remote_chunks.into_iter().map(|chunk| { remote_chunks.into_iter().map(|chunk| {
@ -88,16 +88,16 @@ where
#[instrument(skip_all, fields(instance_name=%self.instance_name))] #[instrument(skip_all, fields(instance_name=%self.instance_name))]
async fn open_write(&self) -> Box<dyn BlobWriter> { async fn open_write(&self) -> Box<dyn BlobWriter> {
// direct writes to the local one. // direct writes to the near one.
self.local.as_ref().open_write().await self.near.as_ref().open_write().await
} }
} }
#[derive(serde::Deserialize, Debug, Clone)] #[derive(serde::Deserialize, Debug, Clone)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
pub struct CombinedBlobServiceConfig { pub struct CombinedBlobServiceConfig {
local: String, near: String,
remote: String, far: String,
} }
impl TryFrom<url::Url> for CombinedBlobServiceConfig { impl TryFrom<url::Url> for CombinedBlobServiceConfig {
@ -119,13 +119,13 @@ impl ServiceBuilder for CombinedBlobServiceConfig {
context: &CompositionContext, context: &CompositionContext,
) -> Result<Arc<dyn BlobService>, Box<dyn std::error::Error + Send + Sync>> { ) -> Result<Arc<dyn BlobService>, Box<dyn std::error::Error + Send + Sync>> {
let (local, remote) = futures::join!( let (local, remote) = futures::join!(
context.resolve(self.local.clone()), context.resolve(self.near.clone()),
context.resolve(self.remote.clone()) context.resolve(self.far.clone())
); );
Ok(Arc::new(CombinedBlobService { Ok(Arc::new(CombinedBlobService {
instance_name: instance_name.to_string(), instance_name: instance_name.to_string(),
local: local?, near: local?,
remote: remote?, far: remote?,
})) }))
} }
} }

View file

@ -64,8 +64,8 @@
//! }, //! },
//! "root": { //! "root": {
//! "type": "combined", //! "type": "combined",
//! "local": "blobstore1", //! "near": "blobstore1",
//! "remote": "blobstore2" //! "far": "blobstore2"
//! } //! }
//! }); //! });
//! //!
@ -555,13 +555,13 @@ mod test {
let blob_services_configs_json = serde_json::json!({ let blob_services_configs_json = serde_json::json!({
"root": { "root": {
"type": "combined", "type": "combined",
"local": "other", "near": "other",
"remote": "other" "far": "other"
}, },
"other": { "other": {
"type": "combined", "type": "combined",
"local": "root", "near": "root",
"remote": "root" "far": "root"
} }
}); });