From 5bb857dabdbec02db94182b8957630cbc27c0aa9 Mon Sep 17 00:00:00 2001 From: Kenneth Allen Date: Fri, 25 Sep 2026 13:44:59 +1000 Subject: [PATCH] Runs --- .gitignore | 1 + Cargo.lock | 94 ++++++++++++++++++++++++++ Cargo.toml | 11 +++ src/handle_slab.rs | 0 src/lib.rs | 164 +++++++++++++++++++++++++++++++++++++++++++++ src/timer.rs | 107 +++++++++++++++++++++++++++++ 6 files changed, 377 insertions(+) create mode 100644 .gitignore create mode 100644 Cargo.lock create mode 100644 Cargo.toml create mode 100644 src/handle_slab.rs create mode 100644 src/lib.rs create mode 100644 src/timer.rs diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ea8c4bf --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/target diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000..870c75e --- /dev/null +++ b/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", +] diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..e57a740 --- /dev/null +++ b/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" diff --git a/src/handle_slab.rs b/src/handle_slab.rs new file mode 100644 index 0000000..e69de29 diff --git a/src/lib.rs b/src/lib.rs new file mode 100644 index 0000000..6c3b8c7 --- /dev/null +++ b/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>>>, +} + +pub struct RuntimeInner { + timer: timer::TimerHeap, + ring: IoUring, + + tasks: Slab, + runqueue: Vec, +} + +pub struct Runtime(Rc>); +struct RuntimeWeak(Weak>); + + +thread_local! { + static RT: RefCell> = 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>>) { + 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(); + } +} diff --git a/src/timer.rs b/src/timer.rs new file mode 100644 index 0000000..fb17107 --- /dev/null +++ b/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>>, + queue: BinaryHeap>, +} + +pub enum Status { + Empty, + Pending(Instant), + Expired(LocalWaker), +} + +impl TimerHeap { + pub fn insert(&mut self, deadline: Instant) -> Weak> { + 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>, +} + +impl Future for Timer { + type Output = (); + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + 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) +}