gix_pack/index/traverse/
with_index.rs

1use std::sync::atomic::{AtomicBool, Ordering};
2
3use gix_features::{parallel, progress::DynNestedProgress};
4
5use super::Error;
6use crate::{
7    cache::delta::traverse,
8    index::{self, traverse::Outcome, util::index_entries_sorted_by_offset_ascending},
9};
10
11/// Traversal options for [`traverse_with_index()`][index::File::traverse_with_index()]
12#[derive(Default)]
13pub struct Options {
14    /// If `Some`, only use the given number of threads. Otherwise, the number of threads to use will be selected based on
15    /// the number of available logical cores.
16    pub thread_limit: Option<usize>,
17    /// The kinds of safety checks to perform.
18    pub check: crate::index::traverse::SafetyCheck,
19    /// If `Some`, rejects individual allocations above the given number of bytes while resolving decoded object and
20    /// delta result buffers. `Some(0)` rejects all non-empty allocations.
21    pub alloc_limit_bytes: Option<usize>,
22}
23
24/// The progress ids used in [`index::File::traverse_with_index()`].
25///
26/// Use this information to selectively extract the progress of interest in case the parent application has custom visualization.
27#[derive(Debug, Copy, Clone)]
28pub enum ProgressId {
29    /// The amount of bytes currently processed to generate a checksum of the *pack data file*.
30    HashPackDataBytes,
31    /// The amount of bytes currently processed to generate a checksum of the *pack index file*.
32    HashPackIndexBytes,
33    /// Collect all object hashes into a vector and sort it by their pack offset.
34    CollectSortedIndexEntries,
35    /// Count the objects processed when building a cache tree from all objects in a pack index.
36    TreeFromOffsetsObjects,
37    /// The amount of objects which were decoded.
38    DecodedObjects,
39    /// The amount of bytes that were decoded in total, as the sum of all bytes to represent all decoded objects.
40    DecodedBytes,
41}
42
43impl From<ProgressId> for gix_features::progress::Id {
44    fn from(v: ProgressId) -> Self {
45        match v {
46            ProgressId::HashPackDataBytes => *b"PTHP",
47            ProgressId::HashPackIndexBytes => *b"PTHI",
48            ProgressId::CollectSortedIndexEntries => *b"PTCE",
49            ProgressId::TreeFromOffsetsObjects => *b"PTDI",
50            ProgressId::DecodedObjects => *b"PTRO",
51            ProgressId::DecodedBytes => *b"PTDB",
52        }
53    }
54}
55
56/// Traversal with index
57impl<T> index::File<T>
58where
59    T: crate::FileData + Sync,
60{
61    /// Iterate through all _decoded objects_ in the given `pack` and handle them with a `Processor`, using an index to reduce waste
62    /// at the cost of memory.
63    ///
64    /// For more details, see the documentation on the [`traverse()`][index::File::traverse()] method.
65    pub fn traverse_with_index<Processor, E, D>(
66        &self,
67        pack: &crate::data::File<D>,
68        mut processor: Processor,
69        progress: &mut dyn DynNestedProgress,
70        should_interrupt: &AtomicBool,
71        Options {
72            check,
73            thread_limit,
74            alloc_limit_bytes,
75        }: Options,
76    ) -> Result<Outcome, Error<E>>
77    where
78        Processor: FnMut(gix_object::Kind, &[u8], &index::Entry, &dyn gix_features::progress::Progress) -> Result<(), E>
79            + Send
80            + Clone,
81        E: std::error::Error + Send + Sync + 'static,
82        D: crate::FileData + Send + Sync,
83    {
84        let (verify_result, traversal_result) = parallel::join(
85            {
86                let mut pack_progress = progress.add_child_with_id(
87                    format!("Hash of pack '{}'", crate::source_name(pack.path())),
88                    ProgressId::HashPackDataBytes.into(),
89                );
90                let mut index_progress = progress.add_child_with_id(
91                    format!("Hash of index '{}'", crate::source_name(&self.path)),
92                    ProgressId::HashPackIndexBytes.into(),
93                );
94                move || {
95                    let res =
96                        self.possibly_verify(pack, check, &mut pack_progress, &mut index_progress, should_interrupt);
97                    if res.is_err() {
98                        should_interrupt.store(true, Ordering::SeqCst);
99                    }
100                    res
101                }
102            },
103            || -> Result<_, Error<_>> {
104                let sorted_entries = index_entries_sorted_by_offset_ascending(
105                    self,
106                    &mut progress.add_child_with_id(
107                        "collecting sorted index".into(),
108                        ProgressId::CollectSortedIndexEntries.into(),
109                    ),
110                ); /* Pack Traverse Collect sorted Entries */
111                let tree = crate::cache::delta::Tree::from_offsets_in_pack(
112                    pack.path(),
113                    sorted_entries.into_iter().map(Entry::from),
114                    &|e| e.index_entry.pack_offset,
115                    &|id| self.lookup(id).map(|idx| self.pack_offset_at_index(idx)),
116                    &mut progress.add_child_with_id("indexing".into(), ProgressId::TreeFromOffsetsObjects.into()),
117                    should_interrupt,
118                    self.object_hash,
119                )?;
120                let mut outcome = digest_statistics(tree.traverse(
121                    |slice, pack| pack.entry_slice(slice),
122                    pack,
123                    pack.pack_end() as u64,
124                    move |data,
125                          progress,
126                          traverse::Context {
127                              entry: pack_entry,
128                              entry_end,
129                              decompressed: bytes,
130                              level,
131                          }| {
132                        let object_kind = pack_entry.header.as_kind().expect("non-delta object");
133                        data.level = level;
134                        data.decompressed_size = pack_entry.decompressed_size;
135                        data.object_kind = object_kind;
136                        data.compressed_size = entry_end - pack_entry.data_offset;
137                        data.object_size = bytes.len() as u64;
138                        let result = index::traverse::process_entry(
139                            check,
140                            object_kind,
141                            bytes,
142                            &data.index_entry,
143                            || {
144                                // TODO: Fix this - we overwrite the header of 'data' which also changes the computed entry size,
145                                // causing index and pack to seemingly mismatch. This is surprising, and should be done differently.
146                                // debug_assert_eq!(&data.index_entry.pack_offset, &pack_entry.pack_offset());
147                                gix_features::hash::crc32(
148                                    pack.entry_slice(data.index_entry.pack_offset..entry_end)
149                                        .expect("slice pointing into the pack (by now data is verified)"),
150                                )
151                            },
152                            progress,
153                            &mut processor,
154                        );
155                        match result {
156                            Err(err @ Error::PackDecode { .. }) if !check.fatal_decode_error() => {
157                                progress.info(format!("Ignoring decode error: {err}"));
158                                Ok(())
159                            }
160                            res => res,
161                        }
162                    },
163                    traverse::Options {
164                        object_progress: Box::new(
165                            progress.add_child_with_id("Resolving".into(), ProgressId::DecodedObjects.into()),
166                        ),
167                        size_progress:
168                            &mut progress.add_child_with_id("Decoding".into(), ProgressId::DecodedBytes.into()),
169                        thread_limit,
170                        should_interrupt,
171                        object_hash: self.object_hash,
172                        alloc_limit_bytes,
173                    },
174                )?);
175                outcome.pack_size = pack.data_len() as u64;
176                Ok(outcome)
177            },
178        );
179        Ok(Outcome {
180            actual_index_checksum: verify_result?,
181            statistics: traversal_result?,
182        })
183    }
184}
185
186struct Entry {
187    index_entry: crate::index::Entry,
188    object_kind: gix_object::Kind,
189    object_size: u64,
190    decompressed_size: u64,
191    compressed_size: u64,
192    level: u16,
193}
194
195impl From<crate::index::Entry> for Entry {
196    fn from(index_entry: crate::index::Entry) -> Self {
197        Entry {
198            index_entry,
199            level: 0,
200            object_kind: gix_object::Kind::Tree,
201            object_size: 0,
202            decompressed_size: 0,
203            compressed_size: 0,
204        }
205    }
206}
207
208fn digest_statistics(traverse::Outcome { roots, children }: traverse::Outcome<Entry>) -> index::traverse::Statistics {
209    let mut res = index::traverse::Statistics::default();
210    let average = &mut res.average;
211    for item in roots.iter().chain(children.iter()) {
212        res.total_compressed_entries_size += item.data.compressed_size;
213        res.total_decompressed_entries_size += item.data.decompressed_size;
214        res.total_object_size += item.data.object_size;
215        *res.objects_per_chain_length
216            .entry(u32::from(item.data.level))
217            .or_insert(0) += 1;
218
219        average.decompressed_size += item.data.decompressed_size;
220        average.compressed_size += item.data.compressed_size as usize;
221        average.object_size += item.data.object_size;
222        average.num_deltas += u32::from(item.data.level);
223        use gix_object::Kind::*;
224        match item.data.object_kind {
225            Blob => res.num_blobs += 1,
226            Tree => res.num_trees += 1,
227            Tag => res.num_tags += 1,
228            Commit => res.num_commits += 1,
229        }
230    }
231
232    let num_nodes = roots.len() + children.len();
233    average.decompressed_size /= num_nodes as u64;
234    average.compressed_size /= num_nodes;
235    average.object_size /= num_nodes as u64;
236    average.num_deltas /= num_nodes as u32;
237
238    res
239}
gix_pack/index/traverse/with_index.rs

gix_pack/index/traverse/
with_index.rs
