Ringing

Cargo.lock

@@ -95,6 +95,8 @@ dependencies = [
  "log",
  "send_wrapper",
  "slab",
+ "slotmap",
+ "unchecked-refcell",
 ]
 
 [[package]]
@@ -323,6 +325,15 @@ version = "0.4.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
 
+[[package]]
+name = "slotmap"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bdd58c3c93c3d278ca835519292445cb4b0d4dc59ccfdf7ceadaab3f8aeb4038"
+dependencies = [
+ "version_check",
+]
+
 [[package]]
 name = "syn"
 version = "2.0.119"
@@ -365,6 +376,12 @@ dependencies = [
  "syn 3.0.6",
 ]
 
+[[package]]
+name = "unchecked-refcell"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "05442d1d94026b1d3e8ca10750c1461ce60731aee72c698526574e0e91c6e7a2"
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.26"
@@ -377,6 +394,12 @@ version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
 
+[[package]]
+name = "version_check"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
+
 [[package]]
 name = "windows-link"
 version = "0.2.1"

Cargo.toml

@@ -4,9 +4,13 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
-env_logger = "0.11.11"
 io-uring = "0.7.15"
 libc = "0.2.189"
 log = "0.4.34"
 send_wrapper = "0.6.0"
 slab = "0.4.12"
+slotmap = "1.1.1"
+unchecked-refcell = "0.2.2"
+
+[dev-dependencies]
+env_logger = "0.11.11"

src/lib.rs

@@ -1,19 +1,31 @@
 #![feature(local_waker)]
 #![feature(waker_fn)]
+#![feature(context_ext)]
 
 use std::{
-    cell::RefCell, pin::Pin, rc::{Rc, Weak}, task::{ContextBuilder, Poll, local_waker_fn, waker_fn}, thread::sleep, time::Instant,
+    ops::DerefMut as _,
+    pin::Pin,
+    ptr,
+    rc::{Rc, Weak},
+    task::{ContextBuilder, Poll, local_waker_fn, waker_fn},
+    time::Instant,
 };
 
 use io_uring::{
     IoUring,
+    opcode::Nop,
+    register::Restriction,
     types::{SubmitArgs, Timespec},
 };
-use libc::{ETIME};
+use libc::ETIME;
 use log::trace;
 use send_wrapper::SendWrapper;
-use slab::Slab;
+use slotmap::{SlotMap, new_key_type};
+use unchecked_refcell::UncheckedRefCell;
 
+use crate::op::Op;
+
+pub mod op;
 mod timer;
 
 struct Task {
@@ -21,41 +33,41 @@ struct Task {
     fut: Option<Pin<Box<dyn Future<Output = ()>>>>,
 }
 
+new_key_type! { struct TaskId; }
+
 pub struct RuntimeInner {
     timer: timer::TimerHeap,
     ring: IoUring,
 
-    tasks: Slab<Task>,
+    tasks: SlotMap<TaskId, Task>,
-    runqueue: Vec<usize>,
+    runqueue: Vec<TaskId>,
 }
 
-pub struct Runtime(Rc<RefCell<RuntimeInner>>);
+pub struct Runtime(Rc<UncheckedRefCell<RuntimeInner>>);
-struct RuntimeWeak(Weak<RefCell<RuntimeInner>>);
+struct RuntimeWeak(Weak<UncheckedRefCell<RuntimeInner>>);
 
 thread_local! {
-    static RT: RefCell<Option<Runtime>> = RefCell::new(None);
+    static RT: UncheckedRefCell<Option<Runtime>> = const { UncheckedRefCell::new(None) };
 }
 
 impl Runtime {
     pub fn new() -> Self {
         let ring = IoUring::builder()
             .setup_cqsize(256)
-            .setup_r_disabled()
+            //.setup_r_disabled()
             .setup_defer_taskrun()
             .setup_single_issuer()
             .setup_no_sqarray()
             .build(128)
             .expect("failed to create ring");
-        let submitter = ring.submitter();
+        //let submitter = ring.submitter();
-        submitter
+        //submitter
-            .register_restrictions(&mut [
+        //.register_restrictions(&mut [Restriction::register_op(Nop::CODE)])
-            // TODO
+        //.expect("failed to register restrictions on ring");
-        ])
+        //submitter
-            .expect("failed to register restrictions on ring");
+        //.register_enable_rings()
-        submitter
+        //.expect("failed to enable ring");
-            .register_enable_rings()
+        Self(Rc::new(UncheckedRefCell::new(RuntimeInner {
-            .expect("failed to enable rings");
-        Self(Rc::new(RefCell::new(RuntimeInner {
             timer: Default::default(),
             ring,
 
@@ -78,28 +90,32 @@ impl Runtime {
     }
 
     pub fn run(&mut self) {
-        RT.set(Some(Runtime(Rc::clone(&self.0))));
+        RT.with(|cell| {
+            let mut lock = cell.borrow_mut();
+            if lock.is_some() {
+                panic!("runtime already initialized on this thread");
+            }
+            *lock = Some(Runtime(Rc::clone(&self.0)))
+        });
+
         let mut lock = self.0.borrow_mut();
         while !lock.tasks.is_empty() {
             // Run until no ready tasks, no timers, no future CQEs
 
-            trace!("Polling tasks");
+            trace!("Polling {} tasks", lock.tasks.len());
-            loop {
+            while let Some(task_id) = lock.runqueue.pop() {
-                let Some(task_id) = lock.runqueue.pop() else {
+                trace!("Polling task {:?}", task_id);
-                    break;
-                };
-                trace!("Polling task {}", task_id);
                 lock.tasks[task_id].ready = false;
 
                 let waker = {
                     let inner = SendWrapper::new(self.downgrade());
                     waker_fn(move || {
                         if let Some(inner) = inner.0.upgrade() {
-                            trace!("Waking {}", task_id);
+                            trace!("Waking {:?}", task_id);
                             let mut lock = inner.borrow_mut();
                             let ready = &mut lock.tasks[task_id].ready;
                             if !*ready {
-                                trace!("Enqueueing {}", task_id);
+                                trace!("Enqueueing {:?}", task_id);
                                 *ready = true;
                                 lock.runqueue.push(task_id);
                             }
@@ -110,24 +126,29 @@ impl Runtime {
                     let inner = SendWrapper::new(self.downgrade());
                     local_waker_fn(move || {
                         if let Some(inner) = inner.0.upgrade() {
-                            trace!("Local waking {}", task_id);
+                            trace!("Local waking {:?}", task_id);
                             let mut lock = inner.borrow_mut();
                             let ready = &mut lock.tasks[task_id].ready;
                             if !*ready {
-                                trace!("Local enqueueing {}", task_id);
+                                trace!("Local enqueueing {:?}", task_id);
                                 *ready = true;
                                 lock.runqueue.push(task_id);
                             }
                         }
                     })
                 };
-                let mut cx = ContextBuilder::from_waker(&waker)
+
-                    .local_waker(&local_waker)
+                // TODO we shouldn't need to take task out
-                    .build();
                 let mut task = lock.tasks[task_id].fut.take().unwrap();
                 drop(lock);
+                let mut cx = ContextBuilder::from_waker(&waker)
+                    .local_waker(&local_waker)
+                    .ext(self)
+                    .build();
+                // Unlock runtime while running user code
                 let outcome = task.as_mut().poll(&mut cx);
                 lock = self.0.borrow_mut();
+
                 match outcome {
                     Poll::Ready(()) => {
                         lock.tasks.remove(task_id);
@@ -169,14 +190,26 @@ impl Runtime {
                 None => 0,
             };
             match lock.ring.submitter().submit_with_args(want, &args) {
-                Ok(n) => {
+                Ok(_) => {}
-                    debug_assert_eq!(n, 0);
-                }
                 Err(err) if err.raw_os_error() == Some(ETIME) && wait.is_some() => {}
                 e => {
                     e.expect("failed to enter ring");
                 }
             }
+
+            loop {
+                let entry = lock.ring.completion().next();
+                let Some(cqe) = entry else {
+                    break;
+                };
+                trace!("Processing CQE {:?}", &cqe);
+                debug_assert!(cqe.result() >= 0);
+                let user_data = ptr::with_exposed_provenance(cqe.user_data().try_into().unwrap());
+                let op: Rc<UncheckedRefCell<Op>> = unsafe { Rc::from_raw(user_data) };
+                drop(lock);
+                Op::handle(op);
+                lock = self.0.borrow_mut();
+            }
         }
         trace!("Exiting");
     }
@@ -186,19 +219,47 @@ impl Runtime {
 mod tests {
     use std::time::Duration;
 
-    use crate::timer::delay;
+    use crate::{op::no_op, timer::delay};
 
     use super::*;
 
+    fn init() {
+        let _ = env_logger::builder().is_test(true).try_init();
+    }
+
     #[test]
     fn it_works() {
-        env_logger::init();
+        init();
         let mut runtime = Runtime::new();
         runtime.spawn(Box::pin(async {
             println!("1");
-            delay(Duration::from_secs(1)).await;
+            no_op().await;
             println!("2");
+            delay(Duration::from_secs(1)).await;
+            println!("3");
+            no_op().await;
+            println!("4");
         }));
         runtime.run();
     }
+
+    #[test]
+    fn concurrency() {
+        init();
+        let mut runtime = Runtime::new();
+        for x in 0..10 {
+            runtime.spawn(Box::pin(async move {
+                println!("Starting {}", x);
+                delay(Duration::from_millis(x * 100)).await;
+                for y in 0..10 {
+                    println!("Checkin {}ms", y * 1000 + x * 100);
+                    delay(Duration::from_secs(1)).await;
+
+                    no_op().await;
+                }
+                println!("Finishing {}", x);
+            }));
+        }
+        runtime.run();
+    }
 }

src/op.rs

@@ -0,0 +1,102 @@
+use std::{
+    mem,
+    ops::DerefMut as _,
+    pin::Pin,
+    rc::Rc,
+    task::{Context, LocalWaker, Poll},
+};
+
+use io_uring::opcode::Nop;
+use log::trace;
+use unchecked_refcell::UncheckedRefCell;
+
+use crate::{RT, Runtime, RuntimeInner, op::NoOp::Cancelled};
+
+pub struct OpEntry(Rc<UncheckedRefCell<Op>>);
+
+pub enum Op {
+    NoOp(NoOp),
+}
+
+enum NoOp {
+    New,
+    Live(LocalWaker),
+    Done,
+    Cancelled,
+}
+
+pub fn no_op() -> OpEntry {
+    OpEntry(Rc::new(UncheckedRefCell::new(Op::NoOp(NoOp::New))))
+}
+
+impl Future for OpEntry {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let mut lock = self.0.borrow_mut();
+        match &mut *lock {
+            Op::NoOp(NoOp::Cancelled) => unreachable!(),
+            Op::NoOp(NoOp::Done) => Poll::Ready(()),
+            Op::NoOp(op @ NoOp::New) => {
+                {
+                    let rt: &Runtime = cx.ext().downcast_ref().unwrap();
+                    let mut lock = rt.0.borrow_mut();
+                    let mut sq = lock.ring.submission();
+                    let entry = Nop::new().build().user_data(
+                        Rc::into_raw(self.0.clone())
+                            .expose_provenance()
+                            .try_into()
+                            .unwrap(),
+                    );
+                    trace!("Pushing SQE {:?}", &entry);
+                    unsafe { sq.push(&entry) }.expect("failed to submit no-op");
+                }
+                *op = NoOp::Live(cx.local_waker().clone());
+                Poll::Pending
+            }
+            Op::NoOp(NoOp::Live(waker)) => {
+                waker.clone_from(cx.local_waker());
+                Poll::Pending
+            }
+        }
+    }
+}
+
+impl Op {
+    pub fn handle(this: Rc<UncheckedRefCell<Self>>) {
+        // TODO don't drop for multishot accepts
+        let mut lock = this.borrow_mut();
+        match lock.deref_mut() {
+            Op::NoOp(op) => match mem::replace(op, NoOp::Done) {
+                NoOp::New | NoOp::Done => unreachable!(),
+                NoOp::Cancelled => {}
+                NoOp::Live(waker) => waker.wake(),
+            },
+        }
+    }
+}
+
+/*
+impl Drop for OpEntry {
+    fn drop(&mut self) {
+        let mut lock = self.0.borrow_mut();
+        match lock.deref_mut() {
+            Op::NoOp(op) => {
+                match mem::replace(op, NoOp::Cancelled) {
+                    NoOp::New | NoOp::Done => {}
+                    NoOp::Cancelled => unreachable!(),
+                    NoOp::Live(_) => {
+                        RT.with(|rt| {
+                            let outer = rt.borrow();
+                            let mut lock = outer.as_ref().expect("future cancelled outside runtime thread").0.borrow_mut();
+                            let mut sq = lock.ring.submission();
+                            let entry = AsyncCancel::new(self.0.as_ptr().addr().try_into().unwrap()).build();
+                            todo!()
+                        })
+                    }
+                }
+            }
+        }
+    }
+}
+*/

src/timer.rs

@@ -103,11 +103,9 @@ impl Drop for Timer {
 
 pub fn delay_until(deadline: Instant) -> Timer {
     Timer {
-        slot: RT.with_borrow(|rt| {
+        slot: RT.with(|rt| {
-            let rt = rt
+            let rt = rt.borrow();
-                .as_ref()
+            let mut lock = rt.as_ref().unwrap().0.as_ref().borrow_mut();
-                .expect("called from outside curio runtime thread");
-            let mut lock = rt.0.as_ref().borrow_mut();
             lock.timer.insert(deadline)
         }),
     }

todo.txt

@@ -0,0 +1,7 @@
+Minimal wakers
+Nop reactor
+Sleep reactor
+File reactor
+Network reactor
+Task soundness
+Unchecked SQ/CQ without debug_assertions