window: adopt async_task for spawn_task

Cargo.lock

@@ -3257,6 +3257,7 @@ name = "window"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "async-task",
  "bitflags 1.2.1",
  "cgl",
  "clipboard",

src/frontend/mod.rs

@@ -64,12 +64,14 @@ pub fn front_end() -> Option<Rc<dyn FrontEnd>> {
     res
 }
 
-pub fn spawn_task<F: std::future::Future<Output = ()> + 'static>(future: F) {
+pub fn spawn_task<F: std::future::Future<Output = ()> + 'static>(
+    future: F,
+) -> async_task::JoinHandle<(), ()> {
     let frontend = front_end().expect("spawn_task must be called from the gui thread");
     if let Some(frontend) = frontend.downcast_ref::<MuxServerFrontEnd>() {
-        frontend.spawn_task(future);
+        frontend.spawn_task(future)
     } else {
-        window::Connection::get().unwrap().spawn_task(future);
+        window::Connection::get().unwrap().spawn_task(future)
     }
 }
 

window/Cargo.toml

@@ -15,6 +15,7 @@ pretty_env_logger = "0.3"
 gl_generator = {version="0.13", optional=true}
 
 [dependencies]
+async-task = "1.2"
 anyhow = "1.0"
 thiserror = "1.0"
 bitflags = "1.0"

window/src/connection.rs

@@ -26,8 +26,10 @@ pub trait ConnectionOps {
 
     fn terminate_message_loop(&self);
     fn run_message_loop(&self) -> Fallible<()>;
-    fn spawn_task<F: std::future::Future<Output = ()> + 'static>(&self, future: F);
-    fn wake_task_by_id(slot: usize);
+    fn spawn_task<F: std::future::Future<Output = ()> + 'static>(
+        &self,
+        future: F,
+    ) -> async_task::JoinHandle<(), ()>;
 
     // TODO: return a handle that can be used to cancel the timer
     fn schedule_timer<F: FnMut() + 'static>(&self, interval: std::time::Duration, callback: F);

window/src/lib.rs

@@ -6,7 +6,6 @@ pub mod connection;
 pub mod input;
 pub mod os;
 mod spawn;
-mod tasks;
 mod timerlist;
 
 #[cfg(all(

window/src/os/macos/connection.rs

@@ -4,7 +4,6 @@
 use super::window::WindowInner;
 use crate::connection::ConnectionOps;
 use crate::spawn::*;
-use crate::tasks::{Task, Tasks};
 use cocoa::appkit::{NSApp, NSApplication, NSApplicationActivationPolicyRegular};
 use cocoa::base::{id, nil};
 use core_foundation::date::CFAbsoluteTimeGetCurrent;
@@ -20,7 +19,6 @@ pub struct Connection {
     ns_app: id,
     pub(crate) windows: RefCell<HashMap<usize, Rc<RefCell<WindowInner>>>>,
     pub(crate) next_window_id: AtomicUsize,
-    tasks: Tasks,
 }
 
 impl Connection {
@@ -35,7 +33,6 @@ impl Connection {
             let conn = Self {
                 ns_app,
                 windows: RefCell::new(HashMap::new()),
-                tasks: Default::default(),
                 next_window_id: AtomicUsize::new(1),
             };
             Ok(conn)
@@ -97,16 +94,11 @@ impl ConnectionOps for Connection {
         Ok(())
     }
 
-    fn spawn_task<F: std::future::Future<Output = ()> + 'static>(&self, future: F) {
-        let id = self.tasks.add_task(Task(Box::pin(future)));
-        Self::wake_task_by_id(id);
-    }
-
-    fn wake_task_by_id(slot: usize) {
-        SpawnQueueExecutor {}.execute(Box::new(move || {
-            let conn = Connection::get().unwrap();
-            conn.tasks.poll_by_slot(slot);
-        }));
+    fn spawn_task<F: std::future::Future<Output = ()> + 'static>(
+        &self,
+        future: F,
+    ) -> async_task::JoinHandle<(), ()> {
+        SPAWN_QUEUE.spawn_task(future)
     }
 
     fn schedule_timer<F: FnMut() + 'static>(&self, interval: std::time::Duration, callback: F) {

window/src/os/wayland/connection.rs

@@ -4,7 +4,6 @@ use super::pointer::*;
 use super::window::*;
 use crate::connection::ConnectionOps;
 use crate::spawn::*;
-use crate::tasks::{Task, Tasks};
 use crate::timerlist::{TimerEntry, TimerList};
 use crate::Connection;
 use anyhow::{bail, Context};
@@ -24,7 +23,6 @@ use toolkit::Environment;
 pub struct WaylandConnection {
     should_terminate: RefCell<bool>,
     timers: RefCell<TimerList>,
-    pub(crate) tasks: Tasks,
     pub(crate) next_window_id: AtomicUsize,
     pub(crate) windows: RefCell<HashMap<usize, Rc<RefCell<WaylandWindowInner>>>>,
 
@@ -103,7 +101,6 @@ impl WaylandConnection {
             environment: RefCell::new(environment),
             should_terminate: RefCell::new(false),
             timers: RefCell::new(TimerList::new()),
-            tasks: Default::default(),
             next_window_id: AtomicUsize::new(1),
             windows: RefCell::new(HashMap::new()),
             seat,
@@ -181,13 +178,11 @@ impl WaylandConnection {
 }
 
 impl ConnectionOps for WaylandConnection {
-    fn spawn_task<F: std::future::Future<Output = ()> + 'static>(&self, future: F) {
-        let id = self.tasks.add_task(Task(Box::pin(future)));
-        Connection::wake_task_by_id(id);
-    }
-
-    fn wake_task_by_id(_slot: usize) {
-        panic!("use Connection::wake_task_by_id instead of WaylandConnection::wake_task_by_id");
+    fn spawn_task<F: std::future::Future<Output = ()> + 'static>(
+        &self,
+        future: F,
+    ) -> async_task::JoinHandle<(), ()> {
+        SPAWN_QUEUE.spawn_task(future)
     }
 
     fn terminate_message_loop(&self) {

window/src/os/windows/connection.rs

@@ -2,7 +2,6 @@
 use super::{HWindow, WindowInner};
 use crate::connection::ConnectionOps;
 use crate::spawn::*;
-use crate::tasks::{Task, Tasks};
 use promise::BasicExecutor;
 use std::cell::RefCell;
 use std::collections::HashMap;
@@ -19,7 +18,6 @@ use winapi::um::winuser::*;
 pub struct Connection {
     event_handle: HANDLE,
     pub(crate) windows: RefCell<HashMap<HWindow, Rc<RefCell<WindowInner>>>>,
-    tasks: Tasks,
     timers: RefCell<HashMap<UINT_PTR, UINT_PTR>>,
 }
 
@@ -58,15 +56,7 @@ impl ConnectionOps for Connection {
     }
 
     fn spawn_task<F: std::future::Future<Output = ()> + 'static>(&self, future: F) {
-        let id = self.tasks.add_task(Task(Box::pin(future)));
-        Self::wake_task_by_id(id);
-    }
-
-    fn wake_task_by_id(slot: usize) {
-        SpawnQueueExecutor {}.execute(Box::new(move || {
-            let conn = Connection::get().unwrap();
-            conn.tasks.poll_by_slot(slot);
-        }));
+        SPAWN_QUEUE.spawn_task(future);
     }
 
     fn schedule_timer<F: FnMut() + 'static>(&self, interval: std::time::Duration, callback: F) {
@@ -112,7 +102,6 @@ impl Connection {
             event_handle,
             windows: RefCell::new(HashMap::new()),
             timers: RefCell::new(HashMap::new()),
-            tasks: Default::default(),
         })
     }
 

window/src/os/x11/connection.rs

@@ -3,7 +3,6 @@ use crate::connection::ConnectionOps;
 use crate::os::x11::window::XWindowInner;
 use crate::os::Connection;
 use crate::spawn::*;
-use crate::tasks::{Task, Tasks};
 use crate::timerlist::{TimerEntry, TimerList};
 use anyhow::{anyhow, bail};
 use mio::unix::EventedFd;
@@ -33,7 +32,6 @@ pub struct XConnection {
     pub(crate) windows: RefCell<HashMap<xcb::xproto::Window, Arc<Mutex<XWindowInner>>>>,
     should_terminate: RefCell<bool>,
     pub(crate) shm_available: bool,
-    pub(crate) tasks: Tasks,
     timers: RefCell<TimerList>,
     pub(crate) visual: xcb::xproto::Visualtype,
 }
@@ -168,13 +166,11 @@ fn server_supports_shm() -> bool {
 }
 
 impl ConnectionOps for XConnection {
-    fn spawn_task<F: std::future::Future<Output = ()> + 'static>(&self, future: F) {
-        let id = self.tasks.add_task(Task(Box::pin(future)));
-        Connection::wake_task_by_id(id);
-    }
-
-    fn wake_task_by_id(_slot: usize) {
-        panic!("call wake_task_by_id on Connection rather than XConnection");
+    fn spawn_task<F: std::future::Future<Output = ()> + 'static>(
+        &self,
+        future: F,
+    ) -> async_task::JoinHandle<(), ()> {
+        SPAWN_QUEUE.spawn_task(future)
     }
 
     fn terminate_message_loop(&self) {
@@ -398,7 +394,6 @@ impl XConnection {
             windows: RefCell::new(HashMap::new()),
             should_terminate: RefCell::new(false),
             shm_available,
-            tasks: Default::default(),
             timers: RefCell::new(TimerList::new()),
             visual,
         };

window/src/os/x_and_wayland.rs

@@ -75,14 +75,6 @@ impl Connection {
         }
     }
 
-    fn do_wake_task_by_id(&self, slot: usize) {
-        match self {
-            Self::X11(x) => x.tasks.poll_by_slot(slot),
-            #[cfg(feature = "wayland")]
-            Self::Wayland(w) => w.tasks.poll_by_slot(slot),
-        };
-    }
-
     pub(crate) fn x11(&self) -> Rc<XConnection> {
         match self {
             Self::X11(x) => Rc::clone(x),
@@ -109,20 +101,13 @@ impl Connection {
 }
 
 impl ConnectionOps for Connection {
-    fn spawn_task<F: std::future::Future<Output = ()> + 'static>(&self, future: F) {
-        match self {
-            Self::X11(x) => x.spawn_task(future),
-            #[cfg(feature = "wayland")]
-            Self::Wayland(w) => w.spawn_task(future),
-        }
+    fn spawn_task<F: std::future::Future<Output = ()> + 'static>(
+        &self,
+        future: F,
+    ) -> async_task::JoinHandle<(), ()> {
+        SPAWN_QUEUE.spawn_task(future)
     }
 
-    fn wake_task_by_id(slot: usize) {
-        SpawnQueueExecutor {}.execute(Box::new(move || {
-            let conn = Connection::get().unwrap();
-            conn.do_wake_task_by_id(slot);
-        }));
-    }
     fn terminate_message_loop(&self) {
         match self {
             Self::X11(x) => x.terminate_message_loop(),

window/src/spawn.rs

@@ -64,6 +64,19 @@ impl SpawnQueue {
         }
     }
 
+    pub fn spawn_task<F: std::future::Future<Output = ()> + 'static>(
+        &self,
+        future: F,
+    ) -> async_task::JoinHandle<(), ()> {
+        let (task, handle) = async_task::spawn_local(
+            future,
+            move |task| SPAWN_QUEUE.spawn(Box::new(move || task.run())),
+            (),
+        );
+        task.schedule();
+        handle
+    }
+
     fn queue_func(&self, f: SpawnFunc, high_pri: bool) {
         let f = InstrumentedSpawnFunc {
             func: f,

window/src/tasks.rs

@@ -1,96 +0,0 @@
-use crate::connection::ConnectionOps;
-use crate::Connection;
-use std::cell::{Cell, RefCell};
-use std::future::Future;
-use std::task::{Context, RawWaker, RawWakerVTable, Waker};
-
-pub struct Task(pub std::pin::Pin<Box<dyn Future<Output = ()>>>);
-
-enum Slot {
-    Vacant { next_vacant: usize },
-    Running(Option<Task>),
-}
-
-#[derive(Default)]
-pub struct Tasks {
-    tasks: RefCell<Vec<Slot>>,
-    next_vacant: Cell<usize>,
-}
-
-impl Tasks {
-    pub fn add_task(&self, task: Task) -> usize {
-        let idx = self.next_vacant.get();
-        let mut tasks = self.tasks.borrow_mut();
-        match tasks.get_mut(idx) {
-            Some(&mut Slot::Vacant { next_vacant }) => {
-                self.next_vacant.set(next_vacant);
-                tasks[idx] = Slot::Running(Some(task));
-            }
-            Some(&mut Slot::Running(_)) => panic!("vacant points to running task"),
-            None => {
-                assert_eq!(idx, tasks.len());
-                tasks.push(Slot::Running(Some(task)));
-                self.next_vacant.set(idx + 1);
-            }
-        }
-        idx
-    }
-
-    pub fn poll_by_slot(&self, slot: usize) -> bool {
-        let mut task = match self.tasks.borrow_mut().get_mut(slot) {
-            Some(&mut Slot::Running(ref mut task)) => task.take().unwrap(),
-            Some(&mut Slot::Vacant { .. }) | None => return false,
-        };
-
-        let waker = TaskWaker::new_waker(slot);
-        let mut context = Context::from_waker(&waker);
-
-        let done = task.0.as_mut().poll(&mut context).is_ready();
-
-        let mut tasks = self.tasks.borrow_mut();
-        if done {
-            tasks[slot] = Slot::Vacant {
-                next_vacant: self.next_vacant.get(),
-            };
-            self.next_vacant.set(slot);
-        } else {
-            tasks[slot] = Slot::Running(Some(task));
-        }
-
-        true
-    }
-}
-
-struct TaskWaker(usize);
-
-static VTBL: RawWakerVTable = RawWakerVTable::new(
-    TaskWaker::waker_clone,
-    TaskWaker::waker_wake,
-    TaskWaker::waker_wake_by_ref,
-    TaskWaker::waker_drop,
-);
-
-impl TaskWaker {
-    fn new_waker(slot: usize) -> Waker {
-        let raw = RawWaker::new(slot as *const (), &VTBL);
-        unsafe { Waker::from_raw(raw) }
-    }
-
-    unsafe fn waker_clone(p: *const ()) -> RawWaker {
-        RawWaker::new(p, &VTBL)
-    }
-
-    unsafe fn waker_wake(p: *const ()) {
-        let id: usize = std::mem::transmute(p);
-        Connection::wake_task_by_id(id);
-    }
-
-    unsafe fn waker_wake_by_ref(p: *const ()) {
-        let id: usize = std::mem::transmute(p);
-        Connection::wake_task_by_id(id);
-    }
-
-    unsafe fn waker_drop(_p: *const ()) {
-        /* no action required */
-    }
-}