Skip to content

Commit a7e61a3

Browse files
RalfJungRalfJung
authored
Merge pull request #4692 from RalfJung/epoll-wakeups
epoll wakeup fixes
2 parents 339caa8 + 963eda6 commit a7e61a3

18 files changed

+155
-140
lines changed

‎src/concurrency/thread.rs‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -515,10 +515,13 @@ impl<'tcx> ThreadManager<'tcx> {
515515
&mut self.threads[self.active_thread].stack
516516
}
517517

518-
pub fn all_stacks(
518+
pub fn all_blocked_stacks(
519519
&self,
520520
) -> impl Iterator<Item = (ThreadId, &[Frame<'tcx, Provenance, FrameExtra<'tcx>>])> {
521-
self.threads.iter_enumerated().map(|(id, t)| (id, &t.stack[..]))
521+
self.threads
522+
.iter_enumerated()
523+
.filter(|(_id, t)| matches!(t.state, ThreadState::Blocked { .. }))
524+
.map(|(id, t)| (id, &t.stack[..]))
522525
}
523526

524527
/// Create a new thread and returns its id.

‎src/diagnostics.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -469,7 +469,7 @@ pub fn report_result<'tcx>(
469469
eprint!("{extra}"); // newlines are already in the string
470470

471471
if show_all_threads {
472-
for (thread, stack) in ecx.machine.threads.all_stacks() {
472+
for (thread, stack) in ecx.machine.threads.all_blocked_stacks() {
473473
if thread != ecx.active_thread() {
474474
let stacktrace = Frame::generate_stacktrace_from_stack(stack);
475475
let (stacktrace, was_pruned) = prune_stacktrace(stacktrace, &ecx.machine);

‎src/shims/unix/linux_like/epoll.rs‎

Lines changed: 31 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
use std::cell::RefCell;
2-
use std::collections::BTreeMap;
2+
use std::collections::{BTreeMap, VecDeque};
33
use std::io;
44
use std::time::Duration;
55

@@ -22,8 +22,8 @@ struct Epoll {
2222
/// "ready" list; instead, a boolean flag in this list tracks which subset is ready. This makes
2323
/// `epoll_wait` less efficient, but also requires less bookkeeping.
2424
interest_list: RefCell<BTreeMap<EpollEventKey, EpollEventInterest>>,
25-
/// A list of thread ids blocked on this epoll instance.
26-
blocked: RefCell<Vec<ThreadId>>,
25+
/// The queue of threads blocked on this epoll instance, and how many events they'd like to get.
26+
queue: RefCell<VecDeque<(ThreadId, u32)>>,
2727
}
2828

2929
impl VisitProvenance for Epoll {
@@ -459,7 +459,9 @@ pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
459459
}
460460
};
461461
// Record this thread as blocked.
462-
epfd.blocked.borrow_mut().push(this.active_thread());
462+
epfd.queue
463+
.borrow_mut()
464+
.push_back((this.active_thread(), maxevents.try_into().unwrap()));
463465
// And block it.
464466
let dest = dest.clone();
465467
// We keep a strong ref to the underlying `Epoll` to make sure it sticks around.
@@ -477,14 +479,15 @@ pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
477479
|this, unblock: UnblockKind| {
478480
match unblock {
479481
UnblockKind::Ready => {
480-
return_ready_list(&epfd, &dest, &event, this)?;
482+
let events = return_ready_list(&epfd, &dest, &event, this)?;
483+
assert!(events > 0, "we got woken up with no events to deliver");
481484
interp_ok(())
482485
},
483486
UnblockKind::TimedOut => {
484487
// Remove the current active thread_id from the blocked thread_id list.
485488
epfd
486-
.blocked.borrow_mut()
487-
.retain(|&id| id != this.active_thread());
489+
.queue.borrow_mut()
490+
.retain(|&(id, _events)| id != this.active_thread());
488491
this.write_int(0, &dest)?;
489492
interp_ok(())
490493
},
@@ -550,7 +553,7 @@ fn update_readiness<'tcx>(
550553
&mut dyn FnMut(&mut EpollEventInterest) -> InterpResult<'tcx>,
551554
) -> InterpResult<'tcx>,
552555
) -> InterpResult<'tcx> {
553-
let mut wakeup = false;
556+
let mut num_ready = 0u32; // how many events we have ready to deliver
554557
for_each_interest(&mut |interest| {
555558
// Update the ready events tracked in this interest.
556559
let new_readiness = interest.relevant_events & active_events;
@@ -565,19 +568,29 @@ fn update_readiness<'tcx>(
565568
ecx.release_clock(|clock| {
566569
interest.clock.join(clock);
567570
})?;
568-
wakeup = true;
571+
num_ready = num_ready.saturating_add(1);
569572
}
570573
interp_ok(())
571574
})?;
572-
if wakeup {
573-
// Wake up threads that may have been waiting for events on this epoll.
574-
// Do this only once for all the interests.
575-
// Edge-triggered notification only notify one thread even if there are
576-
// multiple threads blocked on the same epoll.
577-
if let Some(thread_id) = epoll.blocked.borrow_mut().pop() {
578-
ecx.unblock_thread(thread_id, BlockReason::Epoll)?;
575+
// Edge-triggered notifications only wake up as many threads as are needed to deliver
576+
// all the events.
577+
while num_ready > 0
578+
&& let Some((thread_id, events)) = epoll.queue.borrow_mut().pop_front()
579+
{
580+
ecx.unblock_thread(thread_id, BlockReason::Epoll)?;
581+
// Keep track of how many events we have left to deliver (except if we saturated;
582+
// in that case we just wake up everybody).
583+
if num_ready != u32::MAX {
584+
num_ready = num_ready.saturating_sub(events);
579585
}
580586
}
587+
// Sanity-check: if there are threads left to wake up, then there are no more ready events.
588+
if !epoll.queue.borrow().is_empty() {
589+
assert!(
590+
epoll.interest_list.borrow().values().all(|i| !i.ready),
591+
"there are unconsumed ready events and threads ready to take them"
592+
);
593+
}
581594

582595
interp_ok(())
583596
}
@@ -589,7 +602,7 @@ fn return_ready_list<'tcx>(
589602
dest: &MPlaceTy<'tcx>,
590603
events: &MPlaceTy<'tcx>,
591604
ecx: &mut MiriInterpCx<'tcx>,
592-
) -> InterpResult<'tcx> {
605+
) -> InterpResult<'tcx, i32> {
593606
let mut interest_list = epfd.interest_list.borrow_mut();
594607
let mut num_of_events: i32 = 0;
595608
let mut array_iter = ecx.project_array_fields(events)?;
@@ -629,5 +642,5 @@ fn return_ready_list<'tcx>(
629642
}
630643
}
631644
ecx.write_int(num_of_events, dest)?;
632-
interp_ok(())
645+
interp_ok(num_of_events)
633646
}

‎tests/fail-dep/libc/eventfd_block_read_twice.rs‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,4 @@
11
//@only-target: linux android illumos
2-
//~^ERROR: deadlocked
3-
//~^^ERROR: deadlocked
42
//@compile-flags: -Zmiri-deterministic-concurrency
53
//@error-in-other-file: deadlock
64

‎tests/fail-dep/libc/eventfd_block_read_twice.stderr‎

Lines changed: 1 addition & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -14,24 +14,13 @@ note: inside `main`
1414
LL | thread2.join().unwrap();
1515
| ^^^^^^^^^^^^^^
1616

17-
error: the evaluated program deadlocked
18-
|
19-
= note: this thread got stuck here
20-
= note: (no span available)
21-
2217
error: the evaluated program deadlocked
2318
--> tests/fail-dep/libc/eventfd_block_read_twice.rs:LL:CC
2419
|
2520
LL | let res: i64 = unsafe { libc::read(fd, buf.as_mut_ptr().cast(), 8).try_into().unwrap() };
2621
| ^ this thread got stuck here
2722

28-
error: the evaluated program deadlocked
29-
|
30-
= note: this thread got stuck here
31-
= note: (no span available)
32-
= note: duplicate diagnostic emitted due to `-Z deduplicate-diagnostics=no`
33-
3423
note: some details are omitted, run with `MIRIFLAGS=-Zmiri-backtrace=full` for a verbose backtrace
3524

36-
error: aborting due to 4 previous errors
25+
error: aborting due to 2 previous errors
3726

‎tests/fail-dep/libc/eventfd_block_write_twice.rs‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,4 @@
11
//@only-target: linux android illumos
2-
//~^ERROR: deadlocked
3-
//~^^ERROR: deadlocked
42
//@compile-flags: -Zmiri-deterministic-concurrency
53
//@error-in-other-file: deadlock
64

@@ -38,7 +36,7 @@ fn main() {
3836

3937
let thread2 = thread::spawn(move || {
4038
let sized_8_data = (u64::MAX - 1).to_ne_bytes();
41-
// Write u64::MAX - 1, so the all subsequent write will block.
39+
// Write u64::MAX - 1, so that all subsequent writes will block.
4240
let res: i64 = unsafe {
4341
// This `write` will initially blocked, then get unblocked by thread3, then get blocked again
4442
// because the `write` in thread1 executes first.

‎tests/fail-dep/libc/eventfd_block_write_twice.stderr‎

Lines changed: 1 addition & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -14,24 +14,13 @@ note: inside `main`
1414
LL | thread2.join().unwrap();
1515
| ^^^^^^^^^^^^^^
1616

17-
error: the evaluated program deadlocked
18-
|
19-
= note: this thread got stuck here
20-
= note: (no span available)
21-
2217
error: the evaluated program deadlocked
2318
--> tests/fail-dep/libc/eventfd_block_write_twice.rs:LL:CC
2419
|
2520
LL | libc::write(fd, sized_8_data.as_ptr() as *const libc::c_void, 8).try_into().unwrap()
2621
| ^ this thread got stuck here
2722

28-
error: the evaluated program deadlocked
29-
|
30-
= note: this thread got stuck here
31-
= note: (no span available)
32-
= note: duplicate diagnostic emitted due to `-Z deduplicate-diagnostics=no`
33-
3423
note: some details are omitted, run with `MIRIFLAGS=-Zmiri-backtrace=full` for a verbose backtrace
3524

36-
error: aborting due to 4 previous errors
25+
error: aborting due to 2 previous errors
3726

‎tests/fail-dep/libc/libc_epoll_block_two_thread.rs‎

Lines changed: 27 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,12 @@
11
//@compile-flags: -Zmiri-deterministic-concurrency
2-
//~^ERROR: deadlocked
3-
//~^^ERROR: deadlocked
42
//@only-target: linux android illumos
53
//@error-in-other-file: deadlock
64

75
use std::convert::TryInto;
8-
use std::thread::spawn;
6+
use std::thread;
7+
8+
#[path = "../../utils/libc.rs"]
9+
mod libc_utils;
910

1011
// Using `as` cast since `EPOLLET` wraps around
1112
const EPOLL_IN_OUT_ET: u32 = (libc::EPOLLIN | libc::EPOLLOUT | libc::EPOLLET) as _;
@@ -49,39 +50,37 @@ fn main() {
4950
let epfd = unsafe { libc::epoll_create1(0) };
5051
assert_ne!(epfd, -1);
5152

52-
// Create a socketpair instance.
53-
let mut fds = [-1, -1];
54-
let res = unsafe { libc::socketpair(libc::AF_UNIX, libc::SOCK_STREAM, 0, fds.as_mut_ptr()) };
55-
assert_eq!(res, 0);
53+
// Create an eventfd instance.
54+
let flags = libc::EFD_NONBLOCK | libc::EFD_CLOEXEC;
55+
let fd1 = unsafe { libc::eventfd(0, flags) };
56+
// Make a duplicate so that we have two file descriptors for the same file description.
57+
let fd2 = unsafe { libc::dup(fd1) };
5658

57-
// Register one side of the socketpair with epoll.
58-
let mut ev = libc::epoll_event { events: EPOLL_IN_OUT_ET, u64: fds[0] as u64 };
59-
let res = unsafe { libc::epoll_ctl(epfd, libc::EPOLL_CTL_ADD, fds[0], &mut ev) };
59+
// Register both with epoll.
60+
let mut ev = libc::epoll_event { events: EPOLL_IN_OUT_ET, u64: fd1 as u64 };
61+
let res = unsafe { libc::epoll_ctl(epfd, libc::EPOLL_CTL_ADD, fd1, &mut ev) };
62+
assert_eq!(res, 0);
63+
let mut ev = libc::epoll_event { events: EPOLL_IN_OUT_ET, u64: fd2 as u64 };
64+
let res = unsafe { libc::epoll_ctl(epfd, libc::EPOLL_CTL_ADD, fd2, &mut ev) };
6065
assert_eq!(res, 0);
6166

62-
// epoll_wait to clear notification.
63-
let expected_event = u32::try_from(libc::EPOLLOUT).unwrap();
64-
let expected_value = fds[0] as u64;
65-
check_epoll_wait::<1>(epfd, &[(expected_event, expected_value)], 0);
67+
// Consume the initial events.
68+
let expected = [(libc::EPOLLOUT as u32, fd1 as u64), (libc::EPOLLOUT as u32, fd2 as u64)];
69+
check_epoll_wait::<8>(epfd, &expected, -1);
6670

67-
let expected_event = u32::try_from(libc::EPOLLIN | libc::EPOLLOUT).unwrap();
68-
let expected_value = fds[0] as u64;
69-
let thread1 = spawn(move || {
70-
check_epoll_wait::<1>(epfd, &[(expected_event, expected_value)], -1);
71-
//~^ERROR: deadlocked
71+
let thread1 = thread::spawn(move || {
72+
check_epoll_wait::<2>(epfd, &expected, -1);
7273
});
73-
let thread2 = spawn(move || {
74-
check_epoll_wait::<1>(epfd, &[(expected_event, expected_value)], -1);
74+
let thread2 = thread::spawn(move || {
75+
check_epoll_wait::<2>(epfd, &expected, -1);
76+
//~^ERROR: deadlocked
7577
});
78+
// Yield so the threads are both blocked.
79+
thread::yield_now();
7680

77-
let thread3 = spawn(move || {
78-
// Just a single write, so we only wake up one of them.
79-
let data = "abcde".as_bytes().as_ptr();
80-
let res = unsafe { libc::write(fds[1], data as *const libc::c_void, 5) };
81-
assert!(res > 0 && res <= 5);
82-
});
81+
// Create two events at once.
82+
libc_utils::write_all_from_slice(fd1, &0_u64.to_ne_bytes()).unwrap();
8383

8484
thread1.join().unwrap();
8585
thread2.join().unwrap();
86-
thread3.join().unwrap();
8786
}

‎tests/fail-dep/libc/libc_epoll_block_two_thread.stderr‎

Lines changed: 4 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,3 @@
1-
error: the evaluated program deadlocked
2-
|
3-
= note: this thread got stuck here
4-
= note: (no span available)
5-
61
error: the evaluated program deadlocked
72
--> RUSTLIB/std/src/sys/thread/PLATFORM.rs:LL:CC
83
|
@@ -16,22 +11,16 @@ LL | let ret = unsafe { libc::pthread_join(id, ptr::null_mut()) };
1611
note: inside `main`
1712
--> tests/fail-dep/libc/libc_epoll_block_two_thread.rs:LL:CC
1813
|
19-
LL | thread1.join().unwrap();
14+
LL | thread2.join().unwrap();
2015
| ^^^^^^^^^^^^^^
2116

2217
error: the evaluated program deadlocked
2318
--> tests/fail-dep/libc/libc_epoll_block_two_thread.rs:LL:CC
2419
|
25-
LL | check_epoll_wait::<TAG>(epfd, &[(expected_event, expected_value)], -1);
26-
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ this thread got stuck here
27-
28-
error: the evaluated program deadlocked
29-
|
30-
= note: this thread got stuck here
31-
= note: (no span available)
32-
= note: duplicate diagnostic emitted due to `-Z deduplicate-diagnostics=no`
20+
LL | check_epoll_wait::<TAG>(epfd, &expected, -1);
21+
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ this thread got stuck here
3322

3423
note: some details are omitted, run with `MIRIFLAGS=-Zmiri-backtrace=full` for a verbose backtrace
3524

36-
error: aborting due to 4 previous errors
25+
error: aborting due to 2 previous errors
3726

‎tests/fail-dep/libc/socketpair-close-while-blocked.rs‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
//! This is a regression test for <https://github.com/rust-lang/miri/issues/3947>: we had some
22
//! faulty logic around `release_clock` that led to this code not reporting a data race.
3-
//~^^ERROR: deadlock
43
//@ignore-target: windows # no libc socketpair on Windows
54
//@compile-flags: -Zmiri-deterministic-concurrency
65
//@error-in-other-file: deadlock

0 commit comments

Comments
 (0)