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<dyn std::error::Error + Send + Sync>), #[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<usize>, /// 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<T> { /// The items that have no children in the pack, i.e. base objects. pub roots: Vec<Item<T>>, /// The items that children to a root object, i.e. delta objects. pub children: Vec<Item<T>>,} impl<T> Tree<T>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<u8>) -> 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<ThreadLocal State>) -> Result<(), CustomError>` is a function /// running for each thread receiving fully decoded objects along with contextual information, which either succeeds 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<F, P1, P2, MBFN, S, E>( 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<Outcome<T>, Error> where F: for<'r> Fn(EntryRange, &'r mut Vec<u8>) -> Option<()> + Send + Clone, P1: Progress, P2: Progress, MBFN: Fn(&mut T, &mut <P1 as Progress>::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<T>]); move |thread_index| { ( Vec::<u8>::with_capacity(4096), lock(&object_progress) .add_child_with_id(format!("thread {thread_index}"), git_features::progress::UNKNOWN), 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, }) }}
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<dyn std::error::Error + Send + Sync>),
#[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<usize>,
/// 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<T> {
/// The items that have no children in the pack, i.e. base objects.
pub roots: Vec<Item<T>>,
/// The items that children to a root object, i.e. delta objects.
pub children: Vec<Item<T>>,
}
impl<T> Tree<T>
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<u8>) -> 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<ThreadLocal State>) -> Result<(), CustomError>` is a function
/// running for each thread receiving fully decoded objects along with contextual information, which either succeeds 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<F, P1, P2, MBFN, S, E>(
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<Outcome<T>, Error>
where
F: for<'r> Fn(EntryRange, &'r mut Vec<u8>) -> Option<()> + Send + Clone,
P1: Progress,
P2: Progress,
MBFN: Fn(&mut T, &mut <P1 as Progress>::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<T>]);
move |thread_index| {
(
Vec::<u8>::with_capacity(4096),
lock(&object_progress)
.add_child_with_id(format!("thread {thread_index}"), git_features::progress::UNKNOWN),
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,
})
}
}