More cleanup

lib/runtime/src/platform/macos_aarch64/driver.rs

@@ -99,7 +99,7 @@ pub struct Driver {
     pending_wakes: Cell<u64>,
     pending_timers: Cell<u64>,
     next_fd_token: Cell<u64>,
-    fd_waiters: RefCell<HashMap<FdKey, Vec<FdWaiter>>>,
+    fd_waiters: RefCell<HashMap<FdKey, FdWaiter>>,
 }
 
 /// Creates a new driver and its paired [`ThreadNotifier`].
@@ -241,17 +241,31 @@ impl Driver {
         completion: FdCompletion,
     ) -> io::Result<FdReadinessToken> {
         let key = FdKey { fd, interest };
-        let should_register = !self.fd_waiters.borrow().contains_key(&key);
-        if should_register {
-            self.update_fd_interest(key, libc::EV_ADD | libc::EV_ENABLE)?;
+        let removed_stale_waiter = {
+            let mut waiters = self.fd_waiters.borrow_mut();
+            match waiters.get(&key) {
+                Some(waiter) if !waiter.completion.is_interested() => {
+                    waiters.remove(&key);
+                    true
+                }
+                Some(_) => {
+                    return Err(io::Error::new(
+                        io::ErrorKind::AlreadyExists,
+                        "fd readiness already has a waiter for this interest",
+                    ));
+                }
+                None => false,
+            }
+        };
+        if removed_stale_waiter {
+            let _ = self.update_fd_interest(key, libc::EV_DELETE);
         }
 
         let token = self.allocate_fd_token();
+        self.update_fd_interest(key, libc::EV_ADD | libc::EV_ENABLE | libc::EV_ONESHOT)?;
         self.fd_waiters
             .borrow_mut()
-            .entry(key)
-            .or_default()
-            .push(FdWaiter { token, completion });
+            .insert(key, FdWaiter { token, completion });
         Ok(token)
     }
 
@@ -259,12 +273,9 @@ impl Driver {
         let mut empty_key = None;
         {
             let mut waiters = self.fd_waiters.borrow_mut();
-            for (key, entries) in waiters.iter_mut() {
-                if let Some(index) = entries.iter().position(|entry| entry.token == token) {
-                    entries.swap_remove(index);
-                    if entries.is_empty() {
-                        empty_key = Some(*key);
-                    }
+            for (key, entry) in waiters.iter() {
+                if entry.token == token {
+                    empty_key = Some(*key);
                     break;
                 }
             }
@@ -372,19 +383,13 @@ impl Driver {
 
     fn complete_fd_waiters(&self, fd: RawFd, interest: FdInterest, event: &libc::kevent) {
         let key = FdKey { fd, interest };
-        let waiters = self.fd_waiters.borrow_mut().remove(&key);
-        let Some(waiters) = waiters else {
+        let waiter = self.fd_waiters.borrow_mut().remove(&key);
+        let Some(waiter) = waiter else {
             return;
         };
 
-        let _ = self.update_fd_interest(key, libc::EV_DELETE);
-        let result = fd_event_result(event);
-        for waiter in waiters {
-            waiter.completion.complete(match &result {
-                Ok(()) => Ok(()),
-                Err(error) => Err(io::Error::new(error.kind(), error.to_string())),
-            });
-        }
+        let result = fd_event_result(event, interest);
+        waiter.completion.complete(result);
     }
 }
 
@@ -436,9 +441,14 @@ fn interest_from_filter(filter: i16) -> Option<FdInterest> {
     }
 }
 
-fn fd_event_result(event: &libc::kevent) -> io::Result<()> {
+fn fd_event_result(event: &libc::kevent, interest: FdInterest) -> io::Result<()> {
     if event.flags & libc::EV_ERROR != 0 && event.data != 0 {
         Err(io::Error::from_raw_os_error(event.data as i32))
+    } else if event.flags & libc::EV_EOF != 0 && interest == FdInterest::Writable {
+        Err(io::Error::new(
+            io::ErrorKind::BrokenPipe,
+            "fd write side reached EOF",
+        ))
     } else {
         Ok(())
     }

lib/runtime/src/sys/macos/fs.rs

@@ -21,6 +21,7 @@ use crate::platform::current::runtime::{ThreadHandle, current_thread_handle};
 type BlockingTask = Box<dyn FnOnce() + Send + 'static>;
 
 static BLOCKING_POOL: OnceLock<io::Result<BlockingPool>> = OnceLock::new();
+const BLOCKING_QUEUE_CAPACITY: usize = 1024;
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum ExecutionPath {
@@ -367,16 +368,20 @@ async fn offload<T: Send + 'static>(
 }
 
 struct BlockingPool {
-    sender: mpsc::Sender<BlockingTask>,
+    sender: mpsc::SyncSender<BlockingTask>,
 }
 
 impl BlockingPool {
     fn spawn(&self, task: BlockingTask) -> io::Result<()> {
-        self.sender.send(task).map_err(|_| {
-            io::Error::new(
+        self.sender.try_send(task).map_err(|error| match error {
+            mpsc::TrySendError::Full(_) => io::Error::new(
+                io::ErrorKind::WouldBlock,
+                "filesystem blocking worker queue is full",
+            ),
+            mpsc::TrySendError::Disconnected(_) => io::Error::new(
                 io::ErrorKind::BrokenPipe,
                 "filesystem blocking worker pool has stopped",
-            )
+            ),
         })
     }
 }
@@ -393,7 +398,7 @@ fn blocking_pool() -> io::Result<&'static BlockingPool> {
 }
 
 fn create_blocking_pool() -> io::Result<BlockingPool> {
-    let (sender, receiver) = mpsc::channel::<BlockingTask>();
+    let (sender, receiver) = mpsc::sync_channel::<BlockingTask>(BLOCKING_QUEUE_CAPACITY);
     let receiver = Arc::new(Mutex::new(receiver));
     let worker_count = std::thread::available_parallelism()
         .map(usize::from)