From d0e839016c63fda5351f42a53119cfa25f48c2a5 Mon Sep 17 00:00:00 2001 From: Henrik Date: Wed, 8 Jul 2026 22:42:06 +0200 Subject: [PATCH 1/4] Add native SCHED_FIFO promotion for the Linux build without dbus The Linux build without the `dbus` feature previously fell through to the blanket fallback, where promotion was a no-op that returned Ok, so callers believed they had real-time priority when they did not. Give that build a real implementation: promote the current thread directly with pthread_setschedparam(SCHED_FIFO | SCHED_RESET_ON_FORK) at priority 10, restoring the saved policy on demotion. This needs no D-Bus daemon and works whenever the process may request real-time scheduling (root, CAP_SYS_NICE, or an RLIMIT_RTPRIO budget). The rtkit path and other platforms are untouched. - New rt_linux_native.rs with the native backend, wired into a dedicated cfg branch after the rtkit branch. - The requested priority defaults to 10 and can be overridden with the AUDIO_RT_PRIORITY environment variable (1-99). - Tests exercise the RLIMIT_RTPRIO boundary and the priority override; the fallback CI job raises RLIMIT_RTPRIO so promotion runs for real. - Docs: add a Platforms section and correct API docs that assumed D-Bus was the only Linux backend. --- .github/workflows/rust.yml | 9 +- audio_thread_priority.h | 63 +++++---- src/lib.rs | 282 ++++++++++++++++++++++++++++++++----- src/rt_linux_native.rs | 228 ++++++++++++++++++++++++++++++ 4 files changed, 516 insertions(+), 66 deletions(-) create mode 100644 src/rt_linux_native.rs diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index d09ac54..a5b9ebe 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -81,4 +81,11 @@ jobs: - name: Test fallback (Linux without dbus) shell: bash - run: rustup run ${{ matrix.rust }} cargo test --no-default-features + # The native (no-dbus) path really promotes threads, so the tests need permission to request + # real-time scheduling. Raise RLIMIT_RTPRIO for this shell, as an admin would via systemd + # LimitRTPRIO or /etc/security/limits.conf; the unprivileged test process and its children + # inherit it. This mirrors a real deployment rather than running the tests as root. + run: | + sudo prlimit --pid $$ --rtprio=10:10 + echo "RLIMIT_RTPRIO soft=$(ulimit -Sr) hard=$(ulimit -Hr)" + rustup run ${{ matrix.rust }} cargo test --no-default-features diff --git a/audio_thread_priority.h b/audio_thread_priority.h index b883728..805cad8 100644 --- a/audio_thread_priority.h +++ b/audio_thread_priority.h @@ -28,6 +28,9 @@ extern "C" { * or an upper bound. * audio_samplerate_hz: sample-rate for this audio stream, in Hz * + * On Linux built without D-Bus, the requested priority defaults to 10 and can be + * overridden with the AUDIO_RT_PRIORITY environment variable (an integer 1-99). + * * Returns an opaque handle in case of success, NULL otherwise. */ atp_handle *atp_promote_current_thread_to_real_time(uint32_t audio_buffer_frames, @@ -35,17 +38,17 @@ atp_handle *atp_promote_current_thread_to_real_time(uint32_t audio_buffer_frames /** - * Demotes the current thread promoted to real-time priority via - * `atp_demote_current_thread_from_real_time` to its previous priority. + * Demotes the current thread, promoted to real-time priority via + * `atp_promote_current_thread_to_real_time`, back to its previous priority. * * Returns 0 in case of success, non-zero otherwise. */ int32_t atp_demote_current_thread_from_real_time(atp_handle *handle); /** - * Frees an atp_handle. This is useful when it impractical to call - *`atp_demote_current_thread_from_real_time` on the right thread. Access to the - * handle must be synchronized externaly (or the related thread must have + * Frees an atp_handle. This is useful when it is impractical to call + * `atp_demote_current_thread_from_real_time` on the right thread. Access to the + * handle must be synchronized externally (or the related thread must have * exited). * * Returns 0 in case of success, non-zero otherwise. @@ -55,9 +58,11 @@ int32_t atp_free_handle(atp_handle *handle); /* * Linux-only API. * - * The Linux backend uses DBUS to promote a thread to real-time priority. In - * environment where this is not possible (due to sandboxing), this set of - * functions allow remoting the call to a process that can make DBUS calls. + * 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. * * To do so: * - Set the real-time limit from within the process where a @@ -65,9 +70,9 @@ int32_t atp_free_handle(atp_handle *handle); * before the sandbox lockdown. * - Then, gather information on the thread that will be promoted. * - Serialize this info. - * - Send over the serialized data via an IPC mechanism - * - Deserialize the inf - * - Call `atp_promote_thread_to_real_time` + * - Send over the serialized data via an IPC mechanism. + * - Deserialize the info. + * - Call `atp_promote_thread_to_real_time`. */ #ifdef __linux__ @@ -83,31 +88,34 @@ int32_t atp_free_handle(atp_handle *handle); * * Returns an opaque handle in case of success, NULL otherwise. * - * This call is useful on Linux desktop only, when the process is sandboxed and - * cannot promote itself directly. + * 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_handle *atp_promote_thread_to_real_time(atp_thread_info *thread_info); /** - * Demotes a thread promoted to real-time priority via - * `atp_demote_thread_from_real_time` to its previous priority. + * 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 call is useful on Linux desktop only, when the process is sandboxed and - * cannot promote itself directly. + * 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). */ int32_t atp_demote_thread_from_real_time(atp_thread_info* thread_info); /** - * Gather informations from the calling thread, to be able to promote it from + * Gather information from the calling thread, to be able to promote it from * another thread and/or process. * * Returns a non-null pointer to an `atp_thread_info` structure in case of - * sucess, to be freed later with `atp_free_thread_info`, and NULL otherwise. + * success, to be freed later with `atp_free_thread_info`, and NULL otherwise. * - * This call is useful on Linux desktop only, when the process is sandboxed and - * cannot promote itself directly. + * 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(); @@ -132,14 +140,15 @@ void atp_serialize_thread_info(atp_thread_info *thread_info, uint8_t *bytes); atp_thread_info* atp_deserialize_thread_info(uint8_t *bytes); /** - * Set real-time limit for the calling process. + * Set the real-time computation limit (RLIMIT_RTTIME) for the calling process. * - * This is useful only on Linux desktop, and allows remoting the rtkit DBUS call - * to a process that has access to DBUS. This function has to be called before - * attempting to promote threads from another process. + * This is needed by the rtkit/D-Bus backend before a thread can be promoted + * from another process, and must be called from within that process (it can be + * done before a sandbox lockdown). Without the `dbus` feature no such limit is + * required and this is a no-op. * - * This sets the real-time computation limit. For actually promoting the thread - * to a real-time scheduling class, see `atp_promote_thread_to_real_time`. + * This only sets the limit. For actually promoting the thread to a real-time + * scheduling class, see `atp_promote_thread_to_real_time`. */ int32_t atp_set_real_time_limit(uint32_t audio_buffer_frames, uint32_t audio_samplerate_hz); diff --git a/src/lib.rs b/src/lib.rs index 613f5c1..0f12a93 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,6 +3,20 @@ //! Promote the current thread, or another thread (possibly in another process), to real-time //! priority, suitable for low-latency audio processing. //! +//! # Platforms +//! +//! - **macOS**: the Mach time-constraint scheduling policy. +//! - **Windows**: the Multimedia Class Scheduler Service (MMCSS), "Pro Audio" task. +//! - **Linux** (default, `dbus` feature enabled): rtkit, over D-Bus. This suits unprivileged, +//! sandboxed desktop processes, since rtkit performs the privileged scheduling change on their +//! behalf. +//! - **Linux** (`dbus` feature disabled): direct promotion with `pthread_setschedparam` and the +//! `SCHED_FIFO` policy. This needs no D-Bus daemon, and works whenever the process may request +//! real-time scheduling: running as root, holding `CAP_SYS_NICE`, or with an `RLIMIT_RTPRIO` +//! limit configured (e.g. systemd `LimitRTPRIO` or `/etc/security/limits.conf`). The priority +//! (default 10) can be overridden with the `AUDIO_RT_PRIORITY` environment variable (1-99). +//! - **Other platforms**: a no-op that reports success. +//! //! # Example //! //! ```rust @@ -112,13 +126,28 @@ cfg_if! { #[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; + extern crate libc; + use rt_linux_native::promote_current_thread_to_real_time_internal; + use rt_linux_native::demote_current_thread_from_real_time_internal; + use rt_linux_native::set_real_time_hard_limit_internal as set_real_time_hard_limit; + use rt_linux_native::get_current_thread_info_internal; + use rt_linux_native::promote_thread_to_real_time_internal; + use rt_linux_native::demote_thread_from_real_time_internal; + use rt_linux_native::RtPriorityThreadInfoInternal; + use rt_linux_native::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 = "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::RtPriorityHandleInternal; } else { - // blanket implementations for Android, Linux Desktop without dbus and others + // blanket no-op implementations for platforms without a real-time backend /// Fallback priority handle that performs no-op operations on unsupported platforms. pub struct RtPriorityHandleInternal {} #[derive(Clone, Copy, PartialEq)] @@ -127,11 +156,7 @@ cfg_if! { _dummy: u8 } - cfg_if! { - if #[cfg(not(target_os = "linux"))] { - pub type RtPriorityThreadInfo = RtPriorityThreadInfoInternal; - } - } + pub type RtPriorityThreadInfo = RtPriorityThreadInfoInternal; impl RtPriorityThreadInfo { /// Serialize the thread info to a byte array (fallback implementation). @@ -198,20 +223,21 @@ pub type RtPriorityHandle = RtPriorityHandleInternal; cfg_if! { if #[cfg(target_os = "linux")] { -/// Opaque handle to a thread info. +/// Opaque handle to a thread's scheduling information. /// -/// This can be serialized to raw bytes to be sent via IPC. +/// 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 call is useful on Linux desktop only, when the process is sandboxed and -/// cannot promote itself directly. +/// 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). pub type RtPriorityThreadInfo = RtPriorityThreadInfoInternal; /// Get the calling thread's information, to be able to promote it to real-time from somewhere /// else, later. /// -/// This call is useful on Linux desktop only, when the process is sandboxed and -/// cannot promote itself directly. +/// 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). /// /// # Return value /// @@ -224,8 +250,8 @@ pub fn get_current_thread_info() -> Result [u8; std::mem::size_of::()] { @@ -234,23 +260,23 @@ pub fn thread_info_serialize( /// From a byte buffer, return a `RtPriorityThreadInfo`. /// -/// This call is useful on Linux desktop only, when the process is sandboxed and -/// cannot promote itself directly. +/// 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 serializezd `RtPriorityThreadInfo`. +/// A byte buffer containing a serialized `RtPriorityThreadInfo`. pub fn thread_info_deserialize( bytes: [u8; std::mem::size_of::()], ) -> RtPriorityThreadInfo { RtPriorityThreadInfoInternal::deserialize(bytes) } -/// Get the calling threads' information, to promote it from another process or thread, with a C +/// Get the calling thread's information, to promote it from another process or thread, with a C /// API. /// -/// This is intended to call on the thread that will end up being promoted to real time priority, -/// but that cannot do it itself (probably because of sandboxing reasons). +/// This is intended to be called on the thread that will be promoted to real-time priority, when +/// that thread cannot do so itself (for example because it is sandboxed). /// /// After use, it MUST be freed by calling `atp_free_thread_info`. /// @@ -295,8 +321,9 @@ 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 call is useful on Linux desktop only, when the process is sandboxed, cannot promote itself -/// directly, and the `atp_thread_info` struct must be passed via IPC. +/// 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 /// @@ -314,12 +341,12 @@ pub unsafe extern "C" fn atp_serialize_thread_info( /// From a byte buffer, return a `RtPriorityThreadInfo`, with a C API. /// -/// This call is useful on Linux desktop only, when the process is sandboxed and -/// cannot promote itself directly. +/// 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 serializezd `RtPriorityThreadInfo`. +/// A byte buffer containing a serialized `RtPriorityThreadInfo`. /// /// # Safety /// @@ -333,14 +360,14 @@ pub unsafe extern "C" fn atp_deserialize_thread_info( Box::into_raw(Box::new(atp_thread_info(thread_info))) } -/// Promote a particular thread thread to real-time priority. +/// Promote a particular thread to real-time priority. /// -/// This call is useful on Linux desktop only, when the process is sandboxed and -/// cannot promote itself directly. +/// 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 /// -/// * `thread_info` - informations about the thread to promote, gathered using +/// * `thread_info` - information 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. @@ -466,7 +493,7 @@ pub extern "C" fn atp_set_real_time_limit(audio_buffer_frames: u32, } } -/// Promote the calling thread thread to real-time priority. +/// Promote the calling thread to real-time priority. /// /// # Arguments /// @@ -508,7 +535,7 @@ pub fn demote_current_thread_from_real_time( #[allow(non_camel_case_types)] pub struct atp_handle(RtPriorityHandle); -/// Promote the calling thread thread to real-time priority, with a C API. +/// Promote the calling thread to real-time priority, with a C API. /// /// # Arguments /// @@ -522,10 +549,12 @@ pub struct atp_handle(RtPriorityHandle); /// `audio_samplerate_hz` is zero. It returns an opaque handle, to be passed to /// `atp_demote_current_thread_from_real_time` to demote the thread. /// -/// Additionaly, NULL can be returned in sandboxed processes on Linux, when DBUS cannot be used in -/// the process (for example because the socket to DBUS cannot be created). If this is the case, -/// it's necessary to get the information from the thread to promote and ask another process to -/// promote it (maybe via another privileged process). +/// Additionally, on Linux this returns NULL when the current thread cannot be promoted directly. +/// With the default rtkit/D-Bus backend that happens in sandboxed processes where D-Bus is +/// unreachable (for example because the socket to D-Bus cannot be created); without the `dbus` +/// feature it happens when the process lacks permission to set real-time scheduling. In that case, +/// gather the thread's information with `atp_get_current_thread_info` and have another (privileged) +/// process promote it via `atp_promote_thread_to_real_time`. #[no_mangle] pub extern "C" fn atp_promote_current_thread_to_real_time( audio_buffer_frames: u32, @@ -564,8 +593,8 @@ pub unsafe extern "C" fn atp_demote_current_thread_from_real_time(handle: *mut a /// Frees a handle, with a C API. /// -/// This is useful when it impractical to call `atp_demote_current_thread_from_real_time` on the -/// right thread. Access to the handle must be synchronized externaly, or the thread that was +/// This is useful when it is impractical to call `atp_demote_current_thread_from_real_time` on the +/// right thread. Access to the handle must be synchronized externally, or the thread that was /// promoted to real-time priority must have exited. /// /// # Arguments @@ -594,10 +623,31 @@ mod tests { use super::*; #[cfg(feature = "terminal-logging")] use simple_logger; + + // On the native (no-dbus) Linux build, promotion actually changes the scheduler, so it needs + // permission to request real-time scheduling. When the environment does not grant it (no + // RLIMIT_RTPRIO budget and not privileged), the promotion tests have nothing to exercise and + // skip rather than fail; CI raises the limit so they run for real. + #[cfg(all(target_os = "linux", not(feature = "dbus")))] + fn rt_scheduling_available() -> bool { + match promote_current_thread_to_real_time(0, 44100) { + Ok(handle) => { + let _ = demote_current_thread_from_real_time(handle); + true + } + Err(_) => false, + } + } + #[test] fn it_works() { #[cfg(feature = "terminal-logging")] simple_logger::init().unwrap(); + #[cfg(all(target_os = "linux", not(feature = "dbus")))] + if !rt_scheduling_available() { + eprintln!("skipping it_works: real-time scheduling is not permitted here"); + return; + } { assert!(promote_current_thread_to_real_time(0, 0).is_err()); } @@ -670,9 +720,16 @@ mod tests { if #[cfg(target_os = "linux")] { use nix::unistd::*; use nix::sys::signal::*; + #[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) { @@ -712,7 +769,16 @@ mod tests { } Err(e) => { kill(child, SIGKILL).expect("Could not kill the child?"); - panic!("{}", e); + // Promoting a thread in another process can need privilege + // beyond RLIMIT_RTPRIO (CAP_SYS_NICE) that an unprivileged + // CI does not grant. With the native (no-dbus) backend, + // treat that as a skip rather than a failure. + if cfg!(feature = "dbus") { + panic!("{}", e); + } else { + eprintln!("skipping test_remote_promotion: promoting a thread in another process needs elevated privilege ({e})"); + return; + } } } } @@ -744,6 +810,146 @@ mod tests { } } } + + // Native (no-dbus) path only. These tests change the process-wide RLIMIT_RTPRIO and, for + // the override test, an environment variable, so they run in a forked child to avoid + // racing with the other (parallel) promotion tests. + cfg_if! { + if #[cfg(not(feature = "dbus"))] { + const SCHED_RESET_ON_FORK: libc::c_int = 0x4000_0000; + // Exit codes the forked child reports back to the parent. + const PASSED: i32 = 0; + const FAILED: i32 = 1; + const SKIPPED: i32 = 2; + + fn rtprio_limit() -> libc::rlimit { + let mut lim = unsafe { std::mem::zeroed::() }; + assert_eq!(unsafe { libc::getrlimit(libc::RLIMIT_RTPRIO, &mut lim) }, 0); + lim + } + + fn set_rtprio_soft(soft: libc::rlim_t) { + let mut lim = rtprio_limit(); + lim.rlim_cur = soft; + assert_eq!(unsafe { libc::setrlimit(libc::RLIMIT_RTPRIO, &lim) }, 0); + } + + fn current_scheduler() -> (libc::c_int, libc::c_int) { + let mut policy = 0; + let mut param = unsafe { std::mem::zeroed::() }; + assert_eq!( + unsafe { + libc::pthread_getschedparam( + libc::pthread_self(), + &mut policy, + &mut param, + ) + }, + 0 + ); + (policy, param.sched_priority) + } + + // Run `checks` in a forked child so its RLIMIT_RTPRIO and environment changes + // do not affect the other tests, and turn the child's exit code into a pass, a + // skip, or a panic. + fn run_in_child(name: &str, checks: impl FnOnce() -> i32) { + match unsafe { fork().expect("fork failed") } { + ForkResult::Parent { child } => { + match waitpid(child, None).expect("waitpid") { + WaitStatus::Exited(_, PASSED) => {} + WaitStatus::Exited(_, SKIPPED) => { + eprintln!("skipping {}: needs an unprivileged process with a real-time budget", name); + } + other => panic!("{} child reported a failure: {:?}", name, other), + } + } + ForkResult::Child => std::process::exit(checks()), + } + } + + // Promotion honours RLIMIT_RTPRIO: with the soft limit below the requested + // priority it must be denied, and at the priority it must succeed and actually + // move the thread to SCHED_FIFO. Skipped as root (which bypasses RLIMIT_RTPRIO) + // or when no real-time budget was granted (a plain developer machine; CI raises + // it with `prlimit`). + #[test] + fn test_native_promotion_honours_rlimit() { + const RT_PRIO: libc::c_int = 10; + run_in_child("test_native_promotion_honours_rlimit", || { + if unsafe { libc::geteuid() } == 0 { + return SKIPPED; + } + if rtprio_limit().rlim_max < RT_PRIO as libc::rlim_t { + return SKIPPED; + } + + // Below the requested priority: promotion must be denied. + set_rtprio_soft(RT_PRIO as libc::rlim_t - 1); + if promote_current_thread_to_real_time(0, 44100).is_ok() { + eprintln!("promotion succeeded below the RLIMIT_RTPRIO ceiling"); + return FAILED; + } + + // At the requested priority: promotion must succeed and take effect. + set_rtprio_soft(RT_PRIO as libc::rlim_t); + let handle = match promote_current_thread_to_real_time(0, 44100) { + Ok(handle) => handle, + Err(e) => { + eprintln!("promotion denied at the RLIMIT_RTPRIO ceiling: {e}"); + return FAILED; + } + }; + let (policy, prio) = current_scheduler(); + if policy & !SCHED_RESET_ON_FORK != libc::SCHED_FIFO || prio != RT_PRIO { + eprintln!("unexpected scheduler after promotion: policy={policy} prio={prio}"); + return FAILED; + } + if demote_current_thread_from_real_time(handle).is_err() { + eprintln!("demotion failed"); + return FAILED; + } + PASSED + }); + } + + // The requested priority can be overridden with AUDIO_RT_PRIORITY. Uses a value + // that differs from the default (10) and fits a modest RLIMIT_RTPRIO, and checks + // the thread lands on exactly that priority. + #[test] + fn test_native_priority_override() { + const OVERRIDE: libc::c_int = 7; + run_in_child("test_native_priority_override", || { + if unsafe { libc::geteuid() } == 0 { + return SKIPPED; + } + let hard = rtprio_limit().rlim_max; + if hard < OVERRIDE as libc::rlim_t { + return SKIPPED; + } + set_rtprio_soft(hard); + std::env::set_var("AUDIO_RT_PRIORITY", OVERRIDE.to_string()); + + let handle = match promote_current_thread_to_real_time(0, 44100) { + Ok(handle) => handle, + Err(e) => { + eprintln!("promotion denied with AUDIO_RT_PRIORITY={OVERRIDE}: {e}"); + return FAILED; + } + }; + let (_, prio) = current_scheduler(); + let result = if prio == OVERRIDE { + PASSED + } else { + eprintln!("expected priority {OVERRIDE}, got {prio}"); + FAILED + }; + let _ = demote_current_thread_from_real_time(handle); + result + }); + } + } + } } } } diff --git a/src/rt_linux_native.rs b/src/rt_linux_native.rs new file mode 100644 index 0000000..a89f68a --- /dev/null +++ b/src/rt_linux_native.rs @@ -0,0 +1,228 @@ +/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this file, + * You can obtain one at http://mozilla.org/MPL/2.0/. */ + +//! Native Linux real-time promotion, used when the crate is built without the `dbus` feature. +//! +//! Instead of asking rtkit over D-Bus, this promotes the thread directly with +//! `pthread_setschedparam(SCHED_FIFO)`. It needs no D-Bus and no rtkit daemon, and works whenever +//! the process is allowed to request real-time scheduling: running as root, holding `CAP_SYS_NICE`, +//! or with an `RLIMIT_RTPRIO` limit configured (e.g. systemd `LimitRTPRIO` or +//! `/etc/security/limits.conf`). This is the mechanism JACK and PipeWire's direct mode use. + +extern crate libc; + +use std::io::Error as OSError; + +use crate::AudioThreadPriorityError; + +/// Default real-time priority to request, when `AUDIO_RT_PRIORITY` is unset. Matches the value the +/// rtkit path already asks for. +const RT_PRIO_DEFAULT: libc::c_int = 10; + +/// Environment variable to override the requested real-time priority. Accepts an integer 1-99. +/// Higher values preempt more work but must stay below the audio interface's IRQ threads, or they +/// starve the very threads that deliver the audio. +const RT_PRIORITY_ENV: &str = "AUDIO_RT_PRIORITY"; + +/// The real-time priority to request, from `AUDIO_RT_PRIORITY` or the default. +fn requested_priority() -> libc::c_int { + match std::env::var(RT_PRIORITY_ENV) { + Ok(value) => match value.trim().parse::() { + Ok(priority) if (1..=99).contains(&priority) => priority, + _ => { + log::warn!( + "Ignoring invalid {RT_PRIORITY_ENV}=\"{value}\", expected an integer 1-99. \ + Using default {RT_PRIO_DEFAULT}." + ); + RT_PRIO_DEFAULT + } + }, + Err(_) => RT_PRIO_DEFAULT, + } +} + +/// Prevents threads/processes forked from a real-time thread from inheriting real-time scheduling. +/// Not exposed by libc: +const SCHED_RESET_ON_FORK: libc::c_int = 0x4000_0000; + +// This is different from libc::pid_t, which is 32 bits, and is defined in sys/types.h. +#[allow(non_camel_case_types)] +type kernel_pid_t = libc::c_long; + +#[repr(C)] +#[derive(Clone, Copy)] +pub struct RtPriorityThreadInfoInternal { + /// System-wide thread id (tid), used to promote a thread by id. + thread_id: kernel_pid_t, + /// Process-local thread id, used to restore scheduler characteristics. + pthread_id: libc::pthread_t, + /// The PID of the process containing `thread_id`. + pid: libc::pid_t, + /// The scheduling policy in place before promotion, to restore on demotion. + policy: libc::c_int, + /// The scheduling parameters in place before promotion, to restore on demotion. + param: libc::sched_param, +} + +impl RtPriorityThreadInfoInternal { + /// Serialize a RtPriorityThreadInfoInternal to a byte buffer. + pub fn serialize(&self) -> [u8; std::mem::size_of::()] { + unsafe { std::mem::transmute::()]>(*self) } + } + /// Get an RtPriorityThreadInfoInternal from a byte buffer. + pub fn deserialize(bytes: [u8; std::mem::size_of::()]) -> Self { + unsafe { std::mem::transmute::<[u8; std::mem::size_of::()], Self>(bytes) } + } + /// 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.thread_id == other.thread_id && self.pthread_id == other.pthread_id + } +} + +pub struct RtPriorityHandleInternal { + thread_info: RtPriorityThreadInfoInternal, +} + +/// The POSIX `pthread_*` functions return the error number directly and do not set `errno`, so the +/// return code must be converted with `Error::from_raw_os_error`, not read via `last_os_error`. +fn pthread_error(context: &str, rc: libc::c_int) -> AudioThreadPriorityError { + AudioThreadPriorityError::new(&format!("{}: {}", context, OSError::from_raw_os_error(rc))) +} + +/// The `sched_*` functions are thin syscall wrappers: they return -1 and set `errno`. +fn sched_error(context: &str) -> AudioThreadPriorityError { + AudioThreadPriorityError::new(&format!("{}: {}", context, OSError::last_os_error())) +} + +/// Get the current thread information, capturing enough to promote or demote it later. This mirrors +/// the rtkit path so the same public API works, but the returned struct is only meaningful within +/// this process (the native path promotes directly rather than via a privileged helper). +pub fn get_current_thread_info_internal( +) -> Result { + let thread_id = unsafe { libc::syscall(libc::SYS_gettid) }; + let pthread_id = unsafe { libc::pthread_self() }; + let pid = unsafe { libc::getpid() }; + let mut policy = 0; + let mut param = unsafe { std::mem::zeroed::() }; + + let rc = unsafe { libc::pthread_getschedparam(pthread_id, &mut policy, &mut param) }; + if rc != 0 { + return Err(pthread_error("pthread_getschedparam", rc)); + } + + Ok(RtPriorityThreadInfoInternal { + thread_id, + pthread_id, + pid, + policy, + param, + }) +} + +/// Promote the calling thread to real-time priority using `SCHED_FIFO`. +/// +/// The buffer size and sample rate are unused here (they matter only for the rtkit path, which +/// derives an `RLIMIT_RTTIME` budget from them); the signature is kept to match the other backends. +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 mut param = unsafe { std::mem::zeroed::() }; + param.sched_priority = requested_priority(); + + let rc = unsafe { + libc::pthread_setschedparam( + thread_info.pthread_id, + libc::SCHED_FIFO | SCHED_RESET_ON_FORK, + ¶m, + ) + }; + if rc != 0 { + return Err(pthread_error("could not promote thread", rc)); + } + + Ok(RtPriorityHandleInternal { thread_info }) +} + +/// Restore the calling thread to the scheduling policy and parameters it had before promotion. +pub fn demote_current_thread_from_real_time_internal( + rt_priority_handle: RtPriorityHandleInternal, +) -> Result<(), AudioThreadPriorityError> { + let RtPriorityThreadInfoInternal { + pthread_id, + policy, + param, + .. + } = rt_priority_handle.thread_info; + + // Keep SCHED_RESET_ON_FORK set: the kernel forbids an unprivileged thread from clearing that + // flag once set (and promotion set it), so restoring the bare saved policy would fail with + // EPERM. The flag is harmless on a non-real-time thread. + let rc = + unsafe { libc::pthread_setschedparam(pthread_id, policy | SCHED_RESET_ON_FORK, ¶m) }; + if rc != 0 { + return Err(pthread_error("could not demote thread", rc)); + } + Ok(()) +} + +/// Promote a thread identified by its tid to real-time priority. Promoting a thread other than the +/// caller (in particular in another process) requires the caller to be privileged. +pub fn promote_thread_to_real_time_internal( + thread_info: RtPriorityThreadInfoInternal, + _audio_buffer_frames: u32, + _audio_samplerate_hz: u32, +) -> Result { + let mut param = unsafe { std::mem::zeroed::() }; + param.sched_priority = requested_priority(); + + let rc = unsafe { + libc::sched_setscheduler( + thread_info.thread_id as libc::pid_t, + libc::SCHED_FIFO | SCHED_RESET_ON_FORK, + ¶m, + ) + }; + if rc < 0 { + return Err(sched_error("could not promote thread")); + } + + Ok(RtPriorityHandleInternal { thread_info }) +} + +/// Restore a thread identified by its tid to the policy and parameters it had before promotion. +pub fn demote_thread_from_real_time_internal( + thread_info: RtPriorityThreadInfoInternal, +) -> Result<(), AudioThreadPriorityError> { + // Keep SCHED_RESET_ON_FORK set (see demote_current_thread_from_real_time_internal): clearing it + // as an unprivileged thread would fail with EPERM. + let rc = unsafe { + libc::sched_setscheduler( + thread_info.thread_id as libc::pid_t, + thread_info.policy | SCHED_RESET_ON_FORK, + &thread_info.param, + ) + }; + if rc < 0 { + return Err(sched_error("could not demote thread")); + } + Ok(()) +} + +/// Setting an `RLIMIT_RTTIME` budget is only needed by the rtkit path. The native path relies on +/// the kernel's real-time throttling (`sched_rt_runtime_us`) instead, so this is a no-op. +pub fn set_real_time_hard_limit_internal( + _audio_buffer_frames: u32, + _audio_samplerate_hz: u32, +) -> Result<(), AudioThreadPriorityError> { + Ok(()) +} From e67b105de3ed6657a73331e55e7a4c535655932d Mon Sep 17 00:00:00 2001 From: Henrik Date: Wed, 8 Jul 2026 22:47:35 +0200 Subject: [PATCH 2/4] Fix clippy useless_borrows_in_formatting in Display impl --- src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/lib.rs b/src/lib.rs index 0f12a93..2b5898c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -80,7 +80,7 @@ impl AudioThreadPriorityError { impl fmt::Display for AudioThreadPriorityError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let mut rv = write!(f, "AudioThreadPriorityError: {}", &self.message); + let mut rv = write!(f, "AudioThreadPriorityError: {}", self.message); if let Some(inner) = &self.inner { rv = write!(f, " ({inner})"); } From d0e55f3823e31d96410993b324e03db3b2e8cb22 Mon Sep 17 00:00:00 2001 From: Henrik Date: Thu, 9 Jul 2026 16:17:34 +0200 Subject: [PATCH 3/4] Replace AUDIO_RT_PRIORITY env var with a set_rt_priority() setter A library reading an environment variable it was never told about is a hidden global input that can surprise the embedding application. Replace it with an explicit, optional setter that the caller drives: set_rt_priority(Some(n)) overrides the default of 10, None restores it. Linux no-dbus only; if never called, promotion uses priority 10 as before. --- audio_thread_priority.h | 3 --- src/lib.rs | 11 +++++----- src/rt_linux_native.rs | 45 ++++++++++++++++++++++++----------------- 3 files changed, 32 insertions(+), 27 deletions(-) diff --git a/audio_thread_priority.h b/audio_thread_priority.h index 805cad8..b3383ec 100644 --- a/audio_thread_priority.h +++ b/audio_thread_priority.h @@ -28,9 +28,6 @@ extern "C" { * or an upper bound. * audio_samplerate_hz: sample-rate for this audio stream, in Hz * - * On Linux built without D-Bus, the requested priority defaults to 10 and can be - * overridden with the AUDIO_RT_PRIORITY environment variable (an integer 1-99). - * * Returns an opaque handle in case of success, NULL otherwise. */ atp_handle *atp_promote_current_thread_to_real_time(uint32_t audio_buffer_frames, diff --git a/src/lib.rs b/src/lib.rs index 2b5898c..101670d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -13,8 +13,8 @@ //! - **Linux** (`dbus` feature disabled): direct promotion with `pthread_setschedparam` and the //! `SCHED_FIFO` policy. This needs no D-Bus daemon, and works whenever the process may request //! real-time scheduling: running as root, holding `CAP_SYS_NICE`, or with an `RLIMIT_RTPRIO` -//! limit configured (e.g. systemd `LimitRTPRIO` or `/etc/security/limits.conf`). The priority -//! (default 10) can be overridden with the `AUDIO_RT_PRIORITY` environment variable (1-99). +//! limit configured (e.g. systemd `LimitRTPRIO` or `/etc/security/limits.conf`). The requested +//! priority defaults to 10 and can be changed with `set_rt_priority` (Linux, no-`dbus` only). //! - **Other platforms**: a no-op that reports success. //! //! # Example @@ -138,6 +138,7 @@ cfg_if! { use rt_linux_native::demote_thread_from_real_time_internal; 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::(); @@ -913,7 +914,7 @@ mod tests { }); } - // The requested priority can be overridden with AUDIO_RT_PRIORITY. Uses a value + // The requested priority can be overridden with set_rt_priority. Uses a value // that differs from the default (10) and fits a modest RLIMIT_RTPRIO, and checks // the thread lands on exactly that priority. #[test] @@ -928,12 +929,12 @@ mod tests { return SKIPPED; } set_rtprio_soft(hard); - std::env::set_var("AUDIO_RT_PRIORITY", OVERRIDE.to_string()); + set_rt_priority(Some(OVERRIDE as u8)); let handle = match promote_current_thread_to_real_time(0, 44100) { Ok(handle) => handle, Err(e) => { - eprintln!("promotion denied with AUDIO_RT_PRIORITY={OVERRIDE}: {e}"); + eprintln!("promotion denied at priority {OVERRIDE}: {e}"); return FAILED; } }; diff --git a/src/rt_linux_native.rs b/src/rt_linux_native.rs index a89f68a..3bdc467 100644 --- a/src/rt_linux_native.rs +++ b/src/rt_linux_native.rs @@ -13,32 +13,39 @@ extern crate libc; use std::io::Error as OSError; +use std::sync::atomic::{AtomicU8, Ordering}; use crate::AudioThreadPriorityError; -/// Default real-time priority to request, when `AUDIO_RT_PRIORITY` is unset. Matches the value the -/// rtkit path already asks for. +/// Default real-time priority to request, unless overridden with [`set_rt_priority`]. Matches the +/// value the rtkit path already asks for. const RT_PRIO_DEFAULT: libc::c_int = 10; -/// Environment variable to override the requested real-time priority. Accepts an integer 1-99. -/// Higher values preempt more work but must stay below the audio interface's IRQ threads, or they -/// starve the very threads that deliver the audio. -const RT_PRIORITY_ENV: &str = "AUDIO_RT_PRIORITY"; +/// The real-time priority to request, or 0 to use `RT_PRIO_DEFAULT`. Set via [`set_rt_priority`]. +static RT_PRIORITY: AtomicU8 = AtomicU8::new(0); -/// The real-time priority to request, from `AUDIO_RT_PRIORITY` or the default. +/// Set the real-time priority (1-99) to request when promoting a thread, overriding the default of +/// 10. Pass `None` to restore the default. Values outside 1-99 are ignored with a warning. +/// +/// This is entirely optional: if never called, promotion uses priority 10, the value the rtkit path +/// requests. It is specific to the Linux build without the `dbus` feature; set it before promoting. +pub fn set_rt_priority(priority: Option) { + match priority { + Some(priority) if (1..=99).contains(&priority) => { + RT_PRIORITY.store(priority, Ordering::Relaxed) + } + Some(priority) => { + log::warn!("Ignoring invalid real-time priority {priority}, expected an integer 1-99") + } + None => RT_PRIORITY.store(0, Ordering::Relaxed), + } +} + +/// The real-time priority to request: the value set via [`set_rt_priority`], or the default. fn requested_priority() -> libc::c_int { - match std::env::var(RT_PRIORITY_ENV) { - Ok(value) => match value.trim().parse::() { - Ok(priority) if (1..=99).contains(&priority) => priority, - _ => { - log::warn!( - "Ignoring invalid {RT_PRIORITY_ENV}=\"{value}\", expected an integer 1-99. \ - Using default {RT_PRIO_DEFAULT}." - ); - RT_PRIO_DEFAULT - } - }, - Err(_) => RT_PRIO_DEFAULT, + match RT_PRIORITY.load(Ordering::Relaxed) { + 0 => RT_PRIO_DEFAULT, + priority => priority as libc::c_int, } } From 9305107df405e48b5a6aa1c7cd49d43a07c8d715 Mon Sep 17 00:00:00 2001 From: Henrik Date: Thu, 9 Jul 2026 18:40:49 +0200 Subject: [PATCH 4/4] Address review comments - Drop the `param` field from RtPriorityThreadInfoInternal so the struct has no padding, matching the rtkit path's struct. `serialize`'s transmute no longer reads uninitialized padding bytes (which was UB). Demotion now restores the saved policy with a zeroed param, as the rtkit path already does. - rt_scheduling_available: treat a demotion failure after a successful promotion as fatal rather than ignoring it. - Fix a stale test comment that referred to the removed environment variable. --- src/lib.rs | 9 ++++++--- src/rt_linux_native.rs | 18 ++++++++---------- 2 files changed, 14 insertions(+), 13 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 101670d..1531665 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -633,7 +633,10 @@ mod tests { fn rt_scheduling_available() -> bool { match promote_current_thread_to_real_time(0, 44100) { Ok(handle) => { - let _ = demote_current_thread_from_real_time(handle); + // Demotion must succeed after a successful promotion; otherwise the thread would + // stay real-time and perturb later tests, so treat a failure as fatal here. + demote_current_thread_from_real_time(handle) + .expect("demotion after a successful promotion should succeed"); true } Err(_) => false, @@ -813,8 +816,8 @@ mod tests { } // Native (no-dbus) path only. These tests change the process-wide RLIMIT_RTPRIO and, for - // the override test, an environment variable, so they run in a forked child to avoid - // racing with the other (parallel) promotion tests. + // the override test, the priority set via `set_rt_priority`, so they run in a forked + // child to avoid racing with the other (parallel) promotion tests. cfg_if! { if #[cfg(not(feature = "dbus"))] { const SCHED_RESET_ON_FORK: libc::c_int = 0x4000_0000; diff --git a/src/rt_linux_native.rs b/src/rt_linux_native.rs index 3bdc467..00b3b68 100644 --- a/src/rt_linux_native.rs +++ b/src/rt_linux_native.rs @@ -57,6 +57,8 @@ const SCHED_RESET_ON_FORK: libc::c_int = 0x4000_0000; #[allow(non_camel_case_types)] type kernel_pid_t = libc::c_long; +// The fields are laid out to leave no padding (like the rtkit path's equivalent struct), so +// `serialize` can transmute the whole struct to bytes without reading uninitialized padding. #[repr(C)] #[derive(Clone, Copy)] pub struct RtPriorityThreadInfoInternal { @@ -68,8 +70,6 @@ pub struct RtPriorityThreadInfoInternal { pid: libc::pid_t, /// The scheduling policy in place before promotion, to restore on demotion. policy: libc::c_int, - /// The scheduling parameters in place before promotion, to restore on demotion. - param: libc::sched_param, } impl RtPriorityThreadInfoInternal { @@ -129,7 +129,6 @@ pub fn get_current_thread_info_internal( pthread_id, pid, policy, - param, }) } @@ -160,20 +159,18 @@ pub fn promote_current_thread_to_real_time_internal( Ok(RtPriorityHandleInternal { thread_info }) } -/// Restore the calling thread to the scheduling policy and parameters it had before promotion. +/// Restore the calling thread to the scheduling policy it had before promotion. pub fn demote_current_thread_from_real_time_internal( rt_priority_handle: RtPriorityHandleInternal, ) -> Result<(), AudioThreadPriorityError> { let RtPriorityThreadInfoInternal { - pthread_id, - policy, - param, - .. + pthread_id, policy, .. } = rt_priority_handle.thread_info; // Keep SCHED_RESET_ON_FORK set: the kernel forbids an unprivileged thread from clearing that // flag once set (and promotion set it), so restoring the bare saved policy would fail with // EPERM. The flag is harmless on a non-real-time thread. + let param = unsafe { std::mem::zeroed::() }; let rc = unsafe { libc::pthread_setschedparam(pthread_id, policy | SCHED_RESET_ON_FORK, ¶m) }; if rc != 0 { @@ -206,17 +203,18 @@ pub fn promote_thread_to_real_time_internal( Ok(RtPriorityHandleInternal { thread_info }) } -/// Restore a thread identified by its tid to the policy and parameters it had before promotion. +/// Restore a thread identified by its tid to the scheduling policy it had before promotion. pub fn demote_thread_from_real_time_internal( thread_info: RtPriorityThreadInfoInternal, ) -> Result<(), AudioThreadPriorityError> { // Keep SCHED_RESET_ON_FORK set (see demote_current_thread_from_real_time_internal): clearing it // as an unprivileged thread would fail with EPERM. + let param = unsafe { std::mem::zeroed::() }; let rc = unsafe { libc::sched_setscheduler( thread_info.thread_id as libc::pid_t, thread_info.policy | SCHED_RESET_ON_FORK, - &thread_info.param, + ¶m, ) }; if rc < 0 {