wtemple/ruin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

initial macos porting work

Browse Source
This commit is contained in:
Will Temple
2026-03-23 15:16:20 -04:00
parent 4193457fc4
commit 861bf63621
40 changed files with 4575 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
lib/runtime/src/channel/mpsc.rs
View File

@@ -7,7 +7,7 @@ use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use crate::op::completion::{CompletionFuture, CompletionHandle};
use crate::sys::linux::channel::runtime_waiter;
use crate::sys::current::channel::runtime_waiter;
/// Creates a bounded channel with room for at most `capacity` queued messages.
///

2
lib/runtime/src/channel/oneshot.rs
View File

@@ -6,7 +6,7 @@ use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use crate::op::completion::{CompletionFuture, CompletionHandle};
use crate::sys::linux::channel::runtime_waiter;
use crate::sys::current::channel::runtime_waiter;
/// Creates a single-use channel for transferring one value from a [`Sender`] to a [`Receiver`].
///

41
lib/runtime/src/fd.rs
View File

@@ -3,44 +3,9 @@
use std::io;
use std::os::fd::RawFd;
use crate::op::completion::completion_for_current_thread;
use crate::platform::linux_x86_64::runtime::with_current_driver;
use crate::platform::linux_x86_64::uring::{IORING_OP_POLL_ADD, IoUringCqe};
/// Waits until `fd` becomes readable or reports an error/hangup condition.
pub async fn wait_readable(fd: RawFd) -> io::Result<()> {
submit_poll(fd, libc::POLLIN | libc::POLLERR | libc::POLLHUP).await
}
async fn submit_poll(fd: RawFd, mask: i16) -> io::Result<()> {
let (future, handle) = completion_for_current_thread::<io::Result<()>>();
let callback_handle = handle.clone();
let token = with_current_driver(|driver| {
driver.submit_operation(
move |sqe| {
sqe.opcode = IORING_OP_POLL_ADD;
sqe.fd = fd;
sqe.len = 0;
sqe.op_flags = mask as u32;
},
move |cqe| {
callback_handle.complete(cqe_to_result(cqe));
},
)
})?;
handle.set_cancel(move || {
let _ = with_current_driver(|driver| driver.cancel_operation(token));
});
future.await
}
fn cqe_to_result(cqe: IoUringCqe) -> io::Result<()> {
if cqe.res < 0 {
return Err(io::Error::from_raw_os_error(-cqe.res));
}
Ok(())
crate::sys::current::fd::wait_readable(fd).await
}
#[cfg(test)]
@@ -53,8 +18,8 @@ mod tests {
#[test]
fn wait_readable_resolves_for_pipe() {
let mut fds = [0; 2];
let result = unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) };
assert_eq!(result, 0, "pipe2 should succeed");
let result = unsafe { libc::pipe(fds.as_mut_ptr()) };
assert_eq!(result, 0, "pipe should succeed");
let read_fd = fds[0];
let write_fd = fds[1];

2
lib/runtime/src/fs.rs
View File

@@ -18,7 +18,7 @@ use crate::op::fs::{
FileType as RawFileType, FsOp, MetadataTarget, OpenOptions as OpOpenOptions,
RawDirEntry as OpDirEntry, RawMetadata,
};
use crate::sys::linux::fs as sys_fs;
use crate::sys::current::fs as sys_fs;
struct FileInner {
fd: OwnedFd,

45
lib/runtime/src/lib.rs
View File

@@ -1,7 +1,7 @@
//! Runtime, driver, async I/O, and channel primitives for RUIN.
//!
//! The crate is centered around a single-threaded event loop with explicit worker threads,
//! JavaScript-style microtask/macrotask scheduling, and Linux `io_uring`-backed I/O.
//! JavaScript-style microtask/macrotask scheduling and platform-specific async I/O backends.
//!
//! Most users will start with:
//!
@@ -11,17 +11,22 @@
//!
//! # Platform support
//!
//! `ruin-runtime` currently targets Linux on `x86_64`.
//! `ruin-runtime` currently targets:
//! - Linux `x86_64`
//! - macOS `aarch64`
//!
//! RUIN runtime foundations.
//!
//! This crate provides a Linux x86_64 runtime substrate: the mesh allocator, the driver, and a
//! single-threaded runtime loop with worker-thread task forwarding.
//! This crate provides a platform runtime substrate with a single-threaded runtime loop and
//! worker-thread task forwarding.
#![feature(thread_local)]
#[cfg(not(all(target_os = "linux", target_arch = "x86_64")))]
compile_error!("ruin-runtime currently supports only Linux x86_64.");
#[cfg(not(any(
all(target_os = "linux", target_arch = "x86_64"),
all(target_os = "macos", target_arch = "aarch64")
)))]
compile_error!("ruin-runtime currently supports Linux x86_64 and macOS aarch64.");
extern crate alloc;
@@ -56,11 +61,24 @@ pub use ruin_runtime_proc_macros::async_main;
/// thread before calling [`run`].
pub use ruin_runtime_proc_macros::main;
/// Driver primitives re-exported from the Linux x86_64 backend.
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub use platform::linux_x86_64::driver::{
/// Driver primitives re-exported from the active backend.
#[cfg(any(
all(target_os = "linux", target_arch = "x86_64"),
all(target_os = "macos", target_arch = "aarch64")
))]
pub use platform::current::driver::{
Driver, ReadyEvents, ThreadNotifier, create, create_driver, monotonic_now,
};
/// Runtime/event-loop primitives re-exported from the active backend.
#[cfg(any(
all(target_os = "linux", target_arch = "x86_64"),
all(target_os = "macos", target_arch = "aarch64")
))]
pub use platform::current::runtime::{
IntervalHandle, JoinHandle, ThreadHandle, TimeoutHandle, WorkerHandle, clear_interval,
clear_timeout, current_thread_handle, queue_future, queue_microtask, queue_task, run,
set_interval, set_timeout, spawn_worker, yield_now,
};
/// Public mesh-allocator surface re-exported from the Linux x86_64 backend.
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub use platform::linux_x86_64::mesh_alloc::{
@@ -77,22 +95,17 @@ pub use platform::linux_x86_64::mesh_alloc::{
/// Additional allocator helpers re-exported from the Linux x86_64 backend.
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub use platform::linux_x86_64::mesh_alloc::{FreelistId, bitmaps_meshable};
/// Runtime/event-loop primitives re-exported from the Linux x86_64 backend.
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub use platform::linux_x86_64::runtime::{
IntervalHandle, JoinHandle, ThreadHandle, TimeoutHandle, WorkerHandle, clear_interval,
clear_timeout, current_thread_handle, queue_future, queue_microtask, queue_task, run,
set_interval, set_timeout, spawn_worker, yield_now,
};
/// Returns the default global mesh allocator configuration for this crate.
///
/// This is useful when embedding the allocator in a `#[global_allocator]` static.
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub const fn default_global_allocator() -> GlobalMeshAllocator {
GlobalMeshAllocator::with_default_config()
}
#[cfg(test)]
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
mod tests {
use super::{MeshAllocator, page_size};

103
lib/runtime/src/net.rs
View File

@@ -72,10 +72,10 @@ impl TcpStream {
where
A: ToSocketAddrs + Send + 'static,
{
let addrs = crate::sys::linux::net::resolve_addrs(addr).await?;
let addrs = crate::sys::current::net::resolve_addrs(addr).await?;
let mut last_error = None;
for addr in addrs {
match crate::sys::linux::net::connect_stream(addr).await {
match crate::sys::current::net::connect_stream(addr).await {
Ok(fd) => return Ok(Self::from_owned_fd(fd)),
Err(error) => last_error = Some(error),
}
@@ -92,7 +92,7 @@ impl TcpStream {
/// Connects to `addr`, failing if the deadline elapses first.
pub async fn connect_timeout(addr: &SocketAddr, timeout: Duration) -> io::Result<Self> {
validate_timeout(timeout)?;
crate::sys::linux::net::connect_stream_timeout(*addr, timeout)
crate::sys::current::net::connect_stream_timeout(*addr, timeout)
.await
.map(Self::from_owned_fd)
}
@@ -101,10 +101,10 @@ impl TcpStream {
pub async fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let data = match self.read_timeout_value() {
Some(timeout) => {
crate::sys::linux::net::recv_timeout(self.raw_fd(), buf.len(), 0, timeout).await?
crate::sys::current::net::recv_timeout(self.raw_fd(), buf.len(), 0, timeout).await?
}
None => {
crate::sys::linux::net::recv(NetOp::Recv {
crate::sys::current::net::recv(NetOp::Recv {
fd: self.raw_fd(),
len: buf.len(),
flags: 0,
@@ -136,10 +136,11 @@ impl TcpStream {
pub async fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
match self.write_timeout_value() {
Some(timeout) => {
crate::sys::linux::net::send_timeout(self.raw_fd(), buf.to_vec(), 0, timeout).await
crate::sys::current::net::send_timeout(self.raw_fd(), buf.to_vec(), 0, timeout)
.await
}
None => {
crate::sys::linux::net::send(NetOp::Send {
crate::sys::current::net::send(NetOp::Send {
fd: self.raw_fd(),
data: buf.to_vec(),
flags: 0,
@@ -166,7 +167,7 @@ impl TcpStream {
/// Shuts down the read, write, or both halves of the connection.
pub async fn shutdown(&self, how: Shutdown) -> io::Result<()> {
crate::sys::linux::net::shutdown(NetOp::Shutdown {
crate::sys::current::net::shutdown(NetOp::Shutdown {
fd: self.raw_fd(),
how,
})
@@ -175,39 +176,39 @@ impl TcpStream {
/// Duplicates the underlying stream socket.
pub async fn try_clone(&self) -> io::Result<Self> {
crate::sys::linux::net::duplicate(self.raw_fd())
crate::sys::current::net::duplicate(self.raw_fd())
.await
.map(Self::from_owned_fd)
}
/// Returns the local socket address of this stream.
pub fn local_addr(&self) -> io::Result<SocketAddr> {
crate::sys::linux::net::local_addr(self.raw_fd())
crate::sys::current::net::local_addr(self.raw_fd())
}
/// Returns the remote peer address of this stream.
pub fn peer_addr(&self) -> io::Result<SocketAddr> {
crate::sys::linux::net::peer_addr(self.raw_fd())
crate::sys::current::net::peer_addr(self.raw_fd())
}
/// Reads the current `TCP_NODELAY` setting.
pub fn nodelay(&self) -> io::Result<bool> {
crate::sys::linux::net::nodelay(self.raw_fd())
crate::sys::current::net::nodelay(self.raw_fd())
}
/// Enables or disables `TCP_NODELAY`.
pub fn set_nodelay(&self, enabled: bool) -> io::Result<()> {
crate::sys::linux::net::set_nodelay(self.raw_fd(), enabled)
crate::sys::current::net::set_nodelay(self.raw_fd(), enabled)
}
/// Reads the socket's IP time-to-live value.
pub fn ttl(&self) -> io::Result<u32> {
crate::sys::linux::net::ttl(self.raw_fd())
crate::sys::current::net::ttl(self.raw_fd())
}
/// Sets the socket's IP time-to-live value.
pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
crate::sys::linux::net::set_ttl(self.raw_fd(), ttl)
crate::sys::current::net::set_ttl(self.raw_fd(), ttl)
}
/// Returns the read timeout used by async read operations on this handle.
@@ -269,10 +270,10 @@ impl TcpListener {
where
A: ToSocketAddrs + Send + 'static,
{
let addrs = crate::sys::linux::net::resolve_addrs(addr).await?;
let addrs = crate::sys::current::net::resolve_addrs(addr).await?;
let mut last_error = None;
for addr in addrs {
match crate::sys::linux::net::bind_listener(addr, None).await {
match crate::sys::current::net::bind_listener(addr, None).await {
Ok(fd) => return Ok(Self::from_owned_fd(fd)),
Err(error) => last_error = Some(error),
}
@@ -288,7 +289,8 @@ impl TcpListener {
/// Accepts an incoming connection.
pub async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> {
let accepted = crate::sys::linux::net::accept(NetOp::Accept { fd: self.raw_fd() }).await?;
let accepted =
crate::sys::current::net::accept(NetOp::Accept { fd: self.raw_fd() }).await?;
let stream = TcpStream::from_owned_fd(unsafe { OwnedFd::from_raw_fd(accepted.fd) });
Ok((stream, accepted.peer_addr))
@@ -296,17 +298,17 @@ impl TcpListener {
/// Returns the local socket address of this listener.
pub fn local_addr(&self) -> io::Result<SocketAddr> {
crate::sys::linux::net::local_addr(self.raw_fd())
crate::sys::current::net::local_addr(self.raw_fd())
}
/// Reads the listener socket's IP time-to-live value.
pub fn ttl(&self) -> io::Result<u32> {
crate::sys::linux::net::ttl(self.raw_fd())
crate::sys::current::net::ttl(self.raw_fd())
}
/// Sets the listener socket's IP time-to-live value.
pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
crate::sys::linux::net::set_ttl(self.raw_fd(), ttl)
crate::sys::current::net::set_ttl(self.raw_fd(), ttl)
}
fn from_owned_fd(fd: OwnedFd) -> Self {
@@ -326,10 +328,10 @@ impl UdpSocket {
where
A: ToSocketAddrs + Send + 'static,
{
let addrs = crate::sys::linux::net::resolve_addrs(addr).await?;
let addrs = crate::sys::current::net::resolve_addrs(addr).await?;
let mut last_error = None;
for addr in addrs {
match crate::sys::linux::net::bind_datagram(addr).await {
match crate::sys::current::net::bind_datagram(addr).await {
Ok(fd) => return Ok(Self::from_owned_fd(fd)),
Err(error) => last_error = Some(error),
}
@@ -351,10 +353,10 @@ impl UdpSocket {
where
A: ToSocketAddrs + Send + 'static,
{
let addrs = crate::sys::linux::net::resolve_addrs(addr).await?;
let addrs = crate::sys::current::net::resolve_addrs(addr).await?;
let mut last_error = None;
for addr in addrs {
match crate::sys::linux::net::connect(NetOp::Connect {
match crate::sys::current::net::connect(NetOp::Connect {
fd: self.raw_fd(),
addr,
})
@@ -377,10 +379,11 @@ impl UdpSocket {
pub async fn send(&self, buf: &[u8]) -> io::Result<usize> {
match self.write_timeout_value() {
Some(timeout) => {
crate::sys::linux::net::send_timeout(self.raw_fd(), buf.to_vec(), 0, timeout).await
crate::sys::current::net::send_timeout(self.raw_fd(), buf.to_vec(), 0, timeout)
.await
}
None => {
crate::sys::linux::net::send(NetOp::Send {
crate::sys::current::net::send(NetOp::Send {
fd: self.raw_fd(),
data: buf.to_vec(),
flags: 0,
@@ -394,10 +397,10 @@ impl UdpSocket {
pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
let data = match self.read_timeout_value() {
Some(timeout) => {
crate::sys::linux::net::recv_timeout(self.raw_fd(), buf.len(), 0, timeout).await?
crate::sys::current::net::recv_timeout(self.raw_fd(), buf.len(), 0, timeout).await?
}
None => {
crate::sys::linux::net::recv(NetOp::Recv {
crate::sys::current::net::recv(NetOp::Recv {
fd: self.raw_fd(),
len: buf.len(),
flags: 0,
@@ -414,7 +417,7 @@ impl UdpSocket {
pub async fn peek(&self, buf: &mut [u8]) -> io::Result<usize> {
let data = match self.read_timeout_value() {
Some(timeout) => {
crate::sys::linux::net::recv_timeout(
crate::sys::current::net::recv_timeout(
self.raw_fd(),
buf.len(),
libc::MSG_PEEK,
@@ -423,7 +426,7 @@ impl UdpSocket {
.await?
}
None => {
crate::sys::linux::net::recv(NetOp::Recv {
crate::sys::current::net::recv(NetOp::Recv {
fd: self.raw_fd(),
len: buf.len(),
flags: libc::MSG_PEEK,
@@ -441,13 +444,13 @@ impl UdpSocket {
where
A: ToSocketAddrs + Send + 'static,
{
let addrs = crate::sys::linux::net::resolve_addrs(addr).await?;
let addrs = crate::sys::current::net::resolve_addrs(addr).await?;
let mut last_error = None;
let timeout = self.write_timeout_value();
for addr in addrs {
let result = match timeout {
Some(timeout) => {
crate::sys::linux::net::send_to_timeout(
crate::sys::current::net::send_to_timeout(
self.raw_fd(),
buf.to_vec(),
addr,
@@ -457,7 +460,7 @@ impl UdpSocket {
.await
}
None => {
crate::sys::linux::net::send_to(NetOp::SendTo {
crate::sys::current::net::send_to(NetOp::SendTo {
fd: self.raw_fd(),
target: addr,
data: buf.to_vec(),
@@ -484,11 +487,11 @@ impl UdpSocket {
pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
let datagram = match self.read_timeout_value() {
Some(timeout) => {
crate::sys::linux::net::recv_from_timeout(self.raw_fd(), buf.len(), 0, timeout)
crate::sys::current::net::recv_from_timeout(self.raw_fd(), buf.len(), 0, timeout)
.await?
}
None => {
crate::sys::linux::net::recv_from(NetOp::RecvFrom {
crate::sys::current::net::recv_from(NetOp::RecvFrom {
fd: self.raw_fd(),
len: buf.len(),
flags: 0,
@@ -505,7 +508,7 @@ impl UdpSocket {
pub async fn peek_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
let datagram = match self.read_timeout_value() {
Some(timeout) => {
crate::sys::linux::net::recv_from_timeout(
crate::sys::current::net::recv_from_timeout(
self.raw_fd(),
buf.len(),
libc::MSG_PEEK,
@@ -514,7 +517,7 @@ impl UdpSocket {
.await?
}
None => {
crate::sys::linux::net::recv_from(NetOp::RecvFrom {
crate::sys::current::net::recv_from(NetOp::RecvFrom {
fd: self.raw_fd(),
len: buf.len(),
flags: libc::MSG_PEEK,
@@ -529,39 +532,39 @@ impl UdpSocket {
/// Duplicates the underlying UDP socket.
pub async fn try_clone(&self) -> io::Result<Self> {
crate::sys::linux::net::duplicate(self.raw_fd())
crate::sys::current::net::duplicate(self.raw_fd())
.await
.map(Self::from_owned_fd)
}
/// Returns the local socket address of this socket.
pub fn local_addr(&self) -> io::Result<SocketAddr> {
crate::sys::linux::net::local_addr(self.raw_fd())
crate::sys::current::net::local_addr(self.raw_fd())
}
/// Returns the connected peer address, if the socket has been connected.
pub fn peer_addr(&self) -> io::Result<SocketAddr> {
crate::sys::linux::net::peer_addr(self.raw_fd())
crate::sys::current::net::peer_addr(self.raw_fd())
}
/// Reads the `SO_BROADCAST` setting.
pub fn broadcast(&self) -> io::Result<bool> {
crate::sys::linux::net::broadcast(self.raw_fd())
crate::sys::current::net::broadcast(self.raw_fd())
}
/// Enables or disables `SO_BROADCAST`.
pub fn set_broadcast(&self, enabled: bool) -> io::Result<()> {
crate::sys::linux::net::set_broadcast(self.raw_fd(), enabled)
crate::sys::current::net::set_broadcast(self.raw_fd(), enabled)
}
/// Reads the socket's IP time-to-live value.
pub fn ttl(&self) -> io::Result<u32> {
crate::sys::linux::net::ttl(self.raw_fd())
crate::sys::current::net::ttl(self.raw_fd())
}
/// Sets the socket's IP time-to-live value.
pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
crate::sys::linux::net::set_ttl(self.raw_fd(), ttl)
crate::sys::current::net::set_ttl(self.raw_fd(), ttl)
}
/// Returns the read timeout used by async receive operations on this handle.
@@ -627,13 +630,13 @@ impl HyperRead for TcpStream {
if this.pending_read.is_none() {
this.pending_read = Some(match this.read_timeout_value() {
Some(timeout) => Box::pin(crate::sys::linux::net::recv_timeout(
Some(timeout) => Box::pin(crate::sys::current::net::recv_timeout(
this.raw_fd(),
buf.remaining(),
0,
timeout,
)),
None => crate::sys::linux::net::recv_future(this.raw_fd(), buf.remaining()),
None => crate::sys::current::net::recv_future(this.raw_fd(), buf.remaining()),
});
}
@@ -671,13 +674,13 @@ impl HyperWrite for TcpStream {
if this.pending_write.is_none() {
this.pending_write = Some(match this.write_timeout_value() {
Some(timeout) => Box::pin(crate::sys::linux::net::send_timeout(
Some(timeout) => Box::pin(crate::sys::current::net::send_timeout(
this.raw_fd(),
buf.to_vec(),
0,
timeout,
)),
None => crate::sys::linux::net::send_future(this.raw_fd(), buf.to_vec()),
None => crate::sys::current::net::send_future(this.raw_fd(), buf.to_vec()),
});
}
@@ -707,7 +710,7 @@ impl HyperWrite for TcpStream {
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
let this = self.get_mut();
if this.pending_shutdown.is_none() {
this.pending_shutdown = Some(crate::sys::linux::net::shutdown_future(
this.pending_shutdown = Some(crate::sys::current::net::shutdown_future(
this.raw_fd(),
Shutdown::Write,
));

2
lib/runtime/src/op/completion.rs
View File

@@ -6,7 +6,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};
use crate::platform::linux_x86_64::runtime::{ThreadHandle, current_thread_handle};
use crate::platform::current::runtime::{ThreadHandle, current_thread_handle};
type CancelCallback = Box<dyn FnOnce() + Send + 'static>;

17
lib/runtime/src/platform/linux_x86_64/runtime.rs
View File

@@ -1178,7 +1178,11 @@ fn schedule_immediate_interval(owner: *const ThreadState, id: usize) {
};
(callback.borrow_mut())();
// Re-enqueue for the next turn if still live.
if current_thread().immediate_intervals.borrow().contains_key(&id) {
if current_thread()
.immediate_intervals
.borrow()
.contains_key(&id)
{
schedule_immediate_interval(owner, id);
}
}));
@@ -1207,12 +1211,8 @@ fn dispatch_expired_timers() {
.deadline
.checked_add(interval)
.unwrap_or(Duration::MAX);
let next_timer = TimerNode::interval(
timer.id,
next_deadline,
interval,
Rc::clone(&callback),
);
let next_timer =
TimerNode::interval(timer.id, next_deadline, interval, Rc::clone(&callback));
current_thread().timers.borrow_mut().insert(next_timer);
push_local_macrotask(Box::new(move || {
@@ -1394,8 +1394,7 @@ mod tests {
// turns by awaiting a sleep between checks.
let count = Rc::new(Cell::new(0usize));
let count_clone = Rc::clone(&count);
let handle_slot: Rc<RefCell<Option<super::IntervalHandle>>> =
Rc::new(RefCell::new(None));
let handle_slot: Rc<RefCell<Option<super::IntervalHandle>>> = Rc::new(RefCell::new(None));
let handle_slot_clone = Rc::clone(&handle_slot);
let handle = set_interval(Duration::ZERO, move || {

340
lib/runtime/src/platform/macos_aarch64/driver.rs Normal file
View File

@@ -0,0 +1,340 @@
//! Public runtime driver primitives for macOS.
use std::cell::Cell;
use std::io;
use std::os::fd::RawFd;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
#[derive(Clone)]
struct NotifierInner {
write_fd: RawFd,
closed: Arc<AtomicBool>,
}
impl NotifierInner {
fn notify(&self) -> io::Result<()> {
if self.closed.load(Ordering::Acquire) {
return Err(io::Error::new(
io::ErrorKind::BrokenPipe,
"target runtime driver is closed",
));
}
let byte = 1u8;
let written = unsafe {
libc::write(
self.write_fd,
&byte as *const u8 as *const libc::c_void,
std::mem::size_of::<u8>(),
)
};
if written < 0 {
let error = io::Error::last_os_error();
if error.kind() == io::ErrorKind::WouldBlock {
return Ok(());
}
return Err(error);
}
Ok(())
}
}
#[derive(Clone)]
/// Cross-thread notifier for a runtime thread's driver.
pub struct ThreadNotifier {
inner: NotifierInner,
}
impl ThreadNotifier {
/// Sends a wake notification to the target runtime thread.
pub fn notify(&self) -> io::Result<()> {
self.inner.notify()
}
}
#[derive(Debug, Default, Clone, Copy, Eq, PartialEq)]
/// Readiness information returned by [`Driver::poll`].
pub struct ReadyEvents {
/// One or more timer expirations are pending.
pub timer: bool,
/// One or more cross-thread wake notifications are pending.
pub wake: bool,
}
/// Low-level macOS runtime driver backed by `kqueue` and a wake pipe.
pub struct Driver {
kqueue_fd: RawFd,
wake_read_fd: RawFd,
wake_write_fd: RawFd,
closed: Arc<AtomicBool>,
timer_deadline: Cell<Option<Duration>>,
pending_wakes: Cell<u64>,
pending_timers: Cell<u64>,
}
/// Creates a new driver and its paired [`ThreadNotifier`].
pub fn create() -> io::Result<(Driver, ThreadNotifier)> {
create_driver()
}
/// Creates a new driver and its paired [`ThreadNotifier`].
pub fn create_driver() -> io::Result<(Driver, ThreadNotifier)> {
let kqueue_fd = cvt(unsafe { libc::kqueue() })?;
let mut pipe_fds = [0; 2];
cvt(unsafe { libc::pipe(pipe_fds.as_mut_ptr()) })?;
let wake_read_fd = pipe_fds[0];
let wake_write_fd = pipe_fds[1];
set_nonblocking(wake_read_fd)?;
set_nonblocking(wake_write_fd)?;
let event = libc::kevent {
ident: wake_read_fd as usize,
filter: libc::EVFILT_READ,
flags: libc::EV_ADD | libc::EV_ENABLE,
fflags: 0,
data: 0,
udata: std::ptr::null_mut(),
};
let submitted = unsafe {
libc::kevent(
kqueue_fd,
&event,
1,
std::ptr::null_mut(),
0,
std::ptr::null(),
)
};
if submitted < 0 {
let error = io::Error::last_os_error();
unsafe {
libc::close(wake_read_fd);
libc::close(wake_write_fd);
libc::close(kqueue_fd);
}
return Err(error);
}
let closed = Arc::new(AtomicBool::new(false));
let driver = Driver {
kqueue_fd,
wake_read_fd,
wake_write_fd,
closed: Arc::clone(&closed),
timer_deadline: Cell::new(None),
pending_wakes: Cell::new(0),
pending_timers: Cell::new(0),
};
let notifier = ThreadNotifier {
inner: NotifierInner {
write_fd: wake_write_fd,
closed,
},
};
Ok((driver, notifier))
}
impl Driver {
pub(crate) fn bind_current_thread(&self) {}
pub(crate) fn unbind_current_thread(&self) {}
/// Polls the driver without blocking.
pub fn poll(&self) -> io::Result<Option<ReadyEvents>> {
let mut pending = ReadyEvents::default();
if self.pending_wakes.get() > 0 {
pending.wake = true;
}
if self.pending_timers.get() > 0 {
pending.timer = true;
}
if pending.wake || pending.timer {
return Ok(Some(pending));
}
self.process(Some(Duration::ZERO))
}
/// Blocks until at least one event is available.
pub fn wait(&self) -> io::Result<()> {
let now = monotonic_now()?;
let timeout = self
.timer_deadline
.get()
.map(|deadline| deadline.saturating_sub(now));
let _ = self.process(timeout)?;
Ok(())
}
/// Updates the currently armed timer deadline.
pub fn rearm_timer(&self, deadline: Option<Duration>) -> io::Result<()> {
self.timer_deadline.set(deadline);
Ok(())
}
/// Drains the accumulated wake notification count.
pub fn drain_wake(&self) -> io::Result<u64> {
let wakes = self.pending_wakes.replace(0);
if wakes == 0 {
Err(io::Error::new(
io::ErrorKind::WouldBlock,
"no wake events are pending",
))
} else {
Ok(wakes)
}
}
/// Drains the accumulated timer-expiration count.
pub fn drain_timer(&self) -> io::Result<u64> {
let timers = self.pending_timers.replace(0);
if timers == 0 {
Err(io::Error::new(
io::ErrorKind::WouldBlock,
"no timer events are pending",
))
} else {
Ok(timers)
}
}
fn process(&self, timeout: Option<Duration>) -> io::Result<Option<ReadyEvents>> {
let mut ready = ReadyEvents::default();
let mut events = [unsafe { std::mem::zeroed::<libc::kevent>() }; 16];
let timeout_spec = timeout_to_timespec(timeout);
let timeout_ptr = timeout_spec
.as_ref()
.map_or(std::ptr::null(), |value| value as *const libc::timespec);
let result = unsafe {
libc::kevent(
self.kqueue_fd,
std::ptr::null(),
0,
events.as_mut_ptr(),
events.len() as i32,
timeout_ptr,
)
};
if result < 0 {
let error = io::Error::last_os_error();
if error.kind() != io::ErrorKind::Interrupted {
return Err(error);
}
}
let mut saw_any = false;
let count = result.max(0) as usize;
if count > 0 {
saw_any = true;
for event in events.iter().take(count) {
if event.ident as RawFd == self.wake_read_fd {
ready.wake = true;
let wakes = drain_wake_pipe(self.wake_read_fd)?;
self.pending_wakes
.set(self.pending_wakes.get().saturating_add(wakes));
}
}
}
if let Some(deadline) = self.timer_deadline.get()
&& monotonic_now()? >= deadline
{
ready.timer = true;
saw_any = true;
self.timer_deadline.set(None);
self.pending_timers
.set(self.pending_timers.get().saturating_add(1));
}
if saw_any { Ok(Some(ready)) } else { Ok(None) }
}
}
impl Drop for Driver {
fn drop(&mut self) {
self.closed.store(true, Ordering::Release);
unsafe {
libc::close(self.wake_read_fd);
libc::close(self.wake_write_fd);
libc::close(self.kqueue_fd);
}
}
}
/// Returns the current monotonic clock reading.
pub fn monotonic_now() -> io::Result<Duration> {
let mut now = std::mem::MaybeUninit::<libc::timespec>::uninit();
let result = unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, now.as_mut_ptr()) };
if result < 0 {
return Err(io::Error::last_os_error());
}
let now = unsafe { now.assume_init() };
Ok(Duration::new(now.tv_sec as u64, now.tv_nsec as u32))
}
fn cvt(value: libc::c_int) -> io::Result<libc::c_int> {
if value < 0 {
Err(io::Error::last_os_error())
} else {
Ok(value)
}
}
fn set_nonblocking(fd: RawFd) -> io::Result<()> {
let flags = cvt(unsafe { libc::fcntl(fd, libc::F_GETFL) })?;
cvt(unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) })?;
Ok(())
}
fn timeout_to_timespec(timeout: Option<Duration>) -> Option<libc::timespec> {
timeout.map(|value| libc::timespec {
tv_sec: value.as_secs() as libc::time_t,
tv_nsec: value.subsec_nanos() as libc::c_long,
})
}
fn drain_wake_pipe(fd: RawFd) -> io::Result<u64> {
let mut wakes = 0u64;
let mut buf = [0u8; 256];
loop {
let read = unsafe {
libc::read(
fd,
buf.as_mut_ptr() as *mut libc::c_void,
buf.len() as libc::size_t,
)
};
if read > 0 {
wakes = wakes.saturating_add(read as u64);
continue;
}
if read == 0 {
break;
}
let error = io::Error::last_os_error();
if error.kind() == io::ErrorKind::WouldBlock {
break;
}
if error.kind() == io::ErrorKind::Interrupted {
continue;
}
return Err(error);
}
Ok(wakes.max(1))
}

2
lib/runtime/src/platform/macos_aarch64/mod.rs Normal file
View File

@@ -0,0 +1,2 @@
pub mod driver;
pub mod runtime;

1416
lib/runtime/src/platform/macos_aarch64/runtime.rs Normal file
View File

File diff suppressed because it is too large Load Diff

7
lib/runtime/src/platform/mod.rs
View File

@@ -1,2 +1,9 @@
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub mod linux_x86_64;
#[cfg(all(target_os = "macos", target_arch = "aarch64"))]
pub mod macos_aarch64;
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub use linux_x86_64 as current;
#[cfg(all(target_os = "macos", target_arch = "aarch64"))]
pub use macos_aarch64 as current;

2
lib/runtime/src/sys/linux/channel.rs
View File

@@ -1,7 +1,7 @@
//! Linux channel wake helpers.
use crate::op::completion::{CompletionFuture, CompletionHandle, completion};
use crate::platform::linux_x86_64::runtime::try_current_thread_handle;
use crate::platform::current::runtime::try_current_thread_handle;
pub(crate) fn runtime_waiter<T: Send + 'static>() -> (CompletionFuture<T>, CompletionHandle<T>) {
let owner = try_current_thread_handle()

44
lib/runtime/src/sys/linux/fd.rs Normal file
View File

@@ -0,0 +1,44 @@
//! Linux fd readiness backend.
use std::io;
use std::os::fd::RawFd;
use crate::op::completion::completion_for_current_thread;
use crate::platform::current::runtime::with_current_driver;
use crate::platform::linux_x86_64::uring::{IORING_OP_POLL_ADD, IoUringCqe};
/// Waits until `fd` becomes readable or reports an error/hangup condition.
pub async fn wait_readable(fd: RawFd) -> io::Result<()> {
submit_poll(fd, libc::POLLIN | libc::POLLERR | libc::POLLHUP).await
}
async fn submit_poll(fd: RawFd, mask: i16) -> io::Result<()> {
let (future, handle) = completion_for_current_thread::<io::Result<()>>();
let callback_handle = handle.clone();
let token = with_current_driver(|driver| {
driver.submit_operation(
move |sqe| {
sqe.opcode = IORING_OP_POLL_ADD;
sqe.fd = fd;
sqe.len = 0;
sqe.op_flags = mask as u32;
},
move |cqe| {
callback_handle.complete(cqe_to_result(cqe));
},
)
})?;
handle.set_cancel(move || {
let _ = with_current_driver(|driver| driver.cancel_operation(token));
});
future.await
}
fn cqe_to_result(cqe: IoUringCqe) -> io::Result<()> {
if cqe.res < 0 {
return Err(io::Error::from_raw_os_error(-cqe.res));
}
Ok(())
}

1
lib/runtime/src/sys/linux/mod.rs
View File

@@ -1,5 +1,6 @@
//! Linux backend modules.
pub mod channel;
pub mod fd;
pub mod fs;
pub mod net;

12
lib/runtime/src/sys/linux/net.rs
View File

@@ -803,13 +803,9 @@ where
let (future, handle) = completion_for_current_thread::<io::Result<T>>();
let callback_handle = handle.clone();
let token = with_current_driver(|driver| {
driver.submit_operation_with_linked_timeout(
fill,
timeout,
move |cqe| {
callback_handle.complete(map(cqe));
},
)
driver.submit_operation_with_linked_timeout(fill, timeout, move |cqe| {
callback_handle.complete(map(cqe));
})
})?;
handle.set_cancel(move || {
@@ -1079,7 +1075,6 @@ fn send_sync(fd: RawFd, data: Vec<u8>, flags: i32) -> io::Result<usize> {
cvt_long(written).map(|written| written as usize)
}
fn recv_sync(fd: RawFd, len: usize, flags: i32) -> io::Result<Vec<u8>> {
let mut buffer = vec![0; len];
let read = unsafe {
@@ -1127,7 +1122,6 @@ fn close_sync(fd: RawFd) -> io::Result<()> {
cvt(unsafe { libc::close(fd) }).map(|_| ())
}
/// Wrapper making `Box<libc::iovec>` sendable across the async CQE boundary.
///
/// Safety: `iov_base` points into a `Vec<u8>` that is owned by the same

10
lib/runtime/src/sys/macos/channel.rs Normal file
View File

@@ -0,0 +1,10 @@
//! macOS channel wake helpers.
use crate::op::completion::{CompletionFuture, CompletionHandle, completion};
use crate::platform::current::runtime::try_current_thread_handle;
pub(crate) fn runtime_waiter<T: Send + 'static>() -> (CompletionFuture<T>, CompletionHandle<T>) {
let owner = try_current_thread_handle()
.expect("async channel operations must be polled on a runtime thread");
completion(owner)
}

43
lib/runtime/src/sys/macos/fd.rs Normal file
View File

@@ -0,0 +1,43 @@
//! macOS fd readiness backend.
use std::io;
use std::os::fd::RawFd;
use std::time::Duration;
use crate::op::completion::completion_for_current_thread;
/// Waits until `fd` becomes readable or reports an error/hangup condition.
pub async fn wait_readable(fd: RawFd) -> io::Result<()> {
let (future, handle) = completion_for_current_thread::<io::Result<()>>();
std::thread::spawn(move || {
let mut pollfd = libc::pollfd {
fd,
events: libc::POLLIN | libc::POLLERR | libc::POLLHUP,
revents: 0,
};
loop {
let result = unsafe { libc::poll(&mut pollfd, 1, -1) };
if result > 0 {
handle.complete(Ok(()));
return;
}
if result == 0 {
continue;
}
let error = io::Error::last_os_error();
if error.kind() == io::ErrorKind::Interrupted {
continue;
}
handle.complete(Err(error));
return;
}
});
future.await
}
#[allow(dead_code)]
fn _duration_to_poll_timeout(duration: Duration) -> i32 {
duration.as_millis().min(i32::MAX as u128) as i32
}

427
lib/runtime/src/sys/macos/fs.rs Normal file
View File

@@ -0,0 +1,427 @@
//! macOS filesystem backend.
use std::collections::VecDeque;
use std::ffi::CString;
use std::future::poll_fn;
use std::io;
use std::os::fd::{FromRawFd, OwnedFd};
use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::{FileTypeExt, MetadataExt, OpenOptionsExt};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};
use std::thread;
use crate::op::completion::completion_for_current_thread;
use crate::op::fs::{FileType, FsOp, MetadataTarget, RawDirEntry, RawMetadata};
use crate::platform::current::runtime::{ThreadHandle, current_thread_handle};
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ExecutionPath {
Offload,
}
pub fn execution_path(_op: &FsOp) -> ExecutionPath {
ExecutionPath::Offload
}
pub async fn open(op: FsOp) -> io::Result<OwnedFd> {
let FsOp::Open { path, options } = op else {
unreachable!("open backend called with non-open op");
};
offload(move || {
let mut open = std::fs::OpenOptions::new();
open.read(options.read)
.write(options.write)
.append(options.append)
.truncate(options.truncate)
.create(options.create)
.create_new(options.create_new)
.mode(0o666);
let file = open.open(path)?;
Ok(unsafe { OwnedFd::from_raw_fd(std::os::fd::IntoRawFd::into_raw_fd(file)) })
})
.await
}
pub async fn read(op: FsOp) -> io::Result<Vec<u8>> {
let FsOp::Read { fd, offset, len } = op else {
unreachable!("read backend called with non-read op");
};
offload(move || {
let mut buffer = vec![0; len];
let read = match offset {
Some(offset) => unsafe {
libc::pread(
fd,
buffer.as_mut_ptr().cast::<libc::c_void>(),
len,
offset as libc::off_t,
)
},
None => unsafe { libc::read(fd, buffer.as_mut_ptr().cast::<libc::c_void>(), len) },
};
if read < 0 {
return Err(io::Error::last_os_error());
}
buffer.truncate(read as usize);
Ok(buffer)
})
.await
}
pub async fn write(op: FsOp) -> io::Result<usize> {
let FsOp::Write { fd, offset, data } = op else {
unreachable!("write backend called with non-write op");
};
offload(move || {
let written = match offset {
Some(offset) => unsafe {
libc::pwrite(
fd,
data.as_ptr().cast::<libc::c_void>(),
data.len(),
offset as libc::off_t,
)
},
None => unsafe { libc::write(fd, data.as_ptr().cast::<libc::c_void>(), data.len()) },
};
if written < 0 {
return Err(io::Error::last_os_error());
}
Ok(written as usize)
})
.await
}
pub async fn metadata(op: FsOp) -> io::Result<RawMetadata> {
let FsOp::Metadata {
target,
follow_symlinks,
} = op
else {
unreachable!("metadata backend called with non-metadata op");
};
offload(move || {
let metadata = match target {
MetadataTarget::Path(path) => {
if follow_symlinks {
std::fs::metadata(path)
} else {
std::fs::symlink_metadata(path)
}
}
MetadataTarget::File(fd) => {
let mut stat = unsafe { std::mem::zeroed::<libc::stat>() };
let result = unsafe { libc::fstat(fd, &mut stat) };
if result < 0 {
return Err(io::Error::last_os_error());
}
return Ok(raw_metadata_from_stat(&stat));
}
}?;
Ok(raw_metadata_from_std(&metadata))
})
.await
}
pub async fn sync_all(op: FsOp) -> io::Result<()> {
let FsOp::SyncAll { fd } = op else {
unreachable!("sync_all backend called with non-sync_all op");
};
offload(move || cvt(unsafe { libc::fsync(fd) }).map(|_| ())).await
}
pub async fn sync_data(op: FsOp) -> io::Result<()> {
let FsOp::SyncData { fd } = op else {
unreachable!("sync_data backend called with non-sync_data op");
};
offload(move || cvt(unsafe { libc::fcntl(fd, libc::F_FULLFSYNC) }).map(|_| ())).await
}
pub async fn set_len(op: FsOp) -> io::Result<()> {
let FsOp::SetLen { fd, len } = op else {
unreachable!("set_len backend called with non-set_len op");
};
offload(move || cvt(unsafe { libc::ftruncate(fd, len as libc::off_t) }).map(|_| ())).await
}
pub async fn try_clone(op: FsOp) -> io::Result<OwnedFd> {
let FsOp::Duplicate { fd } = op else {
unreachable!("try_clone backend called with non-duplicate op");
};
offload(move || {
let duplicated = cvt(unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 0) })?;
Ok(unsafe { OwnedFd::from_raw_fd(duplicated) })
})
.await
}
pub async fn create_dir(op: FsOp) -> io::Result<()> {
let FsOp::CreateDir { path, mode } = op else {
unreachable!("create_dir backend called with non-create_dir op");
};
offload(move || {
let c_path = path_to_c_string(path)?;
cvt(unsafe { libc::mkdir(c_path.as_ptr(), mode as libc::mode_t) }).map(|_| ())
})
.await
}
pub async fn remove_file(op: FsOp) -> io::Result<()> {
let FsOp::RemoveFile { path } = op else {
unreachable!("remove_file backend called with non-remove_file op");
};
offload(move || std::fs::remove_file(path)).await
}
pub async fn remove_dir(op: FsOp) -> io::Result<()> {
let FsOp::RemoveDir { path } = op else {
unreachable!("remove_dir backend called with non-remove_dir op");
};
offload(move || std::fs::remove_dir(path)).await
}
pub async fn rename(op: FsOp) -> io::Result<()> {
let FsOp::Rename { from, to } = op else {
unreachable!("rename backend called with non-rename op");
};
offload(move || std::fs::rename(from, to)).await
}
pub async fn close(op: FsOp) -> io::Result<()> {
let FsOp::Close { fd } = op else {
unreachable!("close backend called with non-close op");
};
offload(move || cvt(unsafe { libc::close(fd) }).map(|_| ())).await
}
pub fn read_dir(op: FsOp) -> io::Result<ReadDirStream> {
let FsOp::ReadDir { path } = op else {
unreachable!("read_dir backend called with non-read_dir op");
};
ReadDirStream::new(path)
}
pub struct ReadDirStream {
state: Arc<ReadDirState>,
}
impl ReadDirStream {
fn new(path: PathBuf) -> io::Result<Self> {
let state = Arc::new(ReadDirState::new(current_thread_handle()));
let producer = Arc::clone(&state);
if let Err(error) = thread::Builder::new()
.name("ruin-runtime-read-dir".into())
.spawn(move || produce_dir_entries(path, producer))
{
state.release_pending();
return Err(io::Error::other(error));
}
Ok(Self { state })
}
pub async fn next_entry(&mut self) -> io::Result<Option<RawDirEntry>> {
poll_fn(|cx| self.state.poll_next(cx)).await
}
}
struct ReadDirState {
owner: ThreadHandle,
queue: Mutex<VecDeque<io::Result<RawDirEntry>>>,
done: AtomicBool,
pending: AtomicBool,
wake_queued: AtomicBool,
waker: Mutex<Option<Waker>>,
}
impl ReadDirState {
fn new(owner: ThreadHandle) -> Self {
owner.begin_async_operation();
Self {
owner,
queue: Mutex::new(VecDeque::new()),
done: AtomicBool::new(false),
pending: AtomicBool::new(true),
wake_queued: AtomicBool::new(false),
waker: Mutex::new(None),
}
}
fn push(self: &Arc<Self>, entry: io::Result<RawDirEntry>) {
self.queue.lock().unwrap().push_back(entry);
self.notify();
}
fn finish(self: &Arc<Self>) {
self.done.store(true, Ordering::Release);
self.release_pending();
self.notify();
}
fn release_pending(&self) {
if self.pending.swap(false, Ordering::AcqRel) {
self.owner.finish_async_operation();
}
}
fn notify(self: &Arc<Self>) {
if self.wake_queued.swap(true, Ordering::AcqRel) {
return;
}
let state = Arc::clone(self);
if !self.owner.queue_task(move || {
state.wake_queued.store(false, Ordering::Release);
if let Some(waker) = state.waker.lock().unwrap().take() {
waker.wake();
}
}) {
self.wake_queued.store(false, Ordering::Release);
}
}
fn poll_next(&self, cx: &mut Context<'_>) -> Poll<io::Result<Option<RawDirEntry>>> {
if let Some(entry) = self.queue.lock().unwrap().pop_front() {
return Poll::Ready(entry.map(Some));
}
if self.done.load(Ordering::Acquire) {
return Poll::Ready(Ok(None));
}
*self.waker.lock().unwrap() = Some(cx.waker().clone());
if let Some(entry) = self.queue.lock().unwrap().pop_front() {
let _ = self.waker.lock().unwrap().take();
return Poll::Ready(entry.map(Some));
}
if self.done.load(Ordering::Acquire) {
let _ = self.waker.lock().unwrap().take();
return Poll::Ready(Ok(None));
}
Poll::Pending
}
}
impl Drop for ReadDirStream {
fn drop(&mut self) {
self.state.release_pending();
}
}
fn produce_dir_entries(path: PathBuf, state: Arc<ReadDirState>) {
match std::fs::read_dir(path) {
Ok(entries) => {
for entry in entries {
match entry {
Ok(entry) => {
let file_name = entry.file_name();
state.push(Ok(RawDirEntry {
path: entry.path(),
file_name,
}));
}
Err(error) => state.push(Err(error)),
}
}
state.finish();
}
Err(error) => {
state.push(Err(error));
state.finish();
}
}
}
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>>();
thread::spawn(move || {
handle.complete(work());
});
future.await
}
fn path_to_c_string(path: PathBuf) -> io::Result<CString> {
CString::new(path.as_os_str().as_bytes()).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidInput,
"path contains interior NUL bytes",
)
})
}
fn raw_metadata_from_std(metadata: &std::fs::Metadata) -> RawMetadata {
let file_type = metadata.file_type();
let kind = if file_type.is_file() {
FileType::File
} else if file_type.is_dir() {
FileType::Directory
} else if file_type.is_symlink() {
FileType::Symlink
} else if file_type.is_block_device() {
FileType::BlockDevice
} else if file_type.is_char_device() {
FileType::CharacterDevice
} else if file_type.is_fifo() {
FileType::Fifo
} else if file_type.is_socket() {
FileType::Socket
} else {
FileType::Unknown
};
RawMetadata {
file_type: kind,
mode: (metadata.mode() & 0o7777) as u16,
len: metadata.len(),
}
}
fn raw_metadata_from_stat(stat: &libc::stat) -> RawMetadata {
let kind = match stat.st_mode & libc::S_IFMT {
libc::S_IFREG => FileType::File,
libc::S_IFDIR => FileType::Directory,
libc::S_IFLNK => FileType::Symlink,
libc::S_IFBLK => FileType::BlockDevice,
libc::S_IFCHR => FileType::CharacterDevice,
libc::S_IFIFO => FileType::Fifo,
libc::S_IFSOCK => FileType::Socket,
_ => FileType::Unknown,
};
RawMetadata {
file_type: kind,
mode: stat.st_mode & 0o7777,
len: stat.st_size as u64,
}
}
fn cvt(value: libc::c_int) -> io::Result<libc::c_int> {
if value < 0 {
Err(io::Error::last_os_error())
} else {
Ok(value)
}
}

6
lib/runtime/src/sys/macos/mod.rs Normal file
View File

@@ -0,0 +1,6 @@
//! macOS backend modules.
pub mod channel;
pub mod fd;
pub mod fs;
pub mod net;

779
lib/runtime/src/sys/macos/net.rs Normal file
View File

@@ -0,0 +1,779 @@
//! macOS networking backend.
use std::ffi::c_void;
use std::future::Future;
use std::io;
use std::mem::MaybeUninit;
use std::net::{
Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, SocketAddrV4, SocketAddrV6, ToSocketAddrs,
};
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
use std::pin::Pin;
use std::thread;
use std::time::Duration;
use crate::op::completion::completion_for_current_thread;
use crate::op::net::{AcceptedSocket, NetOp, ReceivedDatagram};
const DEFAULT_LISTENER_BACKLOG: i32 = 1024;
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>>;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ExecutionPath {
Offload,
}
pub fn execution_path(_op: &NetOp) -> ExecutionPath {
ExecutionPath::Offload
}
pub async fn resolve_addrs<A>(addr: A) -> io::Result<Vec<SocketAddr>>
where
A: ToSocketAddrs + Send + 'static,
{
offload(move || {
let addrs = addr.to_socket_addrs()?.collect::<Vec<_>>();
if addrs.is_empty() {
Err(io::Error::new(
io::ErrorKind::InvalidInput,
"address resolved to no socket addresses",
))
} else {
Ok(addrs)
}
})
.await
}
pub async fn socket(op: NetOp) -> io::Result<OwnedFd> {
let NetOp::Socket {
domain,
socket_type,
protocol,
flags,
} = op
else {
unreachable!("socket backend called with non-socket op");
};
offload(move || socket_sync(domain, socket_type, protocol, flags)).await
}
pub async fn connect(op: NetOp) -> io::Result<()> {
let NetOp::Connect { fd, addr } = op else {
unreachable!("connect backend called with non-connect op");
};
offload(move || connect_sync(fd, RawSocketAddr::from_socket_addr(addr))).await
}
pub async fn bind(op: NetOp) -> io::Result<()> {
let NetOp::Bind { fd, addr } = op else {
unreachable!("bind backend called with non-bind op");
};
offload(move || bind_sync(fd, RawSocketAddr::from_socket_addr(addr))).await
}
pub async fn listen(op: NetOp) -> io::Result<()> {
let NetOp::Listen { fd, backlog } = op else {
unreachable!("listen backend called with non-listen op");
};
offload(move || listen_sync(fd, backlog)).await
}
pub async fn accept(op: NetOp) -> io::Result<AcceptedSocket> {
let NetOp::Accept { fd } = op else {
unreachable!("accept backend called with non-accept op");
};
offload(move || accept_sync(fd)).await
}
pub async fn send(op: NetOp) -> io::Result<usize> {
let NetOp::Send { fd, data, flags } = op else {
unreachable!("send backend called with non-send op");
};
offload(move || send_sync(fd, data, flags)).await
}
pub async fn send_to(op: NetOp) -> io::Result<usize> {
let NetOp::SendTo {
fd,
target,
data,
flags,
} = op
else {
unreachable!("send_to backend called with non-send_to op");
};
offload(move || send_to_sync(fd, target, data, flags)).await
}
pub async fn recv(op: NetOp) -> io::Result<Vec<u8>> {
let NetOp::Recv { fd, len, flags } = op else {
unreachable!("recv backend called with non-recv op");
};
offload(move || recv_sync(fd, len, flags)).await
}
pub async fn recv_from(op: NetOp) -> io::Result<ReceivedDatagram> {
let NetOp::RecvFrom { fd, len, flags } = op else {
unreachable!("recv_from backend called with non-recv_from op");
};
offload(move || recv_from_sync(fd, len, flags)).await
}
pub async fn shutdown(op: NetOp) -> io::Result<()> {
let NetOp::Shutdown { fd, how } = op else {
unreachable!("shutdown backend called with non-shutdown op");
};
offload(move || shutdown_sync(fd, how)).await
}
pub async fn close(op: NetOp) -> io::Result<()> {
let NetOp::Close { fd } = op else {
unreachable!("close backend called with non-close op");
};
offload(move || close_sync(fd)).await
}
pub async fn connect_stream(addr: SocketAddr) -> io::Result<OwnedFd> {
match connect_stream_inner(addr).await {
Err(error) if should_try_ipv4_loopback(addr, &error) => {
connect_stream_inner(localhost_v4(addr)).await
}
result => result,
}
}
pub async fn bind_listener(addr: SocketAddr, backlog: Option<i32>) -> io::Result<OwnedFd> {
match bind_listener_inner(addr, backlog).await {
Err(error) if should_try_ipv4_loopback(addr, &error) => {
bind_listener_inner(localhost_v4(addr), backlog).await
}
result => result,
}
}
pub async fn bind_datagram(addr: SocketAddr) -> io::Result<OwnedFd> {
match bind_datagram_inner(addr).await {
Err(error) if should_try_ipv4_loopback(addr, &error) => {
bind_datagram_inner(localhost_v4(addr)).await
}
result => result,
}
}
async fn connect_stream_inner(addr: SocketAddr) -> io::Result<OwnedFd> {
let stream = socket(NetOp::Socket {
domain: socket_domain(addr),
socket_type: libc::SOCK_STREAM,
protocol: 0,
flags: 0,
})
.await?;
connect(NetOp::Connect {
fd: stream.as_raw_fd(),
addr,
})
.await?;
Ok(stream)
}
async fn bind_listener_inner(addr: SocketAddr, backlog: Option<i32>) -> io::Result<OwnedFd> {
let listener = socket(NetOp::Socket {
domain: socket_domain(addr),
socket_type: libc::SOCK_STREAM,
protocol: 0,
flags: 0,
})
.await?;
set_reuse_addr(listener.as_raw_fd(), true)?;
bind(NetOp::Bind {
fd: listener.as_raw_fd(),
addr,
})
.await?;
listen(NetOp::Listen {
fd: listener.as_raw_fd(),
backlog: backlog.unwrap_or(DEFAULT_LISTENER_BACKLOG),
})
.await?;
Ok(listener)
}
async fn bind_datagram_inner(addr: SocketAddr) -> io::Result<OwnedFd> {
let socket = socket(NetOp::Socket {
domain: socket_domain(addr),
socket_type: libc::SOCK_DGRAM,
protocol: 0,
flags: 0,
})
.await?;
bind(NetOp::Bind {
fd: socket.as_raw_fd(),
addr,
})
.await?;
Ok(socket)
}
pub async fn duplicate(fd: RawFd) -> io::Result<OwnedFd> {
offload(move || {
let duplicated = cvt(unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 0) })?;
Ok(unsafe { OwnedFd::from_raw_fd(duplicated) })
})
.await
}
pub async fn recv_timeout(
fd: RawFd,
len: usize,
flags: i32,
timeout: Duration,
) -> io::Result<Vec<u8>> {
offload(move || {
wait_for_fd(fd, libc::POLLIN, timeout)?;
recv_sync(fd, len, flags)
})
.await
}
pub async fn send_timeout(
fd: RawFd,
data: Vec<u8>,
flags: i32,
timeout: Duration,
) -> io::Result<usize> {
offload(move || {
wait_for_fd(fd, libc::POLLOUT, timeout)?;
send_sync(fd, data, flags)
})
.await
}
pub async fn recv_from_timeout(
fd: RawFd,
len: usize,
flags: i32,
timeout: Duration,
) -> io::Result<ReceivedDatagram> {
offload(move || {
wait_for_fd(fd, libc::POLLIN, timeout)?;
recv_from_sync(fd, len, flags)
})
.await
}
pub async fn send_to_timeout(
fd: RawFd,
data: Vec<u8>,
target: SocketAddr,
flags: i32,
timeout: Duration,
) -> io::Result<usize> {
offload(move || {
wait_for_fd(fd, libc::POLLOUT, timeout)?;
send_to_sync(fd, target, data, flags)
})
.await
}
pub async fn connect_stream_timeout(addr: SocketAddr, timeout: Duration) -> io::Result<OwnedFd> {
offload(move || connect_stream_timeout_sync(addr, timeout)).await
}
pub fn local_addr(fd: RawFd) -> io::Result<SocketAddr> {
socket_addr_with(libc::getsockname, fd)
}
pub fn peer_addr(fd: RawFd) -> io::Result<SocketAddr> {
socket_addr_with(libc::getpeername, fd)
}
pub fn nodelay(fd: RawFd) -> io::Result<bool> {
let mut value = 0;
let mut len = std::mem::size_of::<libc::c_int>() as libc::socklen_t;
cvt(unsafe {
libc::getsockopt(
fd,
libc::IPPROTO_TCP,
libc::TCP_NODELAY,
&mut value as *mut libc::c_int as *mut c_void,
&mut len,
)
})?;
Ok(value != 0)
}
pub fn broadcast(fd: RawFd) -> io::Result<bool> {
getsockopt_int(fd, libc::SOL_SOCKET, libc::SO_BROADCAST).map(|value| value != 0)
}
pub fn set_broadcast(fd: RawFd, enabled: bool) -> io::Result<()> {
setsockopt_int(fd, libc::SOL_SOCKET, libc::SO_BROADCAST, enabled.into())
}
pub fn ttl(fd: RawFd) -> io::Result<u32> {
match socket_family(fd)? {
libc::AF_INET => {
getsockopt_int(fd, libc::IPPROTO_IP, libc::IP_TTL).map(|value| value as u32)
}
libc::AF_INET6 => getsockopt_int(fd, libc::IPPROTO_IPV6, libc::IPV6_UNICAST_HOPS)
.map(|value| value as u32),
family => Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("unsupported socket family {family} for TTL"),
)),
}
}
pub fn set_ttl(fd: RawFd, ttl: u32) -> io::Result<()> {
let ttl = i32::try_from(ttl)
.map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "TTL exceeds i32 range"))?;
match socket_family(fd)? {
libc::AF_INET => setsockopt_int(fd, libc::IPPROTO_IP, libc::IP_TTL, ttl),
libc::AF_INET6 => setsockopt_int(fd, libc::IPPROTO_IPV6, libc::IPV6_UNICAST_HOPS, ttl),
family => Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("unsupported socket family {family} for TTL"),
)),
}
}
pub fn set_nodelay(fd: RawFd, enabled: bool) -> io::Result<()> {
let value: libc::c_int = enabled.into();
cvt(unsafe {
libc::setsockopt(
fd,
libc::IPPROTO_TCP,
libc::TCP_NODELAY,
&value as *const libc::c_int as *const c_void,
std::mem::size_of_val(&value) as libc::socklen_t,
)
})
.map(|_| ())
}
pub fn recv_future(fd: RawFd, len: usize) -> RecvFuture {
Box::pin(recv(NetOp::Recv { fd, len, flags: 0 }))
}
pub fn send_future(fd: RawFd, data: Vec<u8>) -> SendFuture {
Box::pin(send(NetOp::Send { fd, data, flags: 0 }))
}
pub fn shutdown_future(fd: RawFd, how: Shutdown) -> ShutdownFuture {
Box::pin(shutdown(NetOp::Shutdown { fd, how }))
}
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)));
}
future.await
}
fn socket_domain(addr: SocketAddr) -> i32 {
match addr {
SocketAddr::V4(_) => libc::AF_INET,
SocketAddr::V6(_) => libc::AF_INET6,
}
}
fn shutdown_how(how: Shutdown) -> i32 {
match how {
Shutdown::Read => libc::SHUT_RD,
Shutdown::Write => libc::SHUT_WR,
Shutdown::Both => libc::SHUT_RDWR,
}
}
fn socket_addr_with(
op: unsafe extern "C" fn(RawFd, *mut libc::sockaddr, *mut libc::socklen_t) -> libc::c_int,
fd: RawFd,
) -> io::Result<SocketAddr> {
let mut storage = MaybeUninit::<libc::sockaddr_storage>::zeroed();
let mut len = std::mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
cvt(unsafe { op(fd, storage.as_mut_ptr().cast::<libc::sockaddr>(), &mut len) })?;
let storage = unsafe { storage.assume_init() };
socket_addr_from_storage(&storage, len)
}
fn set_reuse_addr(fd: RawFd, enabled: bool) -> io::Result<()> {
setsockopt_int(fd, libc::SOL_SOCKET, libc::SO_REUSEADDR, enabled.into())
}
fn socket_family(fd: RawFd) -> io::Result<i32> {
let mut storage = MaybeUninit::<libc::sockaddr_storage>::zeroed();
let mut len = std::mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
cvt(unsafe { libc::getsockname(fd, storage.as_mut_ptr().cast::<libc::sockaddr>(), &mut len) })?;
let storage = unsafe { storage.assume_init() };
Ok(storage.ss_family as i32)
}
fn getsockopt_int(fd: RawFd, level: i32, name: i32) -> io::Result<i32> {
let mut value = 0;
let mut len = std::mem::size_of::<libc::c_int>() as libc::socklen_t;
cvt(unsafe {
libc::getsockopt(
fd,
level,
name,
&mut value as *mut libc::c_int as *mut c_void,
&mut len,
)
})?;
Ok(value)
}
fn setsockopt_int(fd: RawFd, level: i32, name: i32, value: i32) -> io::Result<()> {
cvt(unsafe {
libc::setsockopt(
fd,
level,
name,
&value as *const libc::c_int as *const c_void,
std::mem::size_of_val(&value) as libc::socklen_t,
)
})
.map(|_| ())
}
fn socket_addr_from_storage(
storage: &libc::sockaddr_storage,
len: libc::socklen_t,
) -> io::Result<SocketAddr> {
match storage.ss_family as i32 {
libc::AF_INET => {
if len < std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"sockaddr_in length is truncated",
));
}
let addr = unsafe { *(storage as *const _ as *const libc::sockaddr_in) };
Ok(SocketAddr::V4(SocketAddrV4::new(
Ipv4Addr::from(u32::from_be(addr.sin_addr.s_addr)),
u16::from_be(addr.sin_port),
)))
}
libc::AF_INET6 => {
if len < std::mem::size_of::<libc::sockaddr_in6>() as libc::socklen_t {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"sockaddr_in6 length is truncated",
));
}
let addr = unsafe { *(storage as *const _ as *const libc::sockaddr_in6) };
Ok(SocketAddr::V6(SocketAddrV6::new(
Ipv6Addr::from(addr.sin6_addr.s6_addr),
u16::from_be(addr.sin6_port),
addr.sin6_flowinfo,
addr.sin6_scope_id,
)))
}
family => Err(io::Error::new(
io::ErrorKind::InvalidData,
format!("unsupported socket family {family}"),
)),
}
}
#[derive(Clone, Copy)]
struct RawSocketAddr {
storage: libc::sockaddr_storage,
len: libc::socklen_t,
}
impl RawSocketAddr {
fn from_socket_addr(addr: SocketAddr) -> Self {
match addr {
SocketAddr::V4(addr) => {
let sockaddr = libc::sockaddr_in {
sin_len: std::mem::size_of::<libc::sockaddr_in>() as u8,
sin_family: libc::AF_INET as libc::sa_family_t,
sin_port: addr.port().to_be(),
sin_addr: libc::in_addr {
s_addr: u32::from_ne_bytes(addr.ip().octets()),
},
sin_zero: [0; 8],
};
let mut storage =
unsafe { MaybeUninit::<libc::sockaddr_storage>::zeroed().assume_init() };
unsafe {
std::ptr::write(
&mut storage as *mut libc::sockaddr_storage as *mut libc::sockaddr_in,
sockaddr,
);
}
Self {
storage,
len: std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t,
}
}
SocketAddr::V6(addr) => {
let sockaddr = libc::sockaddr_in6 {
sin6_len: std::mem::size_of::<libc::sockaddr_in6>() as u8,
sin6_family: libc::AF_INET6 as libc::sa_family_t,
sin6_port: addr.port().to_be(),
sin6_flowinfo: addr.flowinfo(),
sin6_addr: libc::in6_addr {
s6_addr: addr.ip().octets(),
},
sin6_scope_id: addr.scope_id(),
};
let mut storage =
unsafe { MaybeUninit::<libc::sockaddr_storage>::zeroed().assume_init() };
unsafe {
std::ptr::write(
&mut storage as *mut libc::sockaddr_storage as *mut libc::sockaddr_in6,
sockaddr,
);
}
Self {
storage,
len: std::mem::size_of::<libc::sockaddr_in6>() as libc::socklen_t,
}
}
}
}
fn as_ptr(&self) -> *const libc::sockaddr {
&self.storage as *const libc::sockaddr_storage as *const libc::sockaddr
}
fn len(&self) -> libc::socklen_t {
self.len
}
}
fn socket_sync(domain: i32, socket_type: i32, protocol: i32, _flags: u32) -> io::Result<OwnedFd> {
let fd = cvt(unsafe { libc::socket(domain, socket_type, protocol) })?;
set_cloexec(fd)?;
Ok(unsafe { OwnedFd::from_raw_fd(fd) })
}
fn connect_sync(fd: RawFd, addr: RawSocketAddr) -> io::Result<()> {
cvt(unsafe { libc::connect(fd, addr.as_ptr(), addr.len()) }).map(|_| ())
}
fn bind_sync(fd: RawFd, addr: RawSocketAddr) -> io::Result<()> {
cvt(unsafe { libc::bind(fd, addr.as_ptr(), addr.len()) }).map(|_| ())
}
fn listen_sync(fd: RawFd, backlog: i32) -> io::Result<()> {
cvt(unsafe { libc::listen(fd, backlog) }).map(|_| ())
}
fn accept_sync(fd: RawFd) -> io::Result<AcceptedSocket> {
let mut storage = MaybeUninit::<libc::sockaddr_storage>::zeroed();
let mut len = std::mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
let accepted_fd =
cvt(unsafe { libc::accept(fd, storage.as_mut_ptr().cast::<libc::sockaddr>(), &mut len) })?;
set_cloexec(accepted_fd)?;
let storage = unsafe { storage.assume_init() };
let peer_addr = socket_addr_from_storage(&storage, len)?;
Ok(AcceptedSocket {
fd: accepted_fd,
peer_addr,
})
}
fn send_sync(fd: RawFd, data: Vec<u8>, flags: i32) -> io::Result<usize> {
let written = unsafe { libc::send(fd, data.as_ptr().cast::<c_void>(), data.len(), flags) };
cvt_long(written).map(|written| written as usize)
}
fn send_to_sync(fd: RawFd, target: SocketAddr, data: Vec<u8>, flags: i32) -> io::Result<usize> {
let addr = RawSocketAddr::from_socket_addr(target);
let written = unsafe {
libc::sendto(
fd,
data.as_ptr().cast::<c_void>(),
data.len(),
flags,
addr.as_ptr(),
addr.len(),
)
};
cvt_long(written).map(|written| written as usize)
}
fn recv_sync(fd: RawFd, len: usize, flags: i32) -> io::Result<Vec<u8>> {
let mut data = vec![0u8; len];
let read = unsafe { libc::recv(fd, data.as_mut_ptr().cast::<c_void>(), len, flags) };
let read = cvt_long(read)? as usize;
data.truncate(read);
Ok(data)
}
fn recv_from_sync(fd: RawFd, len: usize, flags: i32) -> io::Result<ReceivedDatagram> {
let mut data = vec![0u8; len];
let mut storage = MaybeUninit::<libc::sockaddr_storage>::zeroed();
let mut addr_len = std::mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
let read = unsafe {
libc::recvfrom(
fd,
data.as_mut_ptr().cast::<c_void>(),
len,
flags,
storage.as_mut_ptr().cast::<libc::sockaddr>(),
&mut addr_len,
)
};
let read = cvt_long(read)? as usize;
data.truncate(read);
let storage = unsafe { storage.assume_init() };
let peer_addr = socket_addr_from_storage(&storage, addr_len)?;
Ok(ReceivedDatagram { data, peer_addr })
}
fn shutdown_sync(fd: RawFd, how: Shutdown) -> io::Result<()> {
cvt(unsafe { libc::shutdown(fd, shutdown_how(how)) }).map(|_| ())
}
fn close_sync(fd: RawFd) -> io::Result<()> {
cvt(unsafe { libc::close(fd) }).map(|_| ())
}
fn connect_stream_timeout_sync(addr: SocketAddr, timeout: Duration) -> io::Result<OwnedFd> {
let fd = cvt(unsafe { libc::socket(socket_domain(addr), libc::SOCK_STREAM, 0) })?;
set_cloexec(fd)?;
set_nonblocking(fd)?;
let raw = RawSocketAddr::from_socket_addr(addr);
let connect_result = unsafe { libc::connect(fd, raw.as_ptr(), raw.len()) };
if connect_result < 0 {
let error = io::Error::last_os_error();
if error.raw_os_error() != Some(libc::EINPROGRESS) {
let _ = unsafe { libc::close(fd) };
return Err(error);
}
}
let wait = wait_for_fd(fd, libc::POLLOUT, timeout);
if let Err(error) = wait {
let _ = unsafe { libc::close(fd) };
return Err(error);
}
let mut so_error: libc::c_int = 0;
let mut len = std::mem::size_of::<libc::c_int>() as libc::socklen_t;
if unsafe {
libc::getsockopt(
fd,
libc::SOL_SOCKET,
libc::SO_ERROR,
&mut so_error as *mut libc::c_int as *mut c_void,
&mut len,
)
} < 0
{
let error = io::Error::last_os_error();
let _ = unsafe { libc::close(fd) };
return Err(error);
}
if so_error != 0 {
let _ = unsafe { libc::close(fd) };
return Err(io::Error::from_raw_os_error(so_error));
}
clear_nonblocking(fd)?;
Ok(unsafe { OwnedFd::from_raw_fd(fd) })
}
fn wait_for_fd(fd: RawFd, events: i16, timeout: Duration) -> io::Result<()> {
let timeout_ms = timeout.as_millis().min(i32::MAX as u128) as i32;
let mut pfd = libc::pollfd {
fd,
events,
revents: 0,
};
loop {
let result = unsafe { libc::poll(&mut pfd, 1, timeout_ms) };
if result > 0 {
return Ok(());
}
if result == 0 {
return Err(io::Error::new(
io::ErrorKind::TimedOut,
"operation timed out",
));
}
let error = io::Error::last_os_error();
if error.kind() == io::ErrorKind::Interrupted {
continue;
}
return Err(error);
}
}
fn set_cloexec(fd: RawFd) -> io::Result<()> {
let flags = cvt(unsafe { libc::fcntl(fd, libc::F_GETFD) })?;
cvt(unsafe { libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC) })?;
Ok(())
}
fn set_nonblocking(fd: RawFd) -> io::Result<()> {
let flags = cvt(unsafe { libc::fcntl(fd, libc::F_GETFL) })?;
cvt(unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) })?;
Ok(())
}
fn clear_nonblocking(fd: RawFd) -> io::Result<()> {
let flags = cvt(unsafe { libc::fcntl(fd, libc::F_GETFL) })?;
cvt(unsafe { libc::fcntl(fd, libc::F_SETFL, flags & !libc::O_NONBLOCK) })?;
Ok(())
}
fn should_try_ipv4_loopback(addr: SocketAddr, error: &io::Error) -> bool {
matches!(addr, SocketAddr::V6(v6) if v6.ip().is_loopback())
&& matches!(
error.raw_os_error(),
Some(libc::EADDRNOTAVAIL | libc::EAFNOSUPPORT | libc::ENETUNREACH)
)
}
fn localhost_v4(addr: SocketAddr) -> SocketAddr {
SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, addr.port()))
}
fn cvt(value: libc::c_int) -> io::Result<libc::c_int> {
if value < 0 {
Err(io::Error::last_os_error())
} else {
Ok(value)
}
}
fn cvt_long(value: libc::ssize_t) -> io::Result<libc::ssize_t> {
if value < 0 {
Err(io::Error::last_os_error())
} else {
Ok(value)
}
}

7
lib/runtime/src/sys/mod.rs
View File

@@ -2,3 +2,10 @@
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub mod linux;
#[cfg(all(target_os = "macos", target_arch = "aarch64"))]
pub mod macos;
#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
pub use linux as current;
#[cfg(all(target_os = "macos", target_arch = "aarch64"))]
pub use macos as current;