From 75863712b6a3f36182210d59361ab9109a23c9f4 Mon Sep 17 00:00:00 2001 From: "Park.Jiyeon" Date: Wed, 22 Jul 2026 20:56:22 +0800 Subject: [PATCH] service: reduce fscache blob init lock scope Move blob initialization and prefetch work out of the fscache blob lock. Wait for initialization only when read or close needs the blob, so concurrent requests do not block on long I/O. Signed-off-by: Park.Jiyeon --- service/src/fs_cache.rs | 209 +++++++++++++++++++++++++++++++--------- 1 file changed, 166 insertions(+), 43 deletions(-) diff --git a/service/src/fs_cache.rs b/service/src/fs_cache.rs index 5553da153ab..eb5a3eb4da6 100644 --- a/service/src/fs_cache.rs +++ b/service/src/fs_cache.rs @@ -217,6 +217,23 @@ struct FsCacheBlobCache { cache: Option>, config: Arc, file: Arc, + init_state: Arc, +} + +#[derive(Default)] +struct FsCacheBlobInitState { + done: Mutex, + condvar: Condvar, +} + +struct FsCacheBlobInitGuard { + init_state: Option>, +} + +struct FsCachePrefetchPlan { + blob_id: String, + request_size: u64, + blob_size: u64, } impl FsCacheBlobCache { @@ -229,6 +246,43 @@ impl FsCacheBlobCache { } } +impl FsCacheBlobInitState { + fn wait(&self) { + let mut done = self.done.lock().unwrap(); + while !*done { + done = self.condvar.wait(done).unwrap(); + } + } + + fn mark_done(&self) { + let mut done = self.done.lock().unwrap(); + *done = true; + self.condvar.notify_all(); + } +} + +impl FsCacheBlobInitGuard { + fn new(init_state: Arc) -> Self { + Self { + init_state: Some(init_state), + } + } + + fn mark_done(mut self) { + if let Some(init_state) = self.init_state.take() { + init_state.mark_done(); + } + } +} + +impl Drop for FsCacheBlobInitGuard { + fn drop(&mut self) { + if let Some(init_state) = self.init_state.take() { + init_state.mark_done(); + } + } +} + #[derive(Clone)] enum FsCacheObject { Bootstrap(Arc), @@ -474,19 +528,17 @@ impl FsCacheHandler { ) -> String { let mut state = self.state.lock().unwrap(); if let Vacant(e) = state.id_to_object_map.entry(hdr.object_id) { + let init_state = Arc::new(FsCacheBlobInitState::default()); let fsblob = Arc::new(RwLock::new(FsCacheBlobCache { cache: None, config: config.clone(), file: Arc::new(unsafe { File::from_raw_fd(msg.fd as RawFd) }), + init_state: init_state.clone(), })); e.insert((FsCacheObject::DataBlob(fsblob.clone()), msg.fd)); state.id_to_config_map.insert(hdr.object_id, config.clone()); let blob_size = config.blob_info().deref().uncompressed_size(); - let barrier = Arc::new(Barrier::new(2)); - Self::init_blob_cache(fsblob, barrier.clone()); - // make sure that the blobcache init thread have gotten writer lock before user daemon - // receives first request. - barrier.wait(); + Self::init_blob_cache(fsblob, init_state); format!("copen {},{}", hdr.msg_id, blob_size) } else { unsafe { libc::close(msg.fd as i32) }; @@ -494,39 +546,65 @@ impl FsCacheHandler { } } - fn init_blob_cache(fsblob: Arc>, barrier: Arc) { + fn init_blob_cache( + fsblob: Arc>, + init_state: Arc, + ) { thread::spawn(move || { - let mut guard = fsblob.write().unwrap(); - barrier.wait(); - //for now FsCacheBlobCache only init once, should not have blobcache associated with it - assert!(guard.get_blob_cache().is_none()); + let init_guard = FsCacheBlobInitGuard::new(init_state); + let (config, file) = { + let guard = fsblob.read().unwrap(); + // For now FsCacheBlobCache only init once, should not have blobcache associated + // with it. + assert!(guard.get_blob_cache().is_none()); + (guard.config.clone(), guard.file.clone()) + }; + for _ in 0..BLOB_CACHE_INIT_RETRY { - match Self::create_data_blob_object(&guard.config, guard.file.clone()) { + match Self::create_data_blob_object(&config, file.clone()) { Err(e) => { warn!("fscache: create_data_blob_object failed {}", e); thread::sleep(time::Duration::from_millis(BLOB_CACHE_INIT_INTERVAL_MS)); } Ok(blob) => { - guard.set_blob_cache(Some(blob.clone())); - if let Err(e) = Self::do_prefetch(&guard.config, blob.clone()) { - warn!( - "fscache: failed to prefetch data for blob {}, {}", - blob.blob_id(), - e - ); + let prefetch_plan = match Self::start_prefetch(&config, blob.clone()) { + Ok(plan) => plan, + Err(e) => { + warn!( + "fscache: failed to prefetch data for blob {}, {}", + blob.blob_id(), + e + ); + None + } + }; + + { + let mut guard = fsblob.write().unwrap(); + guard.set_blob_cache(Some(blob.clone())); } - break; + init_guard.mark_done(); + + if let Some(plan) = prefetch_plan { + Self::submit_prefetch(blob, plan); + } + return; } } } + + init_guard.mark_done(); }); } - fn do_prefetch(cfg: &DataBlobConfig, blob: Arc) -> Result<()> { + fn start_prefetch( + cfg: &DataBlobConfig, + blob: Arc, + ) -> Result> { let blob_info = cfg.blob_info().deref(); let cache_cfg = cfg.config_v2().get_cache_config()?; if !cache_cfg.prefetch.enable { - return Ok(()); + return Ok(None); } blob.start_prefetch() .map_err(|e| eother!(format!("failed to start prefetch worker, {}", e)))?; @@ -537,29 +615,44 @@ impl FsCacheHandler { Some(s) => s as u64, }; let size = std::cmp::max(0x4_0000u64, size); - let blob_size = blob_info.compressed_data_size(); - let count = blob_size.div_ceil(size); + Ok(Some(FsCachePrefetchPlan { + blob_id: blob_info.blob_id().to_owned(), + request_size: size, + blob_size: blob_info.compressed_data_size(), + })) + } + + fn submit_prefetch(blob: Arc, plan: FsCachePrefetchPlan) { + if !blob.is_prefetch_active() { + return; + } + + let count = plan.blob_size.div_ceil(plan.request_size); let mut blob_req = Vec::with_capacity(count as usize); let mut pre_offset = 0u64; for _i in 0..count { blob_req.push(BlobPrefetchRequest { - blob_id: blob_info.blob_id().to_owned(), + blob_id: plan.blob_id.clone(), offset: pre_offset, - len: cmp::min(size, blob_size - pre_offset), + len: cmp::min(plan.request_size, plan.blob_size - pre_offset), }); - pre_offset += size; - if pre_offset >= blob_size { + pre_offset += plan.request_size; + if pre_offset >= plan.blob_size { break; } } - let id = blob.blob_id(); - info!("fscache: start to prefetch data for blob {}", id); - if let Err(e) = blob.prefetch(blob.clone(), &blob_req, &[]) { - warn!("fscache: failed to prefetch data for blob {}, {}", id, e); + if !blob.is_prefetch_active() { + return; } - Ok(()) + info!("fscache: start to prefetch data for blob {}", plan.blob_id); + if let Err(e) = blob.prefetch(blob.clone(), &blob_req, &[]) { + warn!( + "fscache: failed to prefetch data for blob {}, {}", + plan.blob_id, e + ); + } } /// The `fscache` factory essentially creates a namespace for blob objects cached by the @@ -656,17 +749,31 @@ impl FsCacheHandler { } fn handle_close_request(&self, hdr: &FsCacheMsgHeader) { - let mut state = self.get_state(); + let data_blob = { + let state = self.get_state(); + match state.id_to_object_map.get(&hdr.object_id) { + // Safe to unwrap() because `id_to_config_map` and data blob entries in + // `id_to_object_map` are kept in consistence. + Some((FsCacheObject::DataBlob(fsblob), _)) => Some(( + fsblob.clone(), + state.id_to_config_map.get(&hdr.object_id).unwrap().clone(), + )), + Some((FsCacheObject::Bootstrap(_), _)) | None => None, + } + }; + + if let Some((fsblob, config)) = data_blob { + let (init_state, mut blob) = { + let guard = fsblob.read().unwrap(); + (guard.init_state.clone(), guard.get_blob_cache()) + }; + if blob.is_none() { + init_state.wait(); + blob = fsblob.read().unwrap().get_blob_cache(); + } - if let Some((FsCacheObject::DataBlob(fsblob), _)) = - state.id_to_object_map.remove(&hdr.object_id) - { - // Safe to unwrap() because `id_to_config_map` and `id_to_object_map` is kept - // in consistence. - let config = state.id_to_config_map.remove(&hdr.object_id).unwrap(); let factory_config = config.config_v2(); - let guard = fsblob.read().unwrap(); - match guard.get_blob_cache() { + match blob { Some(blob) => { if let Ok(cache_cfg) = factory_config.get_cache_config() { if cache_cfg.prefetch.enable { @@ -680,6 +787,14 @@ impl FsCacheHandler { _ => warn!("fscache: blob object not ready {}", hdr.object_id), } } + + let mut state = self.get_state(); + match state.id_to_object_map.remove(&hdr.object_id) { + Some((FsCacheObject::DataBlob(_), _)) => { + state.id_to_config_map.remove(&hdr.object_id); + } + Some((FsCacheObject::Bootstrap(_), _)) | None => {} + } } fn handle_read_request(&self, hdr: &FsCacheMsgHeader, msg: &FsCacheMsgRead) { @@ -695,8 +810,16 @@ impl FsCacheHandler { } Some((FsCacheObject::DataBlob(fsblob), u)) => { fd = u; - let guard = fsblob.read().unwrap(); - match guard.get_blob_cache() { + let (init_state, mut blob) = { + let guard = fsblob.read().unwrap(); + (guard.init_state.clone(), guard.get_blob_cache()) + }; + if blob.is_none() { + init_state.wait(); + blob = fsblob.read().unwrap().get_blob_cache(); + } + + match blob { Some(blob) => match blob.get_blob_object() { None => { warn!("fscache: internal error: cached object is not BlobCache objects")