History

[pack #167] remove iterator based count objects impl…

…and backport all capabilities like progress reporting and interruptability to something that's semantically similar.

Sebastian Thiel committed 5 years ago

Commit 7ec2f2b1 parent 0958fc84 files · +47 −319Browse files

Unified · Split

4 files
  1. git-pack/src/data/output/count/iter_from_objects.rs +16 −282
  2. git-pack/src/data/output/count/mod.rs +1 −1
  3. git-pack/tests/pack/data/output/count_and_entries.rs +1 −0
  4. gitoxide-core/src/pack/create.rs +29 −36
git-pack/src/data/output/count/iter_from_objects.rs+16−282
@@ -7,251 +7,6 @@ use git_object::immutable; · 6 unmodified lines
77
88 use crate::{data::output, find, FindExt};
99
10−/// Generate [`Count`][output::Count] from input `objects` with object expansion based on [`options`][Options]
11−/// to learn which objects would be part of a pack.
12−///
13−/// A [`Count`][output::Count] object maintains enough state to greatly accelerate future access of packed objects.
14−///
15−/// * `db` - the object store to use for accessing objects.
16−/// * `make_cache` - a function to create thread-local pack caches
17−/// * `objects_ids`
18−/// * A list of objects ids to add to the pack. Duplication checks are performed so no object is ever added to a pack twice.
19−/// * Objects may be expanded based on the provided [`options`][Options]
20−/// * `progress`
21−/// * a way to obtain progress information
22−/// * `options`
23−/// * more configuration
24−pub fn iter_from_objects<Find, Iter, Oid, Cache>(
25− db: Find,
26− make_cache: impl Fn() -> Cache + Send + Clone + Sync + 'static,
27− objects_ids: Iter,
28− progress: impl Progress,
29− Options {
30− thread_limit,
31− input_object_expansion,
32− chunk_size,
33− }: Options,
34−) -> impl Iterator<Item = Result<Vec<output::Count>, Error<find::existing::Error<Find::Error>>>>
35− + parallel::reduce::Finalize<Reduce = reduce::Statistics<Error<find::existing::Error<Find::Error>>>>
36−where
37− Find: crate::Find + Clone + Send + Sync + 'static,
38− <Find as crate::Find>::Error: Send,
39− Iter: Iterator<Item = Oid> + Send + 'static,
40− Oid: AsRef<oid> + Send + 'static,
41− Cache: crate::cache::DecodeEntry,
42−{
43− let lower_bound = objects_ids.size_hint().0;
44− let (chunk_size, thread_limit, _) = parallel::optimize_chunk_size_and_thread_limit(
45− chunk_size,
46− if lower_bound == 0 { None } else { Some(lower_bound) },
47− thread_limit,
48− None,
49− );
50− let chunks = util::Chunks {
51− iter: objects_ids,
52− size: chunk_size,
53− };
54− let seen_objs = Arc::new(dashmap::DashSet::<ObjectId>::new());
55− let progress = Arc::new(parking_lot::Mutex::new(progress));
56−
57− parallel::reduce::Stepwise::new(
58− chunks,
59− thread_limit,
60− {
61− let progress = Arc::clone(&progress);
62− move |n| {
63− (
64− Vec::new(), // object data buffer
65− Vec::new(), // object data buffer 2 to hold two objects at a time
66− make_cache(), // cache to speed up pack operations
67− {
68− let mut p = progress.lock().add_child(format!("thread {}", n));
69− p.init(None, git_features::progress::count("objects"));
70− p
71− },
72− )
73− }
74− },
75− {
76− let seen_objs = Arc::clone(&seen_objs);
77− move |oids: Vec<Oid>, (buf1, buf2, cache, progress)| {
78− use ObjectExpansion::*;
79− let mut out = Vec::new();
80− let mut tree_traversal_state = git_traverse::tree::breadthfirst::State::default();
81− let mut tree_diff_state = git_diff::tree::State::default();
82− let mut parent_commit_ids = Vec::new();
83− let seen_objs = seen_objs.as_ref();
84− let mut traverse_delegate = tree::traverse::AllUnseen::new(seen_objs);
85− let mut changes_delegate = tree::changes::AllNew::new(seen_objs);
86− let mut outcome = Outcome::default();
87− let stats = &mut outcome;
88−
89− for id in oids.into_iter() {
90− let id = id.as_ref();
91− let obj = db.find_existing(id, buf1, cache)?;
92− stats.input_objects += 1;
93− match input_object_expansion {
94− TreeAdditionsComparedToAncestor => {
95− use git_object::Kind::*;
96− let mut obj = obj;
97− let mut id = id.to_owned();
98−
99− loop {
100− push_obj_count_unique(&mut out, seen_objs, &id, &obj, progress, stats, false);
101− match obj.kind {
102− Tree | Blob => break,
103− Tag => {
104− id = immutable::TagIter::from_bytes(obj.data)
105− .target_id()
106− .expect("every tag has a target");
107− obj = db.find_existing(id, buf1, cache)?;
108− stats.expanded_objects += 1;
109− continue;
110− }
111− Commit => {
112− let current_tree_iter = {
113− let mut commit_iter = immutable::CommitIter::from_bytes(obj.data);
114− let tree_id = commit_iter.tree_id().expect("every commit has a tree");
115− parent_commit_ids.clear();
116− for token in commit_iter {
117− match token {
118− Ok(immutable::commit::iter::Token::Parent { id }) => {
119− parent_commit_ids.push(id)
120− }
121− Ok(_) => break,
122− Err(err) => return Err(Error::CommitDecode(err)),
123− }
124− }
125− let obj = db.find_existing(tree_id, buf1, cache)?;
126− push_obj_count_unique(
127− &mut out, seen_objs, &tree_id, &obj, progress, stats, true,
128− );
129− immutable::TreeIter::from_bytes(obj.data)
130− };
131−
132− let objects = if parent_commit_ids.is_empty() {
133− traverse_delegate.clear();
134− git_traverse::tree::breadthfirst(
135− current_tree_iter,
136− &mut tree_traversal_state,
137− |oid, buf| {
138− stats.decoded_objects += 1;
139− db.find_existing_tree_iter(oid, buf, cache).ok()
140− },
141− &mut traverse_delegate,
142− )
143− .map_err(Error::TreeTraverse)?;
144− &traverse_delegate.objects
145− } else {
146− for commit_id in &parent_commit_ids {
147− let parent_tree_id = {
148− let parent_commit_obj = db.find_existing(commit_id, buf2, cache)?;
149−
150− push_obj_count_unique(
151− &mut out,
152− seen_objs,
153− commit_id,
154− &parent_commit_obj,
155− progress,
156− stats,
157− true,
158− );
159− immutable::CommitIter::from_bytes(parent_commit_obj.data)
160− .tree_id()
161− .expect("every commit has a tree")
162− };
163− let parent_tree = {
164− let parent_tree_obj =
165− db.find_existing(parent_tree_id, buf2, cache)?;
166− push_obj_count_unique(
167− &mut out,
168− seen_objs,
169− &parent_tree_id,
170− &parent_tree_obj,
171− progress,
172− stats,
173− true,
174− );
175− immutable::TreeIter::from_bytes(parent_tree_obj.data)
176− };
177−
178− changes_delegate.clear();
179− git_diff::tree::Changes::from(Some(parent_tree))
180− .needed_to_obtain(
181− current_tree_iter.clone(),
182− &mut tree_diff_state,
183− |oid, buf| {
184− stats.decoded_objects += 1;
185− db.find_existing_tree_iter(oid, buf, cache).ok()
186− },
187− &mut changes_delegate,
188− )
189− .map_err(Error::TreeChanges)?;
190− }
191− &changes_delegate.objects
192− };
193− for id in objects.iter() {
194− out.push(id_to_count(&db, buf2, id, progress, stats));
195− }
196− break;
197− }
198− }
199− }
200− }
201− TreeContents => {
202− use git_object::Kind::*;
203− let mut id: ObjectId = id.into();
204− let mut obj = obj;
205− loop {
206− push_obj_count_unique(&mut out, seen_objs, &id, &obj, progress, stats, false);
207− match obj.kind {
208− Tree => {
209− traverse_delegate.clear();
210− git_traverse::tree::breadthfirst(
211− git_object::immutable::TreeIter::from_bytes(obj.data),
212− &mut tree_traversal_state,
213− |oid, buf| {
214− stats.decoded_objects += 1;
215− db.find_existing_tree_iter(oid, buf, cache).ok()
216− },
217− &mut traverse_delegate,
218− )
219− .map_err(Error::TreeTraverse)?;
220− for id in traverse_delegate.objects.iter() {
221− out.push(id_to_count(&db, buf1, id, progress, stats));
222− }
223− break;
224− }
225− Commit => {
226− id = immutable::CommitIter::from_bytes(obj.data)
227− .tree_id()
228− .expect("every commit has a tree");
229− stats.expanded_objects += 1;
230− obj = db.find_existing(id, buf1, cache)?;
231− continue;
232− }
233− Blob => break,
234− Tag => {
235− id = immutable::TagIter::from_bytes(obj.data)
236− .target_id()
237− .expect("every tag has a target");
238− stats.expanded_objects += 1;
239− obj = db.find_existing(id, buf1, cache)?;
240− continue;
241− }
242− }
243− }
244− }
245− AsIs => push_obj_count_unique(&mut out, seen_objs, id, &obj, progress, stats, false),
246− }
247− }
248− Ok((out, outcome))
249− }
250− },
251− reduce::Statistics::default(),
252− )
253−}
254−
25510 /// Generate [`Count`][output::Count]s from input `objects` with object expansion based on [`options`][Options]
25611 /// to learn which objects would would constitute a pack. This step is required to know exactly how many objects would
25712 /// be in a pack while keeping data around to avoid minimize object database access.
@@ -265,6 +20,8 @@ where · 7 unmodified lines
26520 /// * Objects may be expanded based on the provided [`options`][Options]
26621 /// * `progress`
26722 /// * a way to obtain progress information
23+/// * `should_interrupt`
24+/// * A flag that is set to true if the operation should stop
26825 /// * `options`
26926 /// * more configuration
27027 pub fn objects<Find, Iter, Oid, Cache>(
@@ -272,6 +29,7 @@ pub fn objects<Find, Iter, Oid, Cache>( · 1 unmodified line
27229 make_cache: impl Fn() -> Cache + Send + Sync,
27330 objects_ids: Iter,
27431 progress: impl Progress,
32+ should_interrupt: &AtomicBool,
27533 Options {
27634 thread_limit,
27735 input_object_expansion,
@@ -297,12 +55,13 @@ where · 19 unmodified lines
29755 size: chunk_size,
29856 };
29957 let seen_objs = dashmap::DashSet::<ObjectId>::new();
300− let progress = parking_lot::Mutex::new(progress);
58+ let top_level_progress = Arc::new(parking_lot::Mutex::new(progress));
30159
30260 parallel::in_parallel(
30361 chunks,
30462 thread_limit,
30563 {
64+ let progress = Arc::clone(&top_level_progress);
30665 move |n| {
30766 (
30867 Vec::new(), // object data buffer
@@ -317,6 +76,7 @@ where · 8 unmodified lines
31776 }
31877 },
31978 {
79+ let top_level_progress = Arc::clone(&top_level_progress);
32080 move |oids: Vec<Oid>, (buf1, buf2, cache, progress)| {
32181 use ObjectExpansion::*;
32282 let mut out = Vec::new();
@@ -327,6 +87,9 @@ where · 4 unmodified lines
32787 let mut changes_delegate = tree::changes::AllNew::new(&seen_objs);
32888 let mut outcome = Outcome::default();
32989 let stats = &mut outcome;
90+ if should_interrupt.load(Ordering::Relaxed) {
91+ return Err(Error::Interrupted);
92+ }
33093
33194 for id in oids.into_iter() {
33295 let id = id.as_ref();
@@ -487,10 +250,11 @@ where · 154 unmodified lines
487250 AsIs => push_obj_count_unique(&mut out, &seen_objs, id, &obj, progress, stats, false),
488251 }
489252 }
253+ top_level_progress.lock().inc_by(out.len());
490254 Ok((out, outcome))
491255 }
492256 },
493− reduce::Statistics2::default(),
257+ reduce::Statistics::default(),
494258 )
495259 }
496260
@@ -761,8 +525,11 @@ mod types { · 264 unmodified lines
761525 TreeTraverse(git_traverse::tree::breadthfirst::Error),
762526 #[error(transparent)]
763527 TreeChanges(git_diff::tree::changes::Error),
528+ #[error("Operation interrupted")]
529+ Interrupted,
764530 }
765531 }
532+use std::sync::atomic::{AtomicBool, Ordering};
766533 pub use types::{Error, ObjectExpansion, Options, Outcome};
767534
768535 mod reduce {
@@ -775,46 +542,13 @@ mod reduce { · 6 unmodified lines
775542
776543 pub struct Statistics<E> {
777544 total: Outcome,
545+ counts: Vec<output::Count>,
778546 _err: PhantomData<E>,
779547 }
780548
781549 impl<E> Default for Statistics<E> {
782550 fn default() -> Self {
783551 Statistics {
784− total: Default::default(),
785− _err: PhantomData::default(),
786− }
787− }
788− }
789−
790− impl<Error> parallel::Reduce for Statistics<Error> {
791− type Input = Result<(Vec<output::Count>, Outcome), Error>;
792− type FeedProduce = Vec<output::Count>;
793− type Output = Outcome;
794− type Error = Error;
795−
796− fn feed(&mut self, item: Self::Input) -> Result<Self::FeedProduce, Self::Error> {
797− item.map(|(counts, mut stats)| {
798− stats.total_objects = counts.len();
799− self.total.aggregate(stats);
800− counts
801− })
802− }
803−
804− fn finalize(self) -> Result<Self::Output, Self::Error> {
805− Ok(self.total)
806− }
807− }
808−
809− pub struct Statistics2<E> {
810− total: Outcome,
811− counts: Vec<output::Count>,
812− _err: PhantomData<E>,
813− }
814−
815− impl<E> Default for Statistics2<E> {
816− fn default() -> Self {
817− Statistics2 {
818552 total: Default::default(),
819553 counts: Default::default(),
820554 _err: PhantomData::default(),
@@ -822,7 +556,7 @@ mod reduce { · 1 unmodified line
822556 }
823557 }
824558
825− impl<Error> parallel::Reduce for Statistics2<Error> {
559+ impl<Error> parallel::Reduce for Statistics<Error> {
826560 type Input = Result<(Vec<output::Count>, Outcome), Error>;
827561 type FeedProduce = ();
828562 type Output = (Vec<output::Count>, Outcome);
git-pack/src/data/output/count/mod.rs+1−1
@@ -27,4 +27,4 @@ impl Count { · 26 unmodified lines
2727
2828 ///
2929 pub mod iter_from_objects;
30−pub use iter_from_objects::{iter_from_objects, objects};
30+pub use iter_from_objects::objects;
git-pack/tests/pack/data/output/count_and_entries.rs+1−0
@@ -252,6 +252,7 @@ fn traversals() -> crate::Result { · 251 unmodified lines
252252 })))
253253 .filter(|o| !o.is_null()),
254254 progress::Discard,
255+ &AtomicBool::new(false),
255256 count::iter_from_objects::Options {
256257 input_object_expansion: expansion_mode,
257258 thread_limit: deterministic_count_needs_single_thread,
gitoxide-core/src/pack/create.rs+29−36
@@ -97,6 +97,7 @@ pub fn create<W>( · 96 unmodified lines
9797 where
9898 W: std::io::Write,
9999 {
100+ // TODO: review need for Arc for the counting part, use different kinds of Easy depending on need
100101 let repo = Arc::new(git_repository::discover(repository_path)?);
101102 progress.init(Some(2), progress::steps());
102103 let tips = tips.into_iter();
@@ -108,6 +109,7 @@ where · 5 unmodified lines
108109 let tips = tips
109110 .map(|tip| {
110111 ObjectId::from_hex(&Vec::from_os_str_lossy(tip.as_ref())).or_else({
112+ // TODO: Use Easy when ready…
111113 let packed = repo.refs.packed().ok().flatten();
112114 let refs = &repo.refs;
113115 let repo = Arc::clone(&repo);
@@ -132,12 +134,13 @@ where · 18 unmodified lines
132134 Err(_) => unreachable!("there is only one instance left here"),
133135 };
134136 let iter = Box::new(
137+ // TODO: Easy-based traversal
135138 traverse::commit::Ancestors::new(tips, traverse::commit::ancestors::State::default(), {
136139 let db = Arc::clone(&db);
137140 move |oid, buf| db.find_existing_commit_iter(oid, buf, &mut pack::cache::Never).ok()
138141 })
139− .inspect(move |_| progress.inc())
140− .map(Result::unwrap),
142+ .map(Result::unwrap)
143+ .inspect(move |_| progress.inc()),
141144 ); // todo: should there be a better way, maybe let input support Option or Result
142145 (db, iter)
143146 }
@@ -157,44 +160,34 @@ where · 13 unmodified lines
157160 };
158161
159162 let mut stats = Statistics::default();
160− let chunk_size = 200;
161− let start = Instant::now();
163+ let chunk_size = 1000; // What's a good value for this?
162164 let counts = {
163165 let mut progress = progress.add_child("counting");
164166 progress.init(None, progress::count("objects"));
165− let mut interruptible_counts_iter = interrupt::Iter::new(
166− pack::data::output::count::iter_from_objects(
167− Arc::clone(&db),
168− move || {
169− if nondeterministic_count {
170− Box::new(pack::cache::lru::StaticLinkedList::<64>::default()) as Box<dyn DecodeEntry>
171− } else {
172− Box::new(pack::cache::lru::MemoryCappedHashmap::new(400 * 1024 * 1024)) as Box<dyn DecodeEntry>
173− // todo: Make that configurable
174− }
175− },
176− input,
177− progress.add_child("threads"),
178− pack::data::output::count::iter_from_objects::Options {
179− thread_limit: if nondeterministic_count || matches!(expansion, ObjectExpansion::None) {
180− thread_limit
181− } else {
182− Some(1)
183− },
184− chunk_size,
185− input_object_expansion: expansion.into(),
167+ let (mut counts, count_stats) = pack::data::output::count::objects(
168+ Arc::clone(&db),
169+ move || {
170+ if nondeterministic_count {
171+ Box::new(pack::cache::lru::StaticLinkedList::<64>::default()) as Box<dyn DecodeEntry>
172+ } else {
173+ Box::new(pack::cache::lru::MemoryCappedHashmap::new(400 * 1024 * 1024)) as Box<dyn DecodeEntry>
174+ // todo: Make that configurable
175+ }
176+ },
177+ input,
178+ progress::ThroughputOnDrop::new(progress),
179+ &interrupt::IS_INTERRUPTED,
180+ pack::data::output::count::iter_from_objects::Options {
181+ thread_limit: if nondeterministic_count || matches!(expansion, ObjectExpansion::None) {
182+ thread_limit
183+ } else {
184+ Some(1)
186185 },
187− ),
188− make_cancellation_err,
189− );
190− let mut counts = Vec::new();
191− for c in interruptible_counts_iter.by_ref() {
192− let c = c??;
193− progress.inc_by(c.len());
194− counts.extend(c.into_iter());
195− }
196− stats.counts = interruptible_counts_iter.into_inner().finalize()?;
197− progress.show_throughput(start);
186+ chunk_size,
187+ input_object_expansion: expansion.into(),
188+ },
189+ )?;
190+ stats.counts = count_stats;
198191 counts.shrink_to_fit();
199192 counts
200193 };