From c143a1c9bb3129f15764e5ab2ddc80e027fe5cf4 Mon Sep 17 00:00:00 2001
From: Stefan Kreutz <mail@skreutz.com>
Date: Wed, 21 Dec 2022 10:25:18 +0100
Subject: Test parallel map drop

---
 src/lib.rs | 32 +++++++++++++++++++++++++++++++-
 1 file changed, 31 insertions(+), 1 deletion(-)

(limited to 'src/lib.rs')

diff --git a/src/lib.rs b/src/lib.rs
index 60d3546..7cff65e 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -321,7 +321,10 @@ impl<F: FnMut()> Drop for Canary<F> {
 
 #[cfg(test)]
 mod tests {
-    use std::time::Duration;
+    use std::{
+        sync::{Arc, Mutex},
+        time::Duration,
+    };
 
     use super::*;
 
@@ -476,4 +479,31 @@ mod tests {
         assert_eq!(inflight(17, 13), 4);
         assert_eq!(inflight(13, usize::MAX - 17), 31);
     }
+
+    /// ParallelMap must stop feeding workers when dropped.
+    #[test]
+    fn drop_parallel_map() {
+        let threads = 5;
+        let buffer_size = 20;
+        let consume = 7;
+
+        let counter = Arc::new(Mutex::new(0));
+        let count = counter.clone();
+
+        let mut iter = (0..)
+            .into_iter()
+            .map_parallel_limit(threads, buffer_size, move |i| {
+                if i < consume {
+                    std::thread::sleep(Duration::from_millis(100));
+                }
+                let mut counter = counter.lock().unwrap();
+                *counter += 1;
+                2 * i
+            });
+        for _ in 0..consume {
+            iter.next();
+        }
+        drop(iter);
+        assert!(*count.lock().unwrap() < consume + buffer_size);
+    }
 }
-- 
cgit v1.2.3