Merge pull request #476 from sdroege/futures-0.3

sdroege · web-flow · commit 66a9c53cb1c4 · 2019-04-26T13:01:32.000+03:00
Port to futures 0.3
diff --git a/Cargo.toml b/Cargo.toml
@@ -24,7 +24,7 @@ name = "glib"
 lazy_static = "1.2"
 libc = "0.2"
 bitflags = "1.0"
-futures-preview = { version = "0.2", optional = true }
+futures-preview = { version = "0.3.0-alpha", optional = true }
 glib-sys = { git = "https://github.com/gtk-rs/sys" }
 gobject-sys = { git = "https://github.com/gtk-rs/sys" }
 
diff --git a/src/lib.rs b/src/lib.rs
@@ -77,6 +77,7 @@
 
 #![allow(clippy::doc_markdown)]
 #![allow(clippy::unreadable_literal)]
+#![cfg_attr(feature = "futures", feature(futures_api, arbitrary_self_types))]
 
 #[macro_use]
 extern crate bitflags;
diff --git a/src/main_context_futures.rs b/src/main_context_futures.rs
@@ -3,16 +3,18 @@
 // Licensed under the MIT license, see the LICENSE file or <http://opensource.org/licenses/MIT>
 
 use futures;
-use futures::executor::{Executor, SpawnError};
+use futures::future::{FutureObj, LocalFutureObj};
 use futures::prelude::*;
-use futures::task::{LocalMap, UnsafeWake, Waker};
-use futures::{Async, Future, Never};
+use futures::task::{
+    Context, LocalSpawn, Poll, RawWaker, RawWakerVTable, Spawn, SpawnError, Waker,
+};
 use get_thread_id;
 use glib_sys;
 use std::mem;
 use std::ptr;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use translate::{from_glib_full, from_glib_none, mut_override, ToGlib};
+
 use MainContext;
 use MainLoop;
 use Priority;
@@ -28,27 +30,48 @@ const DONE: usize = 3;
 #[repr(C)]
 struct TaskSource {
     source: glib_sys::GSource,
-    future: Option<(Box<Future<Item = (), Error = Never>>, Box<LocalMap>)>,
+    future: Option<FutureObj<'static, ()>>,
     thread: Option<usize>,
     state: AtomicUsize,
 }
 
-unsafe impl UnsafeWake for TaskSource {
-    unsafe fn clone_raw(&self) -> Waker {
-        Waker::new(glib_sys::g_source_ref(mut_override(&self.source)) as *const TaskSource)
-    }
+static TASK_SOURCE_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
+    TaskSource::clone_raw,
+    TaskSource::wake_raw,
+    TaskSource::wake_by_ref_raw,
+    TaskSource::drop_raw,
+);
 
-    unsafe fn drop_raw(&self) {
-        glib_sys::g_source_unref(mut_override(&self.source));
+impl TaskSource {
+    unsafe fn clone_raw(waker: *const ()) -> RawWaker {
+        let waker = &*(waker as *const TaskSource);
+        glib_sys::g_source_ref(mut_override(&waker.source));
+        RawWaker::new(waker as *const Self as *const (), &TASK_SOURCE_WAKER_VTABLE)
+    }
+    unsafe fn wake_raw(waker: *const ()) {
+        Self::wake_by_ref_raw(waker);
+        Self::drop_raw(waker);
     }
 
-    unsafe fn wake(&self) {
-        if self.state
-            .compare_and_swap(NOT_READY, READY, Ordering::SeqCst) == NOT_READY
+    unsafe fn wake_by_ref_raw(waker: *const ()) {
+        let waker = &*(waker as *const TaskSource);
+        if waker
+            .state
+            .compare_and_swap(NOT_READY, READY, Ordering::SeqCst)
+            == NOT_READY
         {
-            glib_sys::g_source_set_ready_time(mut_override(&self.source), 0);
+            glib_sys::g_source_set_ready_time(mut_override(&waker.source), 0);
         }
     }
+
+    unsafe fn drop_raw(waker: *const ()) {
+        let waker = &*(waker as *const TaskSource);
+        glib_sys::g_source_unref(mut_override(&waker.source));
+    }
+
+    fn as_waker(&self) -> Waker {
+        unsafe { Waker::from_raw(Self::clone_raw(self as *const Self as *const ())) }
+    }
 }
 
 unsafe extern "C" fn prepare(
@@ -66,7 +89,7 @@ unsafe extern "C" fn prepare(
         // XXX: This is not actually correct, we should not dispatch the
         // GSource here already but we need to know its current status so
         // that if it is not ready yet something can register to the waker
-        if let Async::Ready(()) = source.poll() {
+        if let Poll::Ready(()) = source.poll() {
             source.state.store(DONE, Ordering::SeqCst);
             cur = DONE;
         } else {
@@ -105,7 +128,7 @@ unsafe extern "C" fn dispatch(
         .state
         .compare_and_swap(READY, NOT_READY, Ordering::SeqCst);
     if cur == READY {
-        if let Async::Ready(()) = source.poll() {
+        if let Poll::Ready(()) = source.poll() {
             source.state.store(DONE, Ordering::SeqCst);
             cur = DONE;
         } else {
@@ -134,39 +157,31 @@ static SOURCE_FUNCS: glib_sys::GSourceFuncs = glib_sys::GSourceFuncs {
     closure_marshal: None,
 };
 
+unsafe impl Send for TaskSource {}
+unsafe impl Sync for TaskSource {}
+
 impl TaskSource {
     #[allow(clippy::new_ret_no_self)]
-    fn new(
-        priority: Priority,
-        future: Box<Future<Item = (), Error = Never> + 'static + Send>,
-    ) -> Source {
-        unsafe { Self::new_unsafe(priority, None, future) }
-    }
-
-    // NOTE: This does not have the Send bound and requires to be called from the same
-    // thread where the main context is running
-    unsafe fn new_unsafe(
-        priority: Priority,
-        thread: Option<usize>,
-        future: Box<Future<Item = (), Error = Never> + 'static>,
-    ) -> Source {
-        let source = glib_sys::g_source_new(
-            mut_override(&SOURCE_FUNCS),
-            mem::size_of::<TaskSource>() as u32,
-        );
-        {
-            let source = &mut *(source as *mut TaskSource);
-            ptr::write(&mut source.future, Some((future, Box::new(LocalMap::new()))));
-            source.thread = thread;
-            source.state = AtomicUsize::new(INIT);
-        }
+    fn new(priority: Priority, thread: Option<usize>, future: FutureObj<'static, ()>) -> Source {
+        unsafe {
+            let source = glib_sys::g_source_new(
+                mut_override(&SOURCE_FUNCS),
+                mem::size_of::<TaskSource>() as u32,
+            );
+            {
+                let source = &mut *(source as *mut TaskSource);
+                ptr::write(&mut source.future, Some(future));
+                source.thread = thread;
+                source.state = AtomicUsize::new(INIT);
+            }
 
-        glib_sys::g_source_set_priority(source, priority.to_glib());
+            glib_sys::g_source_set_priority(source, priority.to_glib());
 
-        from_glib_full(source)
+            from_glib_full(source)
+        }
     }
 
-    fn poll(&mut self) -> Async<()> {
+    fn poll(&mut self) -> Poll<()> {
         // Make sure that the first time we're polled that the current thread is remembered
         // and from there one we ensure that we're always polled from exactly the same thread.
         //
@@ -178,32 +193,34 @@ impl TaskSource {
                 *thread = Some(get_thread_id());
             }
             &mut Some(thread_id) => {
-                assert_eq!(get_thread_id(), thread_id,
-                           "Task polled on a different thread than before");
+                assert_eq!(
+                    get_thread_id(),
+                    thread_id,
+                    "Task polled on a different thread than before"
+                );
             }
         }
 
-        let waker = unsafe { self.clone_raw() };
+        let waker = self.as_waker();
         let source = &self.source as *const _;
         if let Some(ref mut future) = self.future {
-            let (ref mut future, ref mut local_map) = *future;
-
-            let mut executor: MainContext = unsafe {
-                from_glib_none(glib_sys::g_source_get_context(mut_override(source)))
-            };
+            let mut executor: MainContext =
+                unsafe { from_glib_none(glib_sys::g_source_get_context(mut_override(source))) };
 
-            assert!(executor.is_owner(), "Polling futures only allowed if the thread is owning the MainContext");
+            assert!(
+                executor.is_owner(),
+                "Polling futures only allowed if the thread is owning the MainContext"
+            );
 
             // Clone that we store in the task local data so that
             // it can be retrieved as needed
             executor.push_thread_default();
 
             let res = {
                 let enter = futures::executor::enter().unwrap();
-                let mut context =
-                    futures::task::Context::new(local_map, &waker, &mut executor);
+                let mut context = Context::from_waker(&waker);
 
-                let res = future.poll(&mut context).unwrap_or(Async::Ready(()));
+                let res = future.poll_unpin(&mut context);
 
                 drop(enter);
 
@@ -213,7 +230,7 @@ impl TaskSource {
             executor.pop_thread_default();
             res
         } else {
-            Async::Ready(())
+            Poll::Ready(())
         }
     }
 }
@@ -223,7 +240,7 @@ impl MainContext {
     ///
     /// This can be called from any thread and will execute the future from the thread
     /// where main context is running, e.g. via a `MainLoop`.
-    pub fn spawn<F: Future<Item = (), Error = Never> + Send + 'static>(&self, f: F) {
+    pub fn spawn<F: Future<Output = ()> + Send + 'static>(&self, f: F) {
         self.spawn_with_priority(::PRIORITY_DEFAULT, f);
     }
 
@@ -234,17 +251,21 @@ impl MainContext {
     /// This can be called only from the thread where the main context is running, e.g.
     /// from any other `Future` that is executed on this main context, or after calling
     /// `push_thread_default` or `acquire` on the main context.
-    pub fn spawn_local<F: Future<Item = (), Error = Never> + 'static>(&self, f: F) {
+    pub fn spawn_local<F: Future<Output = ()> + 'static>(&self, f: F) {
         self.spawn_local_with_priority(::PRIORITY_DEFAULT, f);
     }
 
     /// Spawn a new infallible `Future` on the main context, with a non-default priority.
     ///
     /// This can be called from any thread and will execute the future from the thread
     /// where main context is running, e.g. via a `MainLoop`.
-    pub fn spawn_with_priority<F: Future<Item = (), Error = Never> + Send + 'static>(&self, priority: Priority, f: F) {
-        let f = Box::new(f);
-        let source = TaskSource::new(priority, f);
+    pub fn spawn_with_priority<F: Future<Output = ()> + Send + 'static>(
+        &self,
+        priority: Priority,
+        f: F,
+    ) {
+        let f = FutureObj::new(Box::new(f));
+        let source = TaskSource::new(priority, None, f);
         source.attach(Some(&*self));
     }
 
@@ -255,13 +276,23 @@ impl MainContext {
     /// This can be called only from the thread where the main context is running, e.g.
     /// from any other `Future` that is executed on this main context, or after calling
     /// `push_thread_default` or `acquire` on the main context.
-    pub fn spawn_local_with_priority<F: Future<Item = (), Error = Never> + 'static>(&self, priority: Priority, f: F) {
-        assert!(self.is_owner(), "Spawning local futures only allowed on the thread owning the MainContext");
-        let f = Box::new(f);
+    pub fn spawn_local_with_priority<F: Future<Output = ()> + 'static>(
+        &self,
+        priority: Priority,
+        f: F,
+    ) {
+        assert!(
+            self.is_owner(),
+            "Spawning local futures only allowed on the thread owning the MainContext"
+        );
         unsafe {
-            // Ensure that this task is never polled on another thread
-            // than this one where it was spawned now.
-            let source = TaskSource::new_unsafe(priority, Some(get_thread_id()), f);
+            let f = LocalFutureObj::new(Box::new(f));
+            // We ensure here that we only ever run the future on this very task
+            // and that the futures executor is running on this task. Otherwise
+            // we will panic later.
+            // As such we can add the Send impl here safely
+            let f = f.into_future_obj();
+            let source = TaskSource::new(priority, Some(get_thread_id()), f);
             source.attach(Some(&*self));
         }
     }
@@ -274,7 +305,7 @@ impl MainContext {
     /// This must only be called if no `MainLoop` or anything else is running on this specific main
     /// context.
     #[allow(clippy::transmute_ptr_to_ptr)]
-    pub fn block_on<F: Future>(&self, f: F) -> Result<F::Item, F::Error> {
+    pub fn block_on<F: Future>(&self, f: F) -> F::Output {
         let mut res = None;
         let l = MainLoop::new(Some(&*self), false);
         let l_clone = l.clone();
@@ -283,17 +314,17 @@ impl MainContext {
             let f = f.then(|r| {
                 res = Some(r);
                 l_clone.quit();
-                Ok::<(), Never>(())
+                future::ready(())
             });
 
-            let f: *mut Future<Item = (), Error = Never> = Box::into_raw(Box::new(f));
-            // XXX: Transmute to get a 'static lifetime here, super unsafe
-            let f: *mut (Future<Item = (), Error = Never> + 'static) = mem::transmute(f);
-            let f: Box<Future<Item = (), Error = Never> + 'static> = Box::from_raw(f);
+            // Super-unsafe: We transmute here to get rid of the 'static lifetime
+            let f = LocalFutureObj::new(Box::new(f));
+            let f: (LocalFutureObj<'static, ()>) = mem::transmute(f);
+
+            // And ensure that we are only ever running on this very thread.
+            let f = f.into_future_obj();
 
-            // Ensure that this task is never polled on another thread
-            // than this one where it was spawned now.
-            let source = TaskSource::new_unsafe(::PRIORITY_DEFAULT, Some(get_thread_id()), f);
+            let source = TaskSource::new(::PRIORITY_DEFAULT, Some(get_thread_id()), f);
             source.attach(Some(&*self));
         }
 
@@ -303,10 +334,19 @@ impl MainContext {
     }
 }
 
-impl Executor for MainContext {
-    fn spawn(&mut self, f: Box<Future<Item = (), Error = Never> + Send>) -> Result<(), SpawnError> {
-        let f = Box::new(f);
-        let source = TaskSource::new(::PRIORITY_DEFAULT, f);
+impl Spawn for MainContext {
+    fn spawn_obj(&mut self, f: FutureObj<'static, ()>) -> Result<(), SpawnError> {
+        let source = TaskSource::new(::PRIORITY_DEFAULT, None, f);
+        source.attach(Some(&*self));
+        Ok(())
+    }
+}
+
+impl LocalSpawn for MainContext {
+    fn spawn_local_obj(&mut self, f: LocalFutureObj<'static, ()>) -> Result<(), SpawnError> {
+        let source = TaskSource::new(::PRIORITY_DEFAULT, Some(get_thread_id()), unsafe {
+            f.into_future_obj()
+        });
         source.attach(Some(&*self));
         Ok(())
     }
@@ -315,10 +355,9 @@ impl Executor for MainContext {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::thread;
-    use std::sync::mpsc;
-    use futures::future;
     use futures::channel::oneshot;
+    use std::sync::mpsc;
+    use std::thread;
 
     #[test]
     fn test_spawn() {
@@ -329,14 +368,15 @@ mod tests {
         let (o_sender, o_receiver) = oneshot::channel();
 
         let l_clone = l.clone();
-        c.spawn(o_receiver
-                .map_err(|_| unimplemented!())
+        c.spawn(
+            o_receiver
                 .and_then(move |()| {
                     sender.send(()).unwrap();
                     l_clone.quit();
 
-                    Ok(())
+                    future::ok(())
                 })
+                .then(|res| future::ready(res.unwrap())),
         );
 
         thread::spawn(move || {
@@ -357,8 +397,6 @@ mod tests {
         let l_clone = l.clone();
         c.spawn_local(future::lazy(move |_ctx| {
             l_clone.quit();
-
-            Ok(())
         }));
 
         l.run();
diff --git a/src/source_futures.rs b/src/source_futures.rs
