@@ -14,9 +14,8 @@ use std::{
1414
1515use anyhow:: { anyhow, bail} ;
1616use async_trait:: async_trait;
17- use catalyst_types:: conversion:: from_saturating;
17+ use catalyst_types:: { conversion:: from_saturating, mmap_file :: MemoryMapFile } ;
1818use dashmap:: DashSet ;
19- use fmmap:: MmapFileExt ;
2019use memx:: memcmp;
2120use mithril_client:: {
2221 common:: CompressionAlgorithm , snapshot_downloader:: SnapshotDownloader , MithrilResult ,
@@ -134,8 +133,7 @@ impl Inner {
134133 self . ext_size . fetch_add ( entry_size, Ordering :: SeqCst ) ;
135134
136135 // Try and deduplicate the file if we can, otherwise just extract it.
137- if let Ok ( ( prev_mmap, _) ) =
138- Self :: can_deduplicate ( & rel_file, entry_size, prev_file. as_ref ( ) )
136+ if let Ok ( prev_mmap) = Self :: can_deduplicate ( & rel_file, entry_size, prev_file. as_ref ( ) )
139137 {
140138 let expected_file_size = from_saturating ( entry_size) ;
141139 let mut buf: Vec < u8 > = Vec :: with_capacity ( expected_file_size) ;
@@ -225,7 +223,7 @@ impl Inner {
225223 /// Check if a given path from the archive is able to be deduplicated.
226224fn can_deduplicate (
227225 rel_file : & Path , file_size : u64 , prev_file : Option < & PathBuf > ,
228- ) -> MithrilResult < ( fmmap :: MmapFile , u64 ) > {
226+ ) -> MithrilResult < MemoryMapFile > {
229227 // Can't dedup if the current file is not de-dupable (must be immutable)
230228 if rel_file. starts_with ( "immutable" ) {
231229 // Can't dedup if we don't have a previous file to dedup against.
@@ -234,8 +232,8 @@ impl Inner {
234232 // If the current file is not exactly the same as the previous file size, we
235233 // can't dedup.
236234 if file_size == current_size {
237- if let Ok ( pref_file_loaded) = mmap_open_sync ( prev_file) {
238- if pref_file_loaded. 1 == file_size {
235+ if let Ok ( pref_file_loaded) = Self :: mmap_open_sync ( prev_file) {
236+ if pref_file_loaded. size ( ) == file_size {
239237 return Ok ( pref_file_loaded) ;
240238 }
241239 }
@@ -245,6 +243,17 @@ impl Inner {
245243 }
246244 bail ! ( "Can not deduplicate." ) ;
247245 }
246+
247+ /// Open a file using mmap for performance.
248+ fn mmap_open_sync ( path : & Path ) -> MithrilResult < MemoryMapFile > {
249+ match MemoryMapFile :: try_from ( path) {
250+ Ok ( mmap_file) => Ok ( mmap_file) ,
251+ Err ( error) => {
252+ error ! ( error=%error, file=%path. to_string_lossy( ) , "Failed to open file" ) ;
253+ Err ( error. into ( ) )
254+ } ,
255+ }
256+ }
248257}
249258
250259/// A snapshot downloader that accelerates Download using `aria2`.
@@ -302,7 +311,17 @@ impl MithrilTurboDownloader {
302311 let target_dir = target_dir. to_owned ( ) ;
303312
304313 // This is fully synchronous IO, so do it on a sync thread.
305- let result = spawn_blocking ( move || inner. dl_and_dedup ( & location, & target_dir) ) . await ;
314+ let result = spawn_blocking ( move || {
315+ stats:: start_thread (
316+ inner. cfg . chain ,
317+ stats:: thread:: name:: MITHRIL_DL_DEDUP ,
318+ false ,
319+ ) ;
320+ let result = inner. dl_and_dedup ( & location, & target_dir) ;
321+ stats:: stop_thread ( inner. cfg . chain , stats:: thread:: name:: MITHRIL_DL_DEDUP ) ;
322+ result
323+ } )
324+ . await ;
306325
307326 if let Ok ( result) = result {
308327 return result;
@@ -321,20 +340,6 @@ fn get_file_size_sync(file: &Path) -> Option<u64> {
321340 Some ( metadata. len ( ) )
322341}
323342
324- /// Open a file using mmap for performance.
325- fn mmap_open_sync ( path : & Path ) -> MithrilResult < ( fmmap:: MmapFile , u64 ) > {
326- match fmmap:: MmapFile :: open_with_options ( path, fmmap:: Options :: new ( ) . read ( true ) . populate ( ) ) {
327- Ok ( file) => {
328- let len = file. len ( ) as u64 ;
329- Ok ( ( file, len) )
330- } ,
331- Err ( error) => {
332- error ! ( error=%error, file=%path. to_string_lossy( ) , "Failed to open file" ) ;
333- Err ( error. into ( ) )
334- } ,
335- }
336- }
337-
338343#[ async_trait]
339344impl SnapshotDownloader for MithrilTurboDownloader {
340345 async fn download_unpack (
@@ -366,9 +371,9 @@ impl SnapshotDownloader for MithrilTurboDownloader {
366371
367372 async fn probe ( & self , location : & str ) -> MithrilResult < ( ) > {
368373 debug ! ( "Probe Snapshot location='{location}'." ) ;
369-
370374 let dl_config = self . inner . cfg . dl_config . clone ( ) . unwrap_or_default ( ) ;
371- let dl_processor = ParallelDownloadProcessor :: new ( location, dl_config) . await ?;
375+ let dl_processor =
376+ ParallelDownloadProcessor :: new ( location, dl_config, self . inner . cfg . chain ) . await ?;
372377
373378 // Decompress and extract and de-dupe each file in the archive.
374379 stats:: mithril_extract_started ( self . inner . cfg . chain ) ;
0 commit comments