Rust: Support delayed tasks in RunLoop

mojo/public/rust/src/bindings/run_loop.rs

 // Copyright 2016 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 //! This module contains a thread-local run-loop.
 //!
 //! The run-loop may have handles and handlers pre-registers
 //! (and in fact, must) in order to keep running. The run-loop
 //! executes until it has no more handles or handlers on itself,
 //! or until it is told to quit via stop().
 //!
 //! The run-loop waits until some signals on some handle is satisfied,
 //! at which point it wakes up and executes the appropriate handler
 //! method. This handler method may then be used to further populate
 //! or de-populate the run-loop.
+//!
+//! As of yet, the run-loop is NOT thread-safe. Although it is useful
+//! to be able to register tasks or handles from one thread onto
+//! another thread's run-loop, this is as-of-yet unsupported, and
+//! Rust should complain loudly when you try to do any threading here.
 
 use std::cell::RefCell;
 use std::cmp::{PartialEq, Eq, PartialOrd, Ord, Ordering};
 use std::collections::BinaryHeap;
 use std::collections::HashMap;
 use std::i64;
 use std::u32;
 use std::vec::Vec;
 
 use system;
 use system::{Handle, MOJO_INDEFINITE, MojoResult};
 use system::core;
 use system::wait_set;
 
 /// Define the equivalent of MOJO_INDEFINITE for absolute deadlines
 const MOJO_INDEFINITE_ABSOLUTE: system::MojoTimeTicks = 0;
 
 // TODO(mknyszek): The numbers below are arbitrary and come from the C++ bindings,
 // and should probably be changed at some point
 
 /// Initial size of the result buffer.
 const INITIAL_WAIT_SET_NUM_RESULTS: usize = 16;
 
 /// Maximum size of the result buffer.
 const MAXIMUM_WAIT_SET_NUM_RESULTS: usize = 256;
 
 /// Thread-local data structure for keeping track of handles to wait on.
-thread_local!(static TL_RUN_LOOP: RefCell<RunLoop<'static>> = RefCell::new(RunLoop::new()));
+thread_local!(static TL_RUN_LOOP: RefCell<RunLoop<'static, 'static>> = RefCell::new(RunLoop::new()));
 
 /// Token representing handle/callback to wait on for this thread only. This
 /// token only has meaning on the thread in which the handle was registered.
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct Token(u64);
 
 impl Token {
     /// Get the wait token's "cookie" form, suitable for use in a wait set.
     fn as_cookie(&self) -> u64 {
         self.0
@@ skipping 72 matching lines @@
     pub fn deadline(&self) -> system::MojoTimeTicks {
         self.deadline
     }
 
     /// Setter to update the current absolute deadline.
     pub fn set_deadline(&mut self, deadline: system::MojoTimeTicks) {
         self.deadline = deadline
     }
 }
 
+/// A wrapper struct for carrying the task as well as some information about
+/// it.
+struct TaskInfo<'t> {
+    /// The task, boxed up.
+    closure: Box<Fn(&mut RunLoop) + 't>,

[Eric Holk, 2016/08/12 01:36:31]

Can this be an FnOnce? A lot of times that gives you more flexibility it what
you can do in the closure.

[mknyszek, 2016/08/12 08:09:31]

That's true, but unfortunately Rust doesn't support calling a Box<FnOnce> at all
since FnOnce needs to consume itself, but being a trait it is not SIzed. There
is a workaround called FnBox which is an extension of FnOnce meant to be called
from inside a Box, but its currently sitting in nightly as an unstable feature.
In the future Rust plans to fix this by having Box<FnOnce> just work (maybe by
implementing some specialization for the precise type Box<FnOnce>?), but for now
it just doesn't.

As a short-term fix I have still loosened the trait bound on the closure to
FnMut, which should make it a little bit easier to use. Realistically, FnOnce is
all I need, but its not yet available in the context I want to use it in (that
is, from inside a Box).

[Eric Holk, 2016/08/12 16:35:07]

Sounds good!

+
+    /// An absolute deadline in terms of time ticks.
+    ///
+    /// This is the most recently updated deadline that
+    /// we should be watching out for. All others for this
+    /// token may be considered "stale".
+    deadline: system::MojoTimeTicks,
+}
+
+impl<'t> TaskInfo<'t> {
+    /// Executes the task within the info object, consuming it
+    /// in the process.
+    pub fn execute_task(self, run_loop: &mut RunLoop) {
+        (*self.closure)(run_loop);
+    }
+
+    /// Getter for the current absolute deadline held inside.
+    pub fn deadline(&self) -> system::MojoTimeTicks {
+        self.deadline
+    }
+}
+
+impl<'t> PartialEq for TaskInfo<'t> {
+    /// Equality for TaskInfo in terms of its deadline
+    fn eq(&self, other: &TaskInfo) -> bool {
+        self.deadline == other.deadline
+    }
+}
+
+impl<'t> Eq for TaskInfo<'t> {}
+
+impl<'t> PartialOrd for TaskInfo<'t> {
+    /// Partial comparison for TaskInfo in terms of its deadline
+    ///
+    /// Reverses the comparison because the Rust std library only
+    /// offers a max-heap, and we need a min-heap.
+    fn partial_cmp(&self, other: &TaskInfo) -> Option<Ordering> {
+        other.deadline.partial_cmp(&self.deadline)
+    }
+}
+
+impl<'t> Ord for TaskInfo<'t> {
+    /// Implement comparisons for Task Info.
+    ///
+    /// Reverses the comparison because the Rust std library only
+    /// offers a max-heap, and we need a min-heap.
+    fn cmp(&self, other: &Self) -> Ordering {
+        other.deadline.cmp(&self.deadline)
+    }
+}
+
 /// Wrapper struct intended to be used in a priority queue
 /// for efficiently retrieving the next closest deadline.
 #[derive(Clone)]
 struct DeadlineInfo {
     /// The ID of the associated Handler struct in the RunLoop's
     /// hash map.
     token: Token,
 
     /// An absolute deadline in terms of time ticks.
     deadline: system::MojoTimeTicks,
@@ skipping 71 matching lines @@
     } else {
         (deadline - now) as system::MojoDeadline
     }
 }
 
 /// This structure contains all information necessary to wait on handles
 /// asynchronously.
 ///
 /// Ultimately, it should only be a singleton living in
 /// thread-local storage.
-pub struct RunLoop<'h> {
+pub struct RunLoop<'h, 't> {
     /// Running count of the next available token slot.
     token_count: u64,
 
     /// A map of handlers.
     ///
     /// TODO(mknyszek): Try out a Slab allocator instead of a hashmap.
     handlers: HashMap<Token, HandlerInfo<'h>>,
 
+    /// A min-heap of delayed tasks in order to pull the soonest task to
+    /// execute efficiently.
+    tasks: BinaryHeap<TaskInfo<'t>>,
+
     /// A min-heap containing deadlines in order to pull out the soonest
     /// deadline efficiently.
     ///
     /// Warning: may contain "stale" deadlines which are not kept in the
     /// map!
     deadlines: BinaryHeap<DeadlineInfo>,
 
     /// The Mojo structure keeping track of handles and signals.
     ///
     /// This structure must be kept in sync with handlers.
     handle_set: wait_set::WaitSet,
 
     /// A flag that tells the RunLoop whether it should quit.
     should_quit: bool,
 
     /// A flag that indicates whether the RunLoop is running or now
     running: bool,
 }
 
-impl<'h> RunLoop<'h> {
+impl<'h, 't> RunLoop<'h, 't> {
     /// Create a new RunLoop.
-    pub fn new() -> RunLoop<'h> {
+    pub fn new() -> RunLoop<'h, 't> {
         RunLoop {
             token_count: 0,
             handlers: HashMap::new(),
+            tasks: BinaryHeap::new(),
             deadlines: BinaryHeap::new(),
             handle_set: wait_set::WaitSet::new(wsflags!(Create::None)).unwrap(),
             should_quit: false,
             running: false,
         }
     }
 
     /// Generate a new Token for this RunLoop
     fn generate_token(&mut self) -> Token {
         self.token_count = self.token_count.wrapping_add(1);
@@ skipping 78 matching lines @@
         match self.handlers.remove(&token) {
             Some(_) => {
                 let _result = self.handle_set.remove(token.as_cookie());
                 debug_assert_eq!(_result, MojoResult::Okay);
                 true
             }
             None => false,
         }
     }
 
-    /// Uses the binary heap to get the next closest deadline.
+    /// Adds a task to be run by the runloop after some delay.
+    ///
+    /// Returns a token if the delay is valid, otherwise returns None.
+    pub fn post_task<F>(&mut self, task: F, delay: system::MojoTimeTicks) -> Result<(), ()>
+        where F: Fn(&mut RunLoop) + 't

[Eric Holk, 2016/08/12 01:36:31]

If you make the previous one an FnOnce, I think you'll have to do it here too.

+    {
+        let now = core::get_time_ticks_now();
+        if delay > i64::MAX - now {
+            return Err(());
+        }
+        let deadline = now + delay;
+        self.tasks.push(TaskInfo {
+            closure: Box::new(task),
+            deadline: deadline,
+        });
+        Ok(())
+    }
+
+    /// Uses the binary heaps to get the next closest deadline.
     ///
     /// Removes stale deadline entries as they are found, but
     /// does not otherwise modify the heap of deadlines.
     fn get_next_deadline(&mut self) -> system::MojoTimeTicks {
         debug_assert!(!self.handlers.is_empty());
+        let top_task_deadline = match self.tasks.peek() {
+            Some(info) => info.deadline(),
+            None => MOJO_INDEFINITE_ABSOLUTE,
+        };
         let mut top = match self.deadlines.peek() {
             Some(info) => info.clone(),
             None => return MOJO_INDEFINITE_ABSOLUTE,
         };
         while !self.handlers.contains_key(top.token()) {
             self.deadlines.pop();
             top = match self.deadlines.peek() {
                 Some(info) => info.clone(),
                 None => return MOJO_INDEFINITE_ABSOLUTE,
             }
         }
-        top.deadline()
+        if top_task_deadline != MOJO_INDEFINITE_ABSOLUTE &&
+           top_task_deadline < top.deadline() {
+            top_task_deadline
+        } else {
+            top.deadline()
+        }
     }
 
     /// Gets a handler by token to be manipulated in a consistent environment.
     ///
     /// This method provides a method of accessing a handler in order to manipulate
     /// it in a manner that avoids a borrow cycle, that is, it take()s the handler
     /// out of the HashMap, and returns it when manipulation has completed.
     fn get_handler_with<F>(&mut self, token: &Token, invoker: F)
         where F: FnOnce(&mut Self,
                         &mut Box<Handler + 'h>,
@@ skipping 89 matching lines @@
             // Remove the deadline
             self.deadlines.pop();
             // Break if the next deadline has not yet expired.
             top = match self.deadlines.peek() {
                 Some(info) => info.clone(),
                 None => break,
             };
         }
     }
 
+    /// Iterates through all tasks whose deadline has passed and executes
+    /// them, consuming their information object.
+    ///
+    /// These tasks all have access to the RunLoop so that they may reschedule
+    /// themselves or manipulate the RunLoop in some other way.
+    fn execute_ready_tasks(&mut self) {
+        let now = core::get_time_ticks_now();
+        let mut deadline = match self.tasks.peek() {
+            Some(info) => info.deadline(),
+            None => return,
+        };
+        while deadline < now {
+            let top = self.tasks.pop().expect("Sudden change to heap?");
+            top.execute_task(self);
+            if self.should_quit {
+                return;
+            }
+            deadline = match self.tasks.peek() {
+                Some(info) => info.deadline(),
+                None => return,
+            };
+        }
+    }
+
     /// Blocks on handle_set.wait_on_set using the information contained
     /// within itself.
     ///
     /// This method blocks for only as long as the shortest deadline among all
     /// handles this thread has registered. This method returns immediately as
     /// soon as any one handle has its signals satisfied, fails to ever have its
     /// signals satisfied, or reaches its deadline.
     fn wait(&mut self, results_buffer: &mut Vec<system::WaitSetResult>) {
         debug_assert!(!self.handlers.is_empty());
+        self.execute_ready_tasks();
+        // If after executing a task we quit or there are no handles,
+        // we have no reason to continue.
+        if self.handlers.is_empty() || self.should_quit {
+            return;
+        }
         let deadline = self.get_next_deadline();
         let until_deadline = relative_deadline(deadline, core::get_time_ticks_now());
         // Perform the wait
         match self.handle_set.wait_on_set(until_deadline, results_buffer) {
             Ok(max_results) => {
                 self.notify_of_results(results_buffer);
                 // Clear the buffer since we don't need the results anymore.
                 // Helps prevent a copy if we resize the buffer.
                 results_buffer.clear();
                 // Increase the size of the buffer if there are more results
@@ skipping 35 matching lines @@
 
 /// Provides a scope to modify the current thread's runloop.
 pub fn with_current<F>(modifier: F)
     where F: FnOnce(&mut RunLoop)
 {
     TL_RUN_LOOP.with(|ref_runloop| {
         let mut runloop = ref_runloop.borrow_mut();
         modifier(&mut *runloop);
     });
 }