lael-labs/gitoxide

git-features/Cargo.toml+4−4

@@ -20,16 +20,16 @@ io-pipe = ["bytes"] · 19 unmodified lines
20		20
21	[[test]]	21	[[test]]
22	name = "parallel"	22	name = "parallel"
23	path = "tests/parallel.rs"	23	path = "tests/parallel_threaded.rs"
24	required-features = ["parallel"]	24	required-features = ["parallel"]
25		25
26	[[test]]	26	[[test]]
27	name = "parallel_shared"	27	name = "parallel-shared-multi-threaded"
28	path = "tests/parallel_shared.rs"	28	path = "tests/parallel_shared_threaded.rs"
29	required-features = ["parallel"]	29	required-features = ["parallel"]
30		30
31	[[test]]	31	[[test]]
32	name = "parallel_shared_in_serial"	32	name = "parallel-shared-single-threaded"
33	path = "tests/parallel_shared.rs"	33	path = "tests/parallel_shared.rs"
34	required-features = []	34	required-features = []
35		35

git-features/src/parallel/mod.rs+4−245

@@ -128,251 +128,6 @@ pub(crate) fn num_threads(thread_limit: Option<usize>) -> usize { · 127 unmodified lines
128	.unwrap_or_else(logical_cores)	128	.unwrap_or_else(logical_cores)
129	}	129	}
130		130
131	///
132	pub mod reduce {
133
134	#[cfg(feature = "parallel")]
135	mod stepped {
136	use crate::parallel::num_threads;
137
138	/// An iterator adaptor to allow running computations using [`in_parallel()`][crate::parallel::in_parallel()] in a step-wise manner, see the [module docs][crate::parallel]
139	/// for details.
140	pub struct Stepwise<Reduce: super::Reduce> {
141	/// This field is first to assure it's dropped first and cause threads that are dropped next to stop their loops
142	/// as sending results fails when the receiver is dropped.
143	receive_result: std::sync::mpsc::Receiver<Reduce::Input>,
144	/// `join()` will be called on these guards to assure every thread tries to send through a closed channel. When
145	/// that happens, they break out of their loops.
146	_threads: Vec<std::thread::JoinHandle<()>>,
147	/// The reducer is called only in the thread using the iterator, dropping it has no side effects.
148	reducer: Option<Reduce>,
149	}
150
151	impl<Reduce: super::Reduce> Drop for Stepwise<Reduce> {
152	fn drop(&mut self) {
153	let (_, sink) = std::sync::mpsc::channel();
154	drop(std::mem::replace(&mut self.receive_result, sink));
155
156	let mut last_err = None;
157	for handle in std::mem::take(&mut self._threads) {
158	if let Err(err) = handle.join() {
159	last_err = Some(err);
160	};
161	}
162	if let Some(thread_err) = last_err {
163	std::panic::resume_unwind(thread_err);
164	}
165	}
166	}
167
168	impl<Reduce: super::Reduce> Stepwise<Reduce> {
169	/// Instantiate a new iterator and start working in threads.
170	/// For a description of parameters, see [`in_parallel()`][crate::parallel::in_parallel()].
171	pub fn new<InputIter, ThreadStateFn, ConsumeFn, I, O, S>(
172	input: InputIter,
173	thread_limit: Option<usize>,
174	new_thread_state: ThreadStateFn,
175	consume: ConsumeFn,
176	reducer: Reduce,
177	) -> Self
178	where
179	InputIter: Iterator<Item = I> + Send + 'static,
180	ThreadStateFn: Fn(usize) -> S + Send + Clone + 'static,
181	ConsumeFn: Fn(I, &mut S) -> O + Send + Clone + 'static,
182	Reduce: super::Reduce<Input = O> + 'static,
183	I: Send + 'static,
184	O: Send + 'static,
185	{
186	let num_threads = num_threads(thread_limit);
187	let mut threads = Vec::with_capacity(num_threads + 1);
188	let receive_result = {
189	let (send_input, receive_input) = crossbeam_channel::bounded::<I>(num_threads);
190	let (send_result, receive_result) = std::sync::mpsc::sync_channel::<O>(num_threads);
191	for thread_id in 0..num_threads {
192	let handle = std::thread::spawn({
193	let send_result = send_result.clone();
194	let receive_input = receive_input.clone();
195	let new_thread_state = new_thread_state.clone();
196	let consume = consume.clone();
197	move \|\| {
198	let mut state = new_thread_state(thread_id);
199	for item in receive_input {
200	if send_result.send(consume(item, &mut state)).is_err() {
201	break;
202	}
203	}
204	}
205	});
206	threads.push(handle);
207	}
208	threads.push(std::thread::spawn(move \|\| {
209	for item in input {
210	if send_input.send(item).is_err() {
211	break;
212	}
213	}
214	}));
215	receive_result
216	};
217	Stepwise {
218	_threads: threads,
219	receive_result,
220	reducer: Some(reducer),
221	}
222	}
223
224	/// Consume the iterator by finishing its iteration and calling [`Reduce::finalize()`][crate::parallel::Reduce::finalize()].
225	pub fn finalize(mut self) -> Result<Reduce::Output, Reduce::Error> {
226	for value in self.by_ref() {
227	drop(value?);
228	}
229	self.reducer
230	.take()
231	.expect("this is the last call before consumption")
232	.finalize()
233	}
234	}
235
236	impl<Reduce: super::Reduce> Iterator for Stepwise<Reduce> {
237	type Item = Result<Reduce::FeedProduce, Reduce::Error>;
238
239	fn next(&mut self) -> Option<<Self as Iterator>::Item> {
240	self.receive_result
241	.recv()
242	.ok()
243	.and_then(\|input\| self.reducer.as_mut().map(\|r\| r.feed(input)))
244	}
245	}
246	}
247	#[cfg(not(feature = "parallel"))]
248	mod stepped {
249	/// An iterator adaptor to allow running computations using [`in_parallel()`][crate::parallel::in_parallel()] in a step-wise manner, see the [module docs][crate::parallel]
250	/// for details.
251	pub struct Stepwise<InputIter, ConsumeFn, ThreadState, Reduce> {
252	input: InputIter,
253	consume: ConsumeFn,
254	thread_state: ThreadState,
255	reducer: Reduce,
256	}
257
258	impl<InputIter, ConsumeFn, Reduce, I, O, S> Stepwise<InputIter, ConsumeFn, S, Reduce>
259	where
260	InputIter: Iterator<Item = I> + Send,
261	ConsumeFn: Fn(I, &mut S) -> O + Send + Sync,
262	Reduce: super::Reduce<Input = O>,
263	I: Send,
264	O: Send,
265	{
266	/// Instantiate a new iterator.
267	/// For a description of parameters, see [`in_parallel()`][crate::parallel::in_parallel()].
268	pub fn new<ThreadStateFn>(
269	input: InputIter,
270	_thread_limit: Option<usize>,
271	new_thread_state: ThreadStateFn,
272	consume: ConsumeFn,
273	reducer: Reduce,
274	) -> Self
275	where
276	ThreadStateFn: Fn(usize) -> S + Send + Sync,
277	{
278	Stepwise {
279	input,
280	consume,
281	thread_state: new_thread_state(0),
282	reducer,
283	}
284	}
285
286	/// Consume the iterator by finishing its iteration and calling [`Reduce::finalize()`][crate::parallel::Reduce::finalize()].
287	pub fn finalize(mut self) -> Result<Reduce::Output, Reduce::Error> {
288	for value in self.by_ref() {
289	drop(value?);
290	}
291	self.reducer.finalize()
292	}
293	}
294
295	impl<InputIter, ConsumeFn, ThreadState, Reduce, I, O> Iterator for Stepwise<InputIter, ConsumeFn, ThreadState, Reduce>
296	where
297	InputIter: Iterator<Item = I> + Send,
298	ConsumeFn: Fn(I, &mut ThreadState) -> O + Send + Sync,
299	Reduce: super::Reduce<Input = O>,
300	I: Send,
301	O: Send,
302	{
303	type Item = Result<Reduce::FeedProduce, Reduce::Error>;
304
305	fn next(&mut self) -> Option<<Self as Iterator>::Item> {
306	self.input
307	.next()
308	.map(\|input\| self.reducer.feed((self.consume)(input, &mut self.thread_state)))
309	}
310	}
311	}
312	use std::marker::PhantomData;
313	pub use stepped::Stepwise;
314
315	/// An trait for aggregating items commonly produced in threads into a single result, without itself
316	/// needing to be thread safe.
317	pub trait Reduce {
318	/// The type fed to the reducer in the [`feed()`][Reduce::feed()] method.
319	///
320	/// It's produced by a function that may run on multiple threads.
321	type Input;
322	/// The type produced in Ok(…) by [`feed()`][Reduce::feed()].
323	/// Most reducers by nature use `()` here as the value is in the aggregation.
324	/// However, some may use it to collect statistics only and return their Input
325	/// in some form as a result here for [`Stepwise`] to be useful.
326	type FeedProduce;
327	/// The type produced once by the [`finalize()`][Reduce::finalize()] method.
328	///
329	/// For traditional reducers, this is the value produced by the entire operation.
330	/// For those made for step-wise iteration this may be aggregated statistics.
331	type Output;
332	/// The error type to use for all methods of this trait.
333	type Error;
334	/// Called each time a new `item` was produced in order to aggregate it into the final result.
335	///
336	/// If an `Error` is returned, the entire operation will be stopped.
337	fn feed(&mut self, item: Self::Input) -> Result<Self::FeedProduce, Self::Error>;
338	/// Called once once all items where passed to `feed()`, producing the final `Output` of the operation or an `Error`.
339	fn finalize(self) -> Result<Self::Output, Self::Error>;
340	}
341
342	/// An identity reducer for those who want to use [`Stepwise`] or [`in_parallel()`][crate::parallel::in_parallel()]
343	/// without the use of non-threaded reduction of products created in threads.
344	pub struct IdentityWithResult<Input, Error> {
345	_input: PhantomData<Input>,
346	_error: PhantomData<Error>,
347	}
348
349	impl<Input, Error> Default for IdentityWithResult<Input, Error> {
350	fn default() -> Self {
351	IdentityWithResult {
352	_input: Default::default(),
353	_error: Default::default(),
354	}
355	}
356	}
357
358	impl<Input, Error> Reduce for IdentityWithResult<Input, Error> {
359	type Input = Result<Input, Self::Error>;
360	type FeedProduce = Input;
361	type Output = ();
362	type Error = Error;
363
364	fn feed(&mut self, item: Self::Input) -> Result<Self::FeedProduce, Self::Error> {
365	item
366	}
367
368	fn finalize(self) -> Result<Self::Output, Self::Error> {
369	Ok(())
370	}
371	}
372	}
373
374	pub use reduce::Reduce;
375
376	/// Run [`in_parallel()`] only if the given `condition()` returns true when eagerly evaluated.	131	/// Run [`in_parallel()`] only if the given `condition()` returns true when eagerly evaluated.
377	///	132	///
378	/// For parameters, see the documentation of [`in_parallel()`]	133	/// For parameters, see the documentation of [`in_parallel()`]
@@ -395,3 +150,7 @@ where · 16 unmodified lines
395	serial::in_parallel(input, thread_limit, new_thread_state, consume, reducer)	150	serial::in_parallel(input, thread_limit, new_thread_state, consume, reducer)
396	}	151	}
397	}	152	}
		153
		154	///
		155	pub mod reduce;
		156	pub use reduce::Reduce;

git-features/src/parallel/reduce.rs+240−0

@@ -0,0 +1,240 @@
	1	#[cfg(feature = "parallel")]
	2	mod stepped {
	3	use crate::parallel::num_threads;
	4
	5	/// An iterator adaptor to allow running computations using [`in_parallel()`][crate::parallel::in_parallel()] in a step-wise manner, see the [module docs][crate::parallel]
	6	/// for details.
	7	pub struct Stepwise<Reduce: super::Reduce> {
	8	/// This field is first to assure it's dropped first and cause threads that are dropped next to stop their loops
	9	/// as sending results fails when the receiver is dropped.
	10	receive_result: std::sync::mpsc::Receiver<Reduce::Input>,
	11	/// `join()` will be called on these guards to assure every thread tries to send through a closed channel. When
	12	/// that happens, they break out of their loops.
	13	_threads: Vec<std::thread::JoinHandle<()>>,
	14	/// The reducer is called only in the thread using the iterator, dropping it has no side effects.
	15	reducer: Option<Reduce>,
	16	}
	17
	18	impl<Reduce: super::Reduce> Drop for Stepwise<Reduce> {
	19	fn drop(&mut self) {
	20	let (_, sink) = std::sync::mpsc::channel();
	21	drop(std::mem::replace(&mut self.receive_result, sink));
	22
	23	let mut last_err = None;
	24	for handle in std::mem::take(&mut self._threads) {
	25	if let Err(err) = handle.join() {
	26	last_err = Some(err);
	27	};
	28	}
	29	if let Some(thread_err) = last_err {
	30	std::panic::resume_unwind(thread_err);
	31	}
	32	}
	33	}
	34
	35	impl<Reduce: super::Reduce> Stepwise<Reduce> {
	36	/// Instantiate a new iterator and start working in threads.
	37	/// For a description of parameters, see [`in_parallel()`][crate::parallel::in_parallel()].
	38	pub fn new<InputIter, ThreadStateFn, ConsumeFn, I, O, S>(
	39	input: InputIter,
	40	thread_limit: Option<usize>,
	41	new_thread_state: ThreadStateFn,
	42	consume: ConsumeFn,
	43	reducer: Reduce,
	44	) -> Self
	45	where
	46	InputIter: Iterator<Item = I> + Send + 'static,
	47	ThreadStateFn: Fn(usize) -> S + Send + Clone + 'static,
	48	ConsumeFn: Fn(I, &mut S) -> O + Send + Clone + 'static,
	49	Reduce: super::Reduce<Input = O> + 'static,
	50	I: Send + 'static,
	51	O: Send + 'static,
	52	{
	53	let num_threads = num_threads(thread_limit);
	54	let mut threads = Vec::with_capacity(num_threads + 1);
	55	let receive_result = {
	56	let (send_input, receive_input) = crossbeam_channel::bounded::<I>(num_threads);
	57	let (send_result, receive_result) = std::sync::mpsc::sync_channel::<O>(num_threads);
	58	for thread_id in 0..num_threads {
	59	let handle = std::thread::spawn({
	60	let send_result = send_result.clone();
	61	let receive_input = receive_input.clone();
	62	let new_thread_state = new_thread_state.clone();
	63	let consume = consume.clone();
	64	move \|\| {
	65	let mut state = new_thread_state(thread_id);
	66	for item in receive_input {
	67	if send_result.send(consume(item, &mut state)).is_err() {
	68	break;
	69	}
	70	}
	71	}
	72	});
	73	threads.push(handle);
	74	}
	75	threads.push(std::thread::spawn(move \|\| {
	76	for item in input {
	77	if send_input.send(item).is_err() {
	78	break;
	79	}
	80	}
	81	}));
	82	receive_result
	83	};
	84	Stepwise {
	85	_threads: threads,
	86	receive_result,
	87	reducer: Some(reducer),
	88	}
	89	}
	90
	91	/// Consume the iterator by finishing its iteration and calling [`Reduce::finalize()`][crate::parallel::Reduce::finalize()].
	92	pub fn finalize(mut self) -> Result<Reduce::Output, Reduce::Error> {
	93	for value in self.by_ref() {
	94	drop(value?);
	95	}
	96	self.reducer
	97	.take()
	98	.expect("this is the last call before consumption")
	99	.finalize()
	100	}
	101	}
	102
	103	impl<Reduce: super::Reduce> Iterator for Stepwise<Reduce> {
	104	type Item = Result<Reduce::FeedProduce, Reduce::Error>;
	105
	106	fn next(&mut self) -> Option<<Self as Iterator>::Item> {
	107	self.receive_result
	108	.recv()
	109	.ok()
	110	.and_then(\|input\| self.reducer.as_mut().map(\|r\| r.feed(input)))
	111	}
	112	}
	113	}
	114
	115	#[cfg(not(feature = "parallel"))]
	116	mod stepped {
	117	/// An iterator adaptor to allow running computations using [`in_parallel()`][crate::parallel::in_parallel()] in a step-wise manner, see the [module docs][crate::parallel]
	118	/// for details.
	119	pub struct Stepwise<InputIter, ConsumeFn, ThreadState, Reduce> {
	120	input: InputIter,
	121	consume: ConsumeFn,
	122	thread_state: ThreadState,
	123	reducer: Reduce,
	124	}
	125
	126	impl<InputIter, ConsumeFn, Reduce, I, O, S> Stepwise<InputIter, ConsumeFn, S, Reduce>
	127	where
	128	InputIter: Iterator<Item = I> + Send,
	129	ConsumeFn: Fn(I, &mut S) -> O + Send + Sync,
	130	Reduce: super::Reduce<Input = O>,
	131	I: Send,
	132	O: Send,
	133	{
	134	/// Instantiate a new iterator.
	135	/// For a description of parameters, see [`in_parallel()`][crate::parallel::in_parallel()].
	136	pub fn new<ThreadStateFn>(
	137	input: InputIter,
	138	_thread_limit: Option<usize>,
	139	new_thread_state: ThreadStateFn,
	140	consume: ConsumeFn,
	141	reducer: Reduce,
	142	) -> Self
	143	where
	144	ThreadStateFn: Fn(usize) -> S + Send + Sync,
	145	{
	146	Stepwise {
	147	input,
	148	consume,
	149	thread_state: new_thread_state(0),
	150	reducer,
	151	}
	152	}
	153
	154	/// Consume the iterator by finishing its iteration and calling [`Reduce::finalize()`][crate::parallel::Reduce::finalize()].
	155	pub fn finalize(mut self) -> Result<Reduce::Output, Reduce::Error> {
	156	for value in self.by_ref() {
	157	drop(value?);
	158	}
	159	self.reducer.finalize()
	160	}
	161	}
	162
	163	impl<InputIter, ConsumeFn, ThreadState, Reduce, I, O> Iterator for Stepwise<InputIter, ConsumeFn, ThreadState, Reduce>
	164	where
	165	InputIter: Iterator<Item = I> + Send,
	166	ConsumeFn: Fn(I, &mut ThreadState) -> O + Send + Sync,
	167	Reduce: super::Reduce<Input = O>,
	168	I: Send,
	169	O: Send,
	170	{
	171	type Item = Result<Reduce::FeedProduce, Reduce::Error>;
	172
	173	fn next(&mut self) -> Option<<Self as Iterator>::Item> {
	174	self.input
	175	.next()
	176	.map(\|input\| self.reducer.feed((self.consume)(input, &mut self.thread_state)))
	177	}
	178	}
	179	}
	180
	181	use std::marker::PhantomData;
	182	pub use stepped::Stepwise;
	183
	184	/// An trait for aggregating items commonly produced in threads into a single result, without itself
	185	/// needing to be thread safe.
	186	pub trait Reduce {
	187	/// The type fed to the reducer in the [`feed()`][Reduce::feed()] method.
	188	///
	189	/// It's produced by a function that may run on multiple threads.
	190	type Input;
	191	/// The type produced in Ok(…) by [`feed()`][Reduce::feed()].
	192	/// Most reducers by nature use `()` here as the value is in the aggregation.
	193	/// However, some may use it to collect statistics only and return their Input
	194	/// in some form as a result here for [`Stepwise`] to be useful.
	195	type FeedProduce;
	196	/// The type produced once by the [`finalize()`][Reduce::finalize()] method.
	197	///
	198	/// For traditional reducers, this is the value produced by the entire operation.
	199	/// For those made for step-wise iteration this may be aggregated statistics.
	200	type Output;
	201	/// The error type to use for all methods of this trait.
	202	type Error;
	203	/// Called each time a new `item` was produced in order to aggregate it into the final result.
	204	///
	205	/// If an `Error` is returned, the entire operation will be stopped.
	206	fn feed(&mut self, item: Self::Input) -> Result<Self::FeedProduce, Self::Error>;
	207	/// Called once once all items where passed to `feed()`, producing the final `Output` of the operation or an `Error`.
	208	fn finalize(self) -> Result<Self::Output, Self::Error>;
	209	}
	210
	211	/// An identity reducer for those who want to use [`Stepwise`] or [`in_parallel()`][crate::parallel::in_parallel()]
	212	/// without the use of non-threaded reduction of products created in threads.
	213	pub struct IdentityWithResult<Input, Error> {
	214	_input: PhantomData<Input>,
	215	_error: PhantomData<Error>,
	216	}
	217
	218	impl<Input, Error> Default for IdentityWithResult<Input, Error> {
	219	fn default() -> Self {
	220	IdentityWithResult {
	221	_input: Default::default(),
	222	_error: Default::default(),
	223	}
	224	}
	225	}
	226
	227	impl<Input, Error> Reduce for IdentityWithResult<Input, Error> {
	228	type Input = Result<Input, Self::Error>;
	229	type FeedProduce = Input;
	230	type Output = ();
	231	type Error = Error;
	232
	233	fn feed(&mut self, item: Self::Input) -> Result<Self::FeedProduce, Self::Error> {
	234	item
	235	}
	236
	237	fn finalize(self) -> Result<Self::Output, Self::Error> {
	238	Ok(())
	239	}
	240	}

git-features/tests/parallel/mod.rs+101−0

@@ -0,0 +1,101 @@
	1	//! Tests that are working similarly in parallel and serial mode
	2	use git_features::parallel;
	3
	4	#[derive(Default)]
	5	struct Adder {
	6	count: usize,
	7	}
	8
	9	impl parallel::Reduce for Adder {
	10	type Input = usize;
	11	type FeedProduce = usize;
	12	type Output = usize;
	13	type Error = ();
	14
	15	fn feed(&mut self, item: Self::Input) -> Result<Self::FeedProduce, Self::Error> {
	16	self.count += item;
	17	Ok(item)
	18	}
	19
	20	fn finalize(self) -> Result<Self::Output, Self::Error> {
	21	Ok(self.count)
	22	}
	23	}
	24
	25	#[test]
	26	fn in_parallel() {
	27	let res = parallel::in_parallel(
	28	std::iter::from_fn(\|\| Some(1)).take(100),
	29	None,
	30	\|_n\| (),
	31	\|input, _state\| input,
	32	Adder::default(),
	33	)
	34	.expect("successful computation");
	35	assert_eq!(res, 100);
	36	}
	37
	38	#[test]
	39	fn stepped_reduce_next() {
	40	let mut iter = parallel::reduce::Stepwise::new(
	41	std::iter::from_fn(\|\| Some(1)).take(100),
	42	None,
	43	\|_n\| (),
	44	\|input, _state\| input,
	45	Adder::default(),
	46	);
	47
	48	let mut aggregate = 0;
	49	for value in iter.by_ref() {
	50	aggregate += value.expect("success");
	51	}
	52	assert_eq!(aggregate, 100);
	53	}
	54
	55	#[test]
	56	fn stepped_reduce_ref_input_and_consume() {
	57	let seq = std::sync::Arc::new(vec![0usize, 1, 2]);
	58	struct ArcIter(std::sync::Arc<Vec<usize>>, usize);
	59	impl Iterator for ArcIter {
	60	type Item = usize;
	61
	62	fn next(&mut self) -> Option<Self::Item> {
	63	let n = self.0.get(self.1).copied();
	64	self.1 += 1;
	65	n
	66	}
	67	}
	68
	69	let mut iter = parallel::reduce::Stepwise::new(
	70	ArcIter(seq.clone(), 0).enumerate(),
	71	None,
	72	{
	73	let seq = std::sync::Arc::clone(&seq);
	74	move \|_n\| seq.len()
	75	},
	76	{
	77	let seq = std::sync::Arc::clone(&seq);
	78	move \|(idx, ref_val): (usize, usize), _state\| seq[idx] * ref_val
	79	},
	80	Adder::default(),
	81	);
	82
	83	let mut aggregate = 0;
	84	for value in iter.by_ref() {
	85	aggregate += value.expect("success");
	86	}
	87	assert_eq!(aggregate, 5);
	88	}
	89
	90	#[test]
	91	fn stepped_reduce_finalize() {
	92	let iter = parallel::reduce::Stepwise::new(
	93	std::iter::from_fn(\|\| Some(1)).take(100),
	94	None,
	95	\|_n\| (),
	96	\|input, _state\| input,
	97	Adder::default(),
	98	);
	99
	100	assert_eq!(iter.finalize().expect("success"), 100);
	101	}

git-features/tests/parallel_shared.rs+1−101

@@ -1,101 +1 @@
1	//! Tests that are working similarly in parallel and serial mode	1	mod parallel;
2	use git_features::parallel;
3
4	#[derive(Default)]
5	struct Adder {
6	count: usize,
7	}
8
9	impl parallel::Reduce for Adder {
10	type Input = usize;
11	type FeedProduce = usize;
12	type Output = usize;
13	type Error = ();
14
15	fn feed(&mut self, item: Self::Input) -> Result<Self::FeedProduce, Self::Error> {
16	self.count += item;
17	Ok(item)
18	}
19
20	fn finalize(self) -> Result<Self::Output, Self::Error> {
21	Ok(self.count)
22	}
23	}
24
25	#[test]
26	fn in_parallel() {
27	let res = parallel::in_parallel(
28	std::iter::from_fn(\|\| Some(1)).take(100),
29	None,
30	\|_n\| (),
31	\|input, _state\| input,
32	Adder::default(),
33	)
34	.expect("successful computation");
35	assert_eq!(res, 100);
36	}
37
38	#[test]
39	fn stepped_reduce_next() {
40	let mut iter = parallel::reduce::Stepwise::new(
41	std::iter::from_fn(\|\| Some(1)).take(100),
42	None,
43	\|_n\| (),
44	\|input, _state\| input,
45	Adder::default(),
46	);
47
48	let mut aggregate = 0;
49	for value in iter.by_ref() {
50	aggregate += value.expect("success");
51	}
52	assert_eq!(aggregate, 100);
53	}
54
55	#[test]
56	fn stepped_reduce_ref_input_and_consume() {
57	let seq = std::sync::Arc::new(vec![0usize, 1, 2]);
58	struct ArcIter(std::sync::Arc<Vec<usize>>, usize);
59	impl Iterator for ArcIter {
60	type Item = usize;
61
62	fn next(&mut self) -> Option<Self::Item> {
63	let n = self.0.get(self.1).copied();
64	self.1 += 1;
65	n
66	}
67	}
68
69	let mut iter = parallel::reduce::Stepwise::new(
70	ArcIter(seq.clone(), 0).enumerate(),
71	None,
72	{
73	let seq = std::sync::Arc::clone(&seq);
74	move \|_n\| seq.len()
75	},
76	{
77	let seq = std::sync::Arc::clone(&seq);
78	move \|(idx, ref_val): (usize, usize), _state\| seq[idx] * ref_val
79	},
80	Adder::default(),
81	);
82
83	let mut aggregate = 0;
84	for value in iter.by_ref() {
85	aggregate += value.expect("success");
86	}
87	assert_eq!(aggregate, 5);
88	}
89
90	#[test]
91	fn stepped_reduce_finalize() {
92	let iter = parallel::reduce::Stepwise::new(
93	std::iter::from_fn(\|\| Some(1)).take(100),
94	None,
95	\|_n\| (),
96	\|input, _state\| input,
97	Adder::default(),
98	);
99
100	assert_eq!(iter.finalize().expect("success"), 100);
101	}

git-features/tests/parallel_shared_threaded.rs+1−0

@@ -0,0 +1 @@
		1	mod parallel;

git-features/tests/parallel_threaded.rsrenamed from git-features/tests/parallel.rs+0−0

File renamed without changes.