/// Bypass memory blob cache index, fetch blocks from backend and directly
/// mirror them into blob cache file.
/// Continuous chunks may be compressed or not.
Bypass memory blob cache index, fetch blocks from backend and directly
mirror them into blob cache file.
Continuous chunks may be compressed or not.
Bypass memory blob cache index fetch blocks from backend and directly mirror them into blob cache file . Continuous chunks may be compressed or not .
fn prefetch(&self, bios: &mut [RafsBio]) -> Result<usize> {
        let (mut tx, rx) = spmc::channel::<MergedBackendRequest>();
        for num in 0..self.prefetch_worker.threads_count {
            let cache = Arc::clone(&self.cache);
            // Clone cache fulfils our requirement that invoke `read_chunks` and it's
            // hard to move `self` into closure.
            let cache_cloned = Arc::new(self.clone());
            let rx = rx.clone();
            let _thread = thread::Builder::new()
                .name(format!("prefetch_thread_{}", num))
                .spawn(move || {
                    'wait_mr: while let Ok(mr) = rx.recv() {
                        let blob_offset = mr.blob_offset;
                        let blob_size = mr.blob_size;
                        let continuous_chunks = &mr.chunks;
                        let blob_id = &mr.blob_id;
                        let mut issue_batch: bool;
                        trace!(
                            "Merged req id {} req offset {} size {}",
                            blob_id,
                            blob_offset,
                            blob_size
                        );
                        issue_batch = false;
                        // An immature trick here to detect if chunk already resides in
                        // blob cache file. Hopefully, we can have a more clever and agile
                        // way in the future. Principe is that if all chunks are Ready,
                        // abort this Merged Request. It might involve extra stress
                        // to local file system.
                        for c in continuous_chunks {
                            let d_size = c.decompress_size() as usize;
                            let entry = cache.write().expect("Expect cache lock not poisoned").set(
                                blob_id,
                                c.clone(),
                                cache_cloned.backend(),
                            );
                            if let Ok(entry) = entry {
                                let entry = entry.lock().unwrap();
                                if entry.is_ready() {
                                    continue;
                                }
                                let fd = entry.fd;
                                let chunk = entry.chunk.clone();
                                drop(entry);
                                if cache_cloned
                                    .read_blobcache_chunk(
                                        fd,
                                        chunk.as_ref(),
                                        alloc_buf(d_size).as_mut_slice(),
                                        cache_cloned.need_validate(),
                                    )
                                    .is_err()
                                {
                                    // Aha, we have a not integrated chunk here. Issue the entire
                                    // merged request from backend to boost.
                                    issue_batch = true;
                                    break;
                                }
                            }
                        }
                        if !issue_batch {
                            continue 'wait_mr;
                        }
                        if let Ok(chunks) = cache_cloned.read_chunks(
                            blob_id,
                            blob_offset,
                            blob_size as usize,
                            &continuous_chunks,
                        ) {
                            for (i, c) in continuous_chunks.iter().enumerate() {
                                let mut cache_guard =
                                    cache.write().expect("Expect cache lock not poisoned");
                                if let Ok(entry) = cache_guard
                                    .set(blob_id, c.clone(), cache_cloned.backend())
                                    .map_err(|_| error!("Set cache index error!"))
                                {
                                    let mut entry = entry.lock().unwrap();
                                    if !entry.is_ready() {
                                        let offset = if cache_cloned.is_compressed {
                                            entry.chunk.compress_offset()
                                        } else {
                                            entry.chunk.decompress_offset()
                                        };
                                        if let Err(err) = entry.cache(chunks[i].as_slice(), offset)
                                        {
                                            error!("Failed to cache chunk: {}", err);
                                        }
                                    }
                                }
                            }
                        }
                    }
                    info!("Prefetch thread exits.")
                });
        }
        // Ideally, prefetch task can run within a separated thread from loading prefetch table.
        // However, due to current implementation, doing so needs modifying key data structure like
        // `Superblock` on `Rafs`. So let's suspend this action.
        let mut bios = bios.to_vec();
        let merging_size = self.prefetch_worker.merging_size;
        let _thread = thread::Builder::new().spawn({
            move || {
                generate_merged_requests(bios.as_mut_slice(), &mut tx, merging_size);
            }
        });
        Ok(0)
    }
fn prefetch ( & self , bios : & mut [ RafsBio ] ) -> Result < usize > { let ( mut tx , rx ) = spmc :: channel :: < MergedBackendRequest > ( ) ; for num in 0 .. self . prefetch_worker . threads_count { let cache = Arc :: clone ( & self . cache ) ; // Clone cache fulfils our requirement that invoke `read_chunks` and it's // hard to move `self` into closure. let cache_cloned = Arc :: new ( self . clone ( ) ) ; let rx = rx . clone ( ) ; let _thread = thread :: Builder :: new ( ) . name ( format ! ( " " num ) ) . spawn ( move | | { ' wait_mr : while let Ok ( mr ) = rx . recv ( ) { let blob_offset = mr . blob_offset ; let blob_size = mr . blob_size ; let continuous_chunks = & mr . chunks ; let blob_id = & mr . blob_id ; let mut issue_batch : bool ; trace ! ( " " blob_id blob_offset blob_size ) ; issue_batch = false ; // An immature trick here to detect if chunk already resides in // blob cache file. Hopefully, we can have a more clever and agile // way in the future. Principe is that if all chunks are Ready, // abort this Merged Request. It might involve extra stress // to local file system. for c in continuous_chunks { let d_size = c . decompress_size ( ) as usize ; let entry = cache . write ( ) . expect ( " " ) . set ( blob_id , c . clone ( ) , cache_cloned . backend ( ) , ) ; if let Ok ( entry ) = entry { let entry = entry . lock ( ) . unwrap ( ) ; if entry . is_ready ( ) { continue ; } let fd = entry . fd ; let chunk = entry . chunk . clone ( ) ; drop ( entry ) ; if cache_cloned . read_blobcache_chunk ( fd , chunk . as_ref ( ) , alloc_buf ( d_size ) . as_mut_slice ( ) , cache_cloned . need_validate ( ) , ) . is_err ( ) { // Aha, we have a not integrated chunk here. Issue the entire // merged request from backend to boost. issue_batch = true ; break ; } } } if ! issue_batch { continue ' wait_mr ; } if let Ok ( chunks ) = cache_cloned . read_chunks ( blob_id , blob_offset , blob_size as usize , & continuous_chunks , ) { for ( i , c ) in continuous_chunks . iter ( ) . enumerate ( ) { let mut cache_guard = cache . write ( ) . expect ( " " ) ; if let Ok ( entry ) = cache_guard . set ( blob_id , c . clone ( ) , cache_cloned . backend ( ) ) . map_err ( | _ | error ! ( " " ) ) { let mut entry = entry . lock ( ) . unwrap ( ) ; if ! entry . is_ready ( ) { let offset = if cache_cloned . is_compressed { entry . chunk . compress_offset ( ) } else { entry . chunk . decompress_offset ( ) } ; if let Err ( err ) = entry . cache ( chunks [ i ] . as_slice ( ) , offset ) { error ! ( " " err ) ; } } } } } } info ! ( " " ) } ) ; } // Ideally, prefetch task can run within a separated thread from loading prefetch table. // However, due to current implementation, doing so needs modifying key data structure like // `Superblock` on `Rafs`. So let's suspend this action. let mut bios = bios . to_vec ( ) ; let merging_size = self . prefetch_worker . merging_size ; let _thread = thread :: Builder :: new ( ) . spawn ( { move | | { generate_merged_requests ( bios . as_mut_slice ( ) , & mut tx , merging_size ) ; } } ) ; Ok ( 0 ) }
Bypass memory blob cache index, fetch blocks from backend and directly
mirror them into blob cache file.
Bypass memory blob cache index fetch blocks from backend and directly mirror them into blob cache file .
&self
bios
&mut [RafsBio]