Refactor futures executor

src/webcore/executor.rs

@@ -0,0 +1,243 @@
+// This file implements a futures-compatible executor which schedules futures
+// onto the JavaScript event loop. This implementation assumes there is a
+// single thread and is *not* compatible with multiple WebAssembly workers sharing
+// the same address space.
+//
+// TODO: Implement support for multiple threads. This will require a mechanism to
+// wake up another thread, such as the `postMessage` API.
+
+use futures::future::{Future, ExecuteError, Executor};
+use futures::executor::{self, Notify, Spawn};
+use futures::Async;
+use std::collections::VecDeque;
+use std::result::Result as StdResult;
+use std::cell::{Cell, RefCell};
+use std::rc::Rc;
+use std::cmp;
+use webcore::try_from::TryInto;
+use webcore::value::Reference;
+
+
+// TODO: Determine optimal values for these constants
+// Initial capacity of the event queue
+const INITIAL_QUEUE_CAPACITY: usize = 10;
+// Iterations to wait before allowing the queue to shrink
+const QUEUE_SHRINK_DELAY: usize = 25;
+
+
+// This functionality should really be in libstd, because the implementation
+// looks stupid.
+unsafe fn clone_raw< T >( ptr: *const T ) -> Rc< T > {
+    let result = Rc::from_raw( ptr );
+    ::std::mem::forget( result.clone() );
+    result
+}
+
+// Typing this out is tedious
+type BoxedFuture = Box< Future< Item = (), Error = () > + 'static >;
+
+struct SpawnedTask {
+    is_queued: Cell< bool >,
+    spawn: RefCell< Option< Spawn< BoxedFuture > > >,
+}
+
+impl SpawnedTask {
+    fn new< F >( future: F ) -> Rc< Self >
+        where F: Future< Item = (), Error = () > + 'static {
+        Rc::new( Self {
+            is_queued: Cell::new( false ),
+            spawn: RefCell::new( Some( executor::spawn(
+                Box::new( future ) as BoxedFuture
+            ) ) )
+        } )
+    }
+
+    fn poll( &self ) {
+        let mut spawn = self.spawn.borrow_mut();
+
+        // Take the future so that if we panic it gets dropped
+        if let Some( mut spawn_future ) = spawn.take() {
+            // Clear `is_queued` flag so that it will re-queue if poll calls task.notify()
+            self.is_queued.set( false );
+
+            if spawn_future.poll_future_notify( &&EventLoop, self as *const _ as usize ) == Ok( Async::NotReady ) {
+                // Future was not ready, so put it back
+                *spawn = Some( spawn_future );
+            }
+        }
+    }
+
+    fn notify( task: Rc< SpawnedTask > ) {
+        // If not already queued
+        if !task.is_queued.replace( true ) {
+            EventLoop.push_task(task);
+        }
+    }
+}
+
+// A proxy for the JavaScript event loop.
+struct EventLoop;
+
+// There's only one thread, but this lets us tell the compiler that we
+// don't need a `Sync` bound, and also gives us lazy initialization.
+thread_local! {
+    static EVENT_LOOP_INNER: EventLoopInner = EventLoopInner::new();
+}
+
+impl EventLoop {
+    fn drain(&self) {
+        EVENT_LOOP_INNER.with(EventLoopInner::drain)
+    }
+    fn push_task(&self, task: Rc< SpawnedTask >) {
+        EVENT_LOOP_INNER.with(|inner| inner.push_task(task))
+    }
+}
+
+// State relating to the JavaScript event loop. Only one instance ever exists.
+struct EventLoopInner {
+    // Avoid unnecessary allocation and interop by keeping a local
+    // queue of pending tasks.
+    microtask_queue: RefCell< VecDeque< Rc< SpawnedTask > > >,
+    waker: Reference,
+    shrink_counter: Cell<usize>
+}
+
+// Not strictly necessary, but may become relevant in the future
+impl Drop for EventLoopInner {
+    #[inline]
+    fn drop( &mut self ) {
+        js! { @(no_return)
+            @{&self.waker}.drop();
+        }
+    }
+}
+
+impl EventLoopInner {
+    // Initializes the event loop. Only called once.
+    fn new() -> Self {
+        EventLoopInner {
+            microtask_queue: RefCell::new(VecDeque::with_capacity(INITIAL_QUEUE_CAPACITY)),
+            waker: js!(
+                var callback = @{|| EventLoop.drain()};
+                var wrapper = function() {
+                    if (!callback.dropped) { callback() }
+                };
+                var nextTick;
+
+                // Modern browsers can use `MutationObserver` which allows
+                // us to schedule a micro-task without allocating a promise.
+                // https://dom.spec.whatwg.org/#notify-mutation-observers
+                if ( typeof MutationObserver === "function" ) {
+                    var node = document.createTextNode( "0" );
+                    var state = false;
+
+                    new MutationObserver( wrapper ).observe( node, { characterData: true } );
+
+                    nextTick = function() {
+                        state = !state;
+                        node.data = ( state ? "1" : "0" );
+                    };
+
+                // Node.js and other environments
+                } else {
+                    var promise = Promise.resolve( null );
+
+                    nextTick = function() {
+                        promise.then( wrapper );
+                    };
+                }
+
+                nextTick.drop = function() {
+                    callback.dropped = true;
+                    callback.drop();
+                };
+
+                return nextTick;
+            ).try_into().unwrap(),
+            shrink_counter: Cell::new(0)
+        }
+    }
+    // Pushes a task onto the queue
+    fn push_task(&self, task: Rc< SpawnedTask >) {
+        let mut queue = self.microtask_queue.borrow_mut();
+        queue.push_back(task);
+
+        // If the queue was previously empty, then we need to schedule
+        // the queue to be drained.
+        if queue.len() == 1 {
+            self.wake();
+        }
+    }
+    // Invoke the JavaScript waker function
+    fn wake(&self) {
+        js! { @(no_return) @{&self.waker}(); }
+    }
+    // Remove and return a task from the front of the queue
+    fn pop_task(&self) -> Option< Rc< SpawnedTask > > {
+        self.microtask_queue.borrow_mut().pop_front()
+    }
+    // See if it's worth trying to reclaim some space from the queue
+    fn estimate_realloc_capacity(&self) -> Option<usize> {
+        let queue = self.microtask_queue.borrow();
+        // A VecDeque retains a `2^n-1` capacity
+        let half_cap = queue.capacity()/2;
+        // We consider shrinking the queue if it is less than
+        // half full...
+        if half_cap > queue.len() && half_cap > INITIAL_QUEUE_CAPACITY {
+            // ...and if it's been that way for at least
+            // `QUEUE_SHRINK_DELAY` iterations.
+            let shrink_counter = self.shrink_counter.get();
+            if shrink_counter < QUEUE_SHRINK_DELAY {
+                self.shrink_counter.set(shrink_counter + 1);
+            } else {
+                self.shrink_counter.set(0);
+                return Some(cmp::max(queue.len(), INITIAL_QUEUE_CAPACITY));
+            }
+        } else {
+            self.shrink_counter.set(0);
+        }
+        None
+    }
+    // Poll the queue until it is empty
+    fn drain(&self) {
+        let maybe_realloc_capacity = self.estimate_realloc_capacity();
+
+        // Poll all the pending tasks
+        while let Some(task) = self.pop_task() {
+            task.poll();
+        }
+
+        if let Some(realloc_capacity) = maybe_realloc_capacity {
+            // We decided to reclaim some space
+            *self.microtask_queue.borrow_mut() = VecDeque::with_capacity(realloc_capacity);
+        }
+    }
+}
+
+impl< F > Executor< F > for EventLoop where
+    F: Future< Item = (), Error = () > + 'static {
+    fn execute( &self, future: F ) -> StdResult< (), ExecuteError< F > > {
+        SpawnedTask::notify( SpawnedTask::new( future ) );
+        Ok( () )
+    }
+}
+
+impl Notify for EventLoop {
+    fn notify( &self, spawned_id: usize ) {
+        SpawnedTask::notify( unsafe { clone_raw( spawned_id as *const _ ) } );
+    }
+
+    fn clone_id( &self, id: usize ) -> usize {
+        unsafe { Rc::into_raw( clone_raw( id as *const SpawnedTask ) ) as usize }
+    }
+
+    fn drop_id( &self, id: usize ) {
+        unsafe { Rc::from_raw( id as *const SpawnedTask ) };
+    }
+}
+
+#[inline]
+pub fn spawn< F >( future: F ) where
+    F: Future< Item = (), Error = () > + 'static {
+    EventLoop.execute( future ).unwrap();
+}

src/webcore/mod.rs

@@ -23,7 +23,7 @@ pub mod discard;
 pub mod promise_future;
 
 #[cfg(feature = "futures")]
-pub mod promise_executor;
+pub mod executor;
 
 #[cfg(feature = "nightly")]
 pub mod void {

src/webcore/promise_future.rs

@@ -4,7 +4,7 @@ use webcore::try_from::{TryInto, TryFrom};
 use webapi::error;
 use futures::{Future, Poll, Async};
 use futures::unsync::oneshot::Receiver;
-use webcore::promise_executor::spawn;
+use webcore::executor::spawn;
 use webcore::discard::DiscardOnDrop;
 use super::promise::{Promise, DoneHandle};