diff --git a/Cargo.toml b/Cargo.toml index 55e3dd2..d55f4fa 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,7 @@ version = "0.61.2" features = [ "Win32_Foundation", "Win32_System_LibraryLoader", + "Win32_System_Threading", ] [target.'cfg(target_os = "linux")'.dependencies] diff --git a/README.md b/README.md index 42bb3a0..18ade6d 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,27 @@ match promote_current_thread_to_real_time(512, 44100) { ``` +A thread other than the calling one can also be promoted or demoted. This is +useful when the thread that needs to become real-time cannot promote itself +directly, for example because it is sandboxed: + +```rust +use audio_thread_priority::{get_current_thread_info, promote_thread_to_real_time, demote_thread_from_real_time}; + +// ... on the thread that will compute audio and has to be real-time, gather +// serializable information about it: +let thread_info = get_current_thread_info().unwrap(); +// ... send `thread_info` to another thread, or another process, via any IPC +// mechanism, then: +let handle = promote_thread_to_real_time(thread_info, 512, 44100).unwrap(); +// ... and later, from anywhere: +demote_thread_from_real_time(thread_info).unwrap(); +``` + +Promoting a thread other than the caller's, may require elevated privileges +depending on the platform (for example, Linux uses a privileged rtkit D-Bus +service, and macOS/iOS require `task_for_pid` rights). + This library can also be used from C or C++ using the included header and compiling the rust code in the application. By default, a `.a` is compiled to ease linking. @@ -40,4 +61,3 @@ ease linking. # License MPL-2 - diff --git a/atp_test.cpp b/atp_test.cpp index acdd1d2..82c8127 100644 --- a/atp_test.cpp +++ b/atp_test.cpp @@ -3,25 +3,37 @@ #include #include #include +#include #include "audio_thread_priority.h" int main() { -#ifdef __linux__ atp_thread_info* info = atp_get_current_thread_info(); atp_thread_info* info2 = nullptr; - uint8_t buffer[ATP_THREAD_INFO_SIZE]; - atp_serialize_thread_info(info, buffer); + // ATP_THREAD_INFO_SIZE is a runtime `extern size_t`, not a compile-time constant, so it can't + // size a stack array portably (a VLA is a GCC/Clang extension MSVC doesn't support). Heap- + // allocate instead. + std::vector buffer(ATP_THREAD_INFO_SIZE); + atp_serialize_thread_info(info, buffer.data()); - info2 = atp_deserialize_thread_info(buffer); + info2 = atp_deserialize_thread_info(buffer.data()); - int rv = memcmp(info, info2, ATP_THREAD_INFO_SIZE); + // Compare the two structs via a second serialization round-trip rather than memcmp'ing their + // raw in-memory representations: the platform-specific atp_thread_info can contain padding + // bytes between fields (e.g. on macOS) that aren't guaranteed to be identical between two + // independently-constructed instances even when every actual field is equal. + // atp_serialize_thread_info packs fields explicitly with no padding, so comparing its output is + // well-defined. + std::vector buffer2(ATP_THREAD_INFO_SIZE); + atp_serialize_thread_info(info2, buffer2.data()); + int rv = memcmp(buffer.data(), buffer2.data(), ATP_THREAD_INFO_SIZE); assert(!rv); atp_free_thread_info(info); atp_free_thread_info(info2); +#ifdef __linux__ rv = atp_set_real_time_limit(0, 44100); assert(!rv); #endif diff --git a/audio_thread_priority.h b/audio_thread_priority.h index b3383ec..f10fa48 100644 --- a/audio_thread_priority.h +++ b/audio_thread_priority.h @@ -53,26 +53,28 @@ int32_t atp_demote_current_thread_from_real_time(atp_handle *handle); int32_t atp_free_handle(atp_handle *handle); /* - * Linux-only API. + * Promoting/demoting a thread other than the calling one. * - * This set of functions promotes a thread from another process or thread, for - * cases where the thread to promote cannot do so itself (for example because it - * is sandboxed). With the default build the actual promotion goes through the - * rtkit D-Bus service; when the library is built without the `dbus` feature it - * is done directly, and requires the promoting process to be privileged. + * This is useful when the thread that needs to become real-time cannot + * promote itself directly (for example because it is sandboxed), possibly + * because it lives in another process. * * To do so: - * - Set the real-time limit from within the process where a - * thread will be promoted. This is a `setrlimit` call, that can be done - * before the sandbox lockdown. - * - Then, gather information on the thread that will be promoted. + * - Gather information on the thread that will be promoted, by calling + * `atp_get_current_thread_info` on the thread itself. * - Serialize this info. - * - Send over the serialized data via an IPC mechanism. + * - Send over the serialized data via an IPC mechanism, if the thread that + * will do the promotion is in another process. * - Deserialize the info. * - Call `atp_promote_thread_to_real_time`. + * + * Promoting a thread other than the caller's, may require elevated privileges + * depending on the platform (for example, Linux uses a privileged rtkit D-Bus + * service, or requires the promoting process to be privileged when built + * without the `dbus` feature; macOS/iOS require `task_for_pid` rights when the + * thread lives in another process). */ -#ifdef __linux__ /** * Promotes a thread, possibly in another process, to real-time priority. * @@ -83,11 +85,10 @@ int32_t atp_free_handle(atp_handle *handle); * or an upper bound. * audio_samplerate_hz: sample-rate for this audio stream, in Hz * - * Returns an opaque handle in case of success, NULL otherwise. - * - * This is useful on Linux only, to promote a thread from another process or - * thread when the thread to promote cannot do so itself (for example because it - * is sandboxed). + * Returns an opaque handle in case of success, NULL otherwise. Demote the thread with + * `atp_demote_thread_from_real_time` (using `thread_info`, not this handle). The returned handle + * is not needed for that call, but it is still heap-allocated and must be freed with + * `atp_free_handle` to avoid leaking it. */ atp_handle *atp_promote_thread_to_real_time(atp_thread_info *thread_info); @@ -95,11 +96,10 @@ atp_handle *atp_promote_thread_to_real_time(atp_thread_info *thread_info); * Demotes a thread, promoted to real-time priority via * `atp_promote_thread_to_real_time`, back to its previous priority. * - * Returns 0 in case of success, non-zero otherwise. + * This takes the same `thread_info` passed to `atp_promote_thread_to_real_time`, not the + * `atp_handle` it returned -- free that handle separately with `atp_free_handle`. * - * This is useful on Linux only, to promote a thread from another process or - * thread when the thread to promote cannot do so itself (for example because it - * is sandboxed). + * Returns 0 in case of success, non-zero otherwise. */ int32_t atp_demote_thread_from_real_time(atp_thread_info* thread_info); @@ -109,10 +109,6 @@ int32_t atp_demote_thread_from_real_time(atp_thread_info* thread_info); * * Returns a non-null pointer to an `atp_thread_info` structure in case of * success, to be freed later with `atp_free_thread_info`, and NULL otherwise. - * - * This is useful on Linux only, to promote a thread from another process or - * thread when the thread to promote cannot do so itself (for example because it - * is sandboxed). */ atp_thread_info *atp_get_current_thread_info(); @@ -136,6 +132,7 @@ void atp_serialize_thread_info(atp_thread_info *thread_info, uint8_t *bytes); * */ atp_thread_info* atp_deserialize_thread_info(uint8_t *bytes); +#ifdef __linux__ /** * Set the real-time computation limit (RLIMIT_RTTIME) for the calling process. * diff --git a/src/lib.rs b/src/lib.rs index 1531665..77f951d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -105,11 +105,19 @@ cfg_if! { extern crate libc; use rt_mach::promote_current_thread_to_real_time_internal; use rt_mach::demote_current_thread_from_real_time_internal; + use rt_mach::get_current_thread_info_internal; + use rt_mach::promote_thread_to_real_time_internal; + use rt_mach::demote_thread_from_real_time_internal; + use rt_mach::RtPriorityThreadInfoInternal; use rt_mach::RtPriorityHandleInternal; } else if #[cfg(target_os = "windows")] { mod rt_win; use rt_win::promote_current_thread_to_real_time_internal; use rt_win::demote_current_thread_from_real_time_internal; + use rt_win::get_current_thread_info_internal; + use rt_win::promote_thread_to_real_time_internal; + use rt_win::demote_thread_from_real_time_internal; + use rt_win::RtPriorityThreadInfoInternal; use rt_win::RtPriorityHandleInternal; } else if #[cfg(all(target_os = "linux", feature = "dbus"))] { mod rt_linux; @@ -123,9 +131,6 @@ cfg_if! { use rt_linux::demote_thread_from_real_time_internal; use rt_linux::RtPriorityThreadInfoInternal; use rt_linux::RtPriorityHandleInternal; - #[no_mangle] - /// Size of a RtPriorityThreadInfo or atp_thread_info struct, for use in FFI. - pub static ATP_THREAD_INFO_SIZE: usize = std::mem::size_of::(); } else if #[cfg(target_os = "linux")] { // Linux without the `dbus` feature: promote directly with SCHED_FIFO instead of no-oping. mod rt_linux_native; @@ -139,13 +144,14 @@ cfg_if! { use rt_linux_native::RtPriorityThreadInfoInternal; use rt_linux_native::RtPriorityHandleInternal; pub use rt_linux_native::set_rt_priority; - #[no_mangle] - /// Size of a RtPriorityThreadInfo or atp_thread_info struct, for use in FFI. - pub static ATP_THREAD_INFO_SIZE: usize = std::mem::size_of::(); } else if #[cfg(target_os = "android")] { mod rt_android; use rt_android::promote_current_thread_to_real_time_internal; use rt_android::demote_current_thread_from_real_time_internal; + use rt_android::get_current_thread_info_internal; + use rt_android::promote_thread_to_real_time_internal; + use rt_android::demote_thread_from_real_time_internal; + use rt_android::RtPriorityThreadInfoInternal; use rt_android::RtPriorityHandleInternal; } else { // blanket no-op implementations for platforms without a real-time backend @@ -157,16 +163,14 @@ cfg_if! { _dummy: u8 } - pub type RtPriorityThreadInfo = RtPriorityThreadInfoInternal; - - impl RtPriorityThreadInfo { + impl RtPriorityThreadInfoInternal { /// Serialize the thread info to a byte array (fallback implementation). pub fn serialize(&self) -> [u8; 1] { [0] } /// Deserialize thread info from a byte array (fallback implementation). pub fn deserialize(_: [u8; 1]) -> Self { - RtPriorityThreadInfo{_dummy: 0} + RtPriorityThreadInfoInternal{_dummy: 0} } /// Returns the PID of the process containing the thread (fallback: always -1). pub fn pid(&self) -> i32 { @@ -174,15 +178,15 @@ cfg_if! { } } /// Fallback implementation that performs no operation for unsupported platforms. - pub fn promote_current_thread_to_real_time_internal(_: u32, audio_samplerate_hz: u32) -> Result { + pub fn promote_current_thread_to_real_time_internal(_: u32, audio_samplerate_hz: u32) -> Result { if audio_samplerate_hz == 0 { return Err(AudioThreadPriorityError{message: "sample rate is zero".to_string(), inner: None}); } // no-op - Ok(RtPriorityHandle{}) + Ok(RtPriorityHandleInternal{}) } /// Fallback implementation that performs no operation for unsupported platforms. - pub fn demote_current_thread_from_real_time_internal(_: RtPriorityHandle) -> Result<(), AudioThreadPriorityError> { + pub fn demote_current_thread_from_real_time_internal(_: RtPriorityHandleInternal) -> Result<(), AudioThreadPriorityError> { // no-op Ok(()) } @@ -194,51 +198,46 @@ cfg_if! { Ok(()) } /// Fallback implementation that returns dummy thread info for unsupported platforms. - pub fn get_current_thread_info_internal() -> Result { - Ok(RtPriorityThreadInfo{_dummy: 0}) + pub fn get_current_thread_info_internal() -> Result { + Ok(RtPriorityThreadInfoInternal{_dummy: 0}) } /// Fallback implementation that performs no operation for unsupported platforms. pub fn promote_thread_to_real_time_internal( - _: RtPriorityThreadInfo, + _: RtPriorityThreadInfoInternal, _: u32, audio_samplerate_hz: u32, - ) -> Result { + ) -> Result { if audio_samplerate_hz == 0 { return Err(AudioThreadPriorityError::new("sample rate is zero")); } - Ok(RtPriorityHandle{}) + Ok(RtPriorityHandleInternal{}) } /// Fallback implementation that performs no operation for unsupported platforms. - pub fn demote_thread_from_real_time_internal(_: RtPriorityThreadInfo) -> Result<(), AudioThreadPriorityError> { + pub fn demote_thread_from_real_time_internal(_: RtPriorityThreadInfoInternal) -> Result<(), AudioThreadPriorityError> { Ok(()) } - #[no_mangle] - /// Size of a RtPriorityThreadInfo or atp_thread_info struct, for use in FFI. - pub static ATP_THREAD_INFO_SIZE: usize = std::mem::size_of::(); } } /// Opaque handle to a thread handle structure. pub type RtPriorityHandle = RtPriorityHandleInternal; -cfg_if! { - if #[cfg(target_os = "linux")] { /// Opaque handle to a thread's scheduling information. /// -/// This can be serialized to raw bytes and sent to another process via IPC, so that process can -/// promote the thread to real-time priority on its behalf. -/// -/// This is useful on Linux only, to promote a thread from another process or thread when the -/// thread to promote cannot do so itself (for example because it is sandboxed). +/// This can be serialized to raw bytes and sent to another thread or process via IPC, so that +/// thread or process can promote or demote the thread to/from real-time priority on its behalf. pub type RtPriorityThreadInfo = RtPriorityThreadInfoInternal; +#[no_mangle] +/// Size of a RtPriorityThreadInfo or atp_thread_info struct, for use in FFI. +pub static ATP_THREAD_INFO_SIZE: usize = std::mem::size_of::(); /// Get the calling thread's information, to be able to promote it to real-time from somewhere -/// else, later. +/// else, later on another thread, or in another process. /// -/// This is useful on Linux only, to promote a thread from another process or thread when the -/// thread to promote cannot do so itself (for example because it is sandboxed). +/// This is useful when the thread that needs to become real-time cannot promote itself directly, +/// for example because it is sandboxed. /// /// # Return value /// @@ -250,9 +249,6 @@ pub fn get_current_thread_info() -> Result [u8; std::mem::size_of::()] { @@ -261,9 +257,6 @@ pub fn thread_info_serialize( /// From a byte buffer, return a `RtPriorityThreadInfo`. /// -/// This is useful on Linux only, to promote a thread from another process or thread when the -/// thread to promote cannot do so itself (for example because it is sandboxed). -/// /// # Arguments /// /// A byte buffer containing a serialized `RtPriorityThreadInfo`. @@ -322,10 +315,6 @@ pub unsafe extern "C" fn atp_free_thread_info(thread_info: *mut atp_thread_info) /// /// This is exposed in the C API as `ATP_THREAD_INFO_SIZE`. /// -/// This is useful on Linux only, when the thread to promote lives in another process and the -/// `atp_thread_info` struct must be passed to it via IPC (for example a sandboxed process that -/// cannot promote itself). -/// /// # Safety /// /// This function is safe only and only if the first pointer comes from this library, and the @@ -333,7 +322,7 @@ pub unsafe extern "C" fn atp_free_thread_info(thread_info: *mut atp_thread_info) #[no_mangle] pub unsafe extern "C" fn atp_serialize_thread_info( thread_info: *mut atp_thread_info, - bytes: *mut libc::c_void, + bytes: *mut std::ffi::c_void, ) { let thread_info = &mut *thread_info; let source = thread_info.0.serialize(); @@ -342,9 +331,6 @@ pub unsafe extern "C" fn atp_serialize_thread_info( /// From a byte buffer, return a `RtPriorityThreadInfo`, with a C API. /// -/// This is useful on Linux only, to promote a thread from another process or thread when the -/// thread to promote cannot do so itself (for example because it is sandboxed). -/// /// # Arguments /// /// A byte buffer containing a serialized `RtPriorityThreadInfo`. @@ -353,9 +339,7 @@ pub unsafe extern "C" fn atp_serialize_thread_info( /// /// This function is safe only and only if pointer is at least ATP_THREAD_INFO_SIZE bytes long. #[no_mangle] -pub unsafe extern "C" fn atp_deserialize_thread_info( - in_bytes: *mut u8, -) -> *mut atp_thread_info { +pub unsafe extern "C" fn atp_deserialize_thread_info(in_bytes: *mut u8) -> *mut atp_thread_info { let bytes = *(in_bytes as *mut [u8; std::mem::size_of::()]); let thread_info = RtPriorityThreadInfoInternal::deserialize(bytes); Box::into_raw(Box::new(atp_thread_info(thread_info))) @@ -363,21 +347,29 @@ pub unsafe extern "C" fn atp_deserialize_thread_info( /// Promote a particular thread to real-time priority. /// -/// This is useful on Linux only, to promote a thread from another process or thread when the -/// thread to promote cannot do so itself (for example because it is sandboxed). +/// This is useful when the thread to promote is not the calling thread, is possibly in another +/// process, and/or cannot promote itself directly (for example because of sandboxing). +/// +/// Promoting a thread other than the caller's, especially in another process, may require +/// elevated privileges depending on the platform (for example, Linux uses a privileged rtkit +/// D-Bus service, and macOS/iOS require `task_for_pid` rights). /// /// # Arguments /// -/// * `thread_info` - information about the thread to promote, gathered using -/// `get_current_thread_info`. +/// * `thread_info` - informations about the thread to promote, gathered using +/// `get_current_thread_info`. /// * `audio_buffer_frames` - the exact or an upper limit on the number of frames that have to be -/// rendered each callback, or 0 for a sensible default value. +/// rendered each callback, or 0 for a sensible default value. /// * `audio_samplerate_hz` - the sample-rate for this audio stream, in Hz. /// /// # Return value /// -/// This function returns a `Result`, which is an opaque struct to be passed to -/// `demote_current_thread_from_real_time` to revert to the previous thread priority. +/// This function returns a `Result`. Unlike `promote_current_thread_to_real_time`, +/// this handle should NOT be passed to `demote_current_thread_from_real_time`: use +/// `demote_thread_from_real_time(thread_info)` instead, with the same `thread_info` used here, to +/// revert to the previous thread priority (the promoting and demoting call may not happen on the +/// same thread, or even in the same process, so the handle alone isn't sufficient on every +/// platform). pub fn promote_thread_to_real_time( thread_info: RtPriorityThreadInfo, audio_buffer_frames: u32, @@ -386,11 +378,7 @@ pub fn promote_thread_to_real_time( if audio_samplerate_hz == 0 { return Err(AudioThreadPriorityError::new("sample rate is zero")); } - promote_thread_to_real_time_internal( - thread_info, - audio_buffer_frames, - audio_samplerate_hz, - ) + promote_thread_to_real_time_internal(thread_info, audio_buffer_frames, audio_samplerate_hz) } /// Demotes a thread from real-time priority. @@ -398,12 +386,14 @@ pub fn promote_thread_to_real_time( /// # Arguments /// /// * `thread_info` - An opaque struct returned from a successful call to -/// `get_current_thread_info`. +/// `get_current_thread_info`. /// /// # Return value /// /// `Ok` in case of success, `Err` otherwise. -pub fn demote_thread_from_real_time(thread_info: RtPriorityThreadInfo) -> Result<(), AudioThreadPriorityError> { +pub fn demote_thread_from_real_time( + thread_info: RtPriorityThreadInfo, +) -> Result<(), AudioThreadPriorityError> { demote_thread_from_real_time_internal(thread_info) } @@ -418,10 +408,10 @@ pub struct atp_thread_info(RtPriorityThreadInfo); /// /// # Arguments /// -/// `thread_info` - the information of the thread to promote to real-time, gather from calling -/// `atp_get_current_thread_info` on the thread to promote. +/// * `thread_info` - the information of the thread to promote to real-time, gather from calling +/// `atp_get_current_thread_info` on the thread to promote. /// * `audio_buffer_frames` - the exact or an upper limit on the number of frames that have to be -/// rendered each callback, or 0 for a sensible default value. +/// rendered each callback, or 0 for a sensible default value. /// * `audio_samplerate_hz` - the sample-rate for this audio stream, in Hz. /// /// # Return value @@ -458,7 +448,9 @@ pub unsafe extern "C" fn atp_promote_thread_to_real_time( /// /// This function is safe as long as the first pointer comes from this library, or is null. #[no_mangle] -pub unsafe extern "C" fn atp_demote_thread_from_real_time(thread_info: *mut atp_thread_info) -> i32 { +pub unsafe extern "C" fn atp_demote_thread_from_real_time( + thread_info: *mut atp_thread_info, +) -> i32 { if thread_info.is_null() { return 1; } @@ -472,6 +464,10 @@ pub unsafe extern "C" fn atp_demote_thread_from_real_time(thread_info: *mut atp_ /// Set a real-time limit for the calling thread. /// +/// This is only necessary and available on Linux desktop, and allows remoting the rtkit D-Bus +/// call to a process that has access to D-Bus. This function has to be called before attempting +/// to promote threads from another process. +/// /// # Arguments /// /// `audio_buffer_frames` - the number of frames the audio callback has to render each quantum. 0 @@ -481,9 +477,12 @@ pub unsafe extern "C" fn atp_demote_thread_from_real_time(thread_info: *mut atp_ /// # Return value /// /// 0 in case of success, 1 otherwise. +#[cfg(target_os = "linux")] #[no_mangle] -pub extern "C" fn atp_set_real_time_limit(audio_buffer_frames: u32, - audio_samplerate_hz: u32) -> i32 { +pub extern "C" fn atp_set_real_time_limit( + audio_buffer_frames: u32, + audio_samplerate_hz: u32, +) -> i32 { let r = set_real_time_hard_limit(audio_buffer_frames, audio_samplerate_hz); if r.is_err() { return 1; @@ -491,9 +490,6 @@ pub extern "C" fn atp_set_real_time_limit(audio_buffer_frames: u32, 0 } -} -} - /// Promote the calling thread to real-time priority. /// /// # Arguments @@ -720,6 +716,75 @@ mod tests { } } + #[test] + fn test_thread_info_serialization() { + let info = get_current_thread_info().unwrap(); + let bytes = info.serialize(); + let info2 = RtPriorityThreadInfo::deserialize(bytes); + assert!(info == info2); + + let bytes = thread_info_serialize(info); + let info2 = thread_info_deserialize(bytes); + assert!(info == info2); + } + + #[test] + fn test_promote_another_thread() { + use std::sync::mpsc::channel; + + #[cfg(all(target_os = "linux", not(feature = "dbus")))] + if !rt_scheduling_available() { + eprintln!( + "skipping test_promote_another_thread: real-time scheduling is not permitted here" + ); + return; + } + + let (info_tx, info_rx) = channel(); + let (done_tx, done_rx) = channel(); + + let handle = std::thread::spawn(move || { + let info = get_current_thread_info().unwrap(); + info_tx.send(info).unwrap(); + // Keep this thread alive while the main thread promotes and demotes it. + done_rx.recv().unwrap(); + }); + + // Releases and joins the spawned thread on every exit path, including a panic from + // promotion/demotion below (`Drop` still runs while unwinding): otherwise a failing + // assertion here would leak that thread, parked on `done_rx.recv()` forever, and if the + // panic happened after a successful promotion, pinned at real-time priority for the rest + // of the test binary's lifetime. + struct ReleaseOnDrop { + done_tx: std::sync::mpsc::Sender<()>, + handle: Option>, + } + impl Drop for ReleaseOnDrop { + fn drop(&mut self) { + let _ = self.done_tx.send(()); + if let Some(handle) = self.handle.take() { + let _ = handle.join(); + } + } + } + let _release = ReleaseOnDrop { + done_tx, + handle: Some(handle), + }; + + let info = info_rx.recv().unwrap(); + + match promote_thread_to_real_time(info, 512, 44100) { + Ok(_) => {} + Err(e) => panic!("{}", e), + } + + match demote_thread_from_real_time(info) { + Ok(_) => {} + Err(e) => panic!("{}", e), + } + } + cfg_if! { if #[cfg(target_os = "linux")] { use nix::unistd::*; @@ -727,35 +792,6 @@ mod tests { #[cfg(not(feature = "dbus"))] use nix::sys::wait::*; - #[test] - fn test_linux_api() { - #[cfg(not(feature = "dbus"))] - if !rt_scheduling_available() { - eprintln!("skipping test_linux_api: real-time scheduling is not permitted here"); - return; - } - { - let info = get_current_thread_info().unwrap(); - match promote_thread_to_real_time(info, 512, 44100) { - Ok(_) => { } - Err(e) => { - panic!("{}", e); - } - } - } - { - let info = get_current_thread_info().unwrap(); - let bytes = info.serialize(); - let info2 = RtPriorityThreadInfo::deserialize(bytes); - assert!(info == info2); - } - { - let info = get_current_thread_info().unwrap(); - let bytes = thread_info_serialize(info); - let info2 = thread_info_deserialize(bytes); - assert!(info == info2); - } - } #[test] fn test_remote_promotion() { let (rd, wr) = pipe().unwrap(); diff --git a/src/rt_android.rs b/src/rt_android.rs index a9b9ca5..c5b25a9 100644 --- a/src/rt_android.rs +++ b/src/rt_android.rs @@ -11,50 +11,244 @@ const THREAD_PRIORITY_URGENT_AUDIO: libc::c_int = -19; #[derive(Debug)] pub struct RtPriorityHandleInternal { + // The thread this handle was captured for. Always the calling thread for + // `promote_current_thread_to_real_time_internal`, but the promoted thread's tid for + // `promote_thread_to_real_time_internal` -- storing it here (rather than assuming "the + // calling thread" on demotion) means demoting always affects the thread that was actually + // promoted, even if the caller mixes up `demote_current_thread_from_real_time` with + // `demote_thread_from_real_time`. + pid: libc::pid_t, + tid: libc::pid_t, + // See `RtPriorityThreadInfoInternal::start_time`: carried here too so that even a misuse of + // this handle for demotion re-validates thread identity rather than trusting a possibly- + // recycled `tid`. + start_time: u64, previous_priority: libc::c_int, } -pub fn promote_current_thread_to_real_time_internal( - _: u32, - _: u32, -) -> Result { - // Android's Process.setThreadPriority() ultimately calls setpriority(). - // See https://android.googlesource.com/platform/frameworks/base/+/master/core/jni/android_util_Process.cpp#543 - // and https://android.googlesource.com/platform/system/core/+/master/libutils/Threads.cpp#312 +/// Opaque, serializable information about a thread, possibly running in another process, +/// sufficient to promote or demote it to/from real-time priority. +/// +/// Android runs on the Linux kernel, and `setpriority()`/`getpriority()` with `PRIO_PROCESS` +/// already operate on an individual kernel task (thread) given its `tid`, not just the calling +/// thread, so no daemon or extra IPC is needed to target another therad. Doing so across +/// processes additionally requires the caller to have the right privileges (matching uid, or +/// `CAP_SYS_NICE`) to renice that thread, which is not generally available to sandboxed Android +/// apps targeting another process, but works for another thread in the same process. +/// +/// `start_time` (from `/proc//task//stat`) is captured alongside `tid` because Linux/ +/// Android thread ids are reused once a thread exits: without it, a `tid` captured here could, by +/// the time `promote_thread_to_real_time`/`demote_thread_from_real_time` runs, refer to a +/// completely different, unrelated thread that happened to be assigned the same id in the +/// meantime. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct RtPriorityThreadInfoInternal { + pid: libc::pid_t, + tid: libc::pid_t, + previous_priority: libc::c_int, + start_time: u64, +} - // Per https://github.com/android/ndk/issues/1255 - // and https://android.googlesource.com/platform/bionic/+/master/libc/include/pthread.h#388, - // it's acceptable to call setpriority() directly for native threads. +impl RtPriorityThreadInfoInternal { + /// Serialize to a byte buffer. The fields are packed explicitly rather than transmuting the + /// struct, matching the approach used on the other platforms, so the format doesn't silently + /// start reading uninitialized padding if a field is ever reordered or resized. + pub fn serialize(&self) -> [u8; std::mem::size_of::()] { + let pid = self.pid.to_ne_bytes(); + let tid = self.tid.to_ne_bytes(); + let previous_priority = self.previous_priority.to_ne_bytes(); + let start_time = self.start_time.to_ne_bytes(); + + let mut bytes = [0u8; std::mem::size_of::()]; + let fields = pid + .iter() + .chain(&tid) + .chain(&previous_priority) + .chain(&start_time); + for (dst, &src) in bytes.iter_mut().zip(fields) { + *dst = src; + } + bytes + } + /// Reconstruct from a byte buffer produced by `serialize`. + pub fn deserialize(bytes: [u8; std::mem::size_of::()]) -> Self { + fn take(src: &mut impl Iterator) -> [u8; N] { + let mut chunk = [0u8; N]; + for slot in &mut chunk { + *slot = src.next().unwrap(); + } + chunk + } + let mut src = bytes.iter().copied(); + RtPriorityThreadInfoInternal { + pid: libc::pid_t::from_ne_bytes(take(&mut src)), + tid: libc::pid_t::from_ne_bytes(take(&mut src)), + previous_priority: libc::c_int::from_ne_bytes(take(&mut src)), + start_time: u64::from_ne_bytes(take(&mut src)), + } + } + /// Returns the PID of the process containing the thread. + pub fn pid(&self) -> libc::pid_t { + self.pid + } +} + +impl PartialEq for RtPriorityThreadInfoInternal { + // Compares identity only (which thread, in which process, started when), not the captured + // `previous_priority`, matching `rt_mach.rs`/`rt_linux.rs`. `start_time` is included because + // it's exactly what distinguishes the original thread from an unrelated one that later reused + // the same `tid`. + fn eq(&self, other: &Self) -> bool { + self.pid == other.pid && self.tid == other.tid && self.start_time == other.start_time + } +} - let who = unsafe { libc::gettid().try_into().unwrap() }; +/// Read thread `tid` (in process `pid`)'s start time from `/proc//task//stat`, field 22 +/// ("starttime", clock ticks since boot -- see `proc(5)`). Two threads never share a start time, +/// and a given `tid` gets a new one once it's reused by a different thread after the original +/// exits, so comparing this is how thread identity is re-validated across the gap between +/// capturing `RtPriorityThreadInfoInternal` and acting on it later. Reading another process' +/// `/proc//task//stat` is subject to the same permission rules as `setpriority()` on it +/// (same uid, or `CAP_SYS_NICE`/ptrace access), so this fails closed exactly where the underlying +/// promote/demote would have failed anyway. +fn thread_start_time(pid: libc::pid_t, tid: libc::pid_t) -> Result { + let path = format!("/proc/{pid}/task/{tid}/stat"); + let contents = std::fs::read_to_string(&path).map_err(|_| { + AudioThreadPriorityError::new(&format!("could not read {path} (has the thread exited?)")) + })?; + // Format: " () ... ...". `comm` can itself contain + // spaces or parentheses, so only split into fields after the last ')'. + let after_comm = contents + .rfind(')') + .map(|i| &contents[i + 1..]) + .ok_or_else(|| AudioThreadPriorityError::new("unexpected /proc/.../stat format"))?; + // `state` is field 3, the first field after `comm`; `starttime` is field 22, i.e. the token at + // index 22 - 3 = 19 (zero-based) among the fields following `comm`. + after_comm + .split_whitespace() + .nth(19) + .and_then(|s| s.parse::().ok()) + .ok_or_else(|| AudioThreadPriorityError::new("unexpected /proc/.../stat format")) +} +fn get_thread_priority(tid: libc::pid_t) -> Result { + let who = tid + .try_into() + .map_err(|_| AudioThreadPriorityError::new("Invalid thread id"))?; unsafe { (*libc::__errno()) = 0 }; - let previous_priority = unsafe { libc::getpriority(libc::PRIO_PROCESS, who) }; - if previous_priority == -1 && unsafe { *libc::__errno() } != 0 { + let priority = unsafe { libc::getpriority(libc::PRIO_PROCESS, who) }; + if priority == -1 && unsafe { *libc::__errno() } != 0 { return Err(AudioThreadPriorityError::new( - "Failed to get current thread priority", + "Failed to get thread priority", )); } + Ok(priority) +} + +/// Set a thread's scheduling priority via `setpriority()`, after confirming via +/// `thread_start_time` that `tid` still refers to the same thread `expected_start_time` was +/// captured for. `tid`/`expected_start_time` may come from another process' (deserialized) +/// `RtPriorityThreadInfoInternal`, so this reports an error instead of panicking on an +/// out-of-range value rather than trusting it implicitly. Note there's an inherent, narrow race +/// between the check and `setpriority()` below (the thread could exit and its tid be reused in +/// between); this closes the much wider gap between capture and use, but isn't a full guarantee. +fn set_thread_priority( + pid: libc::pid_t, + tid: libc::pid_t, + expected_start_time: u64, + priority: libc::c_int, +) -> Result<(), AudioThreadPriorityError> { + let start_time = thread_start_time(pid, tid)?; + if start_time != expected_start_time { + return Err(AudioThreadPriorityError::new(&format!( + "thread {tid} in process {pid} has exited and its id was reused by another thread" + ))); + } - let r = unsafe { libc::setpriority(libc::PRIO_PROCESS, who, THREAD_PRIORITY_URGENT_AUDIO) }; + let who = tid + .try_into() + .map_err(|_| AudioThreadPriorityError::new("Invalid thread id"))?; + let r = unsafe { libc::setpriority(libc::PRIO_PROCESS, who, priority) }; if r < 0 { return Err(AudioThreadPriorityError::new( - "Failed to set current thread priority", + "Failed to set thread priority", )); } + Ok(()) +} + +/// Get the current thread information, as an opaque struct, that can be serialized and sent +/// across processes, to have another thread promoted to real-time. +pub fn get_current_thread_info_internal( +) -> Result { + let pid = unsafe { libc::getpid() }; + let tid = unsafe { libc::gettid() }; + let previous_priority = get_thread_priority(tid)?; + let start_time = thread_start_time(pid, tid)?; + + Ok(RtPriorityThreadInfoInternal { + pid, + tid, + previous_priority, + start_time, + }) +} - Ok(RtPriorityHandleInternal { previous_priority }) +pub fn promote_current_thread_to_real_time_internal( + audio_buffer_frames: u32, + audio_samplerate_hz: u32, +) -> Result { + let thread_info = get_current_thread_info_internal()?; + promote_thread_to_real_time_internal(thread_info, audio_buffer_frames, audio_samplerate_hz) } pub fn demote_current_thread_from_real_time_internal( h: RtPriorityHandleInternal, ) -> Result<(), AudioThreadPriorityError> { - let who = unsafe { libc::gettid().try_into().unwrap() }; - let r = unsafe { libc::setpriority(libc::PRIO_PROCESS, who, h.previous_priority) }; - if r < 0 { - return Err(AudioThreadPriorityError::new( - "Failed to demote thread priority", - )); - } - Ok(()) + // Per https://github.com/android/ndk/issues/1255 + // and https://android.googlesource.com/platform/bionic/+/master/libc/include/pthread.h#388, + // it's acceptable to call setpriority() directly for native threads. + // + // Uses the identity captured in the handle (rather than assuming "the calling thread") so + // that this still demotes the correct thread even if `h` came from `promote_thread_to_real_time` + // for a thread other than the caller. + set_thread_priority(h.pid, h.tid, h.start_time, h.previous_priority) +} + +/// Promote a thread (possibly in another process) identified by its thread info, to real-time. +pub fn promote_thread_to_real_time_internal( + thread_info: RtPriorityThreadInfoInternal, + _audio_buffer_frames: u32, + _audio_samplerate_hz: u32, +) -> Result { + // Android's Process.setThreadPriority() ultimately calls setpriority(). + // See https://android.googlesource.com/platform/frameworks/base/+/master/core/jni/android_util_Process.cpp#543 + // and https://android.googlesource.com/platform/system/core/+/master/libutils/Threads.cpp#312 + set_thread_priority( + thread_info.pid, + thread_info.tid, + thread_info.start_time, + THREAD_PRIORITY_URGENT_AUDIO, + )?; + + Ok(RtPriorityHandleInternal { + pid: thread_info.pid, + tid: thread_info.tid, + start_time: thread_info.start_time, + previous_priority: thread_info.previous_priority, + }) +} + +/// This can be called by sandboxed code, it restores the priority the thread had when +/// `get_current_thread_info` captured `thread_info`. +pub fn demote_thread_from_real_time_internal( + thread_info: RtPriorityThreadInfoInternal, +) -> Result<(), AudioThreadPriorityError> { + set_thread_priority( + thread_info.pid, + thread_info.tid, + thread_info.start_time, + thread_info.previous_priority, + ) } diff --git a/src/rt_mach.rs b/src/rt_mach.rs index 99e8c9d..75a3064 100644 --- a/src/rt_mach.rs +++ b/src/rt_mach.rs @@ -1,15 +1,42 @@ use crate::AudioThreadPriorityError; -use libc::{pthread_mach_thread_np, pthread_self, thread_policy_t}; +use libc::{pthread_mach_thread_np, pthread_self, pthread_threadid_np, thread_policy_t}; use log::info; use mach2::boolean::boolean_t; use mach2::kern_return::{kern_return_t, KERN_SUCCESS}; +use mach2::mach_port::mach_port_deallocate; use mach2::mach_time::{mach_timebase_info, mach_timebase_info_data_t}; +use mach2::mach_types::{task_t, thread_act_array_t, thread_act_t}; use mach2::message::mach_msg_type_number_t; -use mach2::port::mach_port_t; +use mach2::port::{mach_port_t, MACH_PORT_NULL}; +use mach2::task::task_threads; use mach2::thread_policy::{ thread_policy_get, thread_policy_set, thread_time_constraint_policy_data_t, THREAD_TIME_CONSTRAINT_POLICY, THREAD_TIME_CONSTRAINT_POLICY_COUNT, }; +use mach2::traps::{mach_task_self, task_for_pid}; +use mach2::vm::mach_vm_deallocate; +use mach2::vm_types::natural_t; + +// Not exposed by the `mach2` crate: mirrors `mach/thread_info.h`. +const THREAD_IDENTIFIER_INFO: natural_t = 4; +const THREAD_IDENTIFIER_INFO_COUNT: mach_msg_type_number_t = 6; + +#[repr(C)] +#[derive(Clone, Copy, Default)] +struct ThreadIdentifierInfo { + thread_id: u64, + thread_handle: u64, + dispatch_qaddr: u64, +} + +extern "C" { + fn thread_info( + target_act: thread_act_t, + flavor: natural_t, + thread_info_out: *mut i32, + thread_info_out_count: *mut mach_msg_type_number_t, + ) -> kern_return_t; +} #[derive(Debug)] pub struct RtPriorityHandleInternal { @@ -37,117 +64,359 @@ impl RtPriorityHandleInternal { } } -pub fn demote_current_thread_from_real_time_internal( - rt_priority_handle: RtPriorityHandleInternal, -) -> Result<(), AudioThreadPriorityError> { +/// Opaque, serializable information about a thread sufficient to promote or demote it to/from +/// real-time priority. +/// +/// This doesn't contain a mach port: port names are only meaningful within the task that owns +/// them, so a raw port name can't be shipped across processes. Instead, this carries a stable, +/// system-wide thread identifier (from `pthread_threadid_np`) and the owning process' pid, and +/// the port is (re-)resolved via `task_threads`/`thread_info` whenever it's needed, in whichever +/// process ends up doing the promotion. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct RtPriorityThreadInfoInternal { + pid: libc::pid_t, + thread_id: u64, + previous_time_constraint_policy: thread_time_constraint_policy_data_t, +} + +impl RtPriorityThreadInfoInternal { + /// Serialize to a byte buffer. The fields are packed explicitly rather than transmuting the + /// struct (there's a padding gap between `pid` and `thread_id` since the latter needs 8-byte + /// alignment), so no uninitialized padding bytes are ever read. + pub fn serialize(&self) -> [u8; std::mem::size_of::()] { + let pid = self.pid.to_ne_bytes(); + let thread_id = self.thread_id.to_ne_bytes(); + let period = self.previous_time_constraint_policy.period.to_ne_bytes(); + let computation = self + .previous_time_constraint_policy + .computation + .to_ne_bytes(); + let constraint = self + .previous_time_constraint_policy + .constraint + .to_ne_bytes(); + let preemptible = self + .previous_time_constraint_policy + .preemptible + .to_ne_bytes(); + + let mut bytes = [0u8; std::mem::size_of::()]; + let fields = pid + .iter() + .chain(&thread_id) + .chain(&period) + .chain(&computation) + .chain(&constraint) + .chain(&preemptible); + for (dst, &src) in bytes.iter_mut().zip(fields) { + *dst = src; + } + bytes + } + /// Reconstruct from a byte buffer produced by `serialize`. + pub fn deserialize(bytes: [u8; std::mem::size_of::()]) -> Self { + fn take(src: &mut impl Iterator) -> [u8; N] { + let mut chunk = [0u8; N]; + for slot in &mut chunk { + *slot = src.next().unwrap(); + } + chunk + } + let mut src = bytes.iter().copied(); + RtPriorityThreadInfoInternal { + pid: libc::pid_t::from_ne_bytes(take(&mut src)), + thread_id: u64::from_ne_bytes(take(&mut src)), + previous_time_constraint_policy: thread_time_constraint_policy_data_t { + period: u32::from_ne_bytes(take(&mut src)), + computation: u32::from_ne_bytes(take(&mut src)), + constraint: u32::from_ne_bytes(take(&mut src)), + preemptible: boolean_t::from_ne_bytes(take(&mut src)), + }, + } + } + /// Returns the PID of the process containing the thread. + pub fn pid(&self) -> libc::pid_t { + self.pid + } +} + +impl PartialEq for RtPriorityThreadInfoInternal { + fn eq(&self, other: &Self) -> bool { + self.pid == other.pid && self.thread_id == other.thread_id + } +} + +/// Resolve a `(pid, thread_id)` pair, as captured by `get_current_thread_info_internal`, into a +/// thread port that can be passed to `thread_policy_set`/`thread_policy_get`. +/// +/// This works unconditionally for a thread running in the current process. For a thread in +/// another process, this requires being able to get a send right to that process' task port +/// (via `task_for_pid`), which needs elevated privileges or the +/// `com.apple.security.get-task-allow` entitlement. +/// +/// The returned port is a right owned by the caller, and must be released with +/// `mach_port_deallocate` once done with it. +fn resolve_thread_port( + pid: libc::pid_t, + thread_id: u64, +) -> Result { + let is_current_process = pid == unsafe { libc::getpid() }; + + let task: task_t = if is_current_process { + unsafe { mach_task_self() } + } else { + let mut task_port: mach_port_t = MACH_PORT_NULL; + let kr = unsafe { task_for_pid(mach_task_self(), pid, &mut task_port) }; + if kr != KERN_SUCCESS { + return Err(AudioThreadPriorityError::new(&format!( + "task_for_pid failed for pid {pid} (kern_return_t {kr}): promoting a thread in \ + another process requires elevated privileges or the \ + com.apple.security.get-task-allow entitlement" + ))); + } + task_port + }; + + // `task` is only a right we own (and must release) when it came from `task_for_pid`; + // `mach_task_self()` is a borrowed, static right that must not be deallocated. Centralizing + // this in one closure means every exit path below releases it exactly the same way. + let deallocate_task = || { + if !is_current_process { + unsafe { mach_port_deallocate(mach_task_self(), task) }; + } + }; + + let mut thread_list: thread_act_array_t = std::ptr::null_mut(); + let mut thread_count: mach_msg_type_number_t = 0; + + let kr = unsafe { task_threads(task, &mut thread_list, &mut thread_count) }; + if kr != KERN_SUCCESS { + deallocate_task(); + return Err(AudioThreadPriorityError::new("task_threads failed")); + } + + let mut found: Option = None; + for i in 0..thread_count { + let port = unsafe { *thread_list.add(i as usize) }; + let mut ident = ThreadIdentifierInfo::default(); + let mut count = THREAD_IDENTIFIER_INFO_COUNT; + let kr = unsafe { + thread_info( + port, + THREAD_IDENTIFIER_INFO, + &mut ident as *mut _ as *mut i32, + &mut count, + ) + }; + if kr == KERN_SUCCESS && ident.thread_id == thread_id { + found = Some(port); + // No need to query the remaining threads now that the match is found; just + // deallocate their ports without another `thread_info` call each. + for j in (i + 1)..thread_count { + let port = unsafe { *thread_list.add(j as usize) }; + unsafe { mach_port_deallocate(mach_task_self(), port) }; + } + break; + } else { + unsafe { mach_port_deallocate(mach_task_self(), port) }; + } + } + unsafe { - let mut h = rt_priority_handle; - let rv: kern_return_t = thread_policy_set( - h.tid, - THREAD_TIME_CONSTRAINT_POLICY, - (&mut h.previous_time_constraint_policy) as *mut _ as thread_policy_t, - THREAD_TIME_CONSTRAINT_POLICY_COUNT, + mach_vm_deallocate( + mach_task_self(), + thread_list as u64, + (thread_count as usize * std::mem::size_of::()) as u64, ); - if rv != KERN_SUCCESS { - return Err(AudioThreadPriorityError::new( - "thread demotion error: thread_policy_get: RT", - )); - } + } + + deallocate_task(); - info!("thread {} priority restored.", h.tid); + found.ok_or_else(|| { + AudioThreadPriorityError::new(&format!( + "could not find thread {thread_id} in process {pid} (has it exited?)" + )) + }) +} + +fn get_time_constraint_policy( + port: mach_port_t, +) -> Result { + let mut policy = thread_time_constraint_policy_data_t { + period: 0, + computation: 0, + constraint: 0, + preemptible: 0, + }; + let mut get_default: boolean_t = 0; + let mut count: mach_msg_type_number_t = THREAD_TIME_CONSTRAINT_POLICY_COUNT; + let rv = unsafe { + thread_policy_get( + port, + THREAD_TIME_CONSTRAINT_POLICY, + (&mut policy) as *mut _ as thread_policy_t, + &mut count, + &mut get_default, + ) + }; + if rv != KERN_SUCCESS { + return Err(AudioThreadPriorityError::new( + "thread_policy_get: time_constraint", + )); } + Ok(policy) +} +fn set_time_constraint_policy( + port: mach_port_t, + mut policy: thread_time_constraint_policy_data_t, +) -> Result<(), AudioThreadPriorityError> { + let rv = unsafe { + thread_policy_set( + port, + THREAD_TIME_CONSTRAINT_POLICY, + (&mut policy) as *mut _ as thread_policy_t, + THREAD_TIME_CONSTRAINT_POLICY_COUNT, + ) + }; + if rv != KERN_SUCCESS { + return Err(AudioThreadPriorityError::new( + "thread_policy_set: time_constraint", + )); + } Ok(()) } -pub fn promote_current_thread_to_real_time_internal( +/// The time constraint calculations are somewhat arbitrary for now. +fn compute_time_constraint_policy( audio_buffer_frames: u32, audio_samplerate_hz: u32, -) -> Result { - let mut rt_priority_handle = RtPriorityHandleInternal::new(); - +) -> thread_time_constraint_policy_data_t { let buffer_frames = if audio_buffer_frames > 0 { audio_buffer_frames } else { audio_samplerate_hz / 20 }; - unsafe { - let tid: mach_port_t = pthread_mach_thread_np(pthread_self()); - let mut time_constraints = thread_time_constraint_policy_data_t { - period: 0, - computation: 0, - constraint: 0, - preemptible: 0, - }; + let mut timebase_info = mach_timebase_info_data_t { denom: 0, numer: 0 }; + unsafe { mach_timebase_info(&mut timebase_info) }; - // Get current thread attributes, to revert back to the correct setting later if needed. - rt_priority_handle.tid = tid; + let ms2abs: f32 = ((timebase_info.denom as f32) / timebase_info.numer as f32) * 1000000.; - // false: we want to get the current value, not the default value. If this is `false` after - // returning, it means there are no current settings because of other factor, and the - // default was returned instead. - let mut get_default: boolean_t = 0; - let mut count: mach_msg_type_number_t = THREAD_TIME_CONSTRAINT_POLICY_COUNT; - let mut rv: kern_return_t = thread_policy_get( - tid, - THREAD_TIME_CONSTRAINT_POLICY, - (&mut time_constraints) as *mut _ as thread_policy_t, - &mut count, - &mut get_default, - ); + let cb_duration = buffer_frames as f32 / (audio_samplerate_hz as f32) * 1000.; - if rv != KERN_SUCCESS { - return Err(AudioThreadPriorityError::new( - "thread promotion error: thread_policy_get: time_constraint", - )); - } + // Computation time is half of constraint, per macOS 12 behaviour. And capped at 50ms per + // macOS limits: + // https://github.com/apple-oss-distributions/xnu/blob/e3723e1f17661b24996789d8afc084c0c3303b26/osfmk/kern/thread_policy.c#L408 + // https://github.com/apple-oss-distributions/xnu/blob/e3723e1f17661b24996789d8afc084c0c3303b26/osfmk/kern/sched_prim.c#L822 + const MAX_RT_QUANTUM: f32 = 50.0; + let computation = cb_duration / 2.0; + let computation = if computation > MAX_RT_QUANTUM { + info!("thread computation time capped at {MAX_RT_QUANTUM}ms ({computation}ms requested)."); + MAX_RT_QUANTUM + } else { + computation + }; - rt_priority_handle.previous_time_constraint_policy = time_constraints; + thread_time_constraint_policy_data_t { + period: (cb_duration * ms2abs) as u32, + computation: (computation * ms2abs) as u32, + constraint: (cb_duration * ms2abs) as u32, + preemptible: 1, // true + } +} - let mut timebase_info = mach_timebase_info_data_t { denom: 0, numer: 0 }; - mach_timebase_info(&mut timebase_info); +pub fn demote_current_thread_from_real_time_internal( + rt_priority_handle: RtPriorityHandleInternal, +) -> Result<(), AudioThreadPriorityError> { + set_time_constraint_policy( + rt_priority_handle.tid, + rt_priority_handle.previous_time_constraint_policy, + )?; - let ms2abs: f32 = ((timebase_info.denom as f32) / timebase_info.numer as f32) * 1000000.; + info!("thread {} priority restored.", rt_priority_handle.tid); - // The time constraint calculations are somewhat arbitrary for now. - let cb_duration = buffer_frames as f32 / (audio_samplerate_hz as f32) * 1000.; + Ok(()) +} - // Computation time is half of constraint, per macOS 12 behaviour. And capped at 50ms per macOS limits: - // https://github.com/apple-oss-distributions/xnu/blob/e3723e1f17661b24996789d8afc084c0c3303b26/osfmk/kern/thread_policy.c#L408 - // https://github.com/apple-oss-distributions/xnu/blob/e3723e1f17661b24996789d8afc084c0c3303b26/osfmk/kern/sched_prim.c#L822 - const MAX_RT_QUANTUM: f32 = 50.0; - let computation = cb_duration / 2.0; - let computation = if computation > MAX_RT_QUANTUM { - info!( - "thread computation time capped at {MAX_RT_QUANTUM}ms ({computation}ms requested)." - ); - MAX_RT_QUANTUM - } else { - computation - }; +/// Get the current thread information, as an opaque struct, that can be serialized and sent +/// accross processes. This is enough to capture the current state of the scheduling policy, and +/// an identifier to have another thread promoted to real-time. +pub fn get_current_thread_info_internal( +) -> Result { + let pid = unsafe { libc::getpid() }; - time_constraints = thread_time_constraint_policy_data_t { - period: (cb_duration * ms2abs) as u32, - computation: (computation * ms2abs) as u32, - constraint: (cb_duration * ms2abs) as u32, - preemptible: 1, // true - }; + let mut thread_id: u64 = 0; + // A null (0) `pthread_t` means "the current thread". + if unsafe { pthread_threadid_np(0, &mut thread_id) } != 0 { + return Err(AudioThreadPriorityError::new("pthread_threadid_np failed")); + } - rv = thread_policy_set( - tid, - THREAD_TIME_CONSTRAINT_POLICY, - (&mut time_constraints) as *mut _ as thread_policy_t, - THREAD_TIME_CONSTRAINT_POLICY_COUNT, - ); - if rv != KERN_SUCCESS { - return Err(AudioThreadPriorityError::new( - "thread promotion error: thread_policy_set: time_constraint", - )); - } + let tid: mach_port_t = unsafe { pthread_mach_thread_np(pthread_self()) }; + let previous_time_constraint_policy = get_time_constraint_policy(tid)?; - info!("thread {tid} bumped to real time priority."); - } + Ok(RtPriorityThreadInfoInternal { + pid, + thread_id, + previous_time_constraint_policy, + }) +} + +pub fn promote_current_thread_to_real_time_internal( + audio_buffer_frames: u32, + audio_samplerate_hz: u32, +) -> Result { + let thread_info = get_current_thread_info_internal()?; + let tid: mach_port_t = unsafe { pthread_mach_thread_np(pthread_self()) }; + + let policy = compute_time_constraint_policy(audio_buffer_frames, audio_samplerate_hz); + set_time_constraint_policy(tid, policy)?; + + info!("thread {tid} bumped to real time priority."); + + Ok(RtPriorityHandleInternal { + tid, + previous_time_constraint_policy: thread_info.previous_time_constraint_policy, + }) +} + +/// Promote a thread (possibly in another process) identified by its thread info, to real-time. +pub fn promote_thread_to_real_time_internal( + thread_info: RtPriorityThreadInfoInternal, + audio_buffer_frames: u32, + audio_samplerate_hz: u32, +) -> Result { + let port = resolve_thread_port(thread_info.pid, thread_info.thread_id)?; + + let policy = compute_time_constraint_policy(audio_buffer_frames, audio_samplerate_hz); + let rv = set_time_constraint_policy(port, policy); + + unsafe { mach_port_deallocate(mach_task_self(), port) }; + rv?; + + info!( + "thread {} (pid {}) bumped to real time priority.", + thread_info.thread_id, thread_info.pid + ); + + // The returned handle isn't used to demote this thread: `demote_thread_from_real_time` + // resolves the thread again from `thread_info` instead, since the promoting and demoting + // call may not happen on the same thread, or even in the same process. + Ok(RtPriorityHandleInternal { + tid: MACH_PORT_NULL, + previous_time_constraint_policy: thread_info.previous_time_constraint_policy, + }) +} + +/// This can be called by sandboxed code, it only restores priority to what they were. +pub fn demote_thread_from_real_time_internal( + thread_info: RtPriorityThreadInfoInternal, +) -> Result<(), AudioThreadPriorityError> { + let port = resolve_thread_port(thread_info.pid, thread_info.thread_id)?; + + let rv = set_time_constraint_policy(port, thread_info.previous_time_constraint_policy); - Ok(rt_priority_handle) + unsafe { mach_port_deallocate(mach_task_self(), port) }; + rv } diff --git a/src/rt_win.rs b/src/rt_win.rs index 32a10a2..3935937 100644 --- a/src/rt_win.rs +++ b/src/rt_win.rs @@ -4,22 +4,40 @@ use log::info; use std::sync::OnceLock; use windows_sys::{ w, - Win32::Foundation::{HANDLE, WIN32_ERROR}, + Win32::Foundation::{CloseHandle, FILETIME, HANDLE, WIN32_ERROR}, + Win32::System::Threading::{ + GetCurrentProcessId, GetCurrentThread, GetCurrentThreadId, GetThreadPriority, + GetThreadTimes, OpenThread, SetThreadPriority, THREAD_PRIORITY_TIME_CRITICAL, + THREAD_QUERY_INFORMATION, THREAD_SET_INFORMATION, + }, }; -#[derive(Debug)] -pub struct RtPriorityHandleInternal { - mmcss_task_index: u32, - task_handle: HANDLE, -} +// `GetThreadPriority` returns this sentinel value on failure. It's defined as +// `THREAD_PRIORITY_ERROR_RETURN` in `winbase.h`, but `windows-sys` only exposes it under +// `Win32::System::WindowsProgramming`, a module this crate otherwise has no use for. +const THREAD_PRIORITY_ERROR_RETURN: i32 = i32::MAX; -impl RtPriorityHandleInternal { - fn new(mmcss_task_index: u32, task_handle: HANDLE) -> RtPriorityHandleInternal { - RtPriorityHandleInternal { - mmcss_task_index, - task_handle, - } - } +/// Two different mechanisms are used, depending on whether the thread being promoted is the +/// calling thread or not: +/// - The calling thread is registered with MMCSS (`avrt.dll`), which is the mechanism +/// recommended by Microsoft for pro-audio applications, but that can only ever be used by a +/// thread on itself: none of the `avrt.dll` APIs take a thread handle or id. +/// - A thread other than the caller (possibly in another process) is bumped using the ordinary +/// Win32 thread priority APIs instead, which do accept a `HANDLE` to an arbitrary thread. +#[derive(Debug)] +pub enum RtPriorityHandleInternal { + Mmcss { + mmcss_task_index: u32, + task_handle: HANDLE, + }, + ThreadPriority { + tid: u32, + previous_priority: i32, + // See `RtPriorityThreadInfoInternal::creation_time`: carried here too so that even a + // misuse of this handle for demotion (see `demote_current_thread_from_real_time_internal`) + // re-validates thread identity rather than trusting a possibly-recycled `tid`. + creation_time: u64, + }, } fn avrt() -> Result<&'static AvRtLibrary, AudioThreadPriorityError> { @@ -40,7 +58,10 @@ pub fn promote_current_thread_to_real_time_internal( .set_mm_thread_characteristics(w!("Audio")) .map(|(mmcss_task_index, task_handle)| { info!("task {mmcss_task_index} bumped to real time priority."); - RtPriorityHandleInternal::new(mmcss_task_index, task_handle) + RtPriorityHandleInternal::Mmcss { + mmcss_task_index, + task_handle, + } }) .map_err(|win32_error| { AudioThreadPriorityError::new(&format!( @@ -52,20 +73,234 @@ pub fn promote_current_thread_to_real_time_internal( pub fn demote_current_thread_from_real_time_internal( rt_priority_handle: RtPriorityHandleInternal, ) -> Result<(), AudioThreadPriorityError> { - let RtPriorityHandleInternal { - mmcss_task_index, - task_handle, - } = rt_priority_handle; - avrt()? - .revert_mm_thread_characteristics(task_handle) - .map(|_| { - info!("task {mmcss_task_index} priority restored."); - }) - .map_err(|win32_error| { - AudioThreadPriorityError::new(&format!( - "Unable to restore the thread priority for task {mmcss_task_index} ({win32_error})" - )) - }) + match rt_priority_handle { + RtPriorityHandleInternal::Mmcss { + mmcss_task_index, + task_handle, + } => avrt()? + .revert_mm_thread_characteristics(task_handle) + .map(|_| { + info!("task {mmcss_task_index} priority restored."); + }) + .map_err(|win32_error| { + AudioThreadPriorityError::new(&format!( + "Unable to restore the thread priority for task {mmcss_task_index} ({win32_error})" + )) + }), + RtPriorityHandleInternal::ThreadPriority { + tid, + previous_priority, + creation_time, + } => set_thread_priority(tid, creation_time, previous_priority), + } +} + +/// Opaque, serializable information about a thread, possibly running in another process, +/// sufficient to promote or demote it to/from real-time priority. +/// +/// `creation_time` (from `GetThreadTimes`) is captured alongside `tid` because Windows thread ids +/// are reused once a thread exits: without it, a `tid` captured here could, by the time +/// `promote_thread_to_real_time`/`demote_thread_from_real_time` runs, refer to a completely +/// different, unrelated thread that happened to be assigned the same id in the meantime. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct RtPriorityThreadInfoInternal { + pid: u32, + tid: u32, + previous_priority: i32, + creation_time: u64, +} + +impl RtPriorityThreadInfoInternal { + /// Serialize to a byte buffer. The fields are packed explicitly rather than transmuting the + /// struct, matching the approach used on the other platforms, so the format doesn't silently + /// start reading uninitialized padding if a field is ever reordered or resized. + pub fn serialize(&self) -> [u8; std::mem::size_of::()] { + let pid = self.pid.to_ne_bytes(); + let tid = self.tid.to_ne_bytes(); + let previous_priority = self.previous_priority.to_ne_bytes(); + let creation_time = self.creation_time.to_ne_bytes(); + + let mut bytes = [0u8; std::mem::size_of::()]; + let fields = pid + .iter() + .chain(&tid) + .chain(&previous_priority) + .chain(&creation_time); + for (dst, &src) in bytes.iter_mut().zip(fields) { + *dst = src; + } + bytes + } + /// Reconstruct from a byte buffer produced by `serialize`. + pub fn deserialize(bytes: [u8; std::mem::size_of::()]) -> Self { + fn take(src: &mut impl Iterator) -> [u8; N] { + let mut chunk = [0u8; N]; + for slot in &mut chunk { + *slot = src.next().unwrap(); + } + chunk + } + let mut src = bytes.iter().copied(); + RtPriorityThreadInfoInternal { + pid: u32::from_ne_bytes(take(&mut src)), + tid: u32::from_ne_bytes(take(&mut src)), + previous_priority: i32::from_ne_bytes(take(&mut src)), + creation_time: u64::from_ne_bytes(take(&mut src)), + } + } + /// Returns the PID of the process containing the thread. + pub fn pid(&self) -> i32 { + self.pid as i32 + } +} + +impl PartialEq for RtPriorityThreadInfoInternal { + // Compares identity only (which thread, in which process, created when), not the captured + // `previous_priority`, matching `rt_mach.rs`/`rt_linux.rs`. `creation_time` is included because + // it's exactly what distinguishes the original thread from an unrelated one that later reused + // the same `tid`. + fn eq(&self, other: &Self) -> bool { + self.pid == other.pid && self.tid == other.tid && self.creation_time == other.creation_time + } +} + +fn open_thread(access: u32, tid: u32) -> Result { + let handle = unsafe { OpenThread(access, 0, tid) }; + if handle.is_null() { + return Err(AudioThreadPriorityError::new(&format!( + "OpenThread failed for thread {tid}" + ))); + } + Ok(handle) +} + +/// Read a thread's creation time (as an opaque, monotonically-increasing 64-bit value) via +/// `GetThreadTimes`. Two live threads never share a creation time, and a given `tid` gets a new +/// creation time once it's reused by a different thread, so comparing this is how thread identity +/// is re-validated across the gap between capturing `RtPriorityThreadInfoInternal` and acting on +/// it later. +fn thread_creation_time(handle: HANDLE) -> Result { + let mut creation = FILETIME::default(); + let mut exit = FILETIME::default(); + let mut kernel = FILETIME::default(); + let mut user = FILETIME::default(); + let ok = unsafe { GetThreadTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) }; + if ok == 0 { + return Err(AudioThreadPriorityError::new("GetThreadTimes failed")); + } + Ok(((creation.dwHighDateTime as u64) << 32) | creation.dwLowDateTime as u64) +} + +fn get_thread_priority(tid: u32) -> Result { + let handle = open_thread(THREAD_QUERY_INFORMATION, tid)?; + let priority = unsafe { GetThreadPriority(handle) }; + unsafe { CloseHandle(handle) }; + if priority == THREAD_PRIORITY_ERROR_RETURN { + return Err(AudioThreadPriorityError::new(&format!( + "GetThreadPriority failed for thread {tid}" + ))); + } + Ok(priority) +} + +/// Open `tid` for `THREAD_SET_INFORMATION`, but only after confirming it's still the same thread +/// `expected_creation_time` was captured for -- see `thread_creation_time`. This is what protects +/// `set_thread_priority` from silently acting on an unrelated thread that reused a stale `tid`. +fn open_thread_verified( + tid: u32, + expected_creation_time: u64, +) -> Result { + let handle = open_thread(THREAD_SET_INFORMATION | THREAD_QUERY_INFORMATION, tid)?; + match thread_creation_time(handle) { + Ok(creation_time) if creation_time == expected_creation_time => Ok(handle), + Ok(_) => { + unsafe { CloseHandle(handle) }; + Err(AudioThreadPriorityError::new(&format!( + "thread {tid} has exited and its id was reused by another thread" + ))) + } + Err(e) => { + unsafe { CloseHandle(handle) }; + Err(e) + } + } +} + +fn set_thread_priority( + tid: u32, + expected_creation_time: u64, + priority: i32, +) -> Result<(), AudioThreadPriorityError> { + let handle = open_thread_verified(tid, expected_creation_time)?; + let rv = unsafe { SetThreadPriority(handle, priority) }; + unsafe { CloseHandle(handle) }; + if rv == 0 { + return Err(AudioThreadPriorityError::new(&format!( + "SetThreadPriority failed for thread {tid}" + ))); + } + Ok(()) +} + +/// Get the current thread information, as an opaque struct, that can be serialized and sent +/// accross processes, to have another thread promoted to real-time. +pub fn get_current_thread_info_internal( +) -> Result { + let tid = unsafe { GetCurrentThreadId() }; + let previous_priority = get_thread_priority(tid)?; + // `GetCurrentThread()` is a pseudo-handle valid without opening the thread; no `CloseHandle` + // needed. + let creation_time = thread_creation_time(unsafe { GetCurrentThread() })?; + + Ok(RtPriorityThreadInfoInternal { + pid: unsafe { GetCurrentProcessId() }, + tid, + previous_priority, + creation_time, + }) +} + +/// Promote a thread (possibly in another process) identified by its thread info, to real-time. +/// +/// Unlike `promote_current_thread_to_real_time`, this can't use MMCSS: none of the `avrt.dll` +/// APIs accept a thread id or handle, they only ever act on the calling thread. This instead +/// raises the target thread's priority to `THREAD_PRIORITY_TIME_CRITICAL` via the ordinary Win32 +/// thread priority APIs, using a handle obtained with `OpenThread`, which works for a thread in +/// another process too, given sufficient access rights. +pub fn promote_thread_to_real_time_internal( + thread_info: RtPriorityThreadInfoInternal, + _audio_buffer_frames: u32, + _audio_samplerate_hz: u32, +) -> Result { + set_thread_priority( + thread_info.tid, + thread_info.creation_time, + THREAD_PRIORITY_TIME_CRITICAL, + )?; + + info!( + "thread {} (pid {}) bumped to real time priority.", + thread_info.tid, thread_info.pid + ); + + Ok(RtPriorityHandleInternal::ThreadPriority { + tid: thread_info.tid, + previous_priority: thread_info.previous_priority, + creation_time: thread_info.creation_time, + }) +} + +/// This can be called by sandboxed code, it restores the priority the thread had when +/// `get_current_thread_info` captured `thread_info`. +pub fn demote_thread_from_real_time_internal( + thread_info: RtPriorityThreadInfoInternal, +) -> Result<(), AudioThreadPriorityError> { + set_thread_priority( + thread_info.tid, + thread_info.creation_time, + thread_info.previous_priority, + ) } mod avrt_lib {