wutil-rs/src/notification.rs

@@ -1,10 +1,9 @@
+use crate::sendable::Sendable;
 use std::{
     collections::{btree_map::Entry, BTreeMap},
     ffi::{c_void, CStr},
-    ptr,
     sync::Mutex,
 };
-use crate::sendable::Sendable;
 
 // Helper function for adding the entry `(key, (observer, action))` to `map`.
 fn register<K: Eq + Ord>(
@@ -90,15 +89,21 @@ impl NotificationCenter {
         }
     }
 
-    /// Provides access to the default, process-wide notification center. The
-    /// FFI C API uses this notification center. This is protected behind a
-    /// mutex that is held while `f` is run, so panicking inside of `f` should
-    /// be avoided.
+    /// Provides access to the default, process-wide notification center.
+    ///
+    /// The FFI C API uses this notification center. This is protected behind a
+    /// mutex that is held while `f` is run, so:
+    ///
+    /// * Panicking inside of `f` should be avoided.
+    /// * Triggering notification dispatch recursively (as when a notification
+    ///   callback dispatches another notification) must be done with care. This
+    ///   is handled correctly by [`ffi::WMPostNotificationName`], which does not
+    ///   hold the mutex during callback execution.
     pub fn with_global_default<R>(f: impl FnOnce(&mut Self) -> R) -> R {
         static INSTANCE: Mutex<NotificationCenter> = Mutex::new(NotificationCenter::new());
         f(&mut INSTANCE
             .try_lock()
-            .unwrap())
+            .expect("cannot unlock notification center (recursive notification dispatch?)"))
     }
 
     /// Registers `action` to be invoked and invoked on `observer` when
@@ -145,7 +150,7 @@ impl NotificationCenter {
     pub fn dispatch(&mut self, notification: Notification) {
         if let Some(observers) = self.by_name.get_mut(notification.name) {
             for (observer, action) in observers {
-                let observer = observer.map(|x| x.ptr.as_ptr()).unwrap_or(ptr::null_mut());
+                let observer = Sendable::as_ptr(*observer);
                 unsafe {
                     (action)(observer, &notification);
                 }
@@ -155,7 +160,7 @@ impl NotificationCenter {
         if let Some(source) = notification.source {
             if let Some(observers) = self.exact.get_mut(&(notification.name, source)) {
                 for (observer, action) in observers {
-                    let observer = observer.map(|x| x.ptr.as_ptr()).unwrap_or(ptr::null_mut());
+                    let observer = Sendable::as_ptr(*observer);
                     unsafe {
                         (action)(observer, &notification);
                     }
@@ -163,7 +168,7 @@ impl NotificationCenter {
             }
             if let Some(observers) = self.by_source.get_mut(&source) {
                 for (observer, action) in observers {
-                    let observer = observer.map(|x| x.ptr.as_ptr()).unwrap_or(ptr::null_mut());
+                    let observer = Sendable::as_ptr(*observer);
                     unsafe {
                         (action)(observer, &notification);
                     }
@@ -172,13 +177,57 @@ impl NotificationCenter {
         }
 
         for (observer, action) in &mut self.universal {
-            let observer = observer.map(|x| x.ptr.as_ptr()).unwrap_or(ptr::null_mut());
+            let observer = Sendable::as_ptr(*observer);
             unsafe {
                 (action)(observer, &notification);
             }
         }
     }
 
+    /// Returns an independent callback that dispatches `notification` when
+    /// invoked.
+    ///
+    /// This should be used when `self` is under a non-reentrant lock (such as a
+    /// mutex) which renders it inaccessible when dispatching
+    /// `notification`. Callback pointers are copied into the return value,
+    /// after which the lock on `self` may be relinquished, and then the return
+    /// value may be invoked, and notification dispatch will actually happen.
+    fn dispatch_actions(&self, notification: Notification) -> Box<dyn FnOnce()> {
+        let mut actions: Vec<(Option<Sendable>, Action)> = Vec::new();
+
+        if let Some(observers) = self.by_name.get(notification.name) {
+            for x in observers.iter().copied() {
+                actions.push(x);
+            }
+        }
+
+        if let Some(source) = notification.source {
+            if let Some(observers) = self.exact.get(&(notification.name, source)) {
+                for x in observers.iter().copied() {
+                    actions.push(x);
+                }
+            }
+            if let Some(observers) = self.by_source.get(&source) {
+                for x in observers.iter().copied() {
+                    actions.push(x);
+                }
+            }
+        }
+
+        for x in self.universal.iter().copied() {
+            actions.push(x);
+        }
+
+        Box::new(move || {
+            for (observer, action) in actions.into_iter() {
+                let observer = Sendable::as_ptr(observer);
+                unsafe {
+                    (action)(observer, &notification);
+                }
+            }
+        })
+    }
+
     /// Removes all notification subscriptions that would notify `observer` if they fired.
     pub fn remove_observer(&mut self, observer: Sendable) {
         self.exact.retain(|_, values| {
@@ -319,14 +368,16 @@ pub mod ffi {
         }
         let name = unsafe { CStr::from_ptr(name) };
         let source = unsafe { Sendable::from_nullable(object) };
-        let client_data = unsafe {Sendable::from_nullable(client_data) };
-        NotificationCenter::with_global_default(|c| {
-            c.dispatch(Notification {
+        let client_data = unsafe { Sendable::from_nullable(client_data) };
+
+        let todo = NotificationCenter::with_global_default(|c| {
+            c.dispatch_actions(Notification {
                 name,
                 source,
                 client_data,
             })
         });
+        (todo)();
     }
 
     /// Resets all notifications for the global notification center. Used by
@@ -336,3 +387,67 @@ pub mod ffi {
         NotificationCenter::with_global_default(|c| c.clear());
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::ffi::*;
+    use super::*;
+    use std::ffi::CStr;
+    use std::ptr;
+    use std::sync::atomic;
+
+    #[test]
+    fn reentrant_global_notification() {
+        static NOTIFICATION_A_NAME: &'static CStr = c"MyNotificationA";
+        static NOTIFICATION_B_NAME: &'static CStr = c"MyNotificationB";
+        static DATA_A: atomic::AtomicU16 = atomic::AtomicU16::new(0);
+        static DATA_B: atomic::AtomicU16 = atomic::AtomicU16::new(0);
+
+        unsafe extern "C" fn action_a(observer: *mut c_void, notification: *const Notification) {
+            assert!(observer.is_null());
+            assert!(!notification.is_null());
+            assert_eq!(DATA_A.fetch_add(1, atomic::Ordering::SeqCst), 0);
+            unsafe {
+                // Trigger action_b.
+                WMPostNotificationName(
+                    NOTIFICATION_B_NAME.as_ptr(),
+                    ptr::null_mut(),
+                    ptr::null_mut(),
+                );
+            }
+        }
+
+        unsafe extern "C" fn action_b(observer: *mut c_void, notification: *const Notification) {
+            assert!(observer.is_null());
+            assert!(!notification.is_null());
+            assert_eq!(DATA_A.load(atomic::Ordering::SeqCst), 1);
+            assert_eq!(DATA_B.fetch_add(1, atomic::Ordering::SeqCst), 0);
+        }
+
+        assert_eq!(DATA_A.load(atomic::Ordering::SeqCst), 0);
+        assert_eq!(DATA_B.load(atomic::Ordering::SeqCst), 0);
+
+        unsafe {
+            WMAddNotificationObserver(
+                Some(action_a),
+                ptr::null_mut(),
+                NOTIFICATION_A_NAME.as_ptr(),
+                ptr::null_mut(),
+            );
+            WMAddNotificationObserver(
+                Some(action_b),
+                ptr::null_mut(),
+                NOTIFICATION_B_NAME.as_ptr(),
+                ptr::null_mut(),
+            );
+            WMPostNotificationName(
+                NOTIFICATION_A_NAME.as_ptr(),
+                ptr::null_mut(),
+                ptr::null_mut(),
+            );
+        }
+
+        assert_eq!(DATA_A.load(atomic::Ordering::SeqCst), 1);
+        assert_eq!(DATA_B.load(atomic::Ordering::SeqCst), 1);
+    }
+}