refactor

Sebastian Thiel committed 6 years ago

Commit 6cbb7cc1 parent 64158e03 files · +189 −202Browse files

Unified · Split

3 files

git-odb/src/pack/tree/traverse.rs +0 −202
git-odb/src/pack/tree/traverse/mod.rs +100 −0
git-odb/src/pack/tree/traverse/resolve.rs +89 −0

@@ -1,202 +0,0 @@
1	use crate::{
2	pack::index::write::EntrySlice,
3	pack::tree::{Item, Tree},
4	};
5	use git_features::{parallel, parallel::in_parallel_if, progress::Progress};
6	use git_object::HashKind;
7
8	impl<T> Tree<T>
9	where
10	T: Default + Send,
11	{
12	pub fn traverse<F, P, MBFN, BR, MCFN>(
13	mut self,
14	should_run_in_parallel: impl FnOnce() -> bool,
15	resolve: F,
16	mut progress: P,
17	thread_limit: Option<usize>,
18	pack_entries_end: u64,
19	hash_kind: HashKind,
20	modify_base: MBFN,
21	modify_child: MCFN,
22	) -> Result<Vec<Item<T>>, Box<dyn std::error::Error + Send + Sync>>
23	where
24	F: for<'r> Fn(EntrySlice, &'r mut Vec<u8>) -> Option<()> + Send + Sync,
25	P: Progress,
26	<P as Progress>::SubProgress: Send,
27	MBFN: for<'r> Fn(&'r mut T, &'r [u8], HashKind) -> BR,
28	BR: Clone,
29	MCFN: for<'r> Fn(&'r mut T, BR),
30	{
31	self.pack_entries_end = Some(pack_entries_end);
32	let (chunk_size, thread_limit, _) = parallel::optimize_chunk_size_and_thread_limit(1, None, thread_limit, None);
33	let reduce_progress = parking_lot::Mutex::new(progress.add_child("Resolving"));
34
35	// SAFETY: We are owning 'self', and it's the UnsafeCell which we are supposed to use requiring unsafe on every access now.
36	#[allow(unsafe_code)]
37	let num_objects = unsafe { (*self.items.get()).len() } as u32;
38	in_parallel_if(
39	should_run_in_parallel,
40	self.iter_root_chunks(chunk_size),
41	thread_limit,
42	\|thread_index\| {
43	(
44	Vec::<u8>::with_capacity(4096),
45	reduce_progress.lock().add_child(format!("thread {}", thread_index)),
46	)
47	},
48	// \|root_nodes, state\| apply_deltas(root_nodes, state, &resolver, kind.hash()),
49	\|_root_nodes, _state\| Ok(0),
50	Reducer::new(num_objects, &reduce_progress),
51	)?;
52	Ok(self.into_items())
53	}
54	}
55
56	pub(crate) struct Reducer<'a, P> {
57	item_count: usize,
58	progress: &'a parking_lot::Mutex<P>,
59	start: std::time::Instant,
60	}
61
62	impl<'a, P> Reducer<'a, P>
63	where
64	P: Progress,
65	{
66	pub fn new(num_objects: u32, progress: &'a parking_lot::Mutex<P>) -> Self {
67	progress.lock().init(Some(num_objects), Some("objects"));
68	Reducer {
69	item_count: 0,
70	progress,
71	start: std::time::Instant::now(),
72	}
73	}
74	}
75
76	impl<'a, P> parallel::Reducer for Reducer<'a, P>
77	where
78	P: Progress,
79	{
80	type Input = Result<usize, Box<dyn std::error::Error + Send + Sync>>;
81	type Output = ();
82	type Error = Box<dyn std::error::Error + Send + Sync>;
83
84	fn feed(&mut self, input: Self::Input) -> Result<(), Self::Error> {
85	let input = input?;
86	self.item_count += input;
87	self.progress.lock().set(self.item_count as u32);
88	Ok(())
89	}
90
91	fn finalize(self) -> Result<Self::Output, Self::Error> {
92	self.progress
93	.lock()
94	.show_throughput(self.start, self.item_count as u32, "objects");
95	Ok(())
96	}
97	}
98
99	mod apply {
100	use crate::{
101	hash, loose,
102	pack::{self, index::write::EntrySlice, index::write::Error},
103	zlib,
104	};
105	use git_features::progress::Progress;
106	use git_object::{owned, HashKind};
107	use std::{cell::RefCell, collections::BTreeMap, io};
108
109	pub(crate) fn apply_deltas<T, F, P, MBFN, BR, MCFN>(
110	nodes: Vec<pack::tree::Node<T>>,
111	(bytes_buf, progress): &mut (Vec<u8>, P),
112	resolve: F,
113	hash_kind: HashKind,
114	modify_base: MBFN,
115	modify_child: MCFN,
116	) -> Result<usize, Error>
117	where
118	F: for<'r> Fn(EntrySlice, &'r mut Vec<u8>) -> Option<()> + Send + Sync,
119	P: Progress,
120	MBFN: for<'r> Fn(&'r mut T, &'r [u8], HashKind) -> BR,
121	BR: Clone,
122	MCFN: for<'r> Fn(&'r mut T, BR),
123	T: Default,
124	{
125	let mut decompressed_bytes_by_pack_offset = BTreeMap::new();
126	let bytes_buf = RefCell::new(bytes_buf);
127	let mut num_objects = 0;
128	let decompress_from_resolver = \|slice: EntrySlice\| -> Result<Vec<u8>, Error> {
129	let mut bytes_buf = bytes_buf.borrow_mut();
130	bytes_buf.resize((slice.end - slice.start) as usize, 0);
131	resolve(slice.clone(), &mut bytes_buf).ok_or_else(\|\| Error::ConsumeResolveFailed(slice.start))?;
132	let entry = pack::data::Entry::from_bytes(&bytes_buf, slice.start);
133	decompress_all_at_once(
134	&bytes_buf[entry.header_size() as usize..],
135	entry.decompressed_size as usize,
136	)
137	};
138
139	// Traverse the tree breadth first and loose the data produced for the base as it won't be needed anymore.
140	progress.init(None, Some("objects"));
141
142	// each node is a base, and its children always start out as deltas which become a base after applying them.
143	// These will be pushed onto our stack until all are processed
144	let root_level = 0;
145	let mut nodes: Vec<_> = nodes.into_iter().map(\|n\| (root_level, n)).collect();
146	while let Some((level, mut base)) = nodes.pop() {
147	let base_bytes = if level == root_level {
148	decompress_from_resolver(base.entry_slice())?
149	} else {
150	decompressed_bytes_by_pack_offset
151	.remove(&base.offset())
152	.expect("we store the resolved delta buffer when done")
153	};
154
155	let base_result = modify_base(&mut base.data, &base_bytes, hash_kind);
156	num_objects += 1;
157	for mut child in base.store_changes_then_into_child_iter() {
158	let delta_bytes = decompress_from_resolver(child.entry_slice())?;
159	let (base_size, consumed) = pack::data::decode::delta_header_size_ofs(&delta_bytes);
160	let mut header_ofs = consumed;
161	assert_eq!(
162	base_bytes.len(),
163	base_size as usize,
164	"recorded base size in delta does not match"
165	);
166	let (result_size, consumed) = pack::data::decode::delta_header_size_ofs(&delta_bytes[consumed..]);
167	header_ofs += consumed;
168
169	let mut fully_resolved_delta_bytes = bytes_buf.borrow_mut();
170	fully_resolved_delta_bytes.resize(result_size as usize, 0);
171	pack::data::decode::apply_delta(
172	&base_bytes,
173	&mut fully_resolved_delta_bytes,
174	&delta_bytes[header_ofs..],
175	);
176
177	decompressed_bytes_by_pack_offset.insert(child.offset(), fully_resolved_delta_bytes.to_owned());
178	modify_child(&mut child.data, base_result.clone());
179	nodes.push((level + 1, child));
180	}
181	}
182
183	Ok(num_objects)
184	}
185
186	fn compute_hash(kind: git_object::Kind, bytes: &[u8], hash_kind: HashKind) -> owned::Id {
187	let mut write = hash::Write::new(io::sink(), hash_kind);
188	loose::object::header::encode(kind, bytes.len() as u64, &mut write)
189	.expect("write to sink and hash cannot fail");
190	write.hash.update(bytes);
191	owned::Id::from(write.hash.digest())
192	}
193
194	fn decompress_all_at_once(b: &[u8], decompressed_len: usize) -> Result<Vec<u8>, Error> {
195	let mut out = Vec::new();
196	out.resize(decompressed_len, 0);
197	zlib::Inflate::default()
198	.once(&b, &mut out, true)
199	.map_err(\|err\| Error::ConsumeZlibInflate(err, "Failed to decompress entry"))?;
200	Ok(out)
201	}
202	}

git-odb/src/pack/tree/traverse/mod.rs+100−0

@@ -0,0 +1,100 @@
	1	use crate::{
	2	pack::index::write::EntrySlice,
	3	pack::tree::{Item, Tree},
	4	};
	5	use git_features::{parallel, parallel::in_parallel_if, progress::Progress};
	6	use git_object::HashKind;
	7
	8	mod resolve;
	9
	10	impl<T> Tree<T>
	11	where
	12	T: Default + Send,
	13	{
	14	pub fn traverse<F, P, MBFN, BR, MCFN>(
	15	mut self,
	16	should_run_in_parallel: impl FnOnce() -> bool,
	17	resolve: F,
	18	mut progress: P,
	19	thread_limit: Option<usize>,
	20	pack_entries_end: u64,
	21	hash_kind: HashKind,
	22	modify_base: MBFN,
	23	modify_child: MCFN,
	24	) -> Result<Vec<Item<T>>, Box<dyn std::error::Error + Send + Sync>>
	25	where
	26	F: for<'r> Fn(EntrySlice, &'r mut Vec<u8>) -> Option<()> + Send + Sync,
	27	P: Progress,
	28	<P as Progress>::SubProgress: Send,
	29	MBFN: for<'r> Fn(&'r mut T, &'r [u8], HashKind) -> BR + Send + Sync,
	30	BR: Clone,
	31	MCFN: for<'r> Fn(&'r mut T, BR) + Send + Sync,
	32	{
	33	self.pack_entries_end = Some(pack_entries_end);
	34	let (chunk_size, thread_limit, _) = parallel::optimize_chunk_size_and_thread_limit(1, None, thread_limit, None);
	35	let reduce_progress = parking_lot::Mutex::new(progress.add_child("Resolving"));
	36
	37	// SAFETY: We are owning 'self', and it's the UnsafeCell which we are supposed to use requiring unsafe on every access now.
	38	#[allow(unsafe_code)]
	39	let num_objects = unsafe { (*self.items.get()).len() } as u32;
	40	in_parallel_if(
	41	should_run_in_parallel,
	42	self.iter_root_chunks(chunk_size),
	43	thread_limit,
	44	\|thread_index\| {
	45	(
	46	Vec::<u8>::with_capacity(4096),
	47	reduce_progress.lock().add_child(format!("thread {}", thread_index)),
	48	)
	49	},
	50	\|root_nodes, state\| {
	51	resolve::deltas(root_nodes, state, &resolve, hash_kind, &modify_base, &modify_child).map_err(Into::into)
	52	},
	53	Reducer::new(num_objects, &reduce_progress),
	54	)?;
	55	Ok(self.into_items())
	56	}
	57	}
	58
	59	pub(crate) struct Reducer<'a, P> {
	60	item_count: usize,
	61	progress: &'a parking_lot::Mutex<P>,
	62	start: std::time::Instant,
	63	}
	64
	65	impl<'a, P> Reducer<'a, P>
	66	where
	67	P: Progress,
	68	{
	69	pub fn new(num_objects: u32, progress: &'a parking_lot::Mutex<P>) -> Self {
	70	progress.lock().init(Some(num_objects), Some("objects"));
	71	Reducer {
	72	item_count: 0,
	73	progress,
	74	start: std::time::Instant::now(),
	75	}
	76	}
	77	}
	78
	79	impl<'a, P> parallel::Reducer for Reducer<'a, P>
	80	where
	81	P: Progress,
	82	{
	83	type Input = Result<usize, Box<dyn std::error::Error + Send + Sync>>;
	84	type Output = ();
	85	type Error = Box<dyn std::error::Error + Send + Sync>;
	86
	87	fn feed(&mut self, input: Self::Input) -> Result<(), Self::Error> {
	88	let input = input?;
	89	self.item_count += input;
	90	self.progress.lock().set(self.item_count as u32);
	91	Ok(())
	92	}
	93
	94	fn finalize(self) -> Result<Self::Output, Self::Error> {
	95	self.progress
	96	.lock()
	97	.show_throughput(self.start, self.item_count as u32, "objects");
	98	Ok(())
	99	}
	100	}

git-odb/src/pack/tree/traverse/resolve.rs+89−0

@@ -0,0 +1,89 @@
	1	use crate::{
	2	pack::{self, index::write::EntrySlice, index::write::Error},
	3	zlib,
	4	};
	5	use git_features::progress::Progress;
	6	use git_object::HashKind;
	7	use std::{cell::RefCell, collections::BTreeMap};
	8
	9	pub(crate) fn deltas<T, F, P, MBFN, BR, MCFN>(
	10	nodes: Vec<pack::tree::Node<T>>,
	11	(bytes_buf, progress): &mut (Vec<u8>, P),
	12	resolve: F,
	13	hash_kind: HashKind,
	14	modify_base: MBFN,
	15	modify_child: MCFN,
	16	) -> Result<usize, Error>
	17	where
	18	F: for<'r> Fn(EntrySlice, &'r mut Vec<u8>) -> Option<()> + Send + Sync,
	19	P: Progress,
	20	MBFN: for<'r> Fn(&'r mut T, &'r [u8], HashKind) -> BR,
	21	BR: Clone,
	22	MCFN: for<'r> Fn(&'r mut T, BR),
	23	T: Default,
	24	{
	25	let mut decompressed_bytes_by_pack_offset = BTreeMap::new();
	26	let bytes_buf = RefCell::new(bytes_buf);
	27	let mut num_objects = 0;
	28	let decompress_from_resolver = \|slice: EntrySlice\| -> Result<Vec<u8>, Error> {
	29	let mut bytes_buf = bytes_buf.borrow_mut();
	30	bytes_buf.resize((slice.end - slice.start) as usize, 0);
	31	resolve(slice.clone(), &mut bytes_buf).ok_or_else(\|\| Error::ConsumeResolveFailed(slice.start))?;
	32	let entry = pack::data::Entry::from_bytes(&bytes_buf, slice.start);
	33	decompress_all_at_once(
	34	&bytes_buf[entry.header_size() as usize..],
	35	entry.decompressed_size as usize,
	36	)
	37	};
	38
	39	// Traverse the tree breadth first and loose the data produced for the base as it won't be needed anymore.
	40	progress.init(None, Some("objects"));
	41
	42	// each node is a base, and its children always start out as deltas which become a base after applying them.
	43	// These will be pushed onto our stack until all are processed
	44	let root_level = 0;
	45	let mut nodes: Vec<_> = nodes.into_iter().map(\|n\| (root_level, n)).collect();
	46	while let Some((level, mut base)) = nodes.pop() {
	47	let base_bytes = if level == root_level {
	48	decompress_from_resolver(base.entry_slice())?
	49	} else {
	50	decompressed_bytes_by_pack_offset
	51	.remove(&base.offset())
	52	.expect("we store the resolved delta buffer when done")
	53	};
	54
	55	let base_result = modify_base(&mut base.data, &base_bytes, hash_kind);
	56	num_objects += 1;
	57	for mut child in base.store_changes_then_into_child_iter() {
	58	let delta_bytes = decompress_from_resolver(child.entry_slice())?;
	59	let (base_size, consumed) = pack::data::decode::delta_header_size_ofs(&delta_bytes);
	60	let mut header_ofs = consumed;
	61	assert_eq!(
	62	base_bytes.len(),
	63	base_size as usize,
	64	"recorded base size in delta does not match"
	65	);
	66	let (result_size, consumed) = pack::data::decode::delta_header_size_ofs(&delta_bytes[consumed..]);
	67	header_ofs += consumed;
	68
	69	let mut fully_resolved_delta_bytes = bytes_buf.borrow_mut();
	70	fully_resolved_delta_bytes.resize(result_size as usize, 0);
	71	pack::data::decode::apply_delta(&base_bytes, &mut fully_resolved_delta_bytes, &delta_bytes[header_ofs..]);
	72
	73	decompressed_bytes_by_pack_offset.insert(child.offset(), fully_resolved_delta_bytes.to_owned());
	74	modify_child(&mut child.data, base_result.clone());
	75	nodes.push((level + 1, child));
	76	}
	77	}
	78
	79	Ok(num_objects)
	80	}
	81
	82	fn decompress_all_at_once(b: &[u8], decompressed_len: usize) -> Result<Vec<u8>, Error> {
	83	let mut out = Vec::new();
	84	out.resize(decompressed_len, 0);
	85	zlib::Inflate::default()
	86	.once(&b, &mut out, true)
	87	.map_err(\|err\| Error::ConsumeZlibInflate(err, "Failed to decompress entry"))?;
	88	Ok(out)
	89	}

refactor

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code