Skip to content

Commit f3b12a6

Browse files
committed
Break out on EINTR for shutdown on Unix
This was a mistake in the previous "retry on EINTR" PR where we want to not retry in the case that we're being shut down with `pthread_kill`. Closes #23
1 parent b41247c commit f3b12a6

File tree

5 files changed

+80
-23
lines changed

5 files changed

+80
-23
lines changed

src/lib.rs

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -458,7 +458,13 @@ impl HelperState {
458458
self.lock.lock().unwrap_or_else(|e| e.into_inner())
459459
}
460460

461-
fn for_each_request(&self, mut f: impl FnMut()) {
461+
/// Executes `f` for each request for a token, where `f` is expected to
462+
/// block and then provide the original closure with a token once it's
463+
/// acquired.
464+
///
465+
/// This is an infinite loop until the helper thread is dropped, at which
466+
/// point everything should get interrupted.
467+
fn for_each_request(&self, mut f: impl FnMut(&HelperState)) {
462468
let mut lock = self.lock();
463469

464470
// We only execute while we could receive requests, but as soon as
@@ -477,12 +483,16 @@ impl HelperState {
477483
// wait for a long time for a token.
478484
lock.requests -= 1;
479485
drop(lock);
480-
f();
486+
f(self);
481487
lock = self.lock();
482488
}
483489
lock.consumer_done = true;
484490
self.cvar.notify_one();
485491
}
492+
493+
fn producer_done(&self) -> bool {
494+
self.lock().producer_done
495+
}
486496
}
487497

488498
#[test]

src/unix.rs

Lines changed: 31 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
use libc::c_int;
12
use std::fs::File;
23
use std::io::{self, Read, Write};
34
use std::mem;
@@ -7,7 +8,6 @@ use std::ptr;
78
use std::sync::{Arc, Once};
89
use std::thread::{self, Builder, JoinHandle};
910
use std::time::Duration;
10-
use libc::c_int;
1111

1212
#[derive(Debug)]
1313
pub struct Client {
@@ -102,6 +102,17 @@ impl Client {
102102
}
103103

104104
pub fn acquire(&self) -> io::Result<Acquired> {
105+
// Ignore interrupts and keep trying if that happens
106+
loop {
107+
if let Some(token) = self.acquire_allow_interrupts()? {
108+
return Ok(token);
109+
}
110+
}
111+
}
112+
113+
/// Block waiting for a token, returning `None` if we're interrupted with
114+
/// EINTR.
115+
fn acquire_allow_interrupts(&self) -> io::Result<Option<Acquired>> {
105116
// We don't actually know if the file descriptor here is set in
106117
// blocking or nonblocking mode. AFAIK all released versions of
107118
// `make` use blocking fds for the jobserver, but the unreleased
@@ -130,7 +141,7 @@ impl Client {
130141
if libc::poll(&mut fd, 1, -1) == -1 {
131142
let e = io::Error::last_os_error();
132143
match e.kind() {
133-
io::ErrorKind::Interrupted => continue,
144+
io::ErrorKind::Interrupted => return Ok(None),
134145
_ => return Err(e),
135146
}
136147
}
@@ -139,15 +150,15 @@ impl Client {
139150
}
140151
let mut buf = [0];
141152
match (&self.read).read(&mut buf) {
142-
Ok(1) => return Ok(Acquired { byte: buf[0] }),
153+
Ok(1) => return Ok(Some(Acquired { byte: buf[0] })),
143154
Ok(_) => {
144155
return Err(io::Error::new(
145156
io::ErrorKind::Other,
146157
"early EOF on jobserver pipe",
147158
))
148159
}
149160
Err(e) => match e.kind() {
150-
io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted => continue,
161+
io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted => return Ok(None),
151162
_ => return Err(e),
152163
},
153164
}
@@ -219,7 +230,20 @@ pub(crate) fn spawn_helper(
219230

220231
let state2 = state.clone();
221232
let thread = Builder::new().spawn(move || {
222-
state2.for_each_request(|| f(client.acquire()));
233+
state2.for_each_request(|helper| loop {
234+
match client.inner.acquire_allow_interrupts() {
235+
Ok(Some(data)) => {
236+
break f(Ok(crate::Acquired {
237+
client: client.inner.clone(),
238+
data,
239+
disabled: false,
240+
}))
241+
}
242+
Err(e) => break f(Err(e)),
243+
Ok(None) if helper.producer_done() => break,
244+
Ok(None) => {}
245+
}
246+
});
223247
})?;
224248

225249
Ok(Helper { thread, state })
@@ -239,7 +263,7 @@ impl Helper {
239263
// This signal should interrupt any blocking `read` call with
240264
// `io::ErrorKind::Interrupt` and cause the thread to cleanly exit.
241265
//
242-
// Note that we don'tdo this forever though since there's a chance
266+
// Note that we don't do this forever though since there's a chance
243267
// of bugs, so only do this opportunistically to make a best effort
244268
// at clearing ourselves up.
245269
for _ in 0..100 {
@@ -264,7 +288,7 @@ impl Helper {
264288
}
265289

266290
// If we managed to actually see the consumer get done, then we can
267-
// definitely wait for the thread. Otherwise it's... of in the ether
291+
// definitely wait for the thread. Otherwise it's... off in the ether
268292
// I guess?
269293
if state.consumer_done {
270294
drop(self.thread.join());

src/wasm.rs

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,3 @@
1-
21
use std::io;
32
use std::process::Command;
43
use std::sync::{Arc, Condvar, Mutex};
@@ -76,7 +75,7 @@ pub(crate) fn spawn_helper(
7675
mut f: Box<dyn FnMut(io::Result<crate::Acquired>) + Send>,
7776
) -> io::Result<Helper> {
7877
let thread = Builder::new().spawn(move || {
79-
state.for_each_request(|| f(client.acquire()));
78+
state.for_each_request(|_| f(client.acquire()));
8079
})?;
8180

8281
Ok(Helper { thread: thread })

src/windows.rs

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -53,11 +53,7 @@ extern "system" {
5353
lMaximumCount: LONG,
5454
lpName: *const i8,
5555
) -> HANDLE;
56-
fn OpenSemaphoreA(
57-
dwDesiredAccess: DWORD,
58-
bInheritHandle: BOOL,
59-
lpName: *const i8,
60-
) -> HANDLE;
56+
fn OpenSemaphoreA(dwDesiredAccess: DWORD, bInheritHandle: BOOL, lpName: *const i8) -> HANDLE;
6157
fn WaitForSingleObject(hHandle: HANDLE, dwMilliseconds: DWORD) -> DWORD;
6258
#[link_name = "SystemFunction036"]
6359
fn RtlGenRandom(RandomBuffer: *mut u8, RandomBufferLength: u32) -> u8;
@@ -95,8 +91,7 @@ impl Client {
9591
for _ in 0..100 {
9692
let mut bytes = [0; 4];
9793
getrandom(&mut bytes)?;
98-
let mut name =
99-
format!("__rust_jobserver_semaphore_{}\0", u32::from_ne_bytes(bytes));
94+
let mut name = format!("__rust_jobserver_semaphore_{}\0", u32::from_ne_bytes(bytes));
10095
unsafe {
10196
let create_limit = if limit == 0 { 1 } else { limit };
10297
let r = CreateSemaphoreA(
@@ -218,7 +213,7 @@ pub(crate) fn spawn_helper(
218213
let event2 = event.clone();
219214
let thread = Builder::new().spawn(move || {
220215
let objects = [event2.0, client.inner.sem.0];
221-
state.for_each_request(|| {
216+
state.for_each_request(|_| {
222217
const WAIT_OBJECT_1: u32 = WAIT_OBJECT_0 + 1;
223218
match unsafe { WaitForMultipleObjects(2, objects.as_ptr(), FALSE, INFINITE) } {
224219
WAIT_OBJECT_0 => return,

tests/helper.rs

Lines changed: 33 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,7 @@
1-
extern crate jobserver;
2-
3-
use std::sync::mpsc;
4-
51
use jobserver::Client;
2+
use std::sync::atomic::*;
3+
use std::sync::mpsc;
4+
use std::sync::*;
65

76
macro_rules! t {
87
($e:expr) => {
@@ -46,3 +45,33 @@ fn acquire() {
4645
helper.request_token();
4746
drop(helper);
4847
}
48+
49+
#[test]
50+
fn prompt_shutdown() {
51+
for _ in 0..100 {
52+
let client = jobserver::Client::new(4).unwrap();
53+
let count = Arc::new(AtomicU32::new(0));
54+
let count2 = count.clone();
55+
let tokens = Arc::new(Mutex::new(Vec::new()));
56+
let helper = client
57+
.into_helper_thread(move |token| {
58+
tokens.lock().unwrap().push(token);
59+
count2.fetch_add(1, Ordering::SeqCst);
60+
})
61+
.unwrap();
62+
63+
// Request more tokens than what are available.
64+
for _ in 0..5 {
65+
helper.request_token();
66+
}
67+
// Wait for at least some of the requests to finish.
68+
while count.load(Ordering::SeqCst) < 3 {
69+
std::thread::yield_now();
70+
}
71+
// Drop helper
72+
let t = std::time::Instant::now();
73+
drop(helper);
74+
let d = t.elapsed();
75+
assert!(d.as_secs_f64() < 0.5);
76+
}
77+
}

0 commit comments

Comments
 (0)