/// Preallocates the files listed in the manifest and copies the chunks
/// that we have locally.
/// Returns a set of chunks that still need to be fetched
Preallocates the files listed in the manifest and copies the chunks
that we have locally.
Returns a set of chunks that still need to be fetched
Preallocates the files listed in the manifest and copies the chunks that we have locally . Returns a set of chunks that still need to be fetched
fn initialize_state_on_disk(&mut self, manifest_new: &Manifest) -> HashSet<usize> {
        Self::preallocate_layout(&self.log, &self.root, manifest_new);
        let state_sync_size_fetch = self.metrics.state_sync_size.with_label_values(&["fetch"]);
        let state_sync_size_copy = self.metrics.state_sync_size.with_label_values(&["copy"]);
        let state_sync_size_preallocate = self
            .metrics
            .state_sync_size
            .with_label_values(&["preallocate"]);
        let total_bytes: u64 = manifest_new.file_table.iter().map(|f| f.size_bytes).sum();
        self.metrics
            .state_sync_remaining
            .add(manifest_new.chunk_table.len() as i64);
        // Get the cache line. We now own an Arc on that cache line, so we are extending
        // the lifetime of the data on disk as long as we keep this cache line in scope
        let cache = self.state_sync_refs.cache.read().get();
        // A little helper struct of all we need to know to copy out of
        struct DiffData<'a> {
            manifest_old: &'a Manifest,
            missing_chunks: HashSet<usize>,
            root_old: PathBuf,
            height_old: Height,
            validate_data: bool,
        }
        // Get a DiffData from the cache or checkpoint_ref, or neither
        let diff_data: Option<DiffData> =
            match (cache.as_ref(), self.manifest_with_checkpoint_ref.as_ref()) {
                (Some(cache_entry), Some((checkpoint_manifest, checkpoint_ref))) => {
                    let cache_height = cache_entry.height;
                    let checkpoint_height = checkpoint_ref.0.height;
                    if cache_height > checkpoint_height {
                        // The cache will have missing chunks. However, it likely started from a
                        // DiffScript with the same checkpoint we have now, so there should be more
                        // relevant chunks in the cache than in the checkpoint.
                        // This is just a heuristic however, as the cached chunks might have been
                        // initialized with a DiffScript from an older checkpoint.
                        Some(DiffData {
                            manifest_old: &cache_entry.manifest,
                            missing_chunks: cache_entry.missing_chunks.clone(),
                            // The data at root_old will live at least as long as the
                            // StateSyncCacheEntry, so cloning the path is safe
                            root_old: cache_entry.path().to_path_buf(),
                            height_old: cache_entry.height,
                            validate_data: false,
                        })
                    } else {
                        // This should be a special case that can only happen if the source of the
                        // checkpoint is outside of state sync, as otherwise we would have cleared
                        // the cache upon successfully syncing a state.
                        let checkpoint_old = checkpoint_ref
                            .0
                            .state_layout
                            .checkpoint(checkpoint_height)
                            .unwrap_or_else(|err| {
                                fatal!(
                                    &self.log,
                                    "Failed to get checkpoint path for height {}: {}",
                                    checkpoint_height,
                                    err
                                );
                            });
                        Some(DiffData {
                            manifest_old: checkpoint_manifest,
                            missing_chunks: Default::default(),
                            root_old: checkpoint_old.raw_path().to_path_buf(),
                            height_old: checkpoint_height,
                            validate_data: true,
                        })
                    }
                }
                (Some(cache_entry), None) => Some(DiffData {
                    manifest_old: &cache_entry.manifest,
                    missing_chunks: cache_entry.missing_chunks.clone(),
                    root_old: cache_entry.path().to_path_buf(),
                    height_old: cache_entry.height,
                    validate_data: false,
                }),
                (None, Some((checkpoint_manifest, checkpoint_ref))) => {
                    let checkpoint_height = checkpoint_ref.0.height;
                    let checkpoint_old = checkpoint_ref
                        .0
                        .state_layout
                        .checkpoint(checkpoint_height)
                        .unwrap_or_else(|err| {
                            fatal!(
                                &self.log,
                                "Failed to get checkpoint path for height {}: {}",
                                checkpoint_height,
                                err
                            );
                        });
                    Some(DiffData {
                        manifest_old: checkpoint_manifest,
                        missing_chunks: Default::default(),
                        // The data in root_old will live at least as long as self.checkpoint_ref,
                        // so cloning here is safe
                        root_old: checkpoint_old.raw_path().to_path_buf(),
                        height_old: checkpoint_height,
                        validate_data: !self
                            .state_sync_refs
                            .cache
                            .read()
                            .state_is_fetched(checkpoint_height),
                    })
                }
                (None, None) => None,
            };
        if let Some(DiffData {
            manifest_old,
            missing_chunks,
            root_old,
            height_old,
            validate_data,
        }) = diff_data
        {
            info!(
                self.log,
                "Initializing state sync for height {} based on {} at height {}",
                self.height,
                if missing_chunks.is_empty() {
                    "checkpoint"
                } else {
                    "cache"
                },
                height_old
            );
            let diff_script =
                crate::manifest::diff_manifest(manifest_old, &missing_chunks, manifest_new);
            debug!(
                self.log,
                "State sync diff script (@{} -> @{}): {:?}", height_old, self.height, diff_script
            );
            // diff_script contains indices into the manifest chunk table, but p2p
            // counts the manifest itself as chunk 0, so all other chunk indices are
            // shifted by 1
            let mut fetch_chunks = diff_script.fetch_chunks.iter().map(|i| *i + 1).collect();
            let mut thread_pool = self.thread_pool.lock().unwrap();
            Self::copy_files(
                &self.log,
                &self.metrics,
                &mut thread_pool,
                &root_old,
                &self.root,
                manifest_old,
                manifest_new,
                &diff_script,
                validate_data,
                &mut fetch_chunks,
            );
            Self::copy_chunks(
                &self.log,
                &self.metrics,
                &mut thread_pool,
                &root_old,
                &self.root,
                manifest_old,
                manifest_new,
                &diff_script,
                validate_data,
                &mut fetch_chunks,
            );
            let diff_bytes: u64 = diff_script
                .fetch_chunks
                .iter()
                .map(|i| manifest_new.chunk_table[*i].size_bytes as u64)
                .sum();
            let preallocate_bytes = diff_script.zeros_chunks * crate::manifest::DEFAULT_CHUNK_SIZE;
            state_sync_size_fetch.inc_by(diff_bytes);
            state_sync_size_preallocate.inc_by(preallocate_bytes as u64);
            state_sync_size_copy.inc_by(total_bytes - diff_bytes - preallocate_bytes as u64);
            self.metrics
                .state_sync_remaining
                .sub(diff_script.zeros_chunks as i64);
            fetch_chunks
        } else {
            info!(
                self.log,
                "Initializing state sync for height {} without any caches or previous checkpoints",
                self.height
            );
            let non_zero_chunks = filter_out_zero_chunks(manifest_new);
            let diff_bytes: u64 = non_zero_chunks
                .iter()
                .map(|i| manifest_new.chunk_table[*i].size_bytes as u64)
                .sum();
            state_sync_size_fetch.inc_by(diff_bytes);
            state_sync_size_preallocate.inc_by(total_bytes - diff_bytes);
            let zeros_chunks = manifest_new.chunk_table.len() - non_zero_chunks.len();
            self.metrics.state_sync_remaining.sub(zeros_chunks as i64);
            non_zero_chunks.iter().map(|i| *i + 1).collect()
        }
    }
fn initialize_state_on_disk ( & mut self , manifest_new : & Manifest ) -> HashSet < usize > { Self :: preallocate_layout ( & self . log , & self . root , manifest_new ) ; let state_sync_size_fetch = self . metrics . state_sync_size . with_label_values ( & [ " " ] ) ; let state_sync_size_copy = self . metrics . state_sync_size . with_label_values ( & [ " " ] ) ; let state_sync_size_preallocate = self . metrics . state_sync_size . with_label_values ( & [ " " ] ) ; let total_bytes : u64 = manifest_new . file_table . iter ( ) . map ( | f | f . size_bytes ) . sum ( ) ; self . metrics . state_sync_remaining . add ( manifest_new . chunk_table . len ( ) as i64 ) ; // Get the cache line. We now own an Arc on that cache line, so we are extending // the lifetime of the data on disk as long as we keep this cache line in scope let cache = self . state_sync_refs . cache . read ( ) . get ( ) ; // A little helper struct of all we need to know to copy out of struct DiffData < ' a > { manifest_old : & ' a Manifest , missing_chunks : HashSet < usize > , root_old : PathBuf , height_old : Height , validate_data : bool , } // Get a DiffData from the cache or checkpoint_ref, or neither let diff_data : Option < DiffData > = match ( cache . as_ref ( ) , self . manifest_with_checkpoint_ref . as_ref ( ) ) { ( Some ( cache_entry ) , Some ( ( checkpoint_manifest , checkpoint_ref ) ) ) => { let cache_height = cache_entry . height ; let checkpoint_height = checkpoint_ref . 0 . height ; if cache_height > checkpoint_height { // The cache will have missing chunks. However, it likely started from a // DiffScript with the same checkpoint we have now, so there should be more // relevant chunks in the cache than in the checkpoint. // This is just a heuristic however, as the cached chunks might have been // initialized with a DiffScript from an older checkpoint. Some ( DiffData { manifest_old : & cache_entry . manifest , missing_chunks : cache_entry . missing_chunks . clone ( ) , // The data at root_old will live at least as long as the // StateSyncCacheEntry, so cloning the path is safe root_old : cache_entry . path ( ) . to_path_buf ( ) , height_old : cache_entry . height , validate_data : false , } ) } else { // This should be a special case that can only happen if the source of the // checkpoint is outside of state sync, as otherwise we would have cleared // the cache upon successfully syncing a state. let checkpoint_old = checkpoint_ref . 0 . state_layout . checkpoint ( checkpoint_height ) . unwrap_or_else ( | err | { fatal ! ( self log " " checkpoint_height err ) ; } ) ; Some ( DiffData { manifest_old : checkpoint_manifest , missing_chunks : Default :: default ( ) , root_old : checkpoint_old . raw_path ( ) . to_path_buf ( ) , height_old : checkpoint_height , validate_data : true , } ) } } ( Some ( cache_entry ) , None ) => Some ( DiffData { manifest_old : & cache_entry . manifest , missing_chunks : cache_entry . missing_chunks . clone ( ) , root_old : cache_entry . path ( ) . to_path_buf ( ) , height_old : cache_entry . height , validate_data : false , } ) , ( None , Some ( ( checkpoint_manifest , checkpoint_ref ) ) ) => { let checkpoint_height = checkpoint_ref . 0 . height ; let checkpoint_old = checkpoint_ref . 0 . state_layout . checkpoint ( checkpoint_height ) . unwrap_or_else ( | err | { fatal ! ( self log " " checkpoint_height err ) ; } ) ; Some ( DiffData { manifest_old : checkpoint_manifest , missing_chunks : Default :: default ( ) , // The data in root_old will live at least as long as self.checkpoint_ref, // so cloning here is safe root_old : checkpoint_old . raw_path ( ) . to_path_buf ( ) , height_old : checkpoint_height , validate_data : ! self . state_sync_refs . cache . read ( ) . state_is_fetched ( checkpoint_height ) , } ) } ( None , None ) => None , } ; if let Some ( DiffData { manifest_old , missing_chunks , root_old , height_old , validate_data , } ) = diff_data { info ! ( self log " " self height if missing_chunks is_empty ( ) { " " } else { " " } height_old ) ; let diff_script = crate :: manifest :: diff_manifest ( manifest_old , & missing_chunks , manifest_new ) ; debug ! ( self log " " height_old self height diff_script ) ; // diff_script contains indices into the manifest chunk table, but p2p // counts the manifest itself as chunk 0, so all other chunk indices are // shifted by 1 let mut fetch_chunks = diff_script . fetch_chunks . iter ( ) . map ( | i | * i + 1 ) . collect ( ) ; let mut thread_pool = self . thread_pool . lock ( ) . unwrap ( ) ; Self :: copy_files ( & self . log , & self . metrics , & mut thread_pool , & root_old , & self . root , manifest_old , manifest_new , & diff_script , validate_data , & mut fetch_chunks , ) ; Self :: copy_chunks ( & self . log , & self . metrics , & mut thread_pool , & root_old , & self . root , manifest_old , manifest_new , & diff_script , validate_data , & mut fetch_chunks , ) ; let diff_bytes : u64 = diff_script . fetch_chunks . iter ( ) . map ( | i | manifest_new . chunk_table [ * i ] . size_bytes as u64 ) . sum ( ) ; let preallocate_bytes = diff_script . zeros_chunks * crate :: manifest :: DEFAULT_CHUNK_SIZE ; state_sync_size_fetch . inc_by ( diff_bytes ) ; state_sync_size_preallocate . inc_by ( preallocate_bytes as u64 ) ; state_sync_size_copy . inc_by ( total_bytes - diff_bytes - preallocate_bytes as u64 ) ; self . metrics . state_sync_remaining . sub ( diff_script . zeros_chunks as i64 ) ; fetch_chunks } else { info ! ( self log " " self height ) ; let non_zero_chunks = filter_out_zero_chunks ( manifest_new ) ; let diff_bytes : u64 = non_zero_chunks . iter ( ) . map ( | i | manifest_new . chunk_table [ * i ] . size_bytes as u64 ) . sum ( ) ; state_sync_size_fetch . inc_by ( diff_bytes ) ; state_sync_size_preallocate . inc_by ( total_bytes - diff_bytes ) ; let zeros_chunks = manifest_new . chunk_table . len ( ) - non_zero_chunks . len ( ) ; self . metrics . state_sync_remaining . sub ( zeros_chunks as i64 ) ; non_zero_chunks . iter ( ) . map ( | i | * i + 1 ) . collect ( ) } }
Preallocates the files listed in the manifest and copies the chunks
that we have locally.
Preallocates the files listed in the manifest and copies the chunks that we have locally .
&mut self
manifest_new
&Manifest