epoll: wake up multiple threads if that is needed to deliver all events

Author: RalfJung

src/tools/miri/src/shims/unix/linux_like/epoll.rs

@@ -1,5 +1,5 @@
 use std::cell::RefCell;
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, VecDeque};
 use std::io;
 use std::time::Duration;
 
@@ -22,8 +22,8 @@ struct Epoll {
     /// "ready" list; instead, a boolean flag in this list tracks which subset is ready. This makes
     /// `epoll_wait` less efficient, but also requires less bookkeeping.
     interest_list: RefCell<BTreeMap<EpollEventKey, EpollEventInterest>>,
-    /// A list of thread ids blocked on this epoll instance.
-    blocked: RefCell<Vec<ThreadId>>,
+    /// The queue of threads blocked on this epoll instance, and how many events they'd like to get.
+    queue: RefCell<VecDeque<(ThreadId, u32)>>,
 }
 
 impl VisitProvenance for Epoll {
@@ -459,7 +459,9 @@ pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
                 }
             };
             // Record this thread as blocked.
-            epfd.blocked.borrow_mut().push(this.active_thread());
+            epfd.queue
+                .borrow_mut()
+                .push_back((this.active_thread(), maxevents.try_into().unwrap()));
             // And block it.
             let dest = dest.clone();
             // We keep a strong ref to the underlying `Epoll` to make sure it sticks around.
@@ -483,8 +485,8 @@ pub trait EvalContextExt<'tcx>: crate::MiriInterpCxExt<'tcx> {
                             UnblockKind::TimedOut => {
                                 // Remove the current active thread_id from the blocked thread_id list.
                                 epfd
-                                    .blocked.borrow_mut()
-                                    .retain(|&id| id != this.active_thread());
+                                    .queue.borrow_mut()
+                                    .retain(|&(id, _events)| id != this.active_thread());
                                 this.write_int(0, &dest)?;
                                 interp_ok(())
                             },
@@ -550,7 +552,7 @@ fn update_readiness<'tcx>(
         &mut dyn FnMut(&mut EpollEventInterest) -> InterpResult<'tcx>,
     ) -> InterpResult<'tcx>,
 ) -> InterpResult<'tcx> {
-    let mut wakeup = false;
+    let mut num_ready = 0u32; // how many events we have ready to deliver
     for_each_interest(&mut |interest| {
         // Update the ready events tracked in this interest.
         let new_readiness = interest.relevant_events & active_events;
@@ -565,17 +567,20 @@ fn update_readiness<'tcx>(
             ecx.release_clock(|clock| {
                 interest.clock.join(clock);
             })?;
-            wakeup = true;
+            num_ready = num_ready.saturating_add(1);
         }
         interp_ok(())
     })?;
-    if wakeup {
-        // Wake up threads that may have been waiting for events on this epoll.
-        // Do this only once for all the interests.
-        // Edge-triggered notification only notify one thread even if there are
-        // multiple threads blocked on the same epoll.
-        if let Some(thread_id) = epoll.blocked.borrow_mut().pop() {
-            ecx.unblock_thread(thread_id, BlockReason::Epoll)?;
+    // Edge-triggered notifications only wake up as many threads as are needed to deliver
+    // all the events.
+    while num_ready > 0
+        && let Some((thread_id, events)) = epoll.queue.borrow_mut().pop_front()
+    {
+        ecx.unblock_thread(thread_id, BlockReason::Epoll)?;
+        // Keep track of how many events we have left to deliver (except if we saturated;
+        // in that case we just wake up everybody).
+        if num_ready != u32::MAX {
+            num_ready = num_ready.saturating_sub(events);
         }
     }
 

src/tools/miri/tests/pass-dep/libc/libc-epoll-blocking.rs

@@ -4,7 +4,6 @@
 
 use std::convert::TryInto;
 use std::thread;
-use std::thread::spawn;
 
 #[path = "../../utils/libc.rs"]
 mod libc_utils;
@@ -17,6 +16,7 @@ fn main() {
     test_notification_after_timeout();
     test_epoll_race();
     wakeup_on_new_interest();
+    multiple_events_wake_multiple_threads();
 }
 
 // Using `as` cast since `EPOLLET` wraps around
@@ -90,7 +90,7 @@ fn test_epoll_block_then_unblock() {
     // epoll_wait before triggering notification so it will block then get unblocked before timeout.
     let expected_event = u32::try_from(libc::EPOLLIN | libc::EPOLLOUT).unwrap();
     let expected_value = fds[0] as u64;
-    let thread1 = spawn(move || {
+    let thread1 = thread::spawn(move || {
         thread::yield_now();
         let data = "abcde".as_bytes().as_ptr();
         let res = unsafe { libc_utils::write_all(fds[1], data as *const libc::c_void, 5) };
@@ -210,3 +210,53 @@ fn wakeup_on_new_interest() {
     // This should wake up the thread.
     t.join().unwrap();
 }
+
+/// Ensure that if a single operation triggers multiple events, we wake up enough threads
+/// to consume them all.
+fn multiple_events_wake_multiple_threads() {
+    // Create an epoll instance.
+    let epfd = unsafe { libc::epoll_create1(0) };
+    assert_ne!(epfd, -1);
+
+    // Create an eventfd instance.
+    let flags = libc::EFD_NONBLOCK | libc::EFD_CLOEXEC;
+    let fd1 = unsafe { libc::eventfd(0, flags) };
+    // Make a duplicate so that we have two file descriptors for the same file description.
+    let fd2 = unsafe { libc::dup(fd1) };
+
+    // Register both with epoll.
+    let mut ev = libc::epoll_event { events: EPOLL_IN_OUT_ET, u64: fd1 as u64 };
+    let res = unsafe { libc::epoll_ctl(epfd, libc::EPOLL_CTL_ADD, fd1, &mut ev) };
+    assert_eq!(res, 0);
+    let mut ev = libc::epoll_event { events: EPOLL_IN_OUT_ET, u64: fd2 as u64 };
+    let res = unsafe { libc::epoll_ctl(epfd, libc::EPOLL_CTL_ADD, fd2, &mut ev) };
+    assert_eq!(res, 0);
+    // Consume the initial event.
+    let expected = [(libc::EPOLLOUT as u32, fd1 as u64), (libc::EPOLLOUT as u32, fd2 as u64)];
+    check_epoll_wait::<8>(epfd, &expected, -1);
+
+    // Block two threads on the epoll, both wanting to get just one event.
+    let t1 = thread::spawn(move || {
+        let mut e = libc::epoll_event { events: 0, u64: 0 };
+        let res = unsafe { libc::epoll_wait(epfd, &raw mut e, 1, -1) };
+        assert!(res == 1);
+        (e.events, e.u64)
+    });
+    let t2 = thread::spawn(move || {
+        let mut e = libc::epoll_event { events: 0, u64: 0 };
+        let res = unsafe { libc::epoll_wait(epfd, &raw mut e, 1, -1) };
+        assert!(res == 1);
+        (e.events, e.u64)
+    });
+    // Yield so both threads are waiting now.
+    thread::yield_now();
+
+    // Trigger the eventfd. This triggers two events at once!
+    libc_utils::write_all_from_slice(fd1, &0_u64.to_ne_bytes()).unwrap();
+
+    // Both threads should have been woken up so that both events can be consumed.
+    let e1 = t1.join().unwrap();
+    let e2 = t2.join().unwrap();
+    // Ensure that across the two threads we got both events.
+    assert!(expected == [e1, e2] || expected == [e2, e1]);
+}