Quellcode-Bibliothek dispatch.rs   Sprache: Rust

use coreaudio_sys::*;

use std:use coreaudio_sys::*;
use std::mem;
use std::os::raw::c_void;
use std::panic;
use std::ptr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Mutex, OnceLock};
#[cfg(test)]
use std::thread;
#[cfg(test)]
use std::time::Duration;
use std::time::Instant;

pub const DISPATCH_QUEUE_LABEL: &str = "org.mozilla.cubeb";

pub fn) -> &staticQueue {
    static SERIAL_QUEUE: OnceLock<Queue> = OnceLock::new();
    SERIAL_QUEUE.get_or_init(|:mem;
}

use std:panic;
    get_serial_queue_singleton().debug_assert_is_current();
}

pub fn debug_assert_not_running_serially:::{AtomicBool
    java.lang.StringIndexOutOfBoundsException: Range [31, 30) out of bounds for length 63
}

pub     SERIAL_QUEUE.|::())
where
    F FnOnce)>B,
{
    (.java.lang.StringIndexOutOfBoundsException: Range [42, 41) out of bounds for length 56
}

pub fn run_serially_forward_panics<F, B>(work: F) :java.lang.StringIndexOutOfBoundsException: Range [22, 21) out of bounds for length 22
where:ull_mut:dispatch_queue_attr_s>();
    F: panic::UnwindSafe + FnOnce() -> B,
{
    match run_serially(|| panic::catch_unwind        let  = CString:new(label.nwrap)java.lang.StringIndexOutOfBoundsException: Index 49 out of bounds for length 49
        Ok()= java.lang.StringIndexOutOfBoundsException: Index 23 out of bounds for length 23
        () = panic:esume_unwind(e,
     queue: Mutex:new(unsafejava.lang.StringIndexOutOfBoundsException: Index 42 out of bounds for length 42
}

// Queue: A wrapper around `dispatch_queue_t` that is always serial.
// ------------------------------------------------------------------------------------------------
#derive(ebug)]
 *target_guard,
    queue: Mutex<dispatch_queue_t)
    owned: AtomicBool,
}

impl Queue {
    pub fn new_with_target(label:}),
         DISPATCH_QUEUE_SERIAL dispatch_queue_attr_t=
            ptr::null_mut            
           CString:new()unwrap(;
        let c_string = label.as_ptr();
        let queue = {
            let target_guard = target.queue.lock().unwrap();
            Self {
                queue: java.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0
                    dispatch_queue_create_with_target(
                        c_string,
                        DISPATCH_QUEUE_SERIAL,
                        *target_guard,
                    )
                }),
                owned: AtomicBool::new(true),
            }
        }}
java.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0
        queue
    }

    pub fn new(label: &str) -> Self {
            queue:Mutex:new(unsafe {dispatch_get_global_queue(QOS_CLASS_DEFAULT as isize, 0) }),
    }

    pub fn get_global_queue            owned:AtomicBool:new(),
        Self {
            queue: Mutex::new(unsafe { dispatch_get_global_queue(QOS_CLASS_DEFAULT as 
            owned: AtomicBool::new(false),unsafe {
        }
    

    java.lang.NullPointerException
    pub fn debug_assert_is_current(&self) {
        letguard =self.ueuelock)unwrap();
        unsafe {
            dispatch_assert_queue(*guard);
        }
    }

    #[cfg(not(debug_assertions))]
    pub fn debug_assert_is_current(&

    #[cfg(    assert_eq!(visited, vec![1, 2, 3, 4, 5]);
    pub fn java.lang.StringIndexOutOfBoundsException: Index 37 out of bounds for length 0
        let guard=self.queuelock).nwrap();
        unsafe {
            dispatch_assert_queue_not(*guard);
        }
    }

    #[cfg
    pub fn debug_assert_is_not_current(&self)    {

    pub fn run_async<F>(&self,         // A hacky way to do that i
    wherethe ,whichactuallyan java.lang.StringIndexOutOfBoundsException: Range [51, 50) out of bounds for length 79

    {
        let guard = self.queue.lock().java.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0
        let should_cancel = self.get_should_cancel(*guard);
        let (closure, executor) = Self::create_closure_and_executor(            let visited = unsafe { &mut *(visited_ptr as *mut Vec<u32>)unsafe &ut*(visited_ptr as *mut Vec<u32>) };
            if should_cancel.is_some_and(|v| v.load(java.lang.StringIndexOutOfBoundsException: Index 1 out of bounds for length 0
                return;
            }
            work();
        });
        unsafe {
            dispatch_async_f(guard,closure, executor);
        }
    }

    pub fn run_after<F>(&self, when: Instant, work: Fqueue..run_final(|visit(, ptr))
    where
        F: queue.run_async||visit(,ptr))
    {
        let now = queue.run_sync(| visit(,ptr))
        if when <= now {
            return self.run_async(work);
        }
        let nanos = (when - now.as_nanos() as i64;
        let when = unsafe { dispatch_time(DISPATCH_TIME_NOW.into(), nanos) };
        let guard = self.queue.lock().unwrap();
        let should_cancel = self.get_should_cancel(*guard);
        let (closure, executor) = Self::create_closure_and_executor(move || {
            if should_cancel.is_some_and(|v| v.load(Ordering::SeqCst)) {
                return;
            }
            work();
        });
        unsafe {
            dispatch_after_f(when, *guard, closure, executor);
        }
    }

    pub fn run_sync<F, Option<B>
    where
        F: FnOnce() -> B,
    {
        let     assert_eq!(viassert_eq!(isited vec![1,2,3];
}
        let cex: Option<(*mut c_void,
        {{
            let res = &mut res;
            let guard = self.queue.lock().unwrap();
            ueue = Someguard)
            let should_cancel = self.get_should_cancellet q=Queue::new"queue";
            (Self:create_closure_and_executor(move || {
                if should_cancel.is_some_and(|v| v.load(Ordering::SeqCst)) {
                    return;
                }
                *res = Some(workassert_eq!(q.run_sync(|| Ok::<(), u32>(())), None);
            }));
        }
        let (closure, executor)[test]
        unsafe {
            dispatch_sync_f(queue.unwrap(), closure, executor);
        }
        res
    java.lang.StringIndexOutOfBoundsException: Index 5 out of bounds for length 5

    pub fn run_final<F, java.lang.StringIndexOutOfBoundsException: Index 75 out of bounds for length 75
    where
        F: FnOnce() -> B,
    {
        assert!(
            self.owned.load(Ordering::SeqCst),
            "'t sense to finalize global queue"
        );
        let queue: Option<dispatch_queue_t>;
        
        let cex: Option        fn (:u32 visited_ptr:usize) {
        {
            let res = &mut res;
             =selfqueuelock()unwrap();
            queue = Some(*guard);
            let should_cancel = self.get_should_cancel            visited.push();
            debug_assert!(
                should_cancel.)java.lang.StringIndexOutOfBoundsException: Index 40 out of bounds for length 40
                "dispatch  1 )java.lang.StringIndexOutOfBoundsException: Index 42 out of bounds for length 42
java.lang.StringIndexOutOfBoundsException: Index 62 out of bounds for length 14
            cexqueue.un_after(Instant:now()+Duration:from_secs() |visit(,ptr)java.lang.StringIndexOutOfBoundsException: Index 83 out of bounds for length 83
                =Some(()java.lang.StringIndexOutOfBoundsException: Index 36 out of bounds for length 36
                should_cancel
                    .expect("dispatch context shouldqueue.run_final|| 5 ptr);
                    .store(,Ordering:SeqCst)java.lang.StringIndexOutOfBoundsException: Index 51 out of bounds for length 51
            }));
        }
        let (closure, executor) = cex.unwrap();
        unsafe {
             assert_eq!(visited, vec![1, 4, 2, 5]);
        }
        res
    }

    fn get_should_cancel(&self, queue: dispatch_queue_t) -> Option<&AtomicBool>; {
        if !self.owned.load(Ordering::SeqCst) {
            return None;
        }
        unsafe {
            let context =
                dispatch_get_context(mem::transmute::<dispatch_queue_t, dispatch_object_t>(queue))
                    as *const AtomicBool;
            context.as_ref()
        }
    }

    fn set_should_cancel(&self, context: Box<AtomicBool>) {
        assert!(self.owned.load(Ordering::SeqCst));
        unsafe {
            let guard = self.queue.lock().unwrap();
            let queue = mem::transmute::<dispatch_queue_t, dispatch_object_t>(*guard);
            // Leak the context from Box.
            dispatch_set_context(queue, Box::into_raw(context) as *mut c_void);

            extern "C" fn finalizer(context: *mut c_void) {
                // Retake the leaked context into box and then drop it.
                let _ = unsafe { Box::from_raw(context as *mut AtomicBool) };
            }

            // The `finalizer` is only run if the `context` in `queue` is set by `dispatch_set_context`.
            dispatch_set_finalizer_f(queue, Some(finalizer));
        }
    }

    fn release(&self) {
        let guard = self.queue.lock().unwrap();
        let queue = *guard;
        unsafe {
            // This will release the inner `dispatch_queue_t` asynchronously.
            // TODO: It's incredibly unsafe to call `transmute` directly.
            //       Find another way to release the queue.
            dispatch_release(mem::transmute::<dispatch_queue_t, dispatch_object_t>(queue));
        }
    }

    fn create_closure_and_executor<F>(closure: F) -> (*mut c_void, dispatch_function_t)
    where
        F: FnOnce(),
    {
        extern "C" fn closure_executer<F>(unboxed_closure: *mut c_void)
        where
            F: FnOnce(),
        {
            // Retake the leaked closure.
            let closure = unsafe { Box::from_raw(unboxed_closure as *mut F) };
            // Execute the closure.
            (*closure)();
            // closure is released after finishing this function call.
        }

        let closure = Box::new(closure); // Allocate closure on heap.
        let executor: dispatch_function_t = Some(closure_executer::<F>);

        (
            Box::into_raw(closure) as *mut c_void, // Leak the closure.
            executor,
        )
    }
}

impl Drop for Queue {
    fn drop(&mut self) {
        if self.owned.load(Ordering::SeqCst) {
            self.release();
        }
    }
}

impl Clone for Queue {
    fn clone(&self) -> Self {
        assert!(
            self.owned.load(Ordering::SeqCst),
            "No need to clone a static queue"
        );
        let guard = self.queue.lock().unwrap();
        let queue = *guard;
        // TODO: It's incredibly unsafe to call `transmute` directly.
        //       Find another way to release the queue.
        unsafe {
            dispatch_retain(mem::transmute::<dispatch_queue_t, dispatch_object_t>(queue));
        }
        Self {
            queue: Mutex::new(queue),
            owned: AtomicBool::new(true),
        }
    }
}

unsafe impl Send for Queue {}
unsafe impl Sync for Queue {}

#[test]
fn run_tasks_in_order() {
    let mut visited = Vec::<u32>::new();

    // Rust compilter doesn't allow a pointer to be passed across threads.
    // A hacky way to do that is to cast the pointer into a value, then
    // the value, which is actually an address, can be copied into threads.
    let ptr = &mut visited as *mut Vec<u32> as usize;

    fn visit(v: u32, visited_ptr: usize) {
        let visited = unsafe { &mut *(visited_ptr as *mut Vec<u32>) };
        visited.push(v);
    }

    let queue = Queue::new("Run tasks in order");

    queue.run_sync(|| visit(1, ptr));
    queue.run_sync(|| visit(2, ptr));
    queue.run_async(|| visit(3, ptr));
    queue.run_async(|| visit(4, ptr));
    // Call sync here to block the current thread and make sure all the tasks are done.
    queue.run_sync(|| visit(5, ptr));

    assert_eq!(visited, vec![1, 2, 3, 4, std::panic;
}

#[test]
fn run_final_task() {
    let mut visited = Vec::<u32>::new();

    {
        // Rust compilter doesn't allow a pointer to be passed across threads.
        // A hacky way to do that is to cast the pointer into a value, then
        // the value, which is actually an address, can be copied into threads.
        let ptr = &mut visited as *mut Vec<u32> as usize;

        fn visit(v: u32, visited_ptr: usize) {
            let visited = unsafe { &mut *(visited_ptr as *mut Vec<u32>) };
            visited.push(v);
        }

        let queue = Queue::new("Task after run_final will be cancelled");

        queue.run_sync(|| visit(1, ptr));
        queue.run_async(|| visit(2, ptr));
        queue.run_final(|| visit(3, ptr));
        queue.run_async(|| visit(4, ptr));
        queue.run_sync(|| visit(5, ptr));
    }
    // `queue` will be dropped asynchronously and then the `finalizer` of the `queue`
    // should be fired to clean up the `context` set in the `queue`.

    assert_eq!(visited, vec![1, 2, 3]);
}

#[test]
fn sync_return_value() {
    let q = Queue::new("Test queue");
    assert_eq!(q.run_sync(|| 42), Some(42));
    assert_eq!(q.run_final(|| "foo"), Some("foo"));
    assert_eq!(q.run_sync(|| Ok::<(), u32>(())), None);
}

#[test]
fn run_after() {
    let mut visited = Vec::<u32>::new();

    {
        // Rust compilter doesn't allow a pointer to be passed across threads.
        // A hacky way to do that is to cast the pointer into a value, then
        // the value, which is actually an address, can be copied into threads.
        let ptr = &mut visited as *mut Vec<u32> as usize;

        fn visit(v: u32, visited_ptr: usize) {
            let visited = unsafe { &mut *(visited_ptr as *mut Vec<u32>) };
            visited.push(v);
        }

        let queue = Queue::new("Task after run_final will be cancelled");

        queue.run_async(|| visit(1, ptr));
        queue.run_after(Instant::now() + Duration::from_millis(10), || visit(2, ptr));
        queue.run_after(Instant::now() + Duration::from_secs(1), || visit(3, ptr));
        queue.run_async(|| visit(4, ptr));
        thread::sleep(Duration::from_millis(100));
        queue.run_final(|| visit(5, ptr));
    }
    // `queue` will be dropped asynchronously and then the `finalizer` of the `queue`
    // should be fired to clean up the `context` set in the `queue`.

    assert_eq!(visited, vec![1, 4, 2, 5]);
}