Merge #996

996: par_bridge: use naive locking of the Iterator r=cuviper a=njaard

Fix issue #795

instead of using crossbeam_deque, just lock the Iterator and get the next item, naively. The original code would spin until more data became available, which would cause par_bridge to not be useful in cases where the source Iterator did slow IO or was computationally costly in itself.

This causes there to be virtually no change in runtime in many cases, but in cases where there are a lot CPUs and even slightly slower IO, significantly less runtime.

In pretty much every case, total CPU time was much lower.

I ran the below test program on a 1.5GiB file from a fast SSD and on a machine with 64 cores and mean runtime went from 44s to 30s. In fact, according to hyperfine, average runtime with this patch was lower than _minimum_ runtime without (25s and 41s, respectively). But more importantly, total CPU time was almost 10x less with this patch.

I'm not sure what realistic usage patterns could be worse with this patch and therefor I'd love to hear other opinions.

I wrote a basic test program:

```
use std::io::{BufReader,BufRead};
use rayon::prelude::*;
use std::sync::atomic::{Ordering,AtomicUsize};
use std::os::unix::io::FromRawFd;

fn main() {
  let counter = AtomicUsize::new(0);
  let stdin = BufReader::new(unsafe { std::fs::File::from_raw_fd(0) });

  stdin.lines().par_bridge()
    .for_each( |row| {
      let row = row.unwrap();
      for _ in 0 .. 1000
      {
        let c = row.len();
        counter.fetch_add(c, Ordering::Relaxed);
      }
    }
  );

  println!("{}", counter.load(Ordering::Relaxed));
}
```

Which directly wastes some CPU time (`for _ in 0 .. 1000`). The unsafe is necessary to make the input stream Send. I feed this program a multigigabyte file as part of a test. Once over a relatively slow network and once off an SSD.

Co-authored-by: Charles Samuels <[email protected]>

Author: bors[bot]

src/iter/par_bridge.rs

@@ -1,8 +1,5 @@
-use crossbeam_deque::{Steal, Stealer, Worker};
-
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
-use std::sync::{Mutex, TryLockError};
-use std::thread::yield_now;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Mutex;
 
 use crate::current_num_threads;
 use crate::iter::plumbing::{bridge_unindexed, Folder, UnindexedConsumer, UnindexedProducer};
@@ -79,17 +76,13 @@ where
         C: UnindexedConsumer<Self::Item>,
     {
         let split_count = AtomicUsize::new(current_num_threads());
-        let worker = Worker::new_fifo();
-        let stealer = worker.stealer();
-        let done = AtomicBool::new(false);
-        let iter = Mutex::new((self.iter, worker));
+
+        let iter = Mutex::new(self.iter.fuse());
 
         bridge_unindexed(
             IterParallelProducer {
                 split_count: &split_count,
-                done: &done,
                 iter: &iter,
-                items: stealer,
             },
             consumer,
         )
@@ -98,19 +91,15 @@ where
 
 struct IterParallelProducer<'a, Iter: Iterator> {
     split_count: &'a AtomicUsize,
-    done: &'a AtomicBool,
-    iter: &'a Mutex<(Iter, Worker<Iter::Item>)>,
-    items: Stealer<Iter::Item>,
+    iter: &'a Mutex<std::iter::Fuse<Iter>>,
 }
 
 // manual clone because T doesn't need to be Clone, but the derive assumes it should be
 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,
-            items: self.items.clone(),
         }
     }
 }
@@ -125,24 +114,19 @@ where
         let mut count = self.split_count.load(Ordering::SeqCst);
 
         loop {
-            // Check if the iterator is exhausted *and* we've consumed every item from it.
-            let done = self.done.load(Ordering::SeqCst) && self.items.is_empty();
-
-            match count.checked_sub(1) {
-                Some(new_count) if !done => {
-                    match self.split_count.compare_exchange_weak(
-                        count,
-                        new_count,
-                        Ordering::SeqCst,
-                        Ordering::SeqCst,
-                    ) {
-                        Ok(_) => return (self.clone(), Some(self)),
-                        Err(last_count) => count = last_count,
-                    }
-                }
-                _ => {
-                    return (self, None);
+            // Check if the iterator is exhausted
+            if let Some(new_count) = count.checked_sub(1) {
+                match self.split_count.compare_exchange_weak(
+                    count,
+                    new_count,
+                    Ordering::SeqCst,
+                    Ordering::SeqCst,
+                ) {
+                    Ok(_) => return (self.clone(), Some(self)),
+                    Err(last_count) => count = last_count,
                 }
+            } else {
+                return (self, None);
             }
         }
     }
@@ -152,65 +136,20 @@ where
         F: Folder<Self::Item>,
     {
         loop {
-            match self.items.steal() {
-                Steal::Success(it) => {
+            if let Ok(mut iter) = self.iter.lock() {
+                if let Some(it) = iter.next() {
+                    drop(iter);
                     folder = folder.consume(it);
                     if folder.full() {
                         return folder;
                     }
+                } else {
+                    return folder;
                 }
-                Steal::Empty => {
-                    // Don't storm the mutex if we're already done.
-                    if self.done.load(Ordering::SeqCst) {
-                        // Someone might have pushed more between our `steal()` and `done.load()`
-                        if self.items.is_empty() {
-                            // The iterator is out of items, no use in continuing
-                            return folder;
-                        }
-                    } else {
-                        // our cache is out of items, time to load more from the iterator
-                        match self.iter.try_lock() {
-                            Ok(mut guard) => {
-                                // Check `done` again in case we raced with the previous lock
-                                // holder on its way out.
-                                if self.done.load(Ordering::SeqCst) {
-                                    if self.items.is_empty() {
-                                        return folder;
-                                    }
-                                    continue;
-                                }
-
-                                let count = current_num_threads();
-                                let count = (count * count) * 2;
-
-                                let (ref mut iter, ref worker) = *guard;
-
-                                // while worker.len() < count {
-                                // FIXME the new deque doesn't let us count items.  We can just
-                                // push a number of items, but that doesn't consider active
-                                // stealers elsewhere.
-                                for _ in 0..count {
-                                    if let Some(it) = iter.next() {
-                                        worker.push(it);
-                                    } else {
-                                        self.done.store(true, Ordering::SeqCst);
-                                        break;
-                                    }
-                                }
-                            }
-                            Err(TryLockError::WouldBlock) => {
-                                // someone else has the mutex, just sit tight until it's ready
-                                yield_now(); //TODO: use a thread-pool-aware yield? (#548)
-                            }
-                            Err(TryLockError::Poisoned(_)) => {
-                                // any panics from other threads will have been caught by the pool,
-                                // and will be re-thrown when joined - just exit
-                                return folder;
-                            }
-                        }
-                    }
-                }
-                Steal::Retry => (),
+            } else {
+                // any panics from other threads will have been caught by the pool,
+                // and will be re-thrown when joined - just exit
+                return folder;
             }
         }
     }