Runs

.gitignore

@@ -0,0 +1 @@
+/target

Cargo.lock

@@ -0,0 +1,94 @@
+# This file is automatically @generated by Cargo.
+# It is not intended for manual editing.
+version = 4
+
+[[package]]
+name = "bitflags"
+version = "2.13.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06"
+
+[[package]]
+name = "bitvec"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ddcec3d12c579d40898fe0a9a358a803c23e9c52ca3c425707f81c9436211837"
+dependencies = [
+ "funty",
+ "radium",
+ "tap",
+ "wyz",
+]
+
+[[package]]
+name = "cfg-if"
+version = "1.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600"
+
+[[package]]
+name = "curio"
+version = "0.1.0"
+dependencies = [
+ "bitvec",
+ "io-uring",
+ "libc",
+ "send_wrapper",
+ "slab",
+]
+
+[[package]]
+name = "funty"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c"
+
+[[package]]
+name = "io-uring"
+version = "0.7.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed3bd0ecfbb87805f538bb7b32e5239ca0763890c623e349860ecba69469f2bb"
+dependencies = [
+ "bitflags",
+ "cfg-if",
+ "libc",
+]
+
+[[package]]
+name = "libc"
+version = "0.2.189"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
+
+[[package]]
+name = "radium"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09"
+
+[[package]]
+name = "send_wrapper"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd0b0ec5f1c1ca621c432a25813d8d60c88abe6d3e08a3eb9cf37d97a0fe3d73"
+
+[[package]]
+name = "slab"
+version = "0.4.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
+
+[[package]]
+name = "tap"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369"
+
+[[package]]
+name = "wyz"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "05f360fc0b24296329c78fda852a1e9ae82de9cf7b27dae4b7f62f118f77b9ed"
+dependencies = [
+ "tap",
+]

Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "curio"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+bitvec = "1.1.1"
+io-uring = "0.7.15"
+libc = "0.2.189"
+send_wrapper = "0.6.0"
+slab = "0.4.12"

src/lib.rs

@@ -0,0 +1,164 @@
+#![feature(local_waker)]
+#![feature(waker_fn)]
+#![feature(context_ext)]
+
+use std::{
+    cell::RefCell, io::ErrorKind, pin::{Pin, pin}, rc::{Rc, Weak}, task::{Context, ContextBuilder, Poll, Waker, local_waker_fn, waker_fn}, time::Instant,
+};
+
+use io_uring::{IoUring, types::{SubmitArgs, Timespec}};
+use libc::ETIME;
+use send_wrapper::SendWrapper;
+use slab::Slab;
+
+mod timer;
+
+struct Task {
+    ready: bool,
+    fut: Option<Pin<Box<dyn Future<Output = ()>>>>,
+}
+
+pub struct RuntimeInner {
+    timer: timer::TimerHeap,
+    ring: IoUring,
+
+    tasks: Slab<Task>,
+    runqueue: Vec<usize>,
+}
+
+pub struct Runtime(Rc<RefCell<RuntimeInner>>);
+struct RuntimeWeak(Weak<RefCell<RuntimeInner>>);
+
+
+thread_local! {
+    static RT: RefCell<Option<Runtime>> = RefCell::new(None);
+}
+
+impl Runtime {
+    pub fn new() -> Self {
+        let ring = IoUring::builder()
+            .setup_cqsize(256)
+            .setup_r_disabled()
+            .setup_defer_taskrun()
+            .setup_single_issuer()
+            .setup_no_sqarray()
+            .build(128)
+            .expect("failed to create ring");
+        let submitter = ring.submitter();
+        submitter
+            .register_restrictions(&mut [
+            // TODO
+        ])
+            .expect("failed to register restrictions on ring");
+        submitter
+            .register_enable_rings()
+            .expect("failed to enable rings");
+        Self(Rc::new(RefCell::new(RuntimeInner {
+            timer: Default::default(),
+            ring,
+
+            tasks: Default::default(),
+            runqueue: Default::default(),
+        })))
+    }
+
+    pub fn spawn(&self, fut: Pin<Box<dyn Future<Output = ()>>>) {
+        let mut inner = self.0.borrow_mut();
+        let task_id = inner.tasks.insert(Task { fut: Some(fut), ready: true });
+        inner.runqueue.push(task_id);
+    }
+
+    fn downgrade(&self) -> RuntimeWeak {
+        RuntimeWeak(Rc::downgrade(&self.0))
+    }
+
+    pub fn run(&mut self) {
+        RT.set(Some(Runtime(Rc::clone(&self.0))));
+        while !self.0.borrow().tasks.is_empty() {
+            // Run until no ready tasks, no timers, no future CQEs
+
+            loop {
+                let task_id = {
+                    let mut lock = self.0.borrow_mut();
+                    let Some(task_id) = lock.runqueue.pop() else {
+                        break;
+                    };
+                    lock.tasks[task_id].ready = false;
+                    task_id
+                };
+
+                let waker = {
+                    let inner = SendWrapper::new(self.downgrade());
+                    waker_fn(move || {
+                        if let Some(inner) = inner.0.upgrade() {
+                            let mut lock = inner.borrow_mut();
+                            let ready = &mut lock.tasks[task_id].ready;
+                            if !*ready {
+                                *ready = true;
+                                lock.runqueue.push(task_id);
+                            }
+                        }
+                    })
+                };
+                let mut cx = ContextBuilder::from_waker(&waker)
+                    .build();
+                let mut task = self.0.borrow_mut().tasks[task_id].fut.take().unwrap();
+                match task.as_mut().poll(&mut cx) {
+                    Poll::Ready(()) => {
+                        self.0.borrow_mut().tasks.remove(task_id);
+                    }
+                    Poll::Pending => {
+                        self.0.borrow_mut().tasks[task_id].fut = Some(task);
+                    }
+                }
+            }
+
+            let now = Instant::now();
+            let wait = loop {
+                let mut lock = self.0.borrow_mut();
+                match lock.timer.poll(now) {
+                    timer::Status::Empty => break None,
+                    timer::Status::Pending(deadline) => break Some(Timespec::from(deadline - now)),
+                    timer::Status::Expired(waker) => {
+                        drop(lock);
+                        waker.wake();
+                    }
+                }
+            };
+
+            let mut args = SubmitArgs::new();
+            if let Some(wait) = &wait {
+                args = args.timespec(&wait);
+            }
+            match self.0.borrow_mut().ring.submitter().submit_with_args(1, &args) {
+                Ok(n) => {
+                    debug_assert_eq!(n, 0);
+                }
+                Err(err) if err.raw_os_error() == Some(ETIME) => {},
+                e => {
+                    e.expect("failed to enter ring");
+                },
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+use crate::timer::delay;
+
+use super::*;
+
+    #[test]
+    fn it_works() {
+        let mut runtime = Runtime::new();
+        runtime.spawn(Box::pin(async {
+            println!("1");
+            delay(Duration::from_secs(1)).await;
+            println!("2");
+        }));
+        runtime.run();
+    }
+}

src/timer.rs

@@ -0,0 +1,107 @@
+use std::{borrow::BorrowMut as _, cell::RefCell, cmp::Reverse, collections::BinaryHeap, pin::Pin, rc::{Rc, Weak}, task::{Context, LocalWaker, Poll, Waker}, time::{Duration, Instant}};
+
+use slab::Slab;
+
+use crate::RT;
+
+#[derive(PartialEq, Eq, PartialOrd, Ord)]
+struct Entry {
+    deadline: Instant,
+    slot_id: usize,
+}
+
+enum Slot {
+    Live,
+    Armed(LocalWaker),
+    Cancelled,
+    // Potential future optimization: Delayed(Instant, T),
+}
+
+#[derive(Default)]
+pub struct TimerHeap {
+    slots: Slab<Rc<RefCell<Slot>>>,
+    queue: BinaryHeap<Reverse<Entry>>,
+}
+
+pub enum Status {
+    Empty,
+    Pending(Instant),
+    Expired(LocalWaker),
+}
+
+impl TimerHeap {
+    pub fn insert(&mut self, deadline: Instant) -> Weak<RefCell<Slot>> {
+        let slot = Rc::new(RefCell::new(Slot::Live));
+        let weak = Rc::downgrade(&slot);
+        let slot_id = self.slots.insert(slot);
+        self.queue.push(Reverse(Entry { deadline, slot_id }));
+        weak
+    }
+
+    pub fn poll(&mut self, now: Instant) -> Status {
+        loop {
+            let slot = match self.queue.peek() {
+                None => break Status::Empty,
+                Some(&Reverse(Entry { deadline, slot_id })) => {
+                    if deadline > now {
+                        // We should check if cancelled here
+                        break Status::Pending(deadline);
+                    }
+                    self.queue.pop().unwrap();
+                    self.slots.remove(slot_id)
+                }
+            };
+            match Rc::into_inner(slot).unwrap().into_inner() {
+                Slot::Live => {}
+                Slot::Armed(waker) => break Status::Expired(waker),
+                Slot::Cancelled => {}
+            }
+        }
+    }
+}
+
+pub struct Timer {
+    slot: Weak<RefCell<Slot>>,
+}
+
+impl Future for Timer {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        match self.slot.upgrade() {
+            None => Poll::Ready(()),
+            Some(slot) => {
+                let mut lock = slot.as_ref().borrow_mut();
+                match &**lock.borrow_mut() {
+                    &Slot::Live | &Slot::Armed(_) => {
+                        *lock = Slot::Armed(cx.local_waker().clone());
+                        Poll::Pending
+                    }
+                    &Slot::Cancelled => Poll::Ready(()),
+                }
+            }
+        }
+    }
+}
+
+impl Drop for Timer {
+    fn drop(&mut self) {
+        if let Some(slot) = self.slot.upgrade() {
+            *slot.as_ref().borrow_mut() = Slot::Cancelled;
+        }
+    }
+}
+
+pub fn delay_until(deadline: Instant) -> Timer {
+    Timer {
+        slot: RT.with_borrow(|rt| {
+            let rt = rt.as_ref().expect("called from outside curio runtime thread");
+            let mut lock = rt.0.as_ref().borrow_mut();
+            lock.timer.insert(deadline)
+        }),
+    }
+}
+
+pub fn delay(duration: Duration) -> Timer {
+    delay_until(Instant::now() + duration)
+}