[WIP] Have EventsLoopProxy::wakeup return a Result. Begin linux impl.

X11 and Wayland implementations are now half implemented, however both
still do not correctly break from the inner blocking event dispatch
functions when `wakeup` is called, which they should do.

examples/proxy.rs

@@ -14,7 +14,7 @@ fn main() {
         // Wake up the `events_loop` once every second.
         loop {
             std::thread::sleep(std::time::Duration::from_secs(1));
-            proxy.wakeup();
+            proxy.wakeup().unwrap();
         }
     });
 

src/events.rs

@@ -11,12 +11,11 @@ pub enum Event {
         device_id: DeviceId,
         event: DeviceEvent,
     },
+    Awakened,
 }
 
 #[derive(Clone, Debug)]
 pub enum WindowEvent {
-    // TODO: remove ; can break the lib internally so be careful
-    Awakened,
 
     /// The size of the window has changed.
     Resized(u32, u32),

src/lib.rs

@@ -189,12 +189,7 @@ pub struct ButtonId(u32);
 ///
 /// To wake up an `EventsLoop` from a another thread, see the `EventsLoopProxy` docs.
 pub struct EventsLoop {
-    events_loop: platform::EventsLoop,
+    events_loop: Arc<platform::EventsLoop>,
-}
-
-/// Used to wake up the `EventsLoop` from another thread.
-pub struct EventsLoopProxy {
-    events_loop_proxy: platform::EventsLoopProxy,
 }
 
 impl EventsLoop {
@@ -232,17 +227,41 @@ impl EventsLoop {
     /// thread.
     pub fn create_proxy(&self) -> EventsLoopProxy {
         EventsLoopProxy {
-            events_loop_proxy: platform::EventsLoopProxy::new(&self.events_loop),
+            events_loop_proxy: self.events_loop.create_proxy(),
         }
     }
 }
 
+/// Used to wake up the `EventsLoop` from another thread.
+pub struct EventsLoopProxy {
+    events_loop_proxy: platform::EventsLoopProxy,
+}
+
 impl EventsLoopProxy {
     /// Wake up the `EventsLoop` from which this proxy was created.
     ///
     /// This causes the `EventsLoop` to emit an `Awakened` event.
-    pub fn wakeup(&self) {
+    ///
-        self.events_loop_proxy.wakeup();
+    /// Returns an `Err` if the associated `EventsLoop` no longer exists.
+    pub fn wakeup(&self) -> Result<(), EventsLoopClosed> {
+        self.events_loop_proxy.wakeup()
+    }
+}
+
+/// The error that is returned when an `EventsLoopProxy` attempts to wake up an `EventsLoop` that
+/// no longer exists.
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
+pub struct EventsLoopClosed;
+
+impl std::fmt::Display for EventsLoopClosed {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "{}", std::error::Error::description(self))
+    }
+}
+
+impl std::error::Error for EventsLoopClosed {
+    fn description(&self) -> &str {
+        "Tried to wake up a closed `EventsLoop`"
     }
 }
 

src/platform/linux/mod.rs

@@ -3,9 +3,7 @@
 use std::collections::VecDeque;
 use std::sync::Arc;
 
-use CreationError;
+use {CreationError, CursorState, EventsLoopClosed, MouseCursor};
-use CursorState;
-use MouseCursor;
 use libc;
 
 use self::x11::XConnection;
@@ -309,6 +307,11 @@ pub enum EventsLoop {
     X(x11::EventsLoop)
 }
 
+pub enum EventsLoopProxy {
+    X(x11::EventsLoopProxy),
+    Wayland(wayland::EventsLoopProxy),
+}
+
 impl EventsLoop {
     pub fn new() -> EventsLoop {
         match *UNIX_BACKEND {
@@ -326,6 +329,13 @@ impl EventsLoop {
         }
     }
 
+    pub fn create_proxy(&self) -> EventsLoopProxy {
+        match *self {
+            EventsLoop::Wayland(ref evlp) => EventsLoopProxy::Wayland(evlp.create_proxy()),
+            EventsLoop::X(ref evlp) => EventsLoopProxy::X(evlp.create_proxy()),
+        }
+    }
+
     pub fn interrupt(&self) {
         match *self {
             EventsLoop::Wayland(ref evlp) => evlp.interrupt(),
@@ -351,3 +361,12 @@ impl EventsLoop {
         }
     }
 }
+
+impl EventsLoopProxy {
+    pub fn wakeup(&self) -> Result<(), EventsLoopClosed> {
+        match *self {
+            EventsLoopProxy::Wayland(ref proxy) => proxy.wakeup(),
+            EventsLoopProxy::X(ref proxy) => proxy.wakeup(),
+        }
+    }
+}

src/platform/linux/wayland/event_loop.rs

@@ -1,7 +1,7 @@
-use {WindowEvent as Event, ElementState, MouseButton, MouseScrollDelta, TouchPhase, ModifiersState, KeyboardInput};
+use {WindowEvent as Event, ElementState, MouseButton, MouseScrollDelta, TouchPhase, ModifiersState, KeyboardInput, EventsLoopClosed};
 
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, Weak};
-use std::sync::atomic::AtomicBool;
+use std::sync::atomic::{self, AtomicBool};
 
 use super::{DecoratedHandler, WindowId, DeviceId, WaylandContext};
 
@@ -71,7 +71,38 @@ pub struct EventsLoop {
     interrupted: AtomicBool,
     // trigger cleanup of the dead surfaces
     cleanup_needed: Arc<AtomicBool>,
-    hid: usize
+    // Whether or not there is a pending `Awakened` event to be emitted.
+    pending_wakeup: Arc<AtomicBool>,
+    hid: usize,
+}
+
+// A handle that can be sent across threads and used to wake up the `EventsLoop`.
+//
+// We should only try and wake up the `EventsLoop` if it still exists, so we hold Weak ptrs.
+pub struct EventsLoopProxy {
+    ctxt: Weak<WaylandContext>,
+    pending_wakeup: Weak<AtomicBool>,
+}
+
+impl EventsLoopProxy {
+    // Causes the `EventsLoop` to stop blocking on `run_forever` and emit an `Awakened` event.
+    //
+    // Returns `Err` if the associated `EventsLoop` no longer exists.
+    pub fn wakeup(&self) -> Result<(), EventsLoopClosed> {
+        let ctxt = self.ctxt.upgrade();
+        let wakeup = self.pending_wakeup.upgrade();
+        match (ctxt, wakeup) {
+            (Some(ctxt), Some(wakeup)) => {
+                // Update the `EventsLoop`'s `pending_wakeup` flag.
+                wakeup.store(true, atomic::Ordering::Relaxed);
+                // TODO:
+                // Cause the `EventsLoop` to break from `dispatch` if it is currently blocked.
+                ctxt.display.sync();
+                Ok(())
+            },
+            _ => Err(EventsLoopClosed),
+        }
+    }
 }
 
 impl EventsLoop {
@@ -85,11 +116,19 @@ impl EventsLoop {
             decorated_ids: Mutex::new(Vec::new()),
             sink: sink,
             interrupted: AtomicBool::new(false),
+            pending_wakeup: Arc::new(AtomicBool::new(false)),
             cleanup_needed: Arc::new(AtomicBool::new(false)),
             hid: hid
         }
     }
 
+    pub fn create_proxy(&self) -> EventsLoopProxy {
+        EventsLoopProxy {
+            ctxt: Arc::downgrade(&self.ctxt),
+            pending_wakeup: Arc::downgrade(&self.pending_wakeup),
+        }
+    }
+
     // some internals that Window needs access to
     pub fn get_window_init(&self) -> (Arc<Mutex<EventQueue>>, Arc<AtomicBool>) {
         (self.evq.clone(), self.cleanup_needed.clone())
@@ -120,7 +159,7 @@ impl EventsLoop {
     }
 
     pub fn interrupt(&self) {
-        self.interrupted.store(true, ::std::sync::atomic::Ordering::Relaxed);
+        self.interrupted.store(true, atomic::Ordering::Relaxed);
     }
 
     fn prune_dead_windows(&self) {
@@ -160,6 +199,8 @@ impl EventsLoop {
         self.ctxt.dispatch_pending();
         evq_guard.dispatch_pending().expect("Wayland connection unexpectedly lost");
 
+        self.emit_pending_wakeup();
+
         {
             let mut sink_guard = self.sink.lock().unwrap();
 
@@ -173,7 +214,7 @@ impl EventsLoop {
             unsafe { sink_guard.set_callback(old_cb) };
         }
 
-        if self.cleanup_needed.swap(false, ::std::sync::atomic::Ordering::Relaxed) {
+        if self.cleanup_needed.swap(false, atomic::Ordering::Relaxed) {
             self.prune_dead_windows()
         }
     }
@@ -181,7 +222,7 @@ impl EventsLoop {
     pub fn run_forever<F>(&self, callback: F)
         where F: FnMut(::Event)
     {
-        self.interrupted.store(false, ::std::sync::atomic::Ordering::Relaxed);
+        self.interrupted.store(false, atomic::Ordering::Relaxed);
 
         // send pending requests to the server...
         self.ctxt.flush();
@@ -195,16 +236,19 @@ impl EventsLoop {
         let static_cb = unsafe { ::std::mem::transmute(Box::new(callback) as Box<FnMut(_)>) };
         let old_cb = unsafe { self.sink.lock().unwrap().set_callback(static_cb) };
 
-        while !self.interrupted.load(::std::sync::atomic::Ordering::Relaxed) {
+        while !self.interrupted.load(atomic::Ordering::Relaxed) {
             self.ctxt.dispatch();
             evq_guard.dispatch_pending().expect("Wayland connection unexpectedly lost");
+
+            self.emit_pending_wakeup();
+
             let ids_guard = self.decorated_ids.lock().unwrap();
             self.sink.lock().unwrap().with_callback(
                 |cb| Self::process_resize(&mut evq_guard, &ids_guard, cb)
             );
             self.ctxt.flush();
 
-            if self.cleanup_needed.swap(false, ::std::sync::atomic::Ordering::Relaxed) {
+            if self.cleanup_needed.swap(false, atomic::Ordering::Relaxed) {
                 self.prune_dead_windows()
             }
         }
@@ -212,6 +256,14 @@ impl EventsLoop {
         // replace the old noop callback
         unsafe { self.sink.lock().unwrap().set_callback(old_cb) };
     }
+
+    // If an `EventsLoopProxy` has signalled a wakeup, emit an event and reset the flag.
+    fn emit_pending_wakeup(&self) {
+        if self.pending_wakeup.load(atomic::Ordering::Relaxed) {
+            self.sink.lock().unwrap().with_callback(|cb| cb(::Event::Awakened));
+            self.pending_wakeup.store(false, atomic::Ordering::Relaxed);
+        }
+    }
 }
 
 enum KbdType {

src/platform/linux/wayland/mod.rs

@@ -1,7 +1,7 @@
 #![cfg(any(target_os = "linux", target_os = "dragonfly", target_os = "freebsd", target_os = "openbsd"))]
 
 pub use self::window::{Window, WindowId};
-pub use self::event_loop::EventsLoop;
+pub use self::event_loop::{EventsLoop, EventsLoopProxy};
 pub use self::context::{WaylandContext, MonitorId, get_available_monitors,
                         get_primary_monitor};
 

src/platform/linux/x11/mod.rs

@@ -7,10 +7,11 @@ pub use self::xdisplay::{XConnection, XNotSupported, XError};
 pub mod ffi;
 
 use platform::PlatformSpecificWindowBuilderAttributes;
-use {CreationError, Event, WindowEvent, DeviceEvent, AxisId, ButtonId, KeyboardInput};
+use {CreationError, Event, EventsLoopClosed, WindowEvent, DeviceEvent, AxisId, ButtonId, KeyboardInput};
 
 use std::{mem, ptr, slice};
 use std::sync::{Arc, Mutex, Weak};
+use std::sync::atomic::{self, AtomicBool};
 use std::collections::HashMap;
 use std::ffi::CStr;
 
@@ -29,15 +30,20 @@ mod xdisplay;
 // the one generated by the macro.
 
 pub struct EventsLoop {
-    interrupted: ::std::sync::atomic::AtomicBool,
+    interrupted: AtomicBool,
     display: Arc<XConnection>,
     wm_delete_window: ffi::Atom,
     windows: Mutex<HashMap<WindowId, WindowData>>,
     devices: Mutex<HashMap<DeviceId, Device>>,
     xi2ext: XExtension,
+    pending_wakeup: Arc<AtomicBool>,
     root: ffi::Window,
 }
 
+pub struct EventsLoopProxy {
+    pending_wakeup: Weak<AtomicBool>,
+}
+
 impl EventsLoop {
     pub fn new(display: Arc<XConnection>) -> EventsLoop {
         let wm_delete_window = unsafe { (display.xlib.XInternAtom)(display.display, b"WM_DELETE_WINDOW\0".as_ptr() as *const c_char, 0) };
@@ -73,7 +79,8 @@ impl EventsLoop {
         let root = unsafe { (display.xlib.XDefaultRootWindow)(display.display) };
 
         let result = EventsLoop {
-            interrupted: ::std::sync::atomic::AtomicBool::new(false),
+            interrupted: AtomicBool::new(false),
+            pending_wakeup: Arc::new(AtomicBool::new(false)),
             display: display,
             wm_delete_window: wm_delete_window,
             windows: Mutex::new(HashMap::new()),
@@ -101,8 +108,14 @@ impl EventsLoop {
         result
     }
 
+    pub fn create_proxy(&self) -> EventsLoopProxy {
+        EventsLoopProxy {
+            pending_wakeup: Arc::downgrade(&self.pending_wakeup),
+        }
+    }
+
     pub fn interrupt(&self) {
-        self.interrupted.store(true, ::std::sync::atomic::Ordering::Relaxed);
+        self.interrupted.store(true, atomic::Ordering::Relaxed);
 
         // Push an event on the X event queue so that methods like run_forever will advance.
         let mut xev = ffi::XClientMessageEvent {
@@ -126,7 +139,7 @@ impl EventsLoop {
     pub fn poll_events<F>(&self, mut callback: F)
         where F: FnMut(Event)
     {
-        self.interrupted.store(false, ::std::sync::atomic::Ordering::Relaxed);
+        self.interrupted.store(false, atomic::Ordering::Relaxed);
         let xlib = &self.display.xlib;
 
         let mut xev = unsafe { mem::uninitialized() };
@@ -142,7 +155,7 @@ impl EventsLoop {
                 (xlib.XNextEvent)(self.display.display, &mut xev);
             }
             self.process_event(&mut xev, &mut callback);
-            if self.interrupted.load(::std::sync::atomic::Ordering::Relaxed) {
+            if self.interrupted.load(atomic::Ordering::Relaxed) {
                 break;
             }
         }
@@ -151,7 +164,7 @@ impl EventsLoop {
     pub fn run_forever<F>(&self, mut callback: F)
         where F: FnMut(Event)
     {
-        self.interrupted.store(false, ::std::sync::atomic::Ordering::Relaxed);
+        self.interrupted.store(false, atomic::Ordering::Relaxed);
 
         let xlib = &self.display.xlib;
 
@@ -160,7 +173,7 @@ impl EventsLoop {
         loop {
             unsafe { (xlib.XNextEvent)(self.display.display, &mut xev) }; // Blocks as necessary
             self.process_event(&mut xev, &mut callback);
-            if self.interrupted.load(::std::sync::atomic::Ordering::Relaxed) {
+            if self.interrupted.load(atomic::Ordering::Relaxed) {
                 break;
             }
         }
@@ -197,7 +210,7 @@ impl EventsLoop {
                     callback(Event::WindowEvent { window_id: wid, event: WindowEvent::Closed })
                 } else {
                     // FIXME: Prone to spurious wakeups
-                    callback(Event::WindowEvent { window_id: wid, event: WindowEvent::Awakened })
+                    callback(Event::Awakened)
                 }
             }
 
@@ -523,6 +536,19 @@ impl EventsLoop {
     }
 }
 
+impl EventsLoopProxy {
+    pub fn wakeup(&self) -> Result<(), EventsLoopClosed> {
+        // Update the `EventsLoop`'s `pending_wakeup` flag.
+        match self.pending_wakeup.upgrade() {
+            Some(wakeup) => Ok(wakeup.store(true, atomic::Ordering::Relaxed)),
+            None => Err(EventsLoopClosed),
+        }
+
+        // TODO:
+        // Cause the `EventsLoop` to break if it is currently blocked.
+    }
+}
+
 struct DeviceInfo<'a> {
     display: &'a XConnection,
     info: *const ffi::XIDeviceInfo,