Make use of shared reducer in upcoming indexed verify implementation

Sebastian Thiel committed 6 years ago

Commit 290eae11 parent 10fc88d4 files · +47 −44Browse files

Unified · Split

4 files

git-odb/src/pack/index/verify/indexed.rs +11 −21
git-odb/src/pack/index/verify/lookup.rs +7 −21
git-odb/src/pack/index/verify/mod.rs +7 −2
git-odb/src/pack/index/verify/reduce.rs +22 −0

git-odb/src/pack/index/verify/indexed.rs+11−21

@@ -1,7 +1,7 @@
1	1	use super::{Error, Mode, Outcome};
2	2	use crate::{pack, pack::index};
3	3	use git_features::{
4		− parallel::{self, in_parallel_if},
	4	+ parallel::in_parallel_if,
5	5	progress::{self, Progress},
6	6	};
7	7	use std::time::SystemTime;
@@ -12,7 +12,7 @@ impl index::File { · 4 unmodified lines
12	12	thread_limit: Option<usize>,
13	13	_mode: Mode,
14	14	_make_cache: impl Fn() -> C + Send + Sync,
15		− mut progress: progress::DoOrDiscard<P>,
	15	+ mut root: progress::DoOrDiscard<P>,
16	16	pack: &pack::data::File,
17	17	) -> Result<Outcome, Error>
18	18	where
@@ -21,7 +21,7 @@ impl index::File { · 2 unmodified lines
21	21	C: pack::cache::DecodeEntry,
22	22	{
23	23	let offsets = {
24		− let mut indexing_progress = progress.add_child("preparing pack offsets");
	24	+ let mut indexing_progress = root.add_child("preparing pack offsets");
25	25	indexing_progress.init(Some(self.num_objects), Some("objects"));
26	26	let then = SystemTime::now();
27	27	let iter = self.sorted_offsets().into_iter();
@@ -33,31 +33,21 @@ impl index::File { · 5 unmodified lines
33	33	));
34	34	iter
35	35	};
36		− let tree = pack::graph::DeltaTree::from_sorted_offsets(offsets, pack.path(), progress.add_child("indexing"))?;
	36	+ let tree = pack::graph::DeltaTree::from_sorted_offsets(offsets, pack.path(), root.add_child("indexing"))?;
37	37	let if_there_are_enough_objects = \|\| self.num_objects > 10_000;
38		− struct Reducer;
39	38
40		− impl parallel::Reducer for Reducer {
41		− type Input = ();
42		− type Output = Outcome;
43		− type Error = Error;
44		−
45		− fn feed(&mut self, input: Self::Input) -> Result<(), Self::Error> {
46		− unimplemented!()
47		− }
48		−
49		− fn finalize(self) -> Result<Self::Output, Self::Error> {
50		− unimplemented!()
51		− }
52		− }
53		− let reduce = Reducer;
	39	+ let reduce_progress = std::sync::Mutex::new({
	40	+ let mut p = root.add_child("Checking");
	41	+ p.init(Some(self.num_objects()), Some("objects"));
	42	+ p
	43	+ });
54	44	in_parallel_if(
55	45	if_there_are_enough_objects,
56	46	tree.bases(),
57	47	thread_limit,
58	48	\|_index\| (),
59		− \|node: pack::graph::Node, state: &mut ()\| (),
60		− reduce,
	49	+ \|node: pack::graph::Node, state: &mut ()\| Ok::<_, Error>(Vec::new()),
	50	+ index::verify::Reducer::from_progress(&reduce_progress, pack.data_len()),
61	51	)?;
62	52
63	53	unimplemented!()

git-odb/src/pack/index/verify/lookup.rs+7−21

@@ -1,11 +1,10 @@
1		−use super::{reduce::Reducer, Error, Mode, Outcome};
	1	+use super::{Error, Mode, Outcome, Reducer};
2	2	use crate::{pack, pack::data::decode, pack::index};
3	3	use git_features::{
4	4	parallel::in_parallel_if,
5	5	progress::{self, Progress},
6	6	};
7	7	use git_object::{borrowed, bstr::ByteSlice, owned};
8		−use std::time::Instant;
9	8
10	9	/// Verify and validate the content of the index file
11	10	impl index::File {
@@ -33,11 +32,11 @@ impl index::File { · 21 unmodified lines
33	32	const CHUNK_SIZE: usize = 1000;
34	33	let there_are_enough_entries_to_process = \|\| index_entries.len() > CHUNK_SIZE * 2;
35	34	let input_chunks = index_entries.chunks(CHUNK_SIZE.max(index_entries.len() / CHUNK_SIZE));
36		− let reduce_progress = std::sync::Mutex::new(root.add_child("Checking"));
37		− reduce_progress
38		− .lock()
39		− .unwrap()
40		− .init(Some(self.num_objects()), Some("objects"));
	35	+ let reduce_progress = std::sync::Mutex::new({
	36	+ let mut p = root.add_child("Checking");
	37	+ p.init(Some(self.num_objects()), Some("objects"));
	38	+ p
	39	+ });
41	40	let state_per_thread = \|index\| {
42	41	(
43	42	make_cache(),
@@ -141,20 +140,7 @@ impl index::File { · 97 unmodified lines
141	140	}
142	141	Ok(stats)
143	142	},
144		− Reducer {
145		− progress: &reduce_progress,
146		− then: Instant::now(),
147		− entries_seen: 0,
148		− chunks_seen: 0,
149		− stats: Outcome {
150		− average: decode::Outcome::default_from_kind(git_object::Kind::Tree),
151		− objects_per_chain_length: Default::default(),
152		− total_compressed_entries_size: 0,
153		− total_decompressed_entries_size: 0,
154		− total_object_size: 0,
155		− pack_size: pack.data_len() as u64,
156		− },
157		− },
	143	+ Reducer::from_progress(&reduce_progress, pack.data_len()),
158	144	)
159	145	}
160	146	}

git-odb/src/pack/index/verify/mod.rs+7−2

@@ -107,10 +107,14 @@ pub enum Mode { · 106 unmodified lines
107	107	#[derive(Debug, Eq, PartialEq, Hash, Clone, Copy)]
108	108	pub enum Algorithm {
109	109	/// We lookup each object similarly to what would happen during normal repository use.
110		− /// Uses more compute resources as it will resolve delta chains from back to front, potentially
	110	+ /// Uses more compute resources as it will resolve delta chains from back to front, but start right away
	111	+ /// without indexing or investing any memory in indices.
	112	+ ///
	113	+ /// This option may be well suited for big packs in memory-starved system that support memory mapping.
111	114	Lookup,
112	115	/// Build an index to allow decoding each delta and base exactly once, saving a lot of computational
113		− /// resource at the expense of resident memory, as we will use an additional DeltaTree to make that happen.
	116	+ /// resource at the expense of resident memory, as we will use an additional `DeltaTree` to accelerate
	117	+ /// delta chain resolution.
114	118	DeltaTreeLookup,
115	119	}
116	120
@@ -123,6 +127,7 @@ impl Default for Algorithm { · 6 unmodified lines
123	127	mod indexed;
124	128	mod lookup;
125	129	mod reduce;
	130	+pub(crate) use reduce::Reducer;
126	131
127	132	/// Verify and validate the content of the index file
128	133	impl index::File {

git-odb/src/pack/index/verify/reduce.rs+22−0

@@ -24,6 +24,28 @@ pub struct Reducer<'a, P> { · 23 unmodified lines
24	24	pub stats: verify::Outcome,
25	25	}
26	26
	27	+impl<'a, P> Reducer<'a, P>
	28	+where
	29	+ P: Progress,
	30	+{
	31	+ pub fn from_progress(progress: &'a std::sync::Mutex<P>, pack_data_len_in_bytes: usize) -> Self {
	32	+ Reducer {
	33	+ progress: &progress,
	34	+ then: Instant::now(),
	35	+ entries_seen: 0,
	36	+ chunks_seen: 0,
	37	+ stats: verify::Outcome {
	38	+ average: decode::Outcome::default_from_kind(git_object::Kind::Tree),
	39	+ objects_per_chain_length: Default::default(),
	40	+ total_compressed_entries_size: 0,
	41	+ total_decompressed_entries_size: 0,
	42	+ total_object_size: 0,
	43	+ pack_size: pack_data_len_in_bytes as u64,
	44	+ },
	45	+ }
	46	+ }
	47	+}
	48	+
27	49	impl<'a, P> parallel::Reducer for Reducer<'a, P>
28	50	where
29	51	P: Progress,