diff --git a/Cargo.lock b/Cargo.lock index 7db7b52..c3d4b9d 100644 --- a/Cargo.lock +++ b/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" diff --git a/Cargo.toml b/Cargo.toml index a81d56e..6b4208e 100644 --- a/Cargo.toml +++ b/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" diff --git a/src/handle_slab.rs b/src/handle_slab.rs deleted file mode 100644 index e69de29..0000000 diff --git a/src/lib.rs b/src/lib.rs index 0f9ac31..8db554f 100644 --- a/src/lib.rs +++ b/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>>>, } +new_key_type! { struct TaskId; } + pub struct RuntimeInner { timer: timer::TimerHeap, ring: IoUring, - tasks: Slab, - runqueue: Vec, + tasks: SlotMap, + runqueue: Vec, } -pub struct Runtime(Rc>); -struct RuntimeWeak(Weak>); +pub struct Runtime(Rc>); +struct RuntimeWeak(Weak>); thread_local! { - static RT: RefCell> = RefCell::new(None); + static RT: UncheckedRefCell> = 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(); - 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 { + //let submitter = ring.submitter(); + //submitter + //.register_restrictions(&mut [Restriction::register_op(Nop::CODE)]) + //.expect("failed to register restrictions on ring"); + //submitter + //.register_enable_rings() + //.expect("failed to enable ring"); + Self(Rc::new(UncheckedRefCell::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"); - loop { - let Some(task_id) = lock.runqueue.pop() else { - break; - }; - trace!("Polling task {}", task_id); + trace!("Polling {} tasks", lock.tasks.len()); + while let Some(task_id) = lock.runqueue.pop() { + 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) - .build(); + + // TODO we shouldn't need to take task out 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) => { - debug_assert_eq!(n, 0); - } + Ok(_) => {} 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> = 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(); + } } diff --git a/src/op.rs b/src/op.rs new file mode 100644 index 0000000..2e0c5e6 --- /dev/null +++ b/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>); + +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 { + 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>) { + // 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!() + }) + } + } + } + } + } +} +*/ diff --git a/src/timer.rs b/src/timer.rs index 6750a91..93e2e06 100644 --- a/src/timer.rs +++ b/src/timer.rs @@ -103,11 +103,9 @@ impl Drop for Timer { 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(); + slot: RT.with(|rt| { + let rt = rt.borrow(); + let mut lock = rt.as_ref().unwrap().0.as_ref().borrow_mut(); lock.timer.insert(deadline) }), } diff --git a/todo.txt b/todo.txt new file mode 100644 index 0000000..37e97a3 --- /dev/null +++ b/todo.txt @@ -0,0 +1,7 @@ +Minimal wakers +Nop reactor +Sleep reactor +File reactor +Network reactor +Task soundness +Unchecked SQ/CQ without debug_assertions