feat!: index-backed tree traversal with a form of work-stealing. (#851)

When delta-trees are unbalanced, in pathological cases it's possible that that one thread ends up with more than half of the work. In this case it's required that it manages to spawn its own threads to parallelize the work it has.

Sebastian Thiel committed 3 years ago

Commit d22dd8f1 parent add5ea811 files · +441 −202Browse files

Unified · Split

11 files

gix-pack/Cargo.toml +1 −1
gix-pack/src/bundle/write/mod.rs +13 −16
gix-pack/src/cache/delta/traverse/mod.rs +34 −26
gix-pack/src/cache/delta/traverse/resolve.rs +314 −51
gix-pack/src/cache/delta/traverse/util.rs +7 −5
gix-pack/src/index/traverse/mod.rs +18 −22
gix-pack/src/index/traverse/with_index.rs +16 −20
gix-pack/src/index/traverse/with_lookup.rs +8 −12
gix-pack/src/index/verify.rs +5 −8
gix-pack/src/index/write/mod.rs +9 −14
gix-pack/tests/pack/index.rs +16 −27

gix-pack/Cargo.toml+1−1

@@ -5,7 +5,7 @@ repository = "https://github.com/Byron/gitoxide" · 4 unmodified lines
5	authors = ["Sebastian Thiel <sebastian.thiel@icloud.com>"]	5	authors = ["Sebastian Thiel <sebastian.thiel@icloud.com>"]
6	license = "MIT/Apache-2.0"	6	license = "MIT/Apache-2.0"
7	description = "Implements git packs and related data structures"	7	description = "Implements git packs and related data structures"
8	edition = "2018"	8	edition = "2021"
9	include = ["src/*/", "CHANGELOG.md"]	9	include = ["src/*/", "CHANGELOG.md"]
10	rust-version = "1.64"	10	rust-version = "1.64"
11	autotests = false	11	autotests = false

gix-pack/src/bundle/write/mod.rs+13−16

     /// * the resulting pack may be empty, that is, contains zero objects in some situations. This is a valid reply by a server and should
     ///   be accounted for.
     ///   - Empty packs always have the same name and not handling this case will result in at most one superfluous pack.
-    pub fn write_to_directory<P>(
+    pub fn write_to_directory(
         pack: impl io::BufRead,
         directory: Option<impl AsRef<Path>>,
-        mut progress: P,
+        mut progress: impl Progress,
         should_interrupt: &AtomicBool,
         thin_pack_base_object_lookup_fn: Option<ThinPackLookupFn>,
         options: Options,
-    ) -> Result<Outcome, Error>
+    ) -> Result<Outcome, Error> {
-    where
-        P: Progress,
-    {
         let mut read_progress = progress.add_child_with_id("read pack", ProgressId::ReadPackBytes.into());
         read_progress.init(None, progress::bytes());
         let pack = progress::Read {
     }
 }
+fn resolve_entry(range: data::EntryRange, mapped_file: &memmap2::Mmap) -> Option<&[u8]> {
+    mapped_file.get(range.start as usize..range.end as usize)
+}
 fn new_pack_file_resolver(
     data_file: SharedTempFile,
-) -> io::Result<impl Fn(data::EntryRange, &mut Vec<u8>) -> Option<()> + Send + Clone> {
+) -> io::Result<(
+    impl Fn(data::EntryRange, &memmap2::Mmap) -> Option<&[u8]> + Send + Clone,
+    memmap2::Mmap,
+)> {
     let mut guard = data_file.lock();
     guard.flush()?;
-    let mapped_file = Arc::new(crate::mmap::read_only(
+    let mapped_file = crate::mmap::read_only(&guard.get_mut().with_mut(|f| f.path().to_owned())?)?;
-        &guard.get_mut().with_mut(|f| f.path().to_owned())?,
+    Ok((resolve_entry, mapped_file))
-    )?);
-    let pack_data_lookup = move |range: std::ops::Range<u64>, out: &mut Vec<u8>| -> Option<()> {
-        mapped_file
-            .get(range.start as usize..range.end as usize)
-            .map(|pack_entry| out.copy_from_slice(pack_entry))
-    };
-    Ok(pack_data_lookup)
 }
 struct WriteOutcome {

gix-pack/src/cache/delta/traverse/mod.rs+34−26

 use std::sync::atomic::{AtomicBool, Ordering};
+use gix_features::threading::{Mutable, OwnShared};
 use gix_features::{
     parallel::in_parallel_with_slice,
     progress::{self, Progress},
-    threading::{lock, Mutable, OwnShared},
+    threading,
 };
 use crate::{
         /// 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,
     },
+    #[error("Failed to spawn thread when switching to work-stealing mode")]
+    SpawnThread(#[from] std::io::Error),
 }
 /// Additional context passed to the `inspect_object(…)` function of the [`Tree::traverse()`] method.
-pub struct Context<'a, S> {
+pub struct Context<'a> {
     /// 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,
     ///   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.
+    ///   Note that `node_data` can be modified to allow storing maintaining computation results on a per-object basis. It should contain
+    ///   its own mutable per-thread data as required.
     ///
     /// 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>(
+    pub fn traverse<F, P1, P2, MBFN, E, R>(
         mut self,
         resolve: F,
+        resolve_data: &R,
         pack_entries_end: u64,
-        new_thread_state: impl Fn() -> S + Send + Clone,
         inspect_object: MBFN,
         Options {
             thread_limit,
-            object_progress,
+            mut 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,
+        F: for<'r> FnMut(EntryRange, &'r R) -> Option<&'r [u8]> + Send + Clone,
+        R: Send + Sync,
         P1: Progress,
         P2: Progress,
-        MBFN: Fn(&mut T, &mut <P1 as Progress>::SubProgress, Context<'_, S>) -> Result<(), E> + Send + Clone,
+        MBFN: FnMut(&mut T, &<P1 as Progress>::SubProgress, Context<'_>) -> 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);
+            let progress = &mut 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 object_progress = OwnShared::new(Mutable::new(object_progress));
         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(std::ptr::slice_from_raw_parts_mut(
                     child_items.as_mut_ptr(),
                     child_items.len(),
                 ));
-                move |thread_index| {
+                {
-                    (
+                    let object_progress = object_progress.clone();
-                        Vec::<u8>::with_capacity(4096),
+                    move |thread_index| {
-                        lock(&object_progress)
+                        let _ = &child_items;
-                            .add_child_with_id(format!("thread {thread_index}"), gix_features::progress::UNKNOWN),
+                        resolve::State {
-                        new_thread_state(),
+                            bytes_buf: Vec::<u8>::with_capacity(4096),
-                        resolve.clone(),
+                            delta_bytes: Vec::<u8>::with_capacity(4096),
-                        inspect_object.clone(),
+                            fully_resolved_delta_bytes: Vec::<u8>::with_capacity(4096),
-                        child_items.clone(),
+                            progress: threading::lock(&object_progress)
-                    )
+                                .add_child(format!("gix-pack.delta-traverse.inspect-object.{thread_index}")),
+                            resolve: resolve.clone(),
+                            modify_base: inspect_object.clone(),
+                            child_items: child_items.clone(),
+                        }
+                    }
                 }
             },
             {
-                move |node, state, _threads_left, should_interrupt| {
+                move |node, state, threads_left, should_interrupt| {
                     resolve::deltas(
                         object_counter.clone(),
                         size_counter.clone(),
                         node,
                         state,
+                        resolve_data,
                         object_hash.len_in_bytes(),
+                        threads_left,
                         should_interrupt,
                     )
                 }
             |_| (),
         )?;
-        lock(&object_progress).show_throughput(start);
+        threading::lock(&object_progress).show_throughput(start);
         size_progress.show_throughput(start);
         Ok(Outcome {

gix-pack/src/cache/delta/traverse/resolve.rs+314−51

@@ -1,7 +1,8 @@
1	use std::sync::atomic::AtomicBool;	1	use std::sync::atomic::{AtomicBool, AtomicIsize};
2	use std::{cell::RefCell, collections::BTreeMap, sync::atomic::Ordering};	2	use std::{collections::BTreeMap, sync::atomic::Ordering};
3		3
4	use gix_features::{progress::Progress, zlib};	4	use gix_features::progress::Progress;
		5	use gix_features::{threading, zlib};
5		6
6	use crate::{	7	use crate::{
7	cache::delta::{	8	cache::delta::{
@@ -11,49 +12,59 @@ use crate::{ · 3 unmodified lines
11	},	12	},
12	Item,	13	Item,
13	},	14	},
		15	data,
14	data::EntryRange,	16	data::EntryRange,
15	};	17	};
16		18
		19	pub(crate) struct State<P, F, MBFN, T: Send> {
		20	pub bytes_buf: Vec<u8>,
		21	pub delta_bytes: Vec<u8>,
		22	pub fully_resolved_delta_bytes: Vec<u8>,
		23	pub progress: P,
		24	pub resolve: F,
		25	pub modify_base: MBFN,
		26	pub child_items: ItemSliceSend<Item<T>>,
		27	}
		28
17	#[allow(clippy::too_many_arguments)]	29	#[allow(clippy::too_many_arguments)]
18	pub(crate) fn deltas<T, F, P, MBFN, S, E>(	30	pub(crate) fn deltas<T, F, MBFN, E, R, P>(
19	object_counter: Option<gix_features::progress::StepShared>,	31	object_counter: Option<gix_features::progress::StepShared>,
20	size_counter: Option<gix_features::progress::StepShared>,	32	size_counter: Option<gix_features::progress::StepShared>,
21	node: &mut Item<T>,	33	node: &mut Item<T>,
22	(bytes_buf, ref mut progress, state, resolve, modify_base, child_items): &mut (	34	State {
23	Vec<u8>,	35	bytes_buf,
24	P,	36	delta_bytes,
25	S,	37	fully_resolved_delta_bytes,
26	F,	38	progress,
27	MBFN,	39	resolve,
28	ItemSliceSend<Item<T>>,	40	modify_base,
29	),	41	child_items,
		42	}: &mut State<P, F, MBFN, T>,
		43	resolve_data: &R,
30	hash_len: usize,	44	hash_len: usize,
		45	threads_left: &AtomicIsize,
31	should_interrupt: &AtomicBool,	46	should_interrupt: &AtomicBool,
32	) -> Result<(), Error>	47	) -> Result<(), Error>
33	where	48	where
34	T: Send,	49	T: Send,
35	F: for<'r> Fn(EntryRange, &'r mut Vec<u8>) -> Option<()>,	50	R: Send + Sync,
36	P: Progress,	51	P: Progress,
37	MBFN: Fn(&mut T, &mut P, Context<'_, S>) -> Result<(), E>,	52	F: for<'r> FnMut(EntryRange, &'r R) -> Option<&'r [u8]> + Send + Clone,
		53	MBFN: FnMut(&mut T, &P, Context<'_>) -> Result<(), E> + Send + Clone,
38	E: std::error::Error + Send + Sync + 'static,	54	E: std::error::Error + Send + Sync + 'static,
39	{	55	{
40	let mut decompressed_bytes_by_pack_offset = BTreeMap::new();	56	let mut decompressed_bytes_by_pack_offset = BTreeMap::new();
41	let bytes_buf = RefCell::new(bytes_buf);	57	let mut decompress_from_resolver = \|slice: EntryRange, out: &mut Vec<u8>\| -> Result<(data::Entry, u64), Error> {
42	let decompress_from_resolver = \|slice: EntryRange\| -> Result<(crate::data::Entry, u64, Vec<u8>), Error> {	58	let bytes = resolve(slice.clone(), resolve_data).ok_or(Error::ResolveFailed {
43	let mut bytes_buf = bytes_buf.borrow_mut();
44	bytes_buf.resize((slice.end - slice.start) as usize, 0);
45	resolve(slice.clone(), &mut bytes_buf).ok_or(Error::ResolveFailed {
46	pack_offset: slice.start,	59	pack_offset: slice.start,
47	})?;	60	})?;
48	let entry = crate::data::Entry::from_bytes(&bytes_buf, slice.start, hash_len);	61	let entry = data::Entry::from_bytes(bytes, slice.start, hash_len);
49	let compressed = &bytes_buf[entry.header_size()..];	62	let compressed = &bytes[entry.header_size()..];
50	let decompressed_len = entry.decompressed_size as usize;	63	let decompressed_len = entry.decompressed_size as usize;
51	Ok((entry, slice.end, decompress_all_at_once(compressed, decompressed_len)?))	64	decompress_all_at_once(compressed, decompressed_len, out)?;
		65	Ok((entry, slice.end))
52	};	66	};
53		67
54	// Traverse the tree breadth first and loose the data produced for the base as it won't be needed anymore.
55	progress.init(None, gix_features::progress::count_with_decimals("objects", 2));
56
57	// each node is a base, and its children always start out as deltas which become a base after applying them.	68	// each node is a base, and its children always start out as deltas which become a base after applying them.
58	// These will be pushed onto our stack until all are processed	69	// These will be pushed onto our stack until all are processed
59	let root_level = 0;	70	let root_level = 0;
@@ -61,19 +72,26 @@ where · 1 unmodified line
61	root_level,	72	root_level,
62	Node {	73	Node {
63	item: node,	74	item: node,
64	child_items: child_items.0,	75	child_items: child_items.clone(),
65	},	76	},
66	)];	77	)];
67	while let Some((level, mut base)) = nodes.pop() {	78	while let Some((level, mut base)) = nodes.pop() {
		79	let storage;
68	if should_interrupt.load(Ordering::Relaxed) {	80	if should_interrupt.load(Ordering::Relaxed) {
69	return Err(Error::Interrupted);	81	return Err(Error::Interrupted);
70	}	82	}
71	let (base_entry, entry_end, base_bytes) = if level == root_level {	83	let (base_entry, entry_end, base_bytes) = if level == root_level {
72	decompress_from_resolver(base.entry_slice())?	84	let (a, b) = decompress_from_resolver(base.entry_slice(), bytes_buf)?;
		85	(a, b, &*bytes_buf)
73	} else {	86	} else {
74	decompressed_bytes_by_pack_offset	87	if !bytes_buf.is_empty() {
		88	*bytes_buf = Vec::new();
		89	}
		90	let (a, b, v) = decompressed_bytes_by_pack_offset
75	.remove(&base.offset())	91	.remove(&base.offset())
76	.expect("we store the resolved delta buffer when done")	92	.expect("we store the resolved delta buffer when done");
		93	storage = v;
		94	(a, b, &storage)
77	};	95	};
78		96
79	// anything done here must be repeated further down for leaf-nodes.	97	// anything done here must be repeated further down for leaf-nodes.
@@ -86,8 +104,7 @@ where · 6 unmodified lines
86	Context {	104	Context {
87	entry: &base_entry,	105	entry: &base_entry,
88	entry_end,	106	entry_end,
89	decompressed: &base_bytes,	107	decompressed: base_bytes,
90	state,
91	level,	108	level,
92	},	109	},
93	)	110	)
@@ -99,20 +116,19 @@ where · 5 unmodified lines
99	}	116	}
100		117
101	for mut child in base.into_child_iter() {	118	for mut child in base.into_child_iter() {
102	let (mut child_entry, entry_end, delta_bytes) = decompress_from_resolver(child.entry_slice())?;	119	let (mut child_entry, entry_end) = decompress_from_resolver(child.entry_slice(), delta_bytes)?;
103	let (base_size, consumed) = crate::data::delta::decode_header_size(&delta_bytes);	120	let (base_size, consumed) = data::delta::decode_header_size(delta_bytes);
104	let mut header_ofs = consumed;	121	let mut header_ofs = consumed;
105	assert_eq!(	122	assert_eq!(
106	base_bytes.len(),	123	base_bytes.len(),
107	base_size as usize,	124	base_size as usize,
108	"recorded base size in delta does not match"	125	"recorded base size in delta does match the actual one"
109	);	126	);
110	let (result_size, consumed) = crate::data::delta::decode_header_size(&delta_bytes[consumed..]);	127	let (result_size, consumed) = data::delta::decode_header_size(&delta_bytes[consumed..]);
111	header_ofs += consumed;	128	header_ofs += consumed;
112		129
113	let mut fully_resolved_delta_bytes = bytes_buf.borrow_mut();	130	resize_vec(fully_resolved_delta_bytes, result_size as usize);
114	fully_resolved_delta_bytes.resize(result_size as usize, 0);	131	data::delta::apply(base_bytes, fully_resolved_delta_bytes, &delta_bytes[header_ofs..]);
115	crate::data::delta::apply(&base_bytes, &mut fully_resolved_delta_bytes, &delta_bytes[header_ofs..]);
116		132
117	// FIXME: this actually invalidates the "pack_offset()" computation, which is not obvious to consumers	133	// FIXME: this actually invalidates the "pack_offset()" computation, which is not obvious to consumers
118	// at all	134	// at all
@@ -120,7 +136,7 @@ where · 1 unmodified line
120	if child.has_children() {	136	if child.has_children() {
121	decompressed_bytes_by_pack_offset.insert(	137	decompressed_bytes_by_pack_offset.insert(
122	child.offset(),	138	child.offset(),
123	(child_entry, entry_end, fully_resolved_delta_bytes.to_owned()),	139	(child_entry, entry_end, std::mem::take(fully_resolved_delta_bytes)),
124	);	140	);
125	nodes.push((level + 1, child));	141	nodes.push((level + 1, child));
126	} else {	142	} else {
@@ -130,8 +146,7 @@ where · 3 unmodified lines
130	Context {	146	Context {
131	entry: &child_entry,	147	entry: &child_entry,
132	entry_end,	148	entry_end,
133	decompressed: &fully_resolved_delta_bytes,	149	decompressed: fully_resolved_delta_bytes,
134	state,
135	level: level + 1,	150	level: level + 1,
136	},	151	},
137	)	152	)
@@ -142,25 +157,273 @@ where · 4 unmodified lines
142	.map(\|c\| c.fetch_add(base_bytes.len(), Ordering::SeqCst));	157	.map(\|c\| c.fetch_add(base_bytes.len(), Ordering::SeqCst));
143	}	158	}
144	}	159	}
		160
		161	// After the first round, see if we can use additional threads, and if so we enter multi-threaded mode.
		162	// In it we will keep using new threads as they become available while using this thread for coordination.
		163	// We optimize for a low memory footprint as we are likely to get here if long delta-chains with large objects are involved.
		164	// Try to avoid going into threaded mode if there isn't more than one unit of work anyway.
		165	if nodes.len() > 1 {
		166	if let Ok(initial_threads) =
		167	threads_left.fetch_update(Ordering::SeqCst, Ordering::SeqCst, \|threads_available\| {
		168	(threads_available > 0).then_some(0)
		169	})
		170	{
		171	// Assure no memory is held here.
		172	*bytes_buf = Vec::new();
		173	*delta_bytes = Vec::new();
		174	*fully_resolved_delta_bytes = Vec::new();
		175	return deltas_mt(
		176	initial_threads,
		177	decompressed_bytes_by_pack_offset,
		178	object_counter,
		179	size_counter,
		180	progress,
		181	nodes,
		182	resolve.clone(),
		183	resolve_data,
		184	modify_base.clone(),
		185	hash_len,
		186	threads_left,
		187	should_interrupt,
		188	);
		189	}
		190	}
145	}	191	}
146		192
147	Ok(())	193	Ok(())
148	}	194	}
149		195
150	fn decompress_all_at_once(b: &[u8], decompressed_len: usize) -> Result<Vec<u8>, Error> {	196	/// * `initial_threads` is the threads we may spawn, not accounting for our own thread which is still considered used by the parent
151	let mut out = Vec::with_capacity(decompressed_len);	197	/// system. Since this thread will take a controlling function, we may spawn one more than that. In threaded mode, we will finish
152	// SAFETY:	198	/// all remaining work.
153	// 1. we have reserved `decompressed_len`	199	#[allow(clippy::too_many_arguments)]
154	// 2. zlib is going to write all of `decompressed_len`. On error, none of the the buffer is made available to the user.	200	pub(crate) fn deltas_mt<T, F, MBFN, E, R, P>(
155	#[allow(unsafe_code, clippy::uninit_vec)]	201	mut threads_to_create: isize,
156	unsafe {	202	decompressed_bytes_by_pack_offset: BTreeMap<u64, (data::Entry, u64, Vec<u8>)>,
157	out.set_len(decompressed_len);	203	object_counter: Option<gix_features::progress::StepShared>,
		204	size_counter: Option<gix_features::progress::StepShared>,
		205	progress: &P,
		206	nodes: Vec<(u16, Node<'_, T>)>,
		207	resolve: F,
		208	resolve_data: &R,
		209	modify_base: MBFN,
		210	hash_len: usize,
		211	threads_left: &AtomicIsize,
		212	should_interrupt: &AtomicBool,
		213	) -> Result<(), Error>
		214	where
		215	T: Send,
		216	R: Send + Sync,
		217	P: Progress,
		218	F: for<'r> FnMut(EntryRange, &'r R) -> Option<&'r [u8]> + Send + Clone,
		219	MBFN: FnMut(&mut T, &P, Context<'_>) -> Result<(), E> + Send + Clone,
		220	E: std::error::Error + Send + Sync + 'static,
		221	{
		222	let nodes = gix_features::threading::Mutable::new(nodes);
		223	let decompressed_bytes_by_pack_offset = gix_features::threading::Mutable::new(decompressed_bytes_by_pack_offset);
		224	threads_to_create += 1; // ourselves
		225	let mut returned_ourselves = false;
		226
		227	gix_features::parallel::threads(\|s\| -> Result<(), Error> {
		228	let mut threads = Vec::new();
		229	let poll_interval = std::time::Duration::from_millis(100);
		230	loop {
		231	for tid in 0..threads_to_create {
		232	let thread = gix_features::parallel::build_thread()
		233	.name(format!("gix-pack.traverse_deltas.{tid}"))
		234	.spawn_scoped(s, {
		235	let nodes = &nodes;
		236	let decompressed_bytes_by_pack_offset = &decompressed_bytes_by_pack_offset;
		237	let mut resolve = resolve.clone();
		238	let mut modify_base = modify_base.clone();
		239	let object_counter = object_counter.as_ref();
		240	let size_counter = size_counter.as_ref();
		241
		242	move \|\| -> Result<(), Error> {
		243	let mut bytes_buf = Vec::new();
		244	let mut delta_bytes = Vec::new();
		245	let mut fully_resolved_delta_bytes = Vec::new();
		246	let mut decompress_from_resolver =
		247	\|slice: EntryRange, out: &mut Vec<u8>\| -> Result<(data::Entry, u64), Error> {
		248	let bytes = resolve(slice.clone(), resolve_data).ok_or(Error::ResolveFailed {
		249	pack_offset: slice.start,
		250	})?;
		251	let entry = data::Entry::from_bytes(bytes, slice.start, hash_len);
		252	let compressed = &bytes[entry.header_size()..];
		253	let decompressed_len = entry.decompressed_size as usize;
		254	decompress_all_at_once(compressed, decompressed_len, out)?;
		255	Ok((entry, slice.end))
		256	};
		257
		258	loop {
		259	let (level, mut base) = match threading::lock(nodes).pop() {
		260	Some(v) => v,
		261	None => break,
		262	};
		263	if should_interrupt.load(Ordering::Relaxed) {
		264	return Err(Error::Interrupted);
		265	}
		266	let storage;
		267	let (base_entry, entry_end, base_bytes) = if level == 0 {
		268	let (a, b) = decompress_from_resolver(base.entry_slice(), &mut bytes_buf)?;
		269	(a, b, &*bytes_buf)
		270	} else {
		271	if !bytes_buf.is_empty() {
		272	bytes_buf = Vec::new();
		273	}
		274	let (a, b, v) = threading::lock(decompressed_bytes_by_pack_offset)
		275	.remove(&base.offset())
		276	.expect("we store the resolved delta buffer when done");
		277	storage = v;
		278	(a, b, storage.as_slice())
		279	};
		280
		281	// anything done here must be repeated further down for leaf-nodes.
		282	// This way we avoid retaining their decompressed memory longer than needed (they have no children,
		283	// thus their memory can be released right away, using 18% less peak memory on the linux kernel).
		284	{
		285	modify_base(
		286	base.data(),
		287	progress,
		288	Context {
		289	entry: &base_entry,
		290	entry_end,
		291	decompressed: base_bytes,
		292	level,
		293	},
		294	)
		295	.map_err(\|err\| Box::new(err) as Box<dyn std::error::Error + Send + Sync>)?;
		296	object_counter.as_ref().map(\|c\| c.fetch_add(1, Ordering::SeqCst));
		297	size_counter
		298	.as_ref()
		299	.map(\|c\| c.fetch_add(base_bytes.len(), Ordering::SeqCst));
		300	}
		301
		302	for mut child in base.into_child_iter() {
		303	let (mut child_entry, entry_end) =
		304	decompress_from_resolver(child.entry_slice(), &mut delta_bytes)?;
		305	let (base_size, consumed) = data::delta::decode_header_size(&delta_bytes);
		306	let mut header_ofs = consumed;
		307	assert_eq!(
		308	base_bytes.len(),
		309	base_size as usize,
		310	"recorded base size in delta does match the actual one"
		311	);
		312	let (result_size, consumed) =
		313	data::delta::decode_header_size(&delta_bytes[consumed..]);
		314	header_ofs += consumed;
		315
		316	fully_resolved_delta_bytes.resize(result_size as usize, 0);
		317	data::delta::apply(
		318	base_bytes,
		319	&mut fully_resolved_delta_bytes,
		320	&delta_bytes[header_ofs..],
		321	);
		322
		323	// FIXME: this actually invalidates the "pack_offset()" computation, which is not obvious to consumers
		324	// at all
		325	child_entry.header = base_entry.header; // assign the actual object type, instead of 'delta'
		326	if child.has_children() {
		327	threading::lock(decompressed_bytes_by_pack_offset).insert(
		328	child.offset(),
		329	(child_entry, entry_end, std::mem::take(&mut fully_resolved_delta_bytes)),
		330	);
		331	threading::lock(nodes).push((level + 1, child));
		332	} else {
		333	modify_base(
		334	child.data(),
		335	progress,
		336	Context {
		337	entry: &child_entry,
		338	entry_end,
		339	decompressed: &fully_resolved_delta_bytes,
		340	level: level + 1,
		341	},
		342	)
		343	.map_err(\|err\| Box::new(err) as Box<dyn std::error::Error + Send + Sync>)?;
		344	object_counter.as_ref().map(\|c\| c.fetch_add(1, Ordering::SeqCst));
		345	size_counter
		346	.as_ref()
		347	.map(\|c\| c.fetch_add(base_bytes.len(), Ordering::SeqCst));
		348	}
		349	}
		350	}
		351	Ok(())
		352	}
		353	})?;
		354	threads.push(thread);
		355	}
		356	if threads_left
		357	.fetch_update(Ordering::SeqCst, Ordering::SeqCst, \|threads_available: isize\| {
		358	(threads_available > 0).then(\|\| {
		359	threads_to_create = threads_available.min(threading::lock(&nodes).len() as isize);
		360	threads_available - threads_to_create
		361	})
		362	})
		363	.is_err()
		364	{
		365	threads_to_create = 0;
		366	}
		367
		368	// What we really want to do is either wait for one of our threads to go down
		369	// or for another scheduled thread to become available. Unfortunately we can't do that,
		370	// but may instead find a good way to set the polling interval instead of hard-coding it.
		371	std::thread::sleep(poll_interval);
		372	// Get out of threads are already starving or they would be starving soon as no work is left.
		373	if threads.iter().any(\|t\| t.is_finished()) {
		374	let mut running_threads = Vec::new();
		375	for thread in threads.drain(..) {
		376	if thread.is_finished() {
		377	match thread.join() {
		378	Ok(Err(err)) => return Err(err),
		379	Ok(Ok(())) => {
		380	if !returned_ourselves {
		381	returned_ourselves = true;
		382	} else {
		383	threads_left.fetch_add(1, Ordering::SeqCst);
		384	}
		385	}
		386	Err(err) => {
		387	std::panic::resume_unwind(err);
		388	}
		389	}
		390	} else {
		391	running_threads.push(thread);
		392	}
		393	}
		394	if running_threads.is_empty() && threading::lock(&nodes).is_empty() {
		395	break;
		396	}
		397	threads = running_threads;
		398	}
		399	}
		400
		401	Ok(())
		402	})
		403	}
		404
		405	fn resize_vec(v: &mut Vec<u8>, new_len: usize) {
		406	if new_len > v.len() {
		407	v.reserve_exact(new_len.saturating_sub(v.capacity()) + (v.capacity() - v.len()));
		408	// SAFETY:
		409	// 1. we have reserved enough capacity to fit `new_len`
		410	// 2. the caller is trusted to write into `v` to completely fill `new_len`.
		411	#[allow(unsafe_code, clippy::uninit_vec)]
		412	unsafe {
		413	v.set_len(new_len);
		414	}
		415	} else {
		416	v.truncate(new_len)
158	}	417	}
		418	}
		419
		420	fn decompress_all_at_once(b: &[u8], decompressed_len: usize, out: &mut Vec<u8>) -> Result<(), Error> {
		421	resize_vec(out, decompressed_len);
159	zlib::Inflate::default()	422	zlib::Inflate::default()
160	.once(b, &mut out)	423	.once(b, out)
161	.map_err(\|err\| Error::ZlibInflate {	424	.map_err(\|err\| Error::ZlibInflate {
162	source: err,	425	source: err,
163	message: "Failed to decompress entry",	426	message: "Failed to decompress entry",
164	})?;	427	})?;
165	Ok(out)	428	Ok(())
166	}	429	}

gix-pack/src/cache/delta/traverse/util.rs+7−5

 where
     T: Send;
+/// SAFETY: This would be unsafe if this would ever be abused, but it's used internally and only in a way that assure that the pointers
+///         don't violate aliasing rules.
 impl<T> Clone for ItemSliceSend<T>
 where
     T: Send,
 unsafe impl<T> Send for ItemSliceSend<T> where T: Send {}
 /// An item returned by `iter_root_chunks`, allowing access to the `data` stored alongside nodes in a [`Tree`].
-pub struct Node<'a, T> {
+pub struct Node<'a, T: Send> {
     pub item: &'a mut Item<T>,
-    pub child_items: *mut [Item<T>],
+    pub child_items: ItemSliceSend<Item<T>>,
 }
-impl<'a, T> Node<'a, T> {
+impl<'a, T: Send> Node<'a, T> {
     /// Returns the offset into the pack at which the `Node`s data is located.
     pub fn offset(&self) -> u64 {
         self.item.offset
             // SAFETY: The resulting mutable pointer cannot be yielded by any other node.
             #[allow(unsafe_code)]
             Node {
-                item: unsafe { &mut *(children as *mut Item<T>).add(index as usize) },
+                item: &mut unsafe { &mut *children.0 }[index as usize],
-                child_items: children,
+                child_items: children.clone(),
             }
         })
     }

gix-pack/src/index/traverse/mod.rs+18−22

 use std::sync::atomic::AtomicBool;
-use gix_features::{parallel, progress::Progress};
+use gix_features::{
+    parallel,
+    progress::{Progress, RawProgress},
+};
 use crate::index;
         pack: &crate::data::File,
         progress: P,
         should_interrupt: &AtomicBool,
-        new_processor: impl Fn() -> Processor + Send + Clone,
+        processor: Processor,
         Options {
             traversal,
             thread_limit,
         P: Progress,
         C: crate::cache::DecodeEntry,
         E: std::error::Error + Send + Sync + 'static,
-        Processor: FnMut(
+        Processor: FnMut(gix_object::Kind, &[u8], &index::Entry, &dyn RawProgress) -> Result<(), E> + Send + Clone,
-            gix_object::Kind,
-            &[u8],
-            &index::Entry,
-            &mut <P::SubProgress as Progress>::SubProgress,
-        ) -> Result<(), E>,
         F: Fn() -> C + Send + Clone,
     {
         match traversal {
             Algorithm::Lookup => self.traverse_with_lookup(
-                new_processor,
+                processor,
                 pack,
                 progress,
                 should_interrupt,
             ),
             Algorithm::DeltaTreeLookup => self.traverse_with_index(
                 pack,
-                new_processor,
+                processor,
                 progress,
                 should_interrupt,
-                crate::index::traverse::with_index::Options { check, thread_limit },
+                with_index::Options { check, thread_limit },
             ),
         }
     }
     }
     #[allow(clippy::too_many_arguments)]
-    fn decode_and_process_entry<C, P, E>(
+    fn decode_and_process_entry<C, E>(
         &self,
         check: SafetyCheck,
         pack: &crate::data::File,
         cache: &mut C,
         buf: &mut Vec<u8>,
-        progress: &mut P,
+        progress: &mut dyn RawProgress,
-        index_entry: &crate::index::Entry,
+        index_entry: &index::Entry,
-        processor: &mut impl FnMut(gix_object::Kind, &[u8], &index::Entry, &mut P) -> Result<(), E>,
+        processor: &mut impl FnMut(gix_object::Kind, &[u8], &index::Entry, &dyn RawProgress) -> Result<(), E>,
     ) -> Result<crate::data::decode::entry::Outcome, Error<E>>
     where
         C: crate::cache::DecodeEntry,
-        P: Progress,
         E: std::error::Error + Send + Sync + 'static,
     {
         let pack_entry = pack.entry(index_entry.pack_offset);
             check,
             object_kind,
             buf,
-            progress,
             index_entry,
             || pack.entry_crc32(index_entry.pack_offset, entry_len),
+            progress,
             processor,
         )?;
         Ok(entry_stats)
 }
 #[allow(clippy::too_many_arguments)]
-fn process_entry<P, E>(
+fn process_entry<E>(
     check: SafetyCheck,
     object_kind: gix_object::Kind,
     decompressed: &[u8],
-    progress: &mut P,
+    index_entry: &index::Entry,
-    index_entry: &crate::index::Entry,
     pack_entry_crc32: impl FnOnce() -> u32,
-    processor: &mut impl FnMut(gix_object::Kind, &[u8], &index::Entry, &mut P) -> Result<(), E>,
+    progress: &dyn RawProgress,
+    processor: &mut impl FnMut(gix_object::Kind, &[u8], &index::Entry, &dyn RawProgress) -> Result<(), E>,
 ) -> Result<(), Error<E>>
 where
-    P: Progress,
     E: std::error::Error + Send + Sync + 'static,
 {
     if check.object_checksum() {

gix-pack/src/index/traverse/with_index.rs+16−20

     pub fn traverse_with_index<P, Processor, E>(
         &self,
         pack: &crate::data::File,
-        new_processor: impl Fn() -> Processor + Send + Clone,
+        mut processor: Processor,
         mut progress: P,
         should_interrupt: &AtomicBool,
         Options { check, thread_limit }: Options,
     ) -> Result<Outcome<P>, Error<E>>
     where
         P: Progress,
-        Processor: FnMut(
+        Processor: FnMut(gix_object::Kind, &[u8], &index::Entry, &dyn gix_features::progress::RawProgress) -> Result<(), E>
-            gix_object::Kind,
+            + Send
-            &[u8],
+            + Clone,
-            &index::Entry,
-            &mut <P::SubProgress as Progress>::SubProgress,
-        ) -> Result<(), E>,
         E: std::error::Error + Send + Sync + 'static,
     {
         let (verify_result, traversal_result) = parallel::join(
                     self.object_hash,
                 )?;
                 let mut outcome = digest_statistics(tree.traverse(
-                    |slice, out| pack.entry_slice(slice).map(|entry| out.copy_from_slice(entry)),
+                    |slice, pack| pack.entry_slice(slice),
+                    pack,
                     pack.pack_end() as u64,
-                    new_processor,
+                    move |data,
-                    |data,
+                          progress,
-                     progress,
+                          traverse::Context {
-                     traverse::Context {
+                              entry: pack_entry,
-                         entry: pack_entry,
+                              entry_end,
-                         entry_end,
+                              decompressed: bytes,
-                         decompressed: bytes,
+                              level,
-                         state: ref mut processor,
+                          }| {
-                         level,
-                     }| {
                         let object_kind = pack_entry.header.as_kind().expect("non-delta object");
                         data.level = level;
                         data.decompressed_size = pack_entry.decompressed_size;
                             check,
                             object_kind,
                             bytes,
-                            progress,
                             &data.index_entry,
                             || {
                                 // TODO: Fix this - we overwrite the header of 'data' which also changes the computed entry size,
                                         .expect("slice pointing into the pack (by now data is verified)"),
                                 )
                             },
-                            processor,
+                            progress,
+                            &mut processor,
                         );
                         match result {
                             Err(err @ Error::PackDecode { .. }) if !check.fatal_decode_error() => {

gix-pack/src/index/traverse/with_lookup.rs+8−12

     /// For more details, see the documentation on the [`traverse()`][index::File::traverse()] method.
     pub fn traverse_with_lookup<P, C, Processor, E, F>(
         &self,
-        new_processor: impl Fn() -> Processor + Send + Clone,
+        mut processor: Processor,
         pack: &data::File,
         mut progress: P,
         should_interrupt: &AtomicBool,
         P: Progress,
         C: crate::cache::DecodeEntry,
         E: std::error::Error + Send + Sync + 'static,
-        Processor: FnMut(
+        Processor: FnMut(gix_object::Kind, &[u8], &index::Entry, &dyn gix_features::progress::RawProgress) -> Result<(), E>
-            gix_object::Kind,
+            + Send
-            &[u8],
+            + Clone,
-            &index::Entry,
-            &mut <P::SubProgress as Progress>::SubProgress,
-        ) -> Result<(), E>,
         F: Fn() -> C + Send + Clone,
     {
         let (verify_result, traversal_result) = parallel::join(
                     move |index| {
                         (
                             make_pack_lookup_cache(),
-                            new_processor(),
                             Vec::with_capacity(2048), // decode buffer
                             lock(&reduce_progress)
                                 .add_child_with_id(format!("thread {index}"), gix_features::progress::UNKNOWN), // per thread progress
                     input_chunks,
                     thread_limit,
                     state_per_thread,
-                    |entries: &[index::Entry],
+                    move |entries: &[index::Entry],
-                     (cache, ref mut processor, buf, progress)|
+                          (cache, buf, progress)|
-                     -> Result<Vec<data::decode::entry::Outcome>, Error<_>> {
+                          -> Result<Vec<data::decode::entry::Outcome>, Error<_>> {
                         progress.init(
                             Some(entries.len()),
                             gix_features::progress::count_with_decimals("objects", 2),
                                 buf,
                                 progress,
                                 index_entry,
-                                processor,
+                                &mut processor,
                             );
                             progress.inc();
                             let stat = match result {

gix-pack/src/index/verify.rs+5−8

                     pack,
                     progress,
                     should_interrupt,
-                    || {
+                    {
                         let mut encode_buf = Vec::with_capacity(2048);
                         move |kind, data, index_entry, progress| {
                             Self::verify_entry(verify_mode, &mut encode_buf, kind, data, index_entry, progress)
     }
     #[allow(clippy::too_many_arguments)]
-    fn verify_entry<P>(
+    fn verify_entry(
         verify_mode: Mode,
         encode_buf: &mut Vec<u8>,
         object_kind: gix_object::Kind,
         buf: &[u8],
         index_entry: &index::Entry,
-        progress: &mut P,
+        progress: &dyn gix_features::progress::RawProgress,
-    ) -> Result<(), integrity::Error>
+    ) -> Result<(), integrity::Error> {
-    where
-        P: Progress,
-    {
         if let Mode::HashCrc32Decode | Mode::HashCrc32DecodeEncode = verify_mode {
             use gix_object::Kind::*;
             match object_kind {
                             .expect("writing to a memory buffer never fails");
                         if encode_buf.as_slice() != buf {
                             let mut should_return_error = true;
-                            if let gix_object::Kind::Tree = object_kind {
+                            if let Tree = object_kind {
                                 if buf.as_bstr().find(b"100664").is_some() || buf.as_bstr().find(b"100640").is_some() {
                                     progress.info(format!("Tree object {} would be cleaned up during re-serialization, replacing mode '100664|100640' with '100644'", index_entry.oid));
                                     should_return_error = false

gix-pack/src/index/write/mod.rs+9−14

     /// It should return `None` if the entry cannot be resolved from the pack that produced the `entries` iterator, causing
     /// the write operation to fail.
     #[allow(clippy::too_many_arguments)]
-    pub fn write_data_iter_to_stream<F, F2>(
+    pub fn write_data_iter_to_stream<F, F2, R, P>(
         version: crate::index::Version,
         make_resolver: F,
         entries: impl Iterator<Item = Result<crate::data::input::Entry, crate::data::input::Error>>,
         thread_limit: Option<usize>,
-        mut root_progress: impl Progress,
+        mut root_progress: P,
         out: impl io::Write,
         should_interrupt: &AtomicBool,
         object_hash: gix_hash::Kind,
         pack_version: crate::data::Version,
     ) -> Result<Outcome, Error>
     where
-        F: FnOnce() -> io::Result<F2>,
+        F: FnOnce() -> io::Result<(F2, R)>,
-        F2: for<'r> Fn(crate::data::EntryRange, &'r mut Vec<u8>) -> Option<()> + Send + Clone,
+        R: Send + Sync,
+        F2: for<'r> Fn(crate::data::EntryRange, &'r R) -> Option<&'r [u8]> + Send + Clone,
+        P: Progress,
     {
         if version != crate::index::Version::default() {
             return Err(Error::Unsupported(version));
         root_progress.inc();
-        let resolver = make_resolver()?;
+        let (resolver, pack) = make_resolver()?;
         let sorted_pack_offsets_by_oid = {
             let traverse::Outcome { roots, children } = tree.traverse(
                 resolver,
+                &pack,
                 pack_entries_end,
-                || (),
                 |data,
                  _progress,
                  traverse::Context {
 }
 fn modify_base(entry: &mut TreeEntry, pack_entry: &crate::data::Entry, decompressed: &[u8], hash: gix_hash::Kind) {
-    fn compute_hash(kind: gix_object::Kind, bytes: &[u8], object_hash: gix_hash::Kind) -> gix_hash::ObjectId {
-        let mut hasher = gix_features::hash::hasher(object_hash);
-        hasher.update(&gix_object::encode::loose_header(kind, bytes.len()));
-        hasher.update(bytes);
-        gix_hash::ObjectId::from(hasher.digest())
-    }
     let object_kind = pack_entry.header.as_kind().expect("base object as source of iteration");
-    let id = compute_hash(object_kind, decompressed, hash);
+    let id = gix_object::compute_hash(hash, object_kind, decompressed);
     entry.id = id;
 }

gix-pack/tests/pack/index.rs+16−27

         use gix_features::progress;
         use gix_odb::pack;
-        use gix_pack::{
+        use gix_pack::{data::input, index};
-            data::{input, EntryRange},
-            index,
-        };
         use crate::{
             fixture_path,
             pack::{INDEX_V2, V2_PACKS_AND_INDICES},
         };
+        fn slice_map(entry: gix_pack::data::EntryRange, map: &memmap2::Mmap) -> Option<&[u8]> {
+            map.get(entry.start as usize..entry.end as usize)
+        }
         #[test]
         fn write_to_stream() -> Result<(), Box<dyn std::error::Error>> {
-            fn assert_index_write<F>(
+            fn assert_index_write(
                 mode: &input::Mode,
                 compressed: &input::EntryDataMode,
                 index_path: &&str,
                 data_path: &&str,
-                resolve: F,
+            ) -> Result<(), Box<dyn std::error::Error>> {
-            ) -> Result<(), Box<dyn std::error::Error>>
-            where
-                F: Fn(pack::data::EntryRange, &mut Vec<u8>) -> Option<()> + Send + Clone,
-            {
                 let pack_iter = pack::data::input::BytesToEntriesIter::new_from_header(
                     io::BufReader::new(fs::File::open(fixture_path(data_path))?),
                     *mode,
                 let pack_version = pack_iter.version();
                 let outcome = pack::index::File::write_data_iter_to_stream(
                     desired_kind,
-                    move || Ok(resolve),
+                    || {
+                        let file = std::fs::File::open(fixture_path(data_path))?;
+                        let map = unsafe { memmap2::Mmap::map(&file)? };
+                        Ok((slice_map, map))
+                    },
                     pack_iter,
                     None,
                     progress::Discard,
             for mode in &[input::Mode::AsIs, input::Mode::Verify, input::Mode::Restore] {
                 for compressed in &[input::EntryDataMode::Crc32, input::EntryDataMode::KeepAndCrc32] {
                     for (index_path, data_path) in V2_PACKS_AND_INDICES {
-                        let resolve = {
+                        assert_index_write(mode, compressed, index_path, data_path)?;
-                            let buf = gix_features::threading::OwnShared::new({
-                                let file = std::fs::File::open(fixture_path(data_path))?;
-                                unsafe { memmap2::Mmap::map(&file)? }
-                            });
-                            move |entry: EntryRange, out: &mut Vec<u8>| {
-                                buf.get(entry.start as usize..entry.end as usize)
-                                    .map(|slice| out.copy_from_slice(slice))
-                            }
-                        };
-                        assert_index_write(mode, compressed, index_path, data_path, resolve)?;
                     }
                 }
             }
     let _it_should_work = index
         .traverse_with_index(
             &data,
-            || {
+            |_, _, _, _| {
-                |_, _, _, _| {
+                count.fetch_add(1, Ordering::SeqCst);
-                    count.fetch_add(1, Ordering::SeqCst);
+                Ok::<_, std::io::Error>(())
-                    Ok::<_, std::io::Error>(())
-                }
             },
             progress::Discard,
             &AtomicBool::new(false),