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::atomic::{AtomicBool, Ordering};
10use std::sync::{Arc, Condvar, Mutex};
11use std::thread::JoinHandle;
12use std::time::{Duration, Instant};
13use tokio::sync::mpsc::{Receiver, Sender};
14use tracing::error;
15
16pub trait Backend: Send + Sync + 'static {
17 type Driver: HwWorkerDriver + Send + 'static;
18 type MidiHub: HwMidiHub + Send + 'static;
19
20 const LABEL: &'static str;
21 const WORKER_THREAD_NAME: &'static str;
22 const ASSIST_THREAD_NAME: &'static str;
23 const ASSIST_AUTONOMOUS_ENV: &'static str;
24 const ASSIST_STEP_REQUIRES_REQUEST_CYCLE: bool = false;
25}
26
27#[derive(Debug)]
28pub struct HwWorker<B: Backend> {
29 driver: Arc<UnsafeMutex<B::Driver>>,
30 midi_hub: Arc<UnsafeMutex<B::MidiHub>>,
31 rx: Receiver<Message>,
32 tx: Sender<Message>,
33 cycle_frames: u32,
34 pending_midi_out_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 pending_midi_out_sorted: true,
236 }
237 }
238
239 pub async fn work(mut self) {
240 if let Err(e) = Self::lock_memory_pages() {
241 error!("{} worker memory lock not enabled: {}", B::LABEL, e);
242 }
243 if let Err(e) = Self::configure_rt_thread(B::WORKER_THREAD_NAME, RT_PRIORITY_WORKER) {
244 error!("{} worker realtime priority not enabled: {}", B::LABEL, e);
245 }
246 #[cfg(target_os = "macos")]
247 unsafe {
248 libc::pthread_set_qos_class_self_np(libc::qos_class_t::QOS_CLASS_USER_INTERACTIVE, 0);
249 }
250 let assist_state = Arc::new((Mutex::new(AssistState::default()), Condvar::new()));
251 let assist_handle = Self::start_assist_thread(self.driver.clone(), assist_state.clone());
252 let midi_stop = Arc::new(AtomicBool::new(false));
253 let midi_handle = Self::start_midi_input_thread(
254 self.midi_hub.clone(),
255 self.tx.clone(),
256 self.cycle_frames,
257 midi_stop.clone(),
258 );
259 loop {
260 match self.rx.recv().await {
261 Some(msg) => match msg {
262 Message::Request(crate::message::Action::Quit) => {
263 if !self.pending_midi_out_events.is_empty() {
264 if !self.pending_midi_out_sorted {
265 self.pending_midi_out_events.sort_by(|a, b| {
266 a.event
267 .frame
268 .cmp(&b.event.frame)
269 .then_with(|| a.device.cmp(&b.device))
270 });
271 self.pending_midi_out_sorted = true;
272 }
273 let midi_hub = self.midi_hub.lock();
274 midi_hub.write_events(&self.pending_midi_out_events);
275 self.pending_midi_out_events.clear();
276 }
277 midi_stop.store(true, Ordering::Release);
278 {
279 let midi_hub = self.midi_hub.lock();
280 midi_hub.wake_input_waiter();
281 }
282 let _ = midi_handle.join();
283 Self::stop_assist_thread(&assist_state, assist_handle);
284 return;
285 }
286 Message::TracksFinished => {
287 {
288 if !self.pending_midi_out_events.is_empty() {
289 if !self.pending_midi_out_sorted {
290 self.pending_midi_out_events.sort_by(|a, b| {
291 a.event
292 .frame
293 .cmp(&b.event.frame)
294 .then_with(|| a.device.cmp(&b.device))
295 });
296 self.pending_midi_out_sorted = true;
297 }
298 let midi_hub = self.midi_hub.lock();
299 midi_hub.write_events(&self.pending_midi_out_events);
300 self.pending_midi_out_events.clear();
301 }
302 }
303 if let Err(e) = Self::run_assist_cycle(&assist_state) {
304 error!("{} assist cycle error: {}", B::LABEL, e);
305 }
306 if let Err(e) = self.tx.send(Message::HWFinished).await {
307 error!(
308 "{} worker failed to send HWFinished to engine: {}",
309 B::LABEL,
310 e
311 );
312 }
313 }
314 Message::HWMidiOutEvents(mut events) => {
315 self.pending_midi_out_events.append(&mut events);
316 self.pending_midi_out_sorted = false;
317 }
318 Message::ClearHWMidiOutEvents => {
319 self.pending_midi_out_events.clear();
320 self.pending_midi_out_sorted = true;
321 }
322 _ => {}
323 },
324 None => {
325 midi_stop.store(true, Ordering::Release);
326 {
327 let midi_hub = self.midi_hub.lock();
328 midi_hub.wake_input_waiter();
329 }
330 let _ = midi_handle.join();
331 Self::stop_assist_thread(&assist_state, assist_handle);
332 return;
333 }
334 }
335 }
336 }
337
338 fn start_midi_input_thread(
339 midi_hub: Arc<UnsafeMutex<B::MidiHub>>,
340 tx: Sender<Message>,
341 cycle_frames: u32,
342 stop: Arc<AtomicBool>,
343 ) -> JoinHandle<()> {
344 std::thread::spawn(move || {
345 let mut midi_in_events = Vec::with_capacity(64);
346 while !stop.load(Ordering::Acquire) {
347 {
348 let hub = midi_hub.lock();
349 hub.read_events_blocking_into(&mut midi_in_events);
350 }
351 if midi_in_events.is_empty() {
352 continue;
353 }
354 spread_hw_event_frames(&mut midi_in_events, cycle_frames);
355 let cap = midi_in_events.capacity();
356 let out = std::mem::replace(&mut midi_in_events, Vec::with_capacity(cap.max(64)));
357 if tx.blocking_send(Message::HWMidiEvents(out)).is_err() {
358 break;
359 }
360 }
361 })
362 }
363
364 fn start_assist_thread(
365 driver: Arc<UnsafeMutex<B::Driver>>,
366 assist_state: Arc<(Mutex<AssistState>, Condvar)>,
367 ) -> JoinHandle<()> {
368 let profile = Self::profile_enabled();
369 let autonomous = Self::assist_autonomous_enabled();
370 std::thread::spawn(move || {
371 if let Err(e) = Self::configure_rt_thread(B::ASSIST_THREAD_NAME, RT_PRIORITY_ASSIST) {
372 error!("{} assist realtime priority not enabled: {}", B::LABEL, e);
373 }
374 #[cfg(target_os = "macos")]
375 unsafe {
376 libc::pthread_set_qos_class_self_np(libc::qos_class_t::QOS_CLASS_USER_INITIATED, 0);
377 }
378 let mut profiler = if profile {
379 let (cycle_samples, sample_rate) = {
380 let d = driver.lock();
381 (d.cycle_samples(), d.sample_rate())
382 };
383 error!(
384 "{} profile enabled: cycle_samples={} sample_rate={} expected_cps={:.1}",
385 B::LABEL,
386 cycle_samples,
387 sample_rate,
388 if cycle_samples > 0 {
389 sample_rate as f64 / cycle_samples as f64
390 } else {
391 0.0
392 }
393 );
394 Some(AssistProfiler::new())
395 } else {
396 None
397 };
398 let (lock, cvar) = &*assist_state;
399 loop {
400 let (shutdown, has_request, target, init_complete) = {
401 let st = lock.lock().expect("assist mutex poisoned");
402 (
403 st.shutdown,
404 st.request_seq > st.done_seq,
405 st.request_seq,
406 st.init_complete,
407 )
408 };
409 if shutdown {
410 break;
411 }
412 if has_request {
413 let started = Instant::now();
414 let run_error = {
415 let d = driver.lock();
416 d.run_cycle_for_worker().err().map(|e| e.to_string())
417 };
418 if let Some(p) = profiler.as_mut() {
419 p.cycle_count += 1;
420 if run_error.is_some() {
421 p.cycle_err_count += 1;
422 }
423 p.cycle_time_ns += started.elapsed().as_nanos();
424 let (cycle_samples, sample_rate) = {
425 let d = driver.lock();
426 (d.cycle_samples(), d.sample_rate())
427 };
428 p.maybe_report(cycle_samples, sample_rate, B::LABEL);
429 }
430 let mut st = lock.lock().expect("assist mutex poisoned");
431 st.done_seq = st.done_seq.max(target);
432 if run_error.is_none() {
433 st.init_complete = true;
434 }
435 st.last_error = run_error;
436 cvar.notify_all();
437 continue;
438 }
439
440 if B::ASSIST_STEP_REQUIRES_REQUEST_CYCLE && !init_complete {
441 let st = lock.lock().expect("assist mutex poisoned");
442 if st.shutdown {
443 break;
444 }
445 let wait_started = Instant::now();
446 let _guard = cvar.wait(st).expect("assist condvar failed");
447 if let Some(p) = profiler.as_mut() {
448 p.wait_count += 1;
449 p.wait_time_ns += wait_started.elapsed().as_nanos();
450 }
451 continue;
452 }
453
454 if !autonomous {
455 let st = lock.lock().expect("assist mutex poisoned");
456 if st.shutdown {
457 break;
458 }
459 let wait_started = Instant::now();
460 let _guard = cvar.wait(st).expect("assist condvar failed");
461 if let Some(p) = profiler.as_mut() {
462 p.wait_count += 1;
463 p.wait_time_ns += wait_started.elapsed().as_nanos();
464 }
465 continue;
466 }
467
468 let started = Instant::now();
469 let did_work = {
470 let d = driver.lock();
471 match d.run_assist_step_for_worker() {
472 Ok(v) => v,
473 Err(e) => {
474 if let Some(p) = profiler.as_mut() {
475 p.step_err_count += 1;
476 }
477 let mut st = lock.lock().expect("assist mutex poisoned");
478 st.last_error = Some(e.to_string());
479 cvar.notify_all();
480 false
481 }
482 }
483 };
484 if let Some(p) = profiler.as_mut() {
485 p.step_count += 1;
486 if did_work {
487 p.step_work_count += 1;
488 }
489 p.step_time_ns += started.elapsed().as_nanos();
490 let (cycle_samples, sample_rate) = {
491 let d = driver.lock();
492 (d.cycle_samples(), d.sample_rate())
493 };
494 p.maybe_report(cycle_samples, sample_rate, B::LABEL);
495 }
496 if !did_work {
497 let st = lock.lock().expect("assist mutex poisoned");
498 if st.shutdown {
499 break;
500 }
501 let wait_started = Instant::now();
502 let _guard = cvar.wait(st).expect("assist condvar failed");
503 if let Some(p) = profiler.as_mut() {
504 p.wait_count += 1;
505 p.wait_time_ns += wait_started.elapsed().as_nanos();
506 }
507 }
508 }
509 })
510 }
511
512 fn run_assist_cycle(assist_state: &Arc<(Mutex<AssistState>, Condvar)>) -> Result<(), String> {
513 let (lock, cvar) = &**assist_state;
514 let mut st = lock
515 .lock()
516 .map_err(|_| "assist mutex poisoned".to_string())?;
517 st.request_seq = st.request_seq.saturating_add(1);
518 let target = st.request_seq;
519 cvar.notify_one();
520 while st.done_seq < target && !st.shutdown {
521 st = cvar
522 .wait(st)
523 .map_err(|_| "assist condvar wait failed".to_string())?;
524 }
525 if let Some(err) = st.last_error.take() {
526 return Err(err);
527 }
528 Ok(())
529 }
530
531 fn stop_assist_thread(
532 assist_state: &Arc<(Mutex<AssistState>, Condvar)>,
533 assist_handle: JoinHandle<()>,
534 ) {
535 let (lock, cvar) = &**assist_state;
536 if let Ok(mut st) = lock.lock() {
537 st.shutdown = true;
538 cvar.notify_all();
539 }
540 let _ = assist_handle.join();
541 }
542}
543
544fn spread_hw_event_frames(events: &mut [HwMidiEvent], frames: u32) {
545 if events.len() <= 1 || frames <= 1 {
546 return;
547 }
548 let n = events.len() as u32;
549 for (idx, event) in events.iter_mut().enumerate() {
550 let pos = idx as u32;
551 event.event.frame = ((pos as u64 * (frames - 1) as u64) / n as u64) as u32;
552 }
553}