From 59bf52169c22bd23ae8033488ecbba517242845d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?John=20K=C3=A5re=20Alsaker?= <john.kare.alsaker@gmail.com>
Date: Fri, 17 Jan 2020 12:45:34 +0100
Subject: [PATCH 1/2] Loop until the helper thread is done

---
 src/lib.rs | 21 ++++++++++++---------
 1 file changed, 12 insertions(+), 9 deletions(-)

diff --git a/src/lib.rs b/src/lib.rs
index 96bdd2d..75ba53c 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -80,6 +80,7 @@
 
 use std::env;
 use std::io;
+use std::panic;
 use std::process::Command;
 use std::sync::{Arc, Condvar, Mutex, MutexGuard};
 
@@ -447,7 +448,16 @@ impl HelperState {
             // wait for a long time for a token.
             lock.requests -= 1;
             drop(lock);
-            f();
+
+            // Run `f`, but mark ourself as done if it panics.
+            if let Err(panic) = panic::catch_unwind(panic::AssertUnwindSafe(|| f())) {
+                let mut lock = self.lock();
+                lock.consumer_done = true;
+                self.cvar.notify_one();
+                drop(lock);
+                panic::resume_unwind(panic);
+            }
+
             lock = self.lock();
         }
         lock.consumer_done = true;
@@ -701,14 +711,7 @@ mod imp {
             // of interrupting that, so resort to `pthread_kill` as a fallback.
             // This signal should interrupt any blocking `read` call with
             // `io::ErrorKind::Interrupt` and cause the thread to cleanly exit.
-            //
-            // Note that we don'tdo this forever though since there's a chance
-            // of bugs, so only do this opportunistically to make a best effort
-            // at clearing ourselves up.
-            for _ in 0..100 {
-                if state.consumer_done {
-                    break;
-                }
+            while !state.consumer_done {
                 unsafe {
                     // Ignore the return value here of `pthread_kill`,
                     // apparently on OSX if you kill a dead thread it will

From 57ceb626a7f624a3fb41225936cb114ebd6280e5 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?John=20K=C3=A5re=20Alsaker?= <john.kare.alsaker@gmail.com>
Date: Fri, 17 Jan 2020 13:18:00 +0100
Subject: [PATCH 2/2] Fix helper thread termination

---
 src/lib.rs | 48 ++++++++++++++++++++++++++++++++++++++++--------
 1 file changed, 40 insertions(+), 8 deletions(-)

diff --git a/src/lib.rs b/src/lib.rs
index 75ba53c..0d07696 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -575,6 +575,14 @@ mod imp {
         }
 
         pub fn acquire(&self) -> io::Result<Acquired> {
+            self.try_acquire(&mut || false).unwrap()
+        }
+
+        /// Tries to acquire a token. Returns `None` if the operation was terminated.
+        fn try_acquire(
+            &self,
+            terminated: &mut dyn FnMut() -> bool,
+        ) -> Option<io::Result<Acquired>> {
             // We don't actually know if the file descriptor here is set in
             // blocking or nonblocking mode. AFAIK all released versions of
             // `make` use blocking fds for the jobserver, but the unreleased
@@ -603,8 +611,13 @@ mod imp {
                     if libc::poll(&mut fd, 1, -1) == -1 {
                         let e = io::Error::last_os_error();
                         match e.kind() {
-                            io::ErrorKind::Interrupted => continue,
-                            _ => return Err(e),
+                            io::ErrorKind::Interrupted => {
+                                if terminated() {
+                                    return None;
+                                };
+                                continue;
+                            }
+                            _ => return Some(Err(e)),
                         }
                     }
                     if fd.revents == 0 {
@@ -612,16 +625,22 @@ mod imp {
                     }
                     let mut buf = [0];
                     match (&self.read).read(&mut buf) {
-                        Ok(1) => return Ok(Acquired { byte: buf[0] }),
+                        Ok(1) => return Some(Ok(Acquired { byte: buf[0] })),
                         Ok(_) => {
-                            return Err(io::Error::new(
+                            return Some(Err(io::Error::new(
                                 io::ErrorKind::Other,
                                 "early EOF on jobserver pipe",
-                            ))
+                            )))
                         }
                         Err(e) => match e.kind() {
-                            io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted => continue,
-                            _ => return Err(e),
+                            io::ErrorKind::WouldBlock => continue,
+                            io::ErrorKind::Interrupted => {
+                                if terminated() {
+                                    return None;
+                                };
+                                continue;
+                            }
+                            _ => return Some(Err(e)),
                         },
                     }
                 }
@@ -692,7 +711,20 @@ mod imp {
 
         let state2 = state.clone();
         let thread = Builder::new().spawn(move || {
-            state2.for_each_request(|| f(client.acquire()));
+            state2.for_each_request(|| {
+                let token = if let Some(token) = client
+                    .inner
+                    .try_acquire(&mut || state2.lock().producer_done)
+                {
+                    token.map(|data| super::Acquired {
+                        client: client.inner.clone(),
+                        data: data,
+                    })
+                } else {
+                    return;
+                };
+                f(token)
+            });
         })?;
 
         Ok(Helper { thread, state })