1use crate::{
2 hw::config,
3 hw::traits::{HwMidiHub, HwWorkerDriver},
4 message::{HwMidiEvent, Message},
5 mutex::UnsafeMutex,
6};
7#[cfg(unix)]
8use nix::libc;
9use std::sync::{Arc, Condvar, Mutex};
10use std::thread::JoinHandle;
11use std::time::{Duration, Instant};
12use tokio::sync::mpsc::{Receiver, Sender};
13use tracing::error;
14
15pub trait Backend: Send + Sync + 'static {
16 type Driver: HwWorkerDriver + Send + 'static;
17 type MidiHub: HwMidiHub + Send + 'static;
18
19 const LABEL: &'static str;
20 const WORKER_THREAD_NAME: &'static str;
21 const ASSIST_THREAD_NAME: &'static str;
22 const ASSIST_AUTONOMOUS_ENV: &'static str;
23 const ASSIST_STEP_REQUIRES_REQUEST_CYCLE: bool = false;
24}
25
26#[derive(Debug)]
27pub struct HwWorker<B: Backend> {
28 driver: Arc<UnsafeMutex<B::Driver>>,
29 midi_hub: Arc<UnsafeMutex<B::MidiHub>>,
30 rx: Receiver<Message>,
31 tx: Sender<Message>,
32 cycle_frames: u32,
33 pending_midi_out_events: Vec<HwMidiEvent>,
34 midi_in_events: Vec<HwMidiEvent>,
35 pending_midi_out_sorted: bool,
36}
37
38#[derive(Debug, Default)]
39struct AssistState {
40 shutdown: bool,
41 request_seq: u64,
42 done_seq: u64,
43 init_complete: bool,
44 last_error: Option<String>,
45}
46
47#[cfg(unix)]
48const RT_POLICY: i32 = libc::SCHED_FIFO;
49const RT_PRIORITY_WORKER: i32 = 18;
50const RT_PRIORITY_ASSIST: i32 = 12;
51const PROFILE_INTERVAL: Duration = Duration::from_secs(1);
52
53#[derive(Debug)]
54struct AssistProfiler {
55 report_at: Instant,
56 cycle_count: u64,
57 cycle_err_count: u64,
58 cycle_time_ns: u128,
59 step_count: u64,
60 step_work_count: u64,
61 step_err_count: u64,
62 step_time_ns: u128,
63 wait_count: u64,
64 wait_time_ns: u128,
65}
66
67impl AssistProfiler {
68 fn new() -> Self {
69 Self {
70 report_at: Instant::now() + PROFILE_INTERVAL,
71 cycle_count: 0,
72 cycle_err_count: 0,
73 cycle_time_ns: 0,
74 step_count: 0,
75 step_work_count: 0,
76 step_err_count: 0,
77 step_time_ns: 0,
78 wait_count: 0,
79 wait_time_ns: 0,
80 }
81 }
82
83 fn maybe_report(&mut self, cycle_samples: usize, sample_rate: i32, label: &str) {
84 let now = Instant::now();
85 if now < self.report_at {
86 return;
87 }
88 let cycle_avg_us = if self.cycle_count > 0 {
89 (self.cycle_time_ns / self.cycle_count as u128) as f64 / 1_000.0
90 } else {
91 0.0
92 };
93 let step_avg_us = if self.step_count > 0 {
94 (self.step_time_ns / self.step_count as u128) as f64 / 1_000.0
95 } else {
96 0.0
97 };
98 let wait_avg_us = if self.wait_count > 0 {
99 (self.wait_time_ns / self.wait_count as u128) as f64 / 1_000.0
100 } else {
101 0.0
102 };
103 let expected_cycles_per_sec = if cycle_samples > 0 && sample_rate > 0 {
104 sample_rate as f64 / cycle_samples as f64
105 } else {
106 0.0
107 };
108 error!(
109 "{} profile: expected_cps={:.1} cycles={} cycle_err={} cycle_avg_us={:.1} steps={} steps_work={} step_err={} step_avg_us={:.1} waits={} wait_avg_us={:.1}",
110 label,
111 expected_cycles_per_sec,
112 self.cycle_count,
113 self.cycle_err_count,
114 cycle_avg_us,
115 self.step_count,
116 self.step_work_count,
117 self.step_err_count,
118 step_avg_us,
119 self.wait_count,
120 wait_avg_us
121 );
122 self.report_at = now + PROFILE_INTERVAL;
123 self.cycle_count = 0;
124 self.cycle_err_count = 0;
125 self.cycle_time_ns = 0;
126 self.step_count = 0;
127 self.step_work_count = 0;
128 self.step_err_count = 0;
129 self.step_time_ns = 0;
130 self.wait_count = 0;
131 self.wait_time_ns = 0;
132 }
133}
134
135impl<B: Backend> HwWorker<B> {
136 fn profile_enabled() -> bool {
137 config::env_flag(config::HW_PROFILE_ENV)
138 }
139
140 fn assist_autonomous_enabled() -> bool {
141 config::env_flag(B::ASSIST_AUTONOMOUS_ENV)
142 }
143
144 fn configure_rt_thread(name: &str, priority: i32) -> Result<(), String> {
145 #[cfg(unix)]
146 {
147 let thread = unsafe { libc::pthread_self() };
148 #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "openbsd"))]
149 let c_name = std::ffi::CString::new(name).map_err(|e| e.to_string())?;
150 #[cfg(target_os = "linux")]
151 unsafe {
152 let _ = libc::pthread_setname_np(thread, c_name.as_ptr());
153 }
154 #[cfg(any(target_os = "freebsd", target_os = "openbsd"))]
155 unsafe {
156 libc::pthread_set_name_np(thread, c_name.as_ptr());
157 }
158
159 let param = unsafe {
160 let mut p = std::mem::zeroed::<libc::sched_param>();
161 p.sched_priority = priority;
162 p
163 };
164 let rc = unsafe { libc::pthread_setschedparam(thread, RT_POLICY, ¶m) };
165 if rc != 0 {
166 return Err(format!(
167 "pthread_setschedparam({}, prio {}) failed with errno {}",
168 name, priority, rc
169 ));
170 }
171
172 let mut actual_policy = 0_i32;
173 let mut actual_param = unsafe { std::mem::zeroed::<libc::sched_param>() };
174 let rc = unsafe {
175 libc::pthread_getschedparam(thread, &mut actual_policy, &mut actual_param)
176 };
177 if rc != 0 {
178 return Err(format!(
179 "pthread_getschedparam({}) failed with errno {}",
180 name, rc
181 ));
182 }
183 if actual_policy != RT_POLICY || actual_param.sched_priority != priority {
184 return Err(format!(
185 "realtime verification failed for {}: policy {}, prio {}",
186 name, actual_policy, actual_param.sched_priority
187 ));
188 }
189 Ok(())
190 }
191 #[cfg(not(unix))]
192 {
193 let _ = name;
194 let _ = priority;
195 Err("Realtime thread priority is not supported on this platform".to_string())
196 }
197 }
198
199 fn lock_memory_pages() -> Result<(), String> {
200 #[cfg(unix)]
201 {
202 let rc = unsafe { libc::mlockall(libc::MCL_CURRENT | libc::MCL_FUTURE) };
203 if rc == 0 {
204 Ok(())
205 } else {
206 Err(format!(
207 "mlockall(MCL_CURRENT|MCL_FUTURE) failed: {}",
208 std::io::Error::last_os_error()
209 ))
210 }
211 }
212 #[cfg(not(unix))]
213 {
214 Err("mlockall is not supported on this platform".to_string())
215 }
216 }
217
218 pub fn new(
219 driver: Arc<UnsafeMutex<B::Driver>>,
220 midi_hub: Arc<UnsafeMutex<B::MidiHub>>,
221 rx: Receiver<Message>,
222 tx: Sender<Message>,
223 ) -> Self {
224 let cycle_frames = {
225 let d = driver.lock();
226 d.cycle_samples() as u32
227 };
228 Self {
229 driver,
230 midi_hub,
231 rx,
232 tx,
233 cycle_frames,
234 pending_midi_out_events: vec![],
235 midi_in_events: Vec::with_capacity(64),
236 pending_midi_out_sorted: true,
237 }
238 }
239
240 pub async fn work(mut self) {
241 if let Err(e) = Self::lock_memory_pages() {
242 error!("{} worker memory lock not enabled: {}", B::LABEL, e);
243 }
244 if let Err(e) = Self::configure_rt_thread(B::WORKER_THREAD_NAME, RT_PRIORITY_WORKER) {
245 error!("{} worker realtime priority not enabled: {}", B::LABEL, e);
246 }
247 #[cfg(target_os = "macos")]
248 unsafe {
249 libc::pthread_set_qos_class_self_np(libc::qos_class_t::QOS_CLASS_USER_INTERACTIVE, 0);
250 }
251 let assist_state = Arc::new((Mutex::new(AssistState::default()), Condvar::new()));
252 let assist_handle = Self::start_assist_thread(self.driver.clone(), assist_state.clone());
253 loop {
254 match self.rx.recv().await {
255 Some(msg) => match msg {
256 Message::Request(crate::message::Action::Quit) => {
257 if !self.pending_midi_out_events.is_empty() {
258 if !self.pending_midi_out_sorted {
259 self.pending_midi_out_events.sort_by(|a, b| {
260 a.event
261 .frame
262 .cmp(&b.event.frame)
263 .then_with(|| a.device.cmp(&b.device))
264 });
265 self.pending_midi_out_sorted = true;
266 }
267 let midi_hub = self.midi_hub.lock();
268 midi_hub.write_events(&self.pending_midi_out_events);
269 self.pending_midi_out_events.clear();
270 }
271 Self::stop_assist_thread(&assist_state, assist_handle);
272 return;
273 }
274 Message::TracksFinished => {
275 {
276 let midi_hub = self.midi_hub.lock();
277 midi_hub.read_events_into(&mut self.midi_in_events);
278 }
279 spread_hw_event_frames(&mut self.midi_in_events, self.cycle_frames);
280 if !self.midi_in_events.is_empty() {
281 let cap = self.midi_in_events.capacity();
282 let out = std::mem::replace(
283 &mut self.midi_in_events,
284 Vec::with_capacity(cap.max(64)),
285 );
286 if let Err(e) = self.tx.send(Message::HWMidiEvents(out)).await {
287 error!(
288 "{} worker failed to send HWMidiEvents to engine: {}",
289 B::LABEL,
290 e
291 );
292 }
293 }
294 {
295 if !self.pending_midi_out_events.is_empty() {
296 if !self.pending_midi_out_sorted {
297 self.pending_midi_out_events.sort_by(|a, b| {
298 a.event
299 .frame
300 .cmp(&b.event.frame)
301 .then_with(|| a.device.cmp(&b.device))
302 });
303 self.pending_midi_out_sorted = true;
304 }
305 let midi_hub = self.midi_hub.lock();
306 midi_hub.write_events(&self.pending_midi_out_events);
307 self.pending_midi_out_events.clear();
308 }
309 }
310 if let Err(e) = Self::run_assist_cycle(&assist_state) {
311 error!("{} assist cycle error: {}", B::LABEL, e);
312 }
313 if let Err(e) = self.tx.send(Message::HWFinished).await {
314 error!(
315 "{} worker failed to send HWFinished to engine: {}",
316 B::LABEL,
317 e
318 );
319 }
320 }
321 Message::HWMidiOutEvents(mut events) => {
322 self.pending_midi_out_events.append(&mut events);
323 self.pending_midi_out_sorted = false;
324 }
325 Message::ClearHWMidiOutEvents => {
326 self.pending_midi_out_events.clear();
327 self.pending_midi_out_sorted = true;
328 }
329 _ => {}
330 },
331 None => {
332 Self::stop_assist_thread(&assist_state, assist_handle);
333 return;
334 }
335 }
336 }
337 }
338
339 fn start_assist_thread(
340 driver: Arc<UnsafeMutex<B::Driver>>,
341 assist_state: Arc<(Mutex<AssistState>, Condvar)>,
342 ) -> JoinHandle<()> {
343 let profile = Self::profile_enabled();
344 let autonomous = Self::assist_autonomous_enabled();
345 std::thread::spawn(move || {
346 if let Err(e) = Self::configure_rt_thread(B::ASSIST_THREAD_NAME, RT_PRIORITY_ASSIST) {
347 error!("{} assist realtime priority not enabled: {}", B::LABEL, e);
348 }
349 #[cfg(target_os = "macos")]
350 unsafe {
351 libc::pthread_set_qos_class_self_np(libc::qos_class_t::QOS_CLASS_USER_INITIATED, 0);
352 }
353 let mut profiler = if profile {
354 let (cycle_samples, sample_rate) = {
355 let d = driver.lock();
356 (d.cycle_samples(), d.sample_rate())
357 };
358 error!(
359 "{} profile enabled: cycle_samples={} sample_rate={} expected_cps={:.1}",
360 B::LABEL,
361 cycle_samples,
362 sample_rate,
363 if cycle_samples > 0 {
364 sample_rate as f64 / cycle_samples as f64
365 } else {
366 0.0
367 }
368 );
369 Some(AssistProfiler::new())
370 } else {
371 None
372 };
373 let (lock, cvar) = &*assist_state;
374 loop {
375 let (shutdown, has_request, target, init_complete) = {
376 let st = lock.lock().expect("assist mutex poisoned");
377 (
378 st.shutdown,
379 st.request_seq > st.done_seq,
380 st.request_seq,
381 st.init_complete,
382 )
383 };
384 if shutdown {
385 break;
386 }
387 if has_request {
388 let started = Instant::now();
389 let run_error = {
390 let d = driver.lock();
391 d.run_cycle_for_worker().err().map(|e| e.to_string())
392 };
393 if let Some(p) = profiler.as_mut() {
394 p.cycle_count += 1;
395 if run_error.is_some() {
396 p.cycle_err_count += 1;
397 }
398 p.cycle_time_ns += started.elapsed().as_nanos();
399 let (cycle_samples, sample_rate) = {
400 let d = driver.lock();
401 (d.cycle_samples(), d.sample_rate())
402 };
403 p.maybe_report(cycle_samples, sample_rate, B::LABEL);
404 }
405 let mut st = lock.lock().expect("assist mutex poisoned");
406 st.done_seq = st.done_seq.max(target);
407 if run_error.is_none() {
408 st.init_complete = true;
409 }
410 st.last_error = run_error;
411 cvar.notify_all();
412 continue;
413 }
414
415 if B::ASSIST_STEP_REQUIRES_REQUEST_CYCLE && !init_complete {
416 let st = lock.lock().expect("assist mutex poisoned");
417 if st.shutdown {
418 break;
419 }
420 let wait_started = Instant::now();
421 let _guard = cvar.wait(st).expect("assist condvar failed");
422 if let Some(p) = profiler.as_mut() {
423 p.wait_count += 1;
424 p.wait_time_ns += wait_started.elapsed().as_nanos();
425 }
426 continue;
427 }
428
429 if !autonomous {
430 let st = lock.lock().expect("assist mutex poisoned");
431 if st.shutdown {
432 break;
433 }
434 let wait_started = Instant::now();
435 let _guard = cvar.wait(st).expect("assist condvar failed");
436 if let Some(p) = profiler.as_mut() {
437 p.wait_count += 1;
438 p.wait_time_ns += wait_started.elapsed().as_nanos();
439 }
440 continue;
441 }
442
443 let started = Instant::now();
444 let did_work = {
445 let d = driver.lock();
446 match d.run_assist_step_for_worker() {
447 Ok(v) => v,
448 Err(e) => {
449 if let Some(p) = profiler.as_mut() {
450 p.step_err_count += 1;
451 }
452 let mut st = lock.lock().expect("assist mutex poisoned");
453 st.last_error = Some(e.to_string());
454 cvar.notify_all();
455 false
456 }
457 }
458 };
459 if let Some(p) = profiler.as_mut() {
460 p.step_count += 1;
461 if did_work {
462 p.step_work_count += 1;
463 }
464 p.step_time_ns += started.elapsed().as_nanos();
465 let (cycle_samples, sample_rate) = {
466 let d = driver.lock();
467 (d.cycle_samples(), d.sample_rate())
468 };
469 p.maybe_report(cycle_samples, sample_rate, B::LABEL);
470 }
471 if !did_work {
472 let st = lock.lock().expect("assist mutex poisoned");
473 if st.shutdown {
474 break;
475 }
476 let wait_started = Instant::now();
477 let _guard = cvar.wait(st).expect("assist condvar failed");
478 if let Some(p) = profiler.as_mut() {
479 p.wait_count += 1;
480 p.wait_time_ns += wait_started.elapsed().as_nanos();
481 }
482 }
483 }
484 })
485 }
486
487 fn run_assist_cycle(assist_state: &Arc<(Mutex<AssistState>, Condvar)>) -> Result<(), String> {
488 let (lock, cvar) = &**assist_state;
489 let mut st = lock
490 .lock()
491 .map_err(|_| "assist mutex poisoned".to_string())?;
492 st.request_seq = st.request_seq.saturating_add(1);
493 let target = st.request_seq;
494 cvar.notify_one();
495 while st.done_seq < target && !st.shutdown {
496 st = cvar
497 .wait(st)
498 .map_err(|_| "assist condvar wait failed".to_string())?;
499 }
500 if let Some(err) = st.last_error.take() {
501 return Err(err);
502 }
503 Ok(())
504 }
505
506 fn stop_assist_thread(
507 assist_state: &Arc<(Mutex<AssistState>, Condvar)>,
508 assist_handle: JoinHandle<()>,
509 ) {
510 let (lock, cvar) = &**assist_state;
511 if let Ok(mut st) = lock.lock() {
512 st.shutdown = true;
513 cvar.notify_all();
514 }
515 let _ = assist_handle.join();
516 }
517}
518
519fn spread_hw_event_frames(events: &mut [HwMidiEvent], frames: u32) {
520 if events.len() <= 1 || frames <= 1 {
521 return;
522 }
523 let n = events.len() as u32;
524 for (idx, event) in events.iter_mut().enumerate() {
525 let pos = idx as u32;
526 event.event.frame = ((pos as u64 * (frames - 1) as u64) / n as u64) as u32;
527 }
528}