Refactor dispatch2

Author: pronebird

crates/dispatch2/Cargo.toml

@@ -38,7 +38,7 @@ targets = [
 ]
 
 [features]
-default = ["std"]
+default = ["std", "objc2"]
 
 # Currently not possible to turn off, put here for forwards compatibility.
 std = ["alloc", "bitflags/std", "block2?/std", "libc?/std", "objc2?/std"]

crates/dispatch2/src/ffi.rs

@@ -2,8 +2,10 @@
 
 #![allow(missing_docs, non_camel_case_types)]
 
-use core::ffi::{c_long, c_uint, c_ulong, c_void};
-use std::ptr::addr_of;
+use std::{
+    ffi::{c_long, c_uint, c_ulong, c_void},
+    ptr::addr_of,
+};
 
 #[cfg(feature = "objc2")]
 use objc2::encode::{Encode, Encoding, RefEncode};
@@ -108,10 +110,11 @@ create_opaque_type!(dispatch_io_s, dispatch_io_t);
 
 /// A dispatch queue that executes blocks serially in FIFO order.
 pub const DISPATCH_QUEUE_SERIAL: dispatch_queue_attr_t = core::ptr::null_mut();
+
 /// A dispatch queue that executes blocks concurrently.
-pub static DISPATCH_QUEUE_CONCURRENT: &dispatch_queue_attr_s = {
+pub const DISPATCH_QUEUE_CONCURRENT: dispatch_queue_attr_t = {
     // Safety: immutable external definition
-    unsafe { &_dispatch_queue_attr_concurrent }
+    unsafe { &_dispatch_queue_attr_concurrent as *const _ as dispatch_queue_attr_t }
 };
 
 pub const DISPATCH_APPLY_AUTO: dispatch_queue_t = core::ptr::null_mut();

crates/dispatch2/src/group.rs

@@ -1,38 +1,25 @@
 //! Dispatch group definition.
 
-use std::time::Duration;
+use std::{ffi::c_void, time::Duration};
 
-use core::ffi::c_void;
-
-use super::object::DispatchObject;
-use super::queue::Queue;
-use super::utils::function_wrapper;
-use super::{ffi::*, WaitError};
+use super::{ffi::*, function_wrapper, queue::Queue, rc::Retained, AsRawDispatchObject, WaitError};
 
 /// Dispatch group.
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub struct Group {
-    dispatch_object: DispatchObject<dispatch_group_s>,
+    inner: Retained<dispatch_group_s>,
 }
 
-/// Dispatch group guard.
-#[derive(Debug)]
-pub struct GroupGuard(Group, bool);
-
 impl Group {
     /// Creates a new [Group].
     pub fn new() -> Option<Self> {
         // Safety: valid to call.
         let object = unsafe { dispatch_group_create() };
 
-        if object.is_null() {
-            return None;
-        }
-
-        // Safety: object cannot be null.
-        let dispatch_object = unsafe { DispatchObject::new_owned(object.cast()) };
+        // Safety: retained accepts null pointer.
+        let inner = unsafe { Retained::from_raw(object)? };
 
-        Some(Group { dispatch_object })
+        Some(Group { inner })
     }
 
     /// Submit a function to a [Queue] and associates it with the [Group].
@@ -104,25 +91,44 @@ impl Group {
         GroupGuard(self.clone(), false)
     }
 
-    /// Set the finalizer function for the object.
-    pub fn set_finalizer<F>(&mut self, destructor: F)
-    where
-        F: Send + FnOnce(),
-    {
-        self.dispatch_object.set_finalizer(destructor);
-    }
-
     /// Get the raw [dispatch_group_t] value.
     ///
     /// # Safety
     ///
     /// - Object shouldn't be released manually.
-    pub const unsafe fn as_raw(&self) -> dispatch_group_t {
-        // SAFETY: Upheld by caller
-        unsafe { self.dispatch_object.as_raw() }
+    pub fn as_raw(&self) -> dispatch_group_t {
+        // Safety: Upheld by caller
+        Retained::as_ptr(&self.inner).cast_mut()
     }
 }
 
+impl Clone for Group {
+    fn clone(&self) -> Self {
+        Self {
+            // Safety: pointer must be valid.
+            inner: unsafe {
+                Retained::retain(self.as_raw()).expect("failed to retain dispatch_group")
+            },
+        }
+    }
+}
+
+impl AsRawDispatchObject for Group {
+    fn as_raw_object(&self) -> dispatch_object_t {
+        self.as_raw().cast()
+    }
+}
+
+// Safety: group is inherently safe to move between threads.
+unsafe impl Send for Group {}
+
+// Safety: group is inherently safe to share between threads.
+unsafe impl Sync for Group {}
+
+/// Dispatch group guard.
+#[derive(Debug)]
+pub struct GroupGuard(Group, bool);
+
 impl GroupGuard {
     /// Explicitly indicates that the function in the [Group] finished executing.
     pub fn leave(mut self) {

crates/dispatch2/src/lib.rs

@@ -24,8 +24,6 @@
 // Update in Cargo.toml as well.
 #![doc(html_root_url = "https://docs.rs/dispatch2/0.1.0")]
 
-use self::ffi::dispatch_qos_class_t;
-
 pub mod ffi;
 #[allow(clippy::undocumented_unsafe_blocks)]
 mod generated;
@@ -35,8 +33,22 @@ mod main_thread_bound;
 pub mod object;
 mod once;
 pub mod queue;
+mod rc;
 pub mod semaphore;
-mod utils;
+pub mod workloop;
+
+#[cfg(feature = "objc2")]
+pub use self::main_thread_bound::{run_on_main, MainThreadBound};
+pub use self::once::*;
+pub use group::*;
+pub use object::*;
+pub use queue::*;
+pub use semaphore::*;
+pub use workloop::*;
+
+use std::{ffi::c_void, time::Duration};
+
+use ffi::{dispatch_qos_class_t, dispatch_time, dispatch_time_t, DISPATCH_TIME_NOW};
 
 /// Wait error.
 #[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
@@ -49,14 +61,15 @@ pub enum WaitError {
 }
 
 /// Quality of service that specify the priorities for executing tasks.
-#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
+#[derive(Debug, Default, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
 #[non_exhaustive]
 pub enum QualityOfServiceClass {
     /// Quality of service for user-interactive tasks.
     UserInteractive,
     /// Quality of service for tasks that prevent the user from actively using your app.
     UserInitiated,
     /// Default Quality of service.
+    #[default]
     Default,
     /// Quality of service for tasks that the user does not track actively.
     Utility,
@@ -82,10 +95,44 @@ impl From<QualityOfServiceClass> for dispatch_qos_class_t {
     }
 }
 
-pub use self::group::*;
-#[cfg(feature = "objc2")]
-pub use self::main_thread_bound::{run_on_main, MainThreadBound};
-pub use self::object::*;
-pub use self::once::*;
-pub use self::queue::*;
-pub use self::semaphore::*;
+impl TryFrom<Duration> for dispatch_time_t {
+    type Error = TryFromDurationError;
+
+    fn try_from(value: Duration) -> Result<Self, Self::Error> {
+        let secs = value.as_secs() as i64;
+
+        secs.checked_mul(1_000_000_000)
+            .and_then(|x| x.checked_add(i64::from(value.subsec_nanos())))
+            .map(|delta| {
+                // Safety: delta cannot overflow
+                unsafe { dispatch_time(DISPATCH_TIME_NOW, delta) }
+            })
+            .ok_or(Self::Error::TimeOverflow)
+    }
+}
+
+/// Error returned by [Queue::after].
+#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
+#[non_exhaustive]
+pub enum TryFromDurationError {
+    /// The given timeout value will result in an overflow when converting to dispatch time.
+    TimeOverflow,
+}
+
+/// Error returned by [Queue::set_qos_class_floor] or [WorkloopQueue::set_qos_class_floor].
+#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
+#[non_exhaustive]
+pub enum QualityOfServiceClassFloorError {
+    /// The relative priority is invalid.
+    InvalidRelativePriority,
+}
+
+pub(crate) extern "C" fn function_wrapper<F>(work_boxed: *mut c_void)
+where
+    F: FnOnce(),
+{
+    // Safety: we reconstruct from a Box.
+    let work = unsafe { Box::from_raw(work_boxed.cast::<F>()) };
+
+    (*work)();
+}