use std::sync::atomic::{AtomicBool, Ordering}; use git_features::{ parallel::in_parallel_with_slice, progress::{self, Progress}, threading::{lock, Mutable, OwnShared}, }; use crate::{ cache::delta::{traverse::util::ItemSliceSend, Item, Tree}, data::EntryRange, }; mod resolve; pub(crate) mod util; /// Returned by [`Tree::traverse()`] #[derive(thiserror::Error, Debug)] #[allow(missing_docs)] pub enum Error { #[error("{message}")] ZlibInflate { source: git_features::zlib::inflate::Error, message: &'static str, }, #[error("The resolver failed to obtain the pack entry bytes for the entry at {pack_offset}")] ResolveFailed { pack_offset: u64 }, #[error("One of the object inspectors failed")] Inspect(#[from] Box), #[error("Interrupted")] Interrupted, #[error( "The base at {base_pack_offset} was referred to by a ref-delta, but it was never added to the tree as if the pack was still thin." )] OutOfPackRefDelta { /// The base's offset which was from a resolved ref-delta that didn't actually get added to the tree base_pack_offset: crate::data::Offset, }, } /// Additional context passed to the `inspect_object(…)` function of the [`Tree::traverse()`] method. pub struct Context<'a, S> { /// The pack entry describing the object pub entry: &'a crate::data::Entry, /// The offset at which `entry` ends in the pack, useful to learn about the exact range of `entry` within the pack. pub entry_end: u64, /// The decompressed object itself, ready to be decoded. pub decompressed: &'a [u8], /// Custom state known to the function pub state: &'a mut S, /// The depth at which this object resides in the delta-tree. It represents the amount of base objects, with 0 indicating /// an 'undeltified' object, and higher values indicating delta objects with the given amount of bases. pub level: u16, } /// Options for [`Tree::traverse()`]. pub struct Options<'a, P1, P2> { /// is a progress instance to track progress for each object in the traversal. pub object_progress: P1, /// is a progress instance to track the overall progress. pub size_progress: P2, /// If `Some`, only use the given amount of threads. Otherwise, the amount of threads to use will be selected based on /// the amount of available logical cores. pub thread_limit: Option, /// Abort the operation if the value is `true`. pub should_interrupt: &'a AtomicBool, /// specifies what kind of hashes we expect to be stored in oid-delta entries, which is viable to decoding them /// with the correct size. pub object_hash: git_hash::Kind, } /// The outcome of [`Tree::traverse()`] pub struct Outcome { /// The items that have no children in the pack, i.e. base objects. pub roots: Vec>, /// The items that children to a root object, i.e. delta objects. pub children: Vec>, } impl Tree where T: Send, { /// Traverse this tree of delta objects with a function `inspect_object` to process each object at will. /// /// * `should_run_in_parallel() -> bool` returns true if the underlying pack is big enough to warrant parallel traversal at all. /// * `resolve(EntrySlice, &mut Vec) -> Option<()>` resolves the bytes in the pack for the given `EntrySlice` and stores them in the /// output vector. It returns `Some(())` if the object existed in the pack, or `None` to indicate a resolution error, which would abort the /// operation as well. /// * `pack_entries_end` marks one-past-the-last byte of the last entry in the pack, as the last entries size would otherwise /// be unknown as it's not part of the index file. /// * `new_thread_state() -> State` is a function to create state to be used in each thread, invoked once per thread. /// * `inspect_object(node_data: &mut T, progress: Progress, context: Context) -> Result<(), CustomError>` is a function /// running for each thread receiving fully decoded objects along with contextual information, which either succceeds with `Ok(())` /// or returns a `CustomError`. /// Note that `node_data` can be modified to allow storing maintaining computation results on a per-object basis. /// /// This method returns a vector of all tree items, along with their potentially modified custom node data. /// /// _Note_ that this method consumed the Tree to assure safe parallel traversal with mutation support. pub fn traverse( mut self, resolve: F, pack_entries_end: u64, new_thread_state: impl Fn() -> S + Send + Clone, inspect_object: MBFN, Options { thread_limit, object_progress, mut size_progress, should_interrupt, object_hash, }: Options<'_, P1, P2>, ) -> Result, Error> where F: for<'r> Fn(EntryRange, &'r mut Vec) -> Option<()> + Send + Clone, P1: Progress, P2: Progress, MBFN: Fn(&mut T, &mut ::SubProgress, Context<'_, S>) -> Result<(), E> + Send + Clone, E: std::error::Error + Send + Sync + 'static, { self.set_pack_entries_end_and_resolve_ref_offsets(pack_entries_end)?; let object_progress = OwnShared::new(Mutable::new(object_progress)); let num_objects = self.num_items(); let object_counter = { let mut progress = lock(&object_progress); progress.init(Some(num_objects), progress::count("objects")); progress.counter() }; size_progress.init(None, progress::bytes()); let size_counter = size_progress.counter(); let child_items = self.child_items.as_mut_slice(); let start = std::time::Instant::now(); in_parallel_with_slice( &mut self.root_items, thread_limit, { let object_progress = object_progress.clone(); let child_items = ItemSliceSend(child_items as *mut [Item]); move |thread_index| { ( Vec::::with_capacity(4096), lock(&object_progress).add_child(format!("thread {}", thread_index)), new_thread_state(), resolve.clone(), inspect_object.clone(), ItemSliceSend(child_items.0), ) } }, { move |node, state| { resolve::deltas( object_counter.clone(), size_counter.clone(), node, state, object_hash.len_in_bytes(), ) } }, || (!should_interrupt.load(Ordering::Relaxed)).then(|| std::time::Duration::from_millis(50)), |_| (), )?; lock(&object_progress).show_throughput(start); size_progress.show_throughput(start); Ok(Outcome { roots: self.root_items, children: self.child_items, }) } }