remove the done flag, its function is now done by the `fuse`
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Mutex;
use crate::current_num_threads;
@@ -77,13 +77,11 @@ where
{
let split_count = AtomicUsize::new(current_num_threads());
- let done = AtomicBool::new(false);
let iter = Mutex::new(self.iter.fuse());
bridge_unindexed(
IterParallelProducer {
split_count: &split_count,
- done: &done,
iter: &iter,
},
consumer,
@@ -93,7 +91,6 @@ where
struct IterParallelProducer<'a, Iter: Iterator> {
split_count: &'a AtomicUsize,
- done: &'a AtomicBool,
iter: &'a Mutex<std::iter::Fuse<Iter>>,
}
@@ -102,7 +99,6 @@ impl<'a, Iter: Iterator + 'a> Clone for IterParallelProducer<'a, Iter> {
fn clone(&self) -> Self {
IterParallelProducer {
split_count: self.split_count,
- done: self.done,
iter: self.iter,
}
}
@@ -119,9 +115,7 @@ where
loop {
// Check if the iterator is exhausted
- let done = self.done.load(Ordering::SeqCst);
- match count.checked_sub(1) {
- Some(new_count) if !done => {
+ if let Some(new_count) = count.checked_sub(1) {
match self.split_count.compare_exchange_weak(
count,
new_count,
@@ -131,13 +125,11 @@ where
Ok(_) => return (self.clone(), Some(self)),
Err(last_count) => count = last_count,
}
- }
- _ => {
+ } else {
return (self, None);
}
}
}
- }
fn fold_with<F>(self, mut folder: F) -> F
where
@@ -152,13 +144,11 @@ where
return folder;
}
} else {
- self.done.store(true, Ordering::SeqCst);
return folder;
}
} else {
// any panics from other threads will have been caught by the pool,
// and will be re-thrown when joined - just exit
- self.done.store(true, Ordering::SeqCst);
return folder;
}
}
