apple_cf/
dispatch_queue.rs

1//! Dispatch Queue wrapper for custom queue management
2//!
3//! This module provides a safe Rust wrapper around GCD (Grand Central Dispatch) queues
4//! that can be used with `ScreenCaptureKit` streams.
5//!
6//! ## When to Use Custom Queues
7//!
8//! By default, stream output handlers are called on a system-managed queue. Use a custom
9//! queue when you need:
10//!
11//! - **Priority control** - Use `UserInteractive` `QoS` for low-latency UI updates
12//! - **Thread isolation** - Ensure handlers run on a specific queue
13//! - **Performance tuning** - Adjust queue priority based on your app's needs
14//!
15//! ## Example
16//!
17#![allow(clippy::missing_panics_doc)]
18
19//! ```rust,no_run
20//! use apple_cf::dispatch_queue::{dispatch_async_and_wait, DispatchQueue, DispatchQoS};
21//!
22//! // Create a high-priority queue for frame processing
23//! let queue = DispatchQueue::new("com.myapp.capture", DispatchQoS::UserInteractive);
24//! dispatch_async_and_wait(&queue, || {
25//!     // do queue-bound work here
26//! });
27//! ```
28
29use crate::utils::panic_safe;
30use std::cell::RefCell;
31use std::ffi::c_void;
32use std::fmt;
33use std::marker::PhantomData;
34use std::rc::Rc;
35use std::sync::Arc;
36use std::time::Duration;
37
38/// Quality of Service levels for dispatch queues
39///
40/// These `QoS` levels help the system prioritize work appropriately.
41///
42/// # Examples
43///
44/// ```
45/// use apple_cf::dispatch_queue::{DispatchQueue, DispatchQoS};
46///
47/// // High priority for UI-affecting work
48/// let queue = DispatchQueue::new("com.myapp.ui", DispatchQoS::UserInteractive);
49///
50/// // Lower priority for background tasks
51/// let bg_queue = DispatchQueue::new("com.myapp.background", DispatchQoS::Background);
52/// ```
53#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
54pub enum DispatchQoS {
55    /// Background `QoS` - for maintenance or cleanup tasks
56    Background = 0,
57    /// Utility `QoS` - for tasks that may take some time
58    Utility = 1,
59    /// Default `QoS` - standard priority
60    #[default]
61    Default = 2,
62    /// User Initiated `QoS` - for tasks initiated by the user
63    UserInitiated = 3,
64    /// User Interactive `QoS` - for tasks that affect the UI
65    UserInteractive = 4,
66}
67
68/// A wrapper around GCD `DispatchQueue`
69///
70/// This allows you to provide a custom dispatch queue for stream output handling
71/// instead of using the default queue.
72///
73/// # Example
74///
75/// ```no_run
76/// use apple_cf::dispatch_queue::{DispatchQueue, DispatchQoS};
77///
78/// let queue = DispatchQueue::new("com.myapp.capture", DispatchQoS::UserInteractive);
79/// ```
80pub struct DispatchQueue {
81    ptr: *const c_void,
82}
83
84// SAFETY: `dispatch_queue_t` is documented by Apple as safe to use from any
85// thread.  The Rust wrapper only holds an opaque pointer and never performs
86// interior mutation on it outside of the underlying GCD primitives, which are
87// themselves thread-safe.
88unsafe impl Send for DispatchQueue {}
89unsafe impl Sync for DispatchQueue {}
90
91impl DispatchQueue {
92    /// Creates a new dispatch queue with the specified label and `QoS`
93    ///
94    /// # Arguments
95    ///
96    /// * `label` - A string label for the queue (e.g., "com.myapp.capture")
97    /// * `qos` - The quality of service level for the queue
98    ///
99    /// # Examples
100    ///
101    /// ```
102    /// use apple_cf::dispatch_queue::{DispatchQueue, DispatchQoS};
103    ///
104    /// let queue = DispatchQueue::new("com.myapp.capture", DispatchQoS::UserInteractive);
105    /// // Use the queue with SCStream's add_output_handler_with_queue
106    /// ```
107    ///
108    /// The label is truncated at its first NUL byte.
109    ///
110    /// # Panics
111    ///
112    /// Panics if queue creation fails
113    #[must_use]
114    pub fn new(label: &str, qos: DispatchQoS) -> Self {
115        let c_label = crate::utils::ffi_string::cstring_until_nul(label);
116        let ptr = unsafe { crate::ffi::acf_dispatch_queue_create(c_label.as_ptr(), qos as i32) };
117        assert!(!ptr.is_null(), "Failed to create dispatch queue");
118        Self { ptr }
119    }
120
121    #[must_use]
122    pub fn concurrent(label: &str, qos: DispatchQoS) -> Self {
123        let c_label = crate::utils::ffi_string::cstring_until_nul(label);
124        let ptr = unsafe {
125            crate::ffi::acf_dispatch_queue_create_concurrent(c_label.as_ptr(), qos as i32)
126        };
127        assert!(!ptr.is_null(), "Failed to create dispatch queue");
128        Self { ptr }
129    }
130
131    #[must_use]
132    pub fn main() -> Self {
133        let ptr = unsafe { crate::ffi::acf_dispatch_queue_main() };
134        assert!(!ptr.is_null(), "dispatch main queue is NULL");
135        Self { ptr }
136    }
137
138    #[must_use]
139    pub fn global(qos: DispatchQoS) -> Self {
140        let ptr = unsafe { crate::ffi::acf_dispatch_queue_global(qos as i32) };
141        assert!(!ptr.is_null(), "dispatch global queue is NULL");
142        Self { ptr }
143    }
144
145    /// Returns the raw pointer to the dispatch queue
146    ///
147    /// This is used internally for FFI calls (and for testing)
148    #[must_use]
149    pub const fn as_ptr(&self) -> *const c_void {
150        self.ptr
151    }
152
153    #[must_use]
154    const fn as_mut_ptr(&self) -> *mut c_void {
155        self.ptr.cast_mut()
156    }
157
158    #[allow(clippy::missing_safety_doc)]
159    pub unsafe fn exec_async_f(
160        &self,
161        context: *mut c_void,
162        work: unsafe extern "C" fn(*mut c_void),
163    ) {
164        unsafe { dispatch_async_f(self.as_mut_ptr(), context, work) };
165    }
166
167    #[allow(clippy::missing_safety_doc)]
168    pub unsafe fn exec_after_f(
169        &self,
170        delay: Duration,
171        context: *mut c_void,
172        work: unsafe extern "C" fn(*mut c_void),
173    ) {
174        let delta = i64::try_from(delay.as_nanos()).unwrap_or(i64::MAX);
175        unsafe {
176            dispatch_after_f(
177                dispatch_time(DISPATCH_TIME_NOW, delta),
178                self.as_mut_ptr(),
179                context,
180                work,
181            );
182        };
183    }
184}
185
186const DISPATCH_TIME_NOW: u64 = 0;
187
188extern "C" {
189    fn dispatch_async_f(
190        queue: *mut c_void,
191        context: *mut c_void,
192        work: unsafe extern "C" fn(*mut c_void),
193    );
194    fn dispatch_after_f(
195        when: u64,
196        queue: *mut c_void,
197        context: *mut c_void,
198        work: unsafe extern "C" fn(*mut c_void),
199    );
200    fn dispatch_time(when: u64, delta: i64) -> u64;
201}
202
203crate::utils::retained::cf_retained!(
204    DispatchQueue,
205    field = ptr,
206    retain = crate::ffi::dispatch_queue_retain,
207    release = crate::ffi::dispatch_queue_release,
208    drop = unchecked,
209);
210
211impl fmt::Debug for DispatchQueue {
212    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
213        f.debug_struct("DispatchQueue")
214            .field("ptr", &self.ptr)
215            .finish()
216    }
217}
218
219impl fmt::Display for DispatchQueue {
220    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
221        write!(f, "DispatchQueue")
222    }
223}
224
225fn timeout_ms(timeout: Option<Duration>) -> i64 {
226    timeout.map_or(-1, |duration| {
227        i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)
228    })
229}
230
231struct DispatchOnceTask {
232    site: &'static str,
233    work: Option<Box<dyn FnOnce() + Send + 'static>>,
234}
235
236struct DispatchApplyTask {
237    work: Box<dyn Fn(usize) + Send + Sync + 'static>,
238}
239
240extern "C" fn dispatch_once_trampoline(context: *mut c_void) {
241    if context.is_null() {
242        return;
243    }
244    let mut task = unsafe { Box::from_raw(context.cast::<DispatchOnceTask>()) };
245    if let Some(work) = task.work.take() {
246        panic_safe::catch_user_panic(task.site, work);
247    }
248}
249
250extern "C" fn dispatch_apply_trampoline(iteration: usize, context: *mut c_void) {
251    if context.is_null() {
252        return;
253    }
254    let task = unsafe { &*context.cast::<DispatchApplyTask>() };
255    panic_safe::catch_user_panic("dispatch_apply", || (task.work)(iteration));
256}
257
258/// Submit `work` to `queue` and return immediately.
259pub fn dispatch_async<F>(queue: &DispatchQueue, work: F)
260where
261    F: FnOnce() + Send + 'static,
262{
263    let task = Box::new(DispatchOnceTask {
264        site: "dispatch_async",
265        work: Some(Box::new(work)),
266    });
267    unsafe {
268        crate::ffi::acf_dispatch_async_f(
269            queue.as_mut_ptr(),
270            Box::into_raw(task).cast(),
271            dispatch_once_trampoline,
272        );
273    }
274}
275
276/// Submit `work` to `queue` and wait until it finishes.
277pub fn dispatch_async_and_wait<F>(queue: &DispatchQueue, work: F)
278where
279    F: FnOnce() + Send + 'static,
280{
281    let task = Box::new(DispatchOnceTask {
282        site: "dispatch_async_and_wait",
283        work: Some(Box::new(work)),
284    });
285    unsafe {
286        crate::ffi::acf_dispatch_async_and_wait_f(
287            queue.as_mut_ptr(),
288            Box::into_raw(task).cast(),
289            dispatch_once_trampoline,
290        );
291    }
292}
293
294pub fn dispatch_after<F>(delay: Duration, queue: &DispatchQueue, work: F)
295where
296    F: FnOnce() + Send + 'static,
297{
298    let task = Box::new(DispatchOnceTask {
299        site: "dispatch_after",
300        work: Some(Box::new(work)),
301    });
302    let delay_ns = u64::try_from(delay.as_nanos()).unwrap_or(u64::MAX);
303    unsafe {
304        crate::ffi::acf_dispatch_after_f(
305            delay_ns,
306            queue.as_mut_ptr(),
307            Box::into_raw(task).cast(),
308            dispatch_once_trampoline,
309        );
310    }
311}
312
313/// Run `work` for every `iteration` on `queue`, waiting for all iterations to finish.
314pub fn dispatch_apply<F>(iterations: usize, queue: &DispatchQueue, work: F)
315where
316    F: Fn(usize) + Send + Sync + 'static,
317{
318    if iterations == 0 {
319        return;
320    }
321    let task = Box::new(DispatchApplyTask {
322        work: Box::new(work),
323    });
324    let raw = Box::into_raw(task);
325    unsafe {
326        crate::ffi::acf_dispatch_apply_f(
327            iterations,
328            queue.as_mut_ptr(),
329            raw.cast(),
330            dispatch_apply_trampoline,
331        );
332        drop(Box::from_raw(raw));
333    }
334}
335
336/// Wrapper around `DispatchGroup`.
337#[derive(PartialEq, Eq, Hash)]
338pub struct DispatchGroup {
339    ptr: *mut c_void,
340}
341
342// SAFETY: `dispatch_group_t` is a thread-safe GCD primitive; it is safe to
343// share across threads and to send between threads.
344unsafe impl Send for DispatchGroup {}
345unsafe impl Sync for DispatchGroup {}
346
347impl DispatchGroup {
348    /// Create a new empty group.
349    #[must_use]
350    pub fn new() -> Self {
351        let ptr = unsafe { crate::ffi::acf_dispatch_group_holder_create() };
352        assert!(!ptr.is_null(), "failed to create DispatchGroup");
353        Self { ptr }
354    }
355
356    /// Enter the group.
357    pub fn enter(&self) {
358        unsafe { crate::ffi::acf_dispatch_group_holder_enter(self.ptr) };
359    }
360
361    /// Leave the group.
362    pub fn leave(&self) {
363        unsafe { crate::ffi::acf_dispatch_group_holder_leave(self.ptr) };
364    }
365
366    /// Wait for the group to finish.
367    #[must_use]
368    pub fn wait(&self, timeout: Option<Duration>) -> bool {
369        unsafe { crate::ffi::acf_dispatch_group_holder_wait(self.ptr, timeout_ms(timeout)) }
370    }
371}
372
373impl Default for DispatchGroup {
374    fn default() -> Self {
375        Self::new()
376    }
377}
378
379crate::utils::retained::cf_retained!(
380    DispatchGroup,
381    field = ptr,
382    retain = crate::ffi::acf_object_retain,
383    release = crate::ffi::acf_object_release,
384    drop = unchecked,
385);
386
387impl fmt::Debug for DispatchGroup {
388    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
389        f.debug_struct("DispatchGroup")
390            .field("ptr", &self.ptr)
391            .finish()
392    }
393}
394
395/// Wrapper around `DispatchSemaphore`.
396#[derive(PartialEq, Eq, Hash)]
397pub struct DispatchSemaphore {
398    ptr: *mut c_void,
399}
400
401// SAFETY: `dispatch_semaphore_t` is a thread-safe GCD primitive designed
402// explicitly for cross-thread signalling.
403unsafe impl Send for DispatchSemaphore {}
404unsafe impl Sync for DispatchSemaphore {}
405
406impl DispatchSemaphore {
407    /// Create a semaphore with an initial signal count.
408    #[must_use]
409    pub fn new(value: i64) -> Option<Self> {
410        if value < 0 {
411            return None;
412        }
413        let ptr = unsafe { crate::ffi::acf_dispatch_semaphore_holder_create(value) };
414        if ptr.is_null() {
415            None
416        } else {
417            Some(Self { ptr })
418        }
419    }
420
421    /// Signal the semaphore.
422    #[must_use]
423    pub fn signal(&self) -> i64 {
424        unsafe { crate::ffi::acf_dispatch_semaphore_holder_signal(self.ptr) }
425    }
426
427    /// Wait for the semaphore.
428    #[must_use]
429    pub fn wait(&self, timeout: Option<Duration>) -> bool {
430        unsafe { crate::ffi::acf_dispatch_semaphore_holder_wait(self.ptr, timeout_ms(timeout)) }
431    }
432}
433
434crate::utils::retained::cf_retained!(
435    DispatchSemaphore,
436    field = ptr,
437    retain = crate::ffi::acf_object_retain,
438    release = crate::ffi::acf_object_release,
439    drop = unchecked,
440);
441
442impl fmt::Debug for DispatchSemaphore {
443    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
444        f.debug_struct("DispatchSemaphore")
445            .field("ptr", &self.ptr)
446            .finish()
447    }
448}
449
450/// Minimal timer-backed `DispatchSource` wrapper.
451#[derive(PartialEq, Eq, Hash)]
452pub struct DispatchSource {
453    ptr: *mut c_void,
454}
455
456// SAFETY: `dispatch_source_t` is a thread-safe GCD primitive, and the bridge
457// holder synchronizes its activation, cancellation, and fire-count state.
458unsafe impl Send for DispatchSource {}
459unsafe impl Sync for DispatchSource {}
460
461impl DispatchSource {
462    /// Create a repeating timer source.
463    #[must_use]
464    pub fn timer(interval: Duration, leeway: Duration) -> Self {
465        let interval_ns = u64::try_from(interval.as_nanos()).unwrap_or(u64::MAX);
466        let leeway_ns = u64::try_from(leeway.as_nanos()).unwrap_or(u64::MAX);
467        let ptr =
468            unsafe { crate::ffi::acf_dispatch_source_timer_create_ns(interval_ns, leeway_ns) };
469        assert!(!ptr.is_null(), "failed to create DispatchSource timer");
470        Self { ptr }
471    }
472
473    /// Activate the timer source after creation.
474    ///
475    /// Repeated or concurrent calls are idempotent.
476    pub fn resume(&self) {
477        unsafe { crate::ffi::acf_dispatch_source_timer_resume(self.ptr) };
478    }
479
480    /// Cancel the source.
481    ///
482    /// Repeated or concurrent calls are idempotent. Activating after
483    /// cancellation is a no-op.
484    pub fn cancel(&self) {
485        unsafe { crate::ffi::acf_dispatch_source_timer_cancel(self.ptr) };
486    }
487
488    /// Number of timer firings observed by the bridge.
489    #[must_use]
490    pub fn fire_count(&self) -> u64 {
491        unsafe { crate::ffi::acf_dispatch_source_timer_fire_count(self.ptr) }
492    }
493}
494
495crate::utils::retained::cf_retained!(
496    DispatchSource,
497    field = ptr,
498    retain = crate::ffi::acf_object_retain,
499    release = crate::ffi::acf_object_release,
500    drop = unchecked,
501);
502
503impl fmt::Debug for DispatchSource {
504    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
505        f.debug_struct("DispatchSource")
506            .field("ptr", &self.ptr)
507            .field("fire_count", &self.fire_count())
508            .finish()
509    }
510}
511
512struct DataAddHandler {
513    callback: RefCell<Box<dyn FnMut(u64)>>,
514}
515
516extern "C" fn data_add_event(context: *mut c_void, value: u64) {
517    if context.is_null() {
518        return;
519    }
520    let handler = unsafe { &*context.cast::<DataAddHandler>() };
521    panic_safe::catch_user_panic("dispatch_data_add", || {
522        if let Ok(mut callback) = handler.callback.try_borrow_mut() {
523            callback(value);
524        }
525    });
526}
527
528extern "C" fn data_add_release(context: *mut c_void) {
529    if context.is_null() {
530        return;
531    }
532    panic_safe::catch_user_panic("dispatch_data_add_release", || {
533        drop(unsafe { Box::from_raw(context.cast::<DataAddHandler>()) });
534    });
535}
536
537extern "C" {
538    fn acf_dispatch_source_data_add_is_main_thread() -> bool;
539    fn acf_dispatch_source_data_add_create(
540        queue: *mut c_void,
541        context: *mut c_void,
542        event: extern "C" fn(*mut c_void, u64),
543        release: extern "C" fn(*mut c_void),
544    ) -> *mut c_void;
545    fn acf_dispatch_source_data_add_resume(source: *mut c_void);
546    fn acf_dispatch_source_data_add_cancel(source: *mut c_void);
547    fn acf_dispatch_source_data_add_merge(source: *mut c_void, value: u64);
548}
549
550struct DataAddSourceInner {
551    ptr: *mut c_void,
552}
553
554unsafe impl Send for DataAddSourceInner {}
555unsafe impl Sync for DataAddSourceInner {}
556
557impl Drop for DataAddSourceInner {
558    fn drop(&mut self) {
559        unsafe { crate::ffi::acf_object_release(self.ptr) };
560    }
561}
562
563pub struct DispatchDataAddSource {
564    inner: Arc<DataAddSourceInner>,
565    main_thread: PhantomData<Rc<()>>,
566}
567
568impl DispatchDataAddSource {
569    #[must_use]
570    pub fn new<F>(callback: F) -> Self
571    where
572        F: FnMut(u64) + 'static,
573    {
574        assert!(
575            unsafe { acf_dispatch_source_data_add_is_main_thread() },
576            "data-add sources must be created on the main thread"
577        );
578        unsafe { Self::with_queue(&DispatchQueue::main(), callback) }
579    }
580
581    unsafe fn with_queue<F>(queue: &DispatchQueue, callback: F) -> Self
582    where
583        F: FnMut(u64) + 'static,
584    {
585        let handler = Box::new(DataAddHandler {
586            callback: RefCell::new(Box::new(callback)),
587        });
588        let context = Box::into_raw(handler).cast();
589        let ptr = unsafe {
590            acf_dispatch_source_data_add_create(
591                queue.as_mut_ptr(),
592                context,
593                data_add_event,
594                data_add_release,
595            )
596        };
597        if ptr.is_null() {
598            data_add_release(context);
599            panic!("failed to create data-add source");
600        }
601        Self {
602            inner: Arc::new(DataAddSourceInner { ptr }),
603            main_thread: PhantomData,
604        }
605    }
606
607    #[must_use]
608    pub fn merge_handle(&self) -> DispatchDataAddMergeHandle {
609        DispatchDataAddMergeHandle(Arc::clone(&self.inner))
610    }
611
612    pub fn resume(&self) {
613        unsafe { acf_dispatch_source_data_add_resume(self.inner.ptr) };
614    }
615
616    pub fn cancel(&self) {
617        unsafe { acf_dispatch_source_data_add_cancel(self.inner.ptr) };
618    }
619}
620
621impl Drop for DispatchDataAddSource {
622    fn drop(&mut self) {
623        self.cancel();
624    }
625}
626
627impl fmt::Debug for DispatchDataAddSource {
628    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
629        f.debug_struct("DispatchDataAddSource")
630            .field("ptr", &self.inner.ptr)
631            .finish()
632    }
633}
634
635#[derive(Clone)]
636pub struct DispatchDataAddMergeHandle(Arc<DataAddSourceInner>);
637
638impl DispatchDataAddMergeHandle {
639    pub fn merge_data(&self, value: u64) {
640        if value != 0 {
641            unsafe { acf_dispatch_source_data_add_merge(self.0.ptr, value) };
642        }
643    }
644}
645
646impl fmt::Debug for DispatchDataAddMergeHandle {
647    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
648        f.debug_struct("DispatchDataAddMergeHandle")
649            .field("ptr", &self.0.ptr)
650            .finish()
651    }
652}
653
654#[cfg(test)]
655mod data_add_tests {
656    use super::{
657        dispatch_async, DispatchDataAddMergeHandle, DispatchDataAddSource, DispatchQoS,
658        DispatchQueue,
659    };
660    use std::sync::{
661        atomic::{AtomicUsize, Ordering},
662        mpsc, Arc,
663    };
664    use std::thread;
665    use std::time::Duration;
666
667    struct ReleaseNotification(mpsc::Sender<()>);
668
669    impl Drop for ReleaseNotification {
670        fn drop(&mut self) {
671            let _ = self.0.send(());
672        }
673    }
674
675    #[test]
676    fn data_add_merges_across_threads_and_coalesces_before_resume() {
677        fn assert_send_sync<T: Send + Sync>() {}
678        assert_send_sync::<DispatchDataAddMergeHandle>();
679
680        let queue = DispatchQueue::new(
681            "com.doomfish.apple-cf.data-add-coalesce",
682            DispatchQoS::Default,
683        );
684        let (received_tx, received_rx) = mpsc::channel();
685        let (released_tx, released_rx) = mpsc::channel();
686        let released = ReleaseNotification(released_tx);
687        let handler: Box<dyn FnMut(u64) + Send> = Box::new(move |value| {
688            let _keep_alive = &released;
689            received_tx.send(value).expect("merged count");
690        });
691        let source = unsafe { DispatchDataAddSource::with_queue(&queue, handler) };
692        let handle = source.merge_handle();
693
694        let threads: Vec<_> = (0..8)
695            .map(|_| {
696                let handle = handle.clone();
697                thread::spawn(move || {
698                    for _ in 0..10 {
699                        handle.merge_data(1);
700                    }
701                })
702            })
703            .collect();
704        for thread in threads {
705            thread.join().expect("merge thread");
706        }
707
708        source.resume();
709        assert_eq!(
710            received_rx
711                .recv_timeout(Duration::from_secs(5))
712                .expect("merged event"),
713            80
714        );
715        source.cancel();
716        released_rx
717            .recv_timeout(Duration::from_secs(5))
718            .expect("cancel handler");
719        assert!(received_rx.try_recv().is_err());
720    }
721
722    #[test]
723    fn data_add_cancel_then_merge_keeps_context_until_cancel_handler() {
724        let queue = DispatchQueue::new(
725            "com.doomfish.apple-cf.data-add-cancel",
726            DispatchQoS::Default,
727        );
728        let (entered_tx, entered_rx) = mpsc::channel();
729        let (unblock_tx, unblock_rx) = mpsc::channel();
730        dispatch_async(&queue, move || {
731            entered_tx.send(()).expect("queue entered");
732            unblock_rx
733                .recv_timeout(Duration::from_secs(5))
734                .expect("queue unblocked");
735        });
736        entered_rx
737            .recv_timeout(Duration::from_secs(5))
738            .expect("queue blocked");
739
740        let (received_tx, received_rx) = mpsc::channel();
741        let (released_tx, released_rx) = mpsc::channel();
742        let released = ReleaseNotification(released_tx);
743        let handler: Box<dyn FnMut(u64) + Send> = Box::new(move |value| {
744            let _keep_alive = &released;
745            received_tx.send(value).expect("merged count");
746        });
747        let source = unsafe { DispatchDataAddSource::with_queue(&queue, handler) };
748        let handle = source.merge_handle();
749        source.cancel();
750        handle.merge_data(7);
751        assert!(matches!(
752            released_rx.try_recv(),
753            Err(mpsc::TryRecvError::Empty)
754        ));
755
756        unblock_tx.send(()).expect("unblock queue");
757        released_rx
758            .recv_timeout(Duration::from_secs(5))
759            .expect("cancel handler");
760        handle.merge_data(1);
761        assert!(received_rx.try_recv().is_err());
762    }
763
764    #[test]
765    fn data_add_drop_then_merge_is_inert() {
766        let queue = DispatchQueue::new("com.doomfish.apple-cf.data-add-drop", DispatchQoS::Default);
767        let (entered_tx, entered_rx) = mpsc::channel();
768        let (unblock_tx, unblock_rx) = mpsc::channel();
769        dispatch_async(&queue, move || {
770            entered_tx.send(()).expect("queue entered");
771            unblock_rx
772                .recv_timeout(Duration::from_secs(5))
773                .expect("queue unblocked");
774        });
775        entered_rx
776            .recv_timeout(Duration::from_secs(5))
777            .expect("queue blocked");
778
779        let (received_tx, received_rx) = mpsc::channel();
780        let (released_tx, released_rx) = mpsc::channel();
781        let released = ReleaseNotification(released_tx);
782        let handler: Box<dyn FnMut(u64) + Send> = Box::new(move |value| {
783            let _keep_alive = &released;
784            received_tx.send(value).expect("merged count");
785        });
786        let source = unsafe { DispatchDataAddSource::with_queue(&queue, handler) };
787        let handle = source.merge_handle();
788        source.resume();
789        drop(source);
790
791        let threads: Vec<_> = (0..8)
792            .map(|_| {
793                let handle = handle.clone();
794                thread::spawn(move || {
795                    for _ in 0..100 {
796                        handle.merge_data(1);
797                    }
798                })
799            })
800            .collect();
801        for thread in threads {
802            thread.join().expect("merge thread");
803        }
804        assert!(matches!(
805            released_rx.try_recv(),
806            Err(mpsc::TryRecvError::Empty)
807        ));
808        unblock_tx.send(()).expect("unblock queue");
809        released_rx
810            .recv_timeout(Duration::from_secs(5))
811            .expect("cancel handler");
812        handle.merge_data(1);
813        assert!(received_rx.try_recv().is_err());
814    }
815
816    #[test]
817    fn data_add_handler_panics_stay_inside_c_boundary() {
818        let queue =
819            DispatchQueue::new("com.doomfish.apple-cf.data-add-panic", DispatchQoS::Default);
820        let calls = Arc::new(AtomicUsize::new(0));
821        let observed = Arc::clone(&calls);
822        let (received_tx, received_rx) = mpsc::channel();
823        let (released_tx, released_rx) = mpsc::channel();
824        let released = ReleaseNotification(released_tx);
825        let handler: Box<dyn FnMut(u64) + Send> = Box::new(move |_| {
826            let _keep_alive = &released;
827            let count = observed.fetch_add(1, Ordering::SeqCst) + 1;
828            received_tx.send(count).expect("callback observed");
829            panic!("data-add handler panic");
830        });
831        let source = unsafe { DispatchDataAddSource::with_queue(&queue, handler) };
832        let handle = source.merge_handle();
833        source.resume();
834        for expected in 1..=2 {
835            handle.merge_data(1);
836            assert_eq!(
837                received_rx
838                    .recv_timeout(Duration::from_secs(5))
839                    .expect("callback returned"),
840                expected
841            );
842        }
843        source.cancel();
844        released_rx
845            .recv_timeout(Duration::from_secs(5))
846            .expect("cancel handler");
847        assert_eq!(calls.load(Ordering::SeqCst), 2);
848    }
849}
apple_cf/dispatch_queue.rs

apple_cf/
dispatch_queue.rs
