autd3_rs_core/
rt.rs

1#[cfg(feature = "logging")]
2mod logging;
3
4mod executor;
5pub mod oneshot;
6mod semaphore;
7
8pub use executor::{Executor, block_on};
9pub use semaphore::{Acquire, Semaphore, SemaphorePermit};
10
11use thread_priority::{ThreadPriority, ThreadPriorityValue};
12
13#[derive(Clone, Copy, Debug, PartialEq, Eq)]
14pub struct RtPriority(ThreadPriority);
15
16impl RtPriority {
17    pub const MIN: Self = Self(ThreadPriority::Min);
18    pub const MAX: Self = Self(ThreadPriority::Max);
19
20    #[must_use]
21    pub fn new(value: u8) -> Option<Self> {
22        ThreadPriorityValue::try_from(value)
23            .ok()
24            .map(|v| Self(ThreadPriority::Crossplatform(v)))
25    }
26
27    #[must_use]
28    pub fn value(self) -> Option<u8> {
29        match self.0 {
30            ThreadPriority::Crossplatform(v) => Some(u8::from(v)),
31            _ => None,
32        }
33    }
34
35    #[must_use]
36    pub fn step_below(self) -> Option<Self> {
37        Self::new(self.value()?.checked_sub(1)?)
38    }
39}
40
41#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
42pub struct CoreId {
43    pub id: usize,
44}
45
46impl From<CoreId> for core_affinity::CoreId {
47    fn from(v: CoreId) -> Self {
48        core_affinity::CoreId { id: v.id }
49    }
50}
51
52impl From<core_affinity::CoreId> for CoreId {
53    fn from(v: core_affinity::CoreId) -> Self {
54        CoreId { id: v.id }
55    }
56}
57
58#[cfg(feature = "logging")]
59pub use logging::{LogWriter, TracingGuard, TracingOption, init_tracing};
60
61#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
62#[non_exhaustive]
63pub enum RtSchedulePolicy {
64    Normal,
65    #[default]
66    Fifo,
67    RoundRobin,
68}
69
70#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
71pub struct RtThreadTuning {
72    pub priority: Option<RtPriority>,
73    pub policy: RtSchedulePolicy,
74    pub affinity: Option<CoreId>,
75}
76
77#[cfg(not(target_os = "windows"))]
78const RT_THREAD_PRIORITY: u8 = 80;
79
80#[cfg(target_os = "linux")]
81const RT_PRIORITY_REMEDY: &str = "Grant the capability with \
82     `sudo setcap cap_sys_nice+ep <executable>`, or raise `rtprio` for the user in \
83     /etc/security/limits.conf.";
84
85#[cfg(not(target_os = "linux"))]
86const RT_PRIORITY_REMEDY: &str = "Run with sufficient scheduling privileges.";
87
88#[cfg(target_os = "linux")]
89fn set_rt_priority(
90    priority: ThreadPriority,
91    policy: RtSchedulePolicy,
92) -> Result<(), thread_priority::Error> {
93    use thread_priority::{
94        RealtimeThreadSchedulePolicy, ThreadSchedulePolicy, set_thread_priority_and_policy,
95        thread_native_id,
96    };
97    let policy = match policy {
98        RtSchedulePolicy::Normal => return thread_priority::set_current_thread_priority(priority),
99        RtSchedulePolicy::Fifo => {
100            ThreadSchedulePolicy::Realtime(RealtimeThreadSchedulePolicy::Fifo)
101        }
102        RtSchedulePolicy::RoundRobin => {
103            ThreadSchedulePolicy::Realtime(RealtimeThreadSchedulePolicy::RoundRobin)
104        }
105    };
106    set_thread_priority_and_policy(thread_native_id(), priority, policy)
107}
108
109#[cfg(not(target_os = "linux"))]
110fn set_rt_priority(
111    priority: ThreadPriority,
112    _policy: RtSchedulePolicy,
113) -> Result<(), thread_priority::Error> {
114    thread_priority::set_current_thread_priority(priority)
115}
116
117pub fn apply_thread_tuning(tuning: RtThreadTuning) -> RtThreadTuning {
118    let mut applied = RtThreadTuning {
119        priority: None,
120        policy: tuning.policy,
121        affinity: None,
122    };
123    if let Some(priority) = tuning.priority {
124        match set_rt_priority(priority.0, tuning.policy) {
125            Ok(()) => {
126                tracing::debug!(?priority, policy = ?tuning.policy, "applied RT thread scheduling");
127                applied.priority = Some(priority);
128            }
129            Err(e) => tracing::warn!(
130                "failed to set RT thread priority: {e:?}. The bus will be unstable under load. {}",
131                RT_PRIORITY_REMEDY
132            ),
133        }
134    }
135    if let Some(core) = tuning.affinity {
136        if core_affinity::set_for_current(core.into()) {
137            applied.affinity = Some(core);
138        } else {
139            tracing::warn!("failed to pin RT thread to core {}", core.id);
140        }
141    }
142    applied
143}
144
145#[must_use]
146#[allow(clippy::unnecessary_wraps)]
147pub fn default_rt_priority() -> Option<RtPriority> {
148    #[cfg(target_os = "windows")]
149    {
150        Some(RtPriority(ThreadPriority::Os(
151            thread_priority::WinAPIThreadPriority::TimeCritical.into(),
152        )))
153    }
154    #[cfg(not(target_os = "windows"))]
155    {
156        Some(RtPriority::new(RT_THREAD_PRIORITY).expect("0..=99 is a valid thread priority"))
157    }
158}
159
160#[cfg(test)]
161mod tests {
162    use super::*;
163
164    fn crossplatform(value: u8) -> RtPriority {
165        RtPriority::new(value).unwrap()
166    }
167
168    #[test]
169    fn a_step_below_is_one_lower() {
170        assert_eq!(crossplatform(80).step_below(), Some(crossplatform(79)));
171    }
172
173    #[test]
174    fn only_the_crossplatform_ladder_steps_down() {
175        assert_eq!(RtPriority::MAX.step_below(), None);
176        assert_eq!(RtPriority::MIN.step_below(), None);
177        assert_eq!(crossplatform(0).step_below(), None);
178    }
179
180    #[test]
181    fn only_the_crossplatform_ladder_has_a_value() {
182        assert_eq!(crossplatform(80).value(), Some(80));
183        assert_eq!(RtPriority::MAX.value(), None);
184        assert_eq!(RtPriority::MIN.value(), None);
185    }
186}
autd3_rs_core/rt.rs

autd3_rs_core/
rt.rs
