Review pass 3

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

@@ -9,6 +9,7 @@ use std::net::{
 };
 use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
 use std::pin::Pin;
+use std::sync::{Arc, Mutex, OnceLock, mpsc};
 use std::thread;
 use std::time::Duration;
 
@@ -16,10 +17,14 @@ use crate::op::completion::completion_for_current_thread;
 use crate::op::net::{AcceptedSocket, NetOp, ReceivedDatagram};
 
 const DEFAULT_LISTENER_BACKLOG: i32 = 1024;
+const DNS_QUEUE_CAPACITY: usize = 256;
 
 type RecvFuture = Pin<Box<dyn Future<Output = io::Result<Vec<u8>>> + 'static>>;
 type SendFuture = Pin<Box<dyn Future<Output = io::Result<usize>> + 'static>>;
 type ShutdownFuture = Pin<Box<dyn Future<Output = io::Result<()>> + 'static>>;
+type BlockingTask = Box<dyn FnOnce() + Send + 'static>;
+
+static DNS_POOL: OnceLock<io::Result<BlockingPool>> = OnceLock::new();
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum ExecutionPath {
@@ -391,16 +396,73 @@ async fn offload<T: Send + 'static>(
     work: impl FnOnce() -> io::Result<T> + Send + 'static,
 ) -> io::Result<T> {
     let (future, handle) = completion_for_current_thread::<io::Result<T>>();
-    let handle_for_thread = handle.clone();
-    if let Err(error) = thread::Builder::new()
-        .name("ruin-runtime-net-offload".into())
-        .spawn(move || handle_for_thread.complete(work()))
-    {
-        handle.complete(Err(io::Error::other(error)));
+    if let Err(error) = dns_pool().and_then(|pool| {
+        pool.spawn(Box::new({
+            let handle = handle.clone();
+            move || handle.complete(work())
+        }))
+    }) {
+        handle.complete(Err(error));
     }
     future.await
 }
 
+struct BlockingPool {
+    sender: mpsc::SyncSender<BlockingTask>,
+}
+
+impl BlockingPool {
+    fn spawn(&self, task: BlockingTask) -> io::Result<()> {
+        self.sender.try_send(task).map_err(|error| match error {
+            mpsc::TrySendError::Full(_) => io::Error::new(
+                io::ErrorKind::WouldBlock,
+                "DNS resolver blocking worker queue is full",
+            ),
+            mpsc::TrySendError::Disconnected(_) => io::Error::new(
+                io::ErrorKind::BrokenPipe,
+                "DNS resolver blocking worker pool has stopped",
+            ),
+        })
+    }
+}
+
+fn dns_pool() -> io::Result<&'static BlockingPool> {
+    match DNS_POOL.get_or_init(create_dns_pool) {
+        Ok(pool) => Ok(pool),
+        Err(error) => Err(io::Error::new(error.kind(), error.to_string())),
+    }
+}
+
+fn create_dns_pool() -> io::Result<BlockingPool> {
+    let (sender, receiver) = mpsc::sync_channel::<BlockingTask>(DNS_QUEUE_CAPACITY);
+    let receiver = Arc::new(Mutex::new(receiver));
+    let worker_count = std::thread::available_parallelism()
+        .map(usize::from)
+        .unwrap_or(2)
+        .clamp(2, 4);
+
+    for index in 0..worker_count {
+        let receiver = Arc::clone(&receiver);
+        thread::Builder::new()
+            .name(format!("ruin-runtime-dns-{index}"))
+            .spawn(move || {
+                loop {
+                    let task = {
+                        let receiver = receiver.lock().expect("DNS worker queue mutex poisoned");
+                        receiver.recv()
+                    };
+                    match task {
+                        Ok(task) => task(),
+                        Err(_) => break,
+                    }
+                }
+            })
+            .map_err(io::Error::other)?;
+    }
+
+    Ok(BlockingPool { sender })
+}
+
 async fn io_timeout<T>(
     timeout: Duration,
     future: impl Future<Output = io::Result<T>>,