chore: clean up scheduler code

This commit is contained in:
Jonathan Kelley committed 2022-11-09 10:58:11 -08:00
1 parent c096057dd3
commit f5bc137f01
17 files changed
+439 -294

No files matched your search

-1
View File
@@ -1 +0,0 @@
-35
View File
@@ -1,35 +0,0 @@
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>,
/// Tasks created with cx.spawn
pub tasks: RefCell<Slab<Rc<LocalTask>>>,
/// Async components
pub leaves: RefCell<Slab<Rc<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()),
}))
}
}
+27 -24
View File
@@ -1,16 +1,11 @@
use slab::Slab;
use std::sync::Arc;
use crate::ScopeId;
use slab::Slab;
mod bumpslab;
mod handle;
mod suspense;
mod task;
mod wait;
mod waker;
pub use handle::*;
pub use suspense::*;
pub use task::*;
pub use waker::RcWake;
@@ -36,26 +31,34 @@ pub enum SchedulerMsg {
SuspenseNotified(SuspenseId),
}
pub struct Scheduler {
rx: futures_channel::mpsc::UnboundedReceiver<SchedulerMsg>,
ready_suspense: Vec<ScopeId>,
pub handle: SchedulerHandle,
use std::{cell::RefCell, rc::Rc};
#[derive(Clone)]
pub(crate) struct Scheduler(Rc<HandleInner>);
impl std::ops::Deref for Scheduler {
type Target = HandleInner;
fn deref(&self) -> &Self::Target {
&self.0
}
}
pub struct HandleInner {
pub sender: futures_channel::mpsc::UnboundedSender<SchedulerMsg>,
/// Tasks created with cx.spawn
pub tasks: RefCell<Slab<Rc<LocalTask>>>,
/// Async components
pub leaves: RefCell<Slab<Rc<SuspenseLeaf>>>,
}
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) {
//
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()),
}))
}
}
+5 -9
View File
@@ -1,16 +1,15 @@
use std::{
cell::{Cell, RefCell},
collections::HashSet,
pin::Pin,
rc::Rc,
};
use super::{waker::RcWake, SchedulerMsg};
use crate::{
innerlude::{Mutation, Renderer},
innerlude::{Mutation, Mutations},
Element, ScopeId,
};
use futures_task::Waker;
use futures_util::Future;
#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
@@ -22,7 +21,7 @@ pub type SuspenseContext = Rc<RefCell<SuspenseBoundary>>;
pub struct SuspenseBoundary {
pub id: ScopeId,
pub waiting_on: HashSet<SuspenseId>,
pub mutations: Renderer<'static>,
pub mutations: Mutations<'static>,
}
impl SuspenseBoundary {
@@ -30,7 +29,7 @@ impl SuspenseBoundary {
Rc::new(RefCell::new(Self {
id,
waiting_on: Default::default(),
mutations: Renderer::new(0),
mutations: Mutations::new(0),
}))
}
}
@@ -46,10 +45,7 @@ pub struct SuspenseLeaf {
impl RcWake for SuspenseLeaf {
fn wake_by_ref(arc_self: &Rc<Self>) {
// if arc_self.notified.get() {
// return;
// }
// arc_self.notified.set(true);
arc_self.notified.set(true);
_ = arc_self
.tx
.unbounded_send(SchedulerMsg::SuspenseNotified(arc_self.id));
+23 -5
View File
@@ -7,15 +7,16 @@ use std::{
process::Output,
rc::Rc,
sync::Arc,
task::Poll,
};
use futures_task::{waker, ArcWake, Context, RawWaker, RawWakerVTable, Waker};
use futures_util::{pin_mut, Future, FutureExt};
use slab::Slab;
use std::task::{Context, RawWaker, RawWakerVTable, Waker};
use crate::{Element, ScopeId};
use super::{waker::RcWake, HandleInner, SchedulerHandle, SchedulerMsg};
use super::{waker::RcWake, HandleInner, Scheduler, SchedulerMsg};
#[cfg_attr(feature = "serialize", derive(serde::Serialize, serde::Deserialize))]
#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
@@ -24,12 +25,29 @@ pub struct TaskId(pub usize);
/// the task itself is the waker
pub struct LocalTask {
id: TaskId,
scope: ScopeId,
tx: futures_channel::mpsc::UnboundedSender<SchedulerMsg>,
pub id: TaskId,
pub scope: ScopeId,
pub tx: futures_channel::mpsc::UnboundedSender<SchedulerMsg>,
// todo: use rc and weak, or the bump slab instead of unsafecell
pub task: UnsafeCell<Pin<Box<dyn Future<Output = ()> + 'static>>>,
}
impl LocalTask {
pub fn progress(self: &Rc<Self>) -> bool {
let waker = self.waker();
let mut cx = Context::from_waker(&waker);
// safety: the waker owns its task and everythig is single threaded
let fut = unsafe { &mut *self.task.get() };
match fut.poll_unpin(&mut cx) {
Poll::Ready(_) => true,
_ => false,
}
}
}
impl HandleInner {
pub fn spawn(&self, scope: ScopeId, task: impl Future<Output = ()> + 'static) -> TaskId {
let mut tasks = self.tasks.borrow_mut();
+24 -24
View File
@@ -1,9 +1,10 @@
use futures_task::Context;
use futures_util::{FutureExt, StreamExt};
use std::task::{Context, Poll};
use crate::{
diff::DirtyScope,
factory::RenderReturn,
innerlude::{Mutation, Renderer, SuspenseContext},
innerlude::{Mutation, Mutations, SuspenseContext},
VNode, VirtualDom,
};
@@ -14,25 +15,29 @@ impl 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) {
// todo: make sure the scheduler queue is completely drained
loop {
match self.scheduler.rx.next().await.unwrap() {
SchedulerMsg::Event => todo!(),
SchedulerMsg::Immediate(_) => todo!(),
match self.rx.next().await.unwrap() {
SchedulerMsg::Event => break,
SchedulerMsg::Immediate(id) => {
let height = self.scopes[id.0].height;
self.dirty_scopes.insert(DirtyScope { height, id });
break;
}
SchedulerMsg::DirtyAll => todo!(),
SchedulerMsg::TaskNotified(id) => {
let mut tasks = self.scheduler.handle.tasks.borrow_mut();
let local_task = &tasks[id.0];
let mut tasks = self.scheduler.tasks.borrow_mut();
let task = &tasks[id.0];
// attach the waker to itself
// todo: don't make a new waker every time, make it once and then just clone it
let waker = local_task.waker();
let mut cx = Context::from_waker(&waker);
// If the task completes...
if task.progress() {
// Remove it from the scope so we dont try to double drop it when the scope dropes
self.scopes[task.scope.0].spawned_tasks.remove(&id);
// safety: the waker owns its task and everythig is single threaded
let fut = unsafe { &mut *local_task.task.get() };
if let futures_task::Poll::Ready(_) = fut.poll_unpin(&mut cx) {
// Remove it from the scheduler
tasks.remove(id.0);
}
}
@@ -42,7 +47,6 @@ impl VirtualDom {
let leaf = self
.scheduler
.handle
.leaves
.borrow_mut()
.get(id.0)
@@ -55,16 +59,14 @@ impl VirtualDom {
let waker = leaf.waker();
let mut cx = Context::from_waker(&waker);
let fut = unsafe { &mut *leaf.task };
let mut pinned = unsafe { std::pin::Pin::new_unchecked(fut) };
// Safety: the future is always pinned to the bump arena
let mut pinned = unsafe { std::pin::Pin::new_unchecked(&mut *leaf.task) };
let as_pinned_mut = &mut pinned;
// the component finished rendering and gave us nodes
// we should attach them to that component and then render its children
// continue rendering the tree until we hit yet another suspended component
if let futures_task::Poll::Ready(new_nodes) = as_pinned_mut.poll_unpin(&mut cx)
{
if let Poll::Ready(new_nodes) = as_pinned_mut.poll_unpin(&mut cx) {
let boundary = &self.scopes[leaf.scope_id.0]
.consume_context::<SuspenseContext>()
.unwrap();
@@ -87,7 +89,7 @@ impl VirtualDom {
if let RenderReturn::Sync(Some(template)) = ret {
let mutations = &mut fiber.mutations;
let template: &VNode = unsafe { std::mem::transmute(template) };
let mutations: &mut Renderer =
let mutations: &mut Mutations =
unsafe { std::mem::transmute(mutations) };
self.scope_stack.push(scope_id);
@@ -103,8 +105,6 @@ impl VirtualDom {
}
}
}
// now proces any events. If we end up running a component and it generates mutations, then we should run those mutations
}
}
}
+1 -1
View File
@@ -1,4 +1,4 @@
use futures_task::{RawWaker, RawWakerVTable, Waker};
use std::task::{RawWaker, RawWakerVTable, Waker};
use std::{mem, rc::Rc};
pub trait RcWake: Sized {