feat: suspense!

packages/core/src/scheduler/handle.rs

@@ -0,0 +1,31 @@
+use futures_util::Future;
+use slab::Slab;
+use std::{cell::RefCell, pin::Pin, rc::Rc, sync::Arc};
+
+use super::{LocalTask, SchedulerMsg, SuspenseLeaf};
+
+#[derive(Clone)]
+pub struct SchedulerHandle(Rc<HandleInner>);
+
+impl std::ops::Deref for SchedulerHandle {
+    type Target = HandleInner;
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+pub struct HandleInner {
+    pub sender: futures_channel::mpsc::UnboundedSender<SchedulerMsg>,
+    pub tasks: RefCell<Slab<LocalTask>>,
+    pub leaves: RefCell<Slab<SuspenseLeaf>>,
+}
+
+impl SchedulerHandle {
+    pub fn new(sender: futures_channel::mpsc::UnboundedSender<SchedulerMsg>) -> Self {
+        Self(Rc::new(HandleInner {
+            sender,
+            tasks: RefCell::new(Slab::new()),
+            leaves: RefCell::new(Slab::new()),
+        }))
+    }
+}

packages/core/src/scheduler/mod.rs

@@ -0,0 +1,58 @@
+use std::sync::Arc;
+
+use crate::ScopeId;
+
+mod handle;
+mod suspense;
+mod task;
+mod wait;
+
+pub use handle::*;
+use slab::Slab;
+pub use suspense::*;
+pub use task::*;
+
+/// The type of message that can be sent to the scheduler.
+///
+/// These messages control how the scheduler will process updates to the UI.
+#[derive(Debug)]
+pub enum SchedulerMsg {
+    /// Events from the Renderer
+    Event,
+
+    /// Immediate updates from Components that mark them as dirty
+    Immediate(ScopeId),
+
+    /// Mark all components as dirty and update them
+    DirtyAll,
+
+    /// A task has woken and needs to be progressed
+    TaskNotified(TaskId),
+
+    /// A task has woken and needs to be progressed
+    SuspenseNotified(SuspenseId),
+}
+
+pub struct Scheduler {
+    rx: futures_channel::mpsc::UnboundedReceiver<SchedulerMsg>,
+    ready_suspense: Vec<ScopeId>,
+    pub handle: SchedulerHandle,
+}
+
+impl Scheduler {
+    pub fn new() -> Self {
+        let (tx, rx) = futures_channel::mpsc::unbounded();
+        Self {
+            rx,
+            handle: SchedulerHandle::new(tx),
+            ready_suspense: Default::default(),
+        }
+    }
+
+    /// Waits for a future to complete that marks the virtualdom as dirty
+    ///
+    /// Not all messages will mark a virtualdom as dirty, so this waits for a message that has side-effects that do
+    pub fn wait_for_work(&mut self) {
+        //
+    }
+}

packages/core/src/scheduler/suspense.rs

@@ -0,0 +1,71 @@
+use std::{collections::HashSet, rc::Rc};
+
+use futures_task::{RawWaker, RawWakerVTable, Waker};
+use futures_util::Future;
+
+use crate::{innerlude::Mutation, Element, Scope, ScopeId};
+
+use super::SchedulerMsg;
+
+#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
+pub struct SuspenseId(pub usize);
+
+pub type SuspenseContext = Rc<SuspenseBoundary>;
+/// Essentially a fiber in React
+pub struct SuspenseBoundary {
+    pub id: ScopeId,
+    waiting_on: HashSet<SuspenseId>,
+    mutations: Vec<Mutation<'static>>,
+}
+
+impl SuspenseBoundary {
+    pub fn new(id: ScopeId) -> Self {
+        Self {
+            id,
+            waiting_on: Default::default(),
+            mutations: Default::default(),
+        }
+    }
+}
+
+/*
+
+
+many times the future will be ready every time it's polled, so we can spin on it until it doesnt wake us up immediately
+
+
+*/
+pub struct SuspenseLeaf {
+    pub id: SuspenseId,
+    pub scope: ScopeId,
+    pub boundary: ScopeId,
+    pub tx: futures_channel::mpsc::UnboundedSender<SchedulerMsg>,
+
+    pub task: *mut dyn Future<Output = Element<'static>>,
+}
+
+pub fn make_suspense_waker(task: &SuspenseLeaf) -> Waker {
+    let raw = RawWaker::new(task as *const SuspenseLeaf as *const _, task_vtable());
+    unsafe { Waker::from_raw(raw) }
+}
+
+fn task_vtable() -> &'static RawWakerVTable {
+    &RawWakerVTable::new(clone, wake, wake_by_ref, drop_task)
+}
+
+unsafe fn clone(data: *const ()) -> RawWaker {
+    RawWaker::new(data as *const (), task_vtable())
+}
+unsafe fn wake(data: *const ()) {
+    wake_by_ref(data);
+}
+unsafe fn wake_by_ref(data: *const ()) {
+    let task = &*(data as *const SuspenseLeaf);
+    task.tx
+        .unbounded_send(SchedulerMsg::SuspenseNotified(task.id))
+        .expect("Scheduler should exist");
+}
+
+unsafe fn drop_task(_data: *const ()) {
+    // doesnt do anything
+}

packages/core/src/scheduler/task.rs

@@ -0,0 +1,81 @@
+use std::{cell::RefCell, mem, ops::DerefMut, pin::Pin, process::Output, rc::Rc, sync::Arc};
+
+use futures_task::{waker, ArcWake, Context, RawWaker, RawWakerVTable, Waker};
+use futures_util::{pin_mut, Future, FutureExt};
+use slab::Slab;
+
+use crate::ScopeId;
+
+use super::{HandleInner, SchedulerHandle, SchedulerMsg};
+
+#[cfg_attr(feature = "serialize", derive(serde::Serialize, serde::Deserialize))]
+#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
+pub struct TaskId(pub usize);
+
+/// the task itself is the waker
+
+#[derive(Clone)]
+pub struct LocalTask {
+    id: TaskId,
+    scope: ScopeId,
+    tx: futures_channel::mpsc::UnboundedSender<SchedulerMsg>,
+    pub task: *mut dyn Future<Output = ()>,
+}
+
+impl HandleInner {
+    pub fn spawn(&self, scope: ScopeId, task: impl Future<Output = ()> + 'static) -> TaskId {
+        let mut tasks = self.tasks.borrow_mut();
+
+        let entry = tasks.vacant_entry();
+        let task_id = TaskId(entry.key());
+
+        entry.insert(LocalTask {
+            id: task_id,
+            tx: self.sender.clone(),
+            task: Box::into_raw(Box::new(task)),
+            scope,
+        });
+
+        self.sender
+            .unbounded_send(SchedulerMsg::TaskNotified(task_id))
+            .expect("Scheduler should exist");
+
+        task_id
+    }
+
+    // drops the future
+    pub fn remove(&self, id: TaskId) {
+        //
+    }
+
+    // Aborts the future
+    pub fn abort(&self, id: TaskId) {
+        //
+    }
+}
+
+pub fn make_task_waker(task: &LocalTask) -> Waker {
+    let raw = RawWaker::new(task as *const LocalTask as *const _, task_vtable());
+    unsafe { Waker::from_raw(raw) }
+}
+
+fn task_vtable() -> &'static RawWakerVTable {
+    &RawWakerVTable::new(clone, wake, wake_by_ref, drop_task)
+}
+
+unsafe fn clone(data: *const ()) -> RawWaker {
+    RawWaker::new(data as *const (), task_vtable())
+}
+unsafe fn wake(data: *const ()) {
+    wake_by_ref(data);
+}
+unsafe fn wake_by_ref(data: *const ()) {
+    let task = &*(data as *const LocalTask);
+    task.tx
+        .unbounded_send(SchedulerMsg::TaskNotified(task.id))
+        .expect("Scheduler should exist");
+}
+
+unsafe fn drop_task(_data: *const ()) {
+    // doesnt do anything
+}

packages/core/src/scheduler/wait.rs

@@ -0,0 +1,52 @@
+use std::{ops::DerefMut, pin::Pin};
+
+use futures_task::Context;
+use futures_util::StreamExt;
+
+use crate::{innerlude::make_task_waker, VirtualDom};
+
+use super::SchedulerMsg;
+
+impl VirtualDom {
+    /// Wait for futures internal to the virtualdom
+    ///
+    /// This is cancel safe, so if the future is dropped, you can push events into the virtualdom
+    pub async fn wait_for_work(&mut self) {
+        loop {
+            let msg = self.scheduler.rx.next().await.unwrap();
+
+            println!("msg received: {:?}", msg);
+
+            match msg {
+                SchedulerMsg::Event => todo!(),
+                SchedulerMsg::Immediate(_) => todo!(),
+                SchedulerMsg::DirtyAll => todo!(),
+                SchedulerMsg::TaskNotified(id) => {
+                    let mut tasks = self.scheduler.handle.tasks.borrow_mut();
+                    let local_task = &tasks[id.0];
+
+                    // // attach the waker to itself
+                    let waker = make_task_waker(local_task);
+                    let mut cx = Context::from_waker(&waker);
+                    let mut fut = unsafe { &mut *local_task.task };
+
+                    let pinned = unsafe { Pin::new_unchecked(fut.deref_mut()) };
+
+                    match pinned.poll(&mut cx) {
+                        futures_task::Poll::Ready(_) => {
+                            // remove the task
+                            tasks.remove(id.0);
+                        }
+                        futures_task::Poll::Pending => {}
+                    }
+
+                    if tasks.is_empty() {
+                        return;
+                    }
+                }
+                // SchedulerMsg::TaskNotified(id) => {},
+                SchedulerMsg::SuspenseNotified(_) => todo!(),
+            }
+        }
+    }
+}