rickub
nandi/jolt-nativepublic⑂ Fork 0
⑂ cfd3e3677bed92e80cfe447469a249cfdc0b1401
Branches
main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/nandi/jolt-native.git
git clone ssh://git@rickub.com/nandi/jolt-native.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

stream.rs · 616 lines · 20.9 KBRust Blame HistoryRaw
Lift freeq's AV media plane out of sleek 90f8b89 nandi 19d ago1use super::windows_err_to_cpal_err;
2use crate::traits::StreamTrait;
3use crate::{
4 BackendSpecificError, BufferSize, Data, FrameCount, InputCallbackInfo, OutputCallbackInfo,
5 PauseStreamError, PlayStreamError, SampleFormat, SampleRate, StreamError,
6};
7use std::mem;
8use std::ptr;
9use std::sync::mpsc::{channel, Receiver, SendError, Sender};
10use std::thread::{self, JoinHandle};
11use std::time::Duration;
12use windows::Win32::Foundation;
13use windows::Win32::Foundation::WAIT_OBJECT_0;
14use windows::Win32::Media::Audio;
15use windows::Win32::System::SystemServices;
16use windows::Win32::System::Threading;
17
18pub struct Stream {
19 /// The high-priority audio processing thread calling callbacks.
20 /// Option used for moving out in destructor.
21 ///
22 /// TODO: Actually set the thread priority.
23 thread: Option<JoinHandle<()>>,
24
25 // Commands processed by the `run()` method that is currently running.
26 // `pending_scheduled_event` must be signalled whenever a command is added here, so that it
27 // will get picked up.
28 commands: Sender<Command>,
29
30 // This event is signalled after a new entry is added to `commands`, so that the `run()`
31 // method can be notified.
32 pending_scheduled_event: Foundation::HANDLE,
33
34 // Callback size in frames.
35 period_frames: FrameCount,
36}
37
38// SAFETY: Windows Event HANDLEs are safe to send between threads - they are designed for
39// synchronization. All fields of Stream are Send:
40// - JoinHandle<()> is Send
41// - Sender<Command> is Send
42// - Foundation::HANDLE is Send (Windows synchronization primitive)
43// See: https://learn.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-createeventa
44unsafe impl Send for Stream {}
45
46// SAFETY: Windows Event HANDLEs are safe to access from multiple threads simultaneously.
47// All synchronization operations (SetEvent, WaitForSingleObject) are thread-safe.
48// All fields of Stream are Sync:
49// - JoinHandle<()> is Sync
50// - Sender<Command> is Sync (uses internal synchronization)
51// - Foundation::HANDLE for event objects supports concurrent access
52// The audio thread owns all COM objects, so no cross-thread COM access occurs.
53unsafe impl Sync for Stream {}
54
55// Compile-time assertion that Stream is Send and Sync
56crate::assert_stream_send!(Stream);
57crate::assert_stream_sync!(Stream);
58
59struct RunContext {
60 // Streams that have been created in this event loop.
61 stream: StreamInner,
62
63 // Handles corresponding to the `event` field of each element of `voices`. Must always be in
64 // sync with `voices`, except that the first element is always `pending_scheduled_event`.
65 handles: Vec<Foundation::HANDLE>,
66
67 commands: Receiver<Command>,
68}
69
70// Once we start running the eventloop, the RunContext will not be moved.
71unsafe impl Send for RunContext {}
72
73pub enum Command {
74 PlayStream,
75 PauseStream,
76 Terminate,
77}
78
79pub enum AudioClientFlow {
80 Render {
81 render_client: Audio::IAudioRenderClient,
82 },
83 Capture {
84 capture_client: Audio::IAudioCaptureClient,
85 },
86}
87
88pub struct StreamInner {
89 pub audio_client: Audio::IAudioClient,
90 pub audio_clock: Audio::IAudioClock,
91 pub client_flow: AudioClientFlow,
92 // Event that is signalled by WASAPI whenever audio data must be written.
93 pub event: Foundation::HANDLE,
94 // True if the stream is currently playing. False if paused.
95 pub playing: bool,
96 // Number of frames of audio data in the underlying buffer allocated by WASAPI.
97 pub max_frames_in_buffer: FrameCount,
98 // Callback size in frames.
99 pub period_frames: FrameCount,
100 // Number of bytes that each frame occupies.
101 pub bytes_per_frame: u16,
102 // The configuration with which the stream was created.
103 pub config: crate::StreamConfig,
104 // The sample format with which the stream was created.
105 pub sample_format: SampleFormat,
106 // Hardware pipeline latency.
107 pub stream_latency: Duration,
108}
109
110impl Stream {
111 pub(crate) fn new_input<D, E>(
112 stream_inner: StreamInner,
113 mut data_callback: D,
114 mut error_callback: E,
115 ) -> Stream
116 where
117 D: FnMut(&Data, &InputCallbackInfo) + Send + 'static,
118 E: FnMut(StreamError) + Send + 'static,
119 {
120 let pending_scheduled_event = unsafe {
121 Threading::CreateEventA(None, false, false, windows::core::PCSTR(ptr::null()))
122 }
123 .expect("cpal: could not create input stream event");
124 let (tx, rx) = channel();
125
126 let period_frames = stream_inner.period_frames;
127
128 let run_context = RunContext {
129 handles: vec![pending_scheduled_event, stream_inner.event],
130 stream: stream_inner,
131 commands: rx,
132 };
133
134 let thread = thread::Builder::new()
135 .name("cpal_wasapi_in".to_owned())
136 .spawn(move || run_input(run_context, &mut data_callback, &mut error_callback))
137 .unwrap();
138
139 Stream {
140 thread: Some(thread),
141 commands: tx,
142 pending_scheduled_event,
143 period_frames,
144 }
145 }
146
147 pub(crate) fn new_output<D, E>(
148 stream_inner: StreamInner,
149 mut data_callback: D,
150 mut error_callback: E,
151 ) -> Stream
152 where
153 D: FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static,
154 E: FnMut(StreamError) + Send + 'static,
155 {
156 let pending_scheduled_event = unsafe {
157 Threading::CreateEventA(None, false, false, windows::core::PCSTR(ptr::null()))
158 }
159 .expect("cpal: could not create output stream event");
160 let (tx, rx) = channel();
161
162 let period_frames = stream_inner.period_frames;
163
164 let run_context = RunContext {
165 handles: vec![pending_scheduled_event, stream_inner.event],
166 stream: stream_inner,
167 commands: rx,
168 };
169
170 let thread = thread::Builder::new()
171 .name("cpal_wasapi_out".to_owned())
172 .spawn(move || run_output(run_context, &mut data_callback, &mut error_callback))
173 .unwrap();
174
175 Stream {
176 thread: Some(thread),
177 commands: tx,
178 pending_scheduled_event,
179 period_frames,
180 }
181 }
182
183 fn push_command(&self, command: Command) -> Result<(), SendError<Command>> {
184 self.commands.send(command)?;
185 unsafe {
186 Threading::SetEvent(self.pending_scheduled_event).unwrap();
187 }
188 Ok(())
189 }
190}
191
192impl Drop for Stream {
193 fn drop(&mut self) {
194 if self.push_command(Command::Terminate).is_ok() {
195 if let Some(handle) = self.thread.take() {
196 let _ = handle.join();
197 }
198 unsafe {
199 let _ = Foundation::CloseHandle(self.pending_scheduled_event);
200 }
201 }
202 }
203}
204
205impl StreamTrait for Stream {
206 fn play(&self) -> Result<(), PlayStreamError> {
207 self.push_command(Command::PlayStream)
208 .map_err(|_| crate::error::PlayStreamError::DeviceNotAvailable)?;
209 Ok(())
210 }
211
212 fn pause(&self) -> Result<(), PauseStreamError> {
213 self.push_command(Command::PauseStream)
214 .map_err(|_| crate::error::PauseStreamError::DeviceNotAvailable)?;
215 Ok(())
216 }
217
218 fn buffer_size(&self) -> Option<FrameCount> {
219 Some(self.period_frames)
220 }
221}
222
223impl Drop for StreamInner {
224 fn drop(&mut self) {
225 unsafe {
226 let _ = Foundation::CloseHandle(self.event);
227 }
228 }
229}
230
231// Process any pending commands that are queued within the `RunContext`.
232// Returns `true` if the loop should continue running, `false` if it should terminate.
233fn process_commands(run_context: &mut RunContext) -> Result<bool, StreamError> {
234 // Process the pending commands.
235 for command in run_context.commands.try_iter() {
236 match command {
237 Command::PlayStream => unsafe {
238 if !run_context.stream.playing {
239 run_context
240 .stream
241 .audio_client
242 .Start()
243 .map_err(windows_err_to_cpal_err::<StreamError>)?;
244 run_context.stream.playing = true;
245 }
246 },
247 Command::PauseStream => unsafe {
248 if run_context.stream.playing {
249 run_context
250 .stream
251 .audio_client
252 .Stop()
253 .map_err(windows_err_to_cpal_err::<StreamError>)?;
254 run_context.stream.playing = false;
255 }
256 },
257 Command::Terminate => {
258 return Ok(false);
259 }
260 }
261 }
262
263 Ok(true)
264}
265// Wait for any of the given handles to be signalled.
266//
267// Returns the index of the `handle` that was signalled, or an `Err` if
268// `WaitForMultipleObjectsEx` fails.
269//
270// This is called when the `run` thread is ready to wait for the next event. The
271// next event might be some command submitted by the user (the first handle) or
272// might indicate that one of the streams is ready to deliver or receive audio.
273fn wait_for_handle_signal(handles: &[Foundation::HANDLE]) -> Result<usize, BackendSpecificError> {
274 debug_assert!(handles.len() <= SystemServices::MAXIMUM_WAIT_OBJECTS as usize);
275 let result = unsafe {
276 Threading::WaitForMultipleObjectsEx(
277 handles,
278 false, // Don't wait for all, just wait for the first
279 Threading::INFINITE, // TODO: allow setting a timeout
280 false, // irrelevant parameter here
281 )
282 };
283 if result == Foundation::WAIT_FAILED {
284 let err = unsafe { Foundation::GetLastError() };
285 let description = format!("`WaitForMultipleObjectsEx failed: {:?}", err);
286 let err = BackendSpecificError { description };
287 return Err(err);
288 }
289 // Notifying the corresponding task handler.
290 let handle_idx = (result.0 - WAIT_OBJECT_0.0) as usize;
291 Ok(handle_idx)
292}
293
294// Get the number of available frames that are available for writing/reading.
295fn get_available_frames(stream: &StreamInner) -> Result<FrameCount, StreamError> {
296 unsafe {
297 let padding = stream
298 .audio_client
299 .GetCurrentPadding()
300 .map_err(windows_err_to_cpal_err::<StreamError>)?;
301 Ok(stream.max_frames_in_buffer - padding)
302 }
303}
304
305fn run_input(
306 mut run_ctxt: RunContext,
307 data_callback: &mut dyn FnMut(&Data, &InputCallbackInfo),
308 error_callback: &mut dyn FnMut(StreamError),
309) {
310 boost_current_thread_priority(
311 run_ctxt.stream.config.buffer_size,
312 run_ctxt.stream.config.sample_rate,
313 );
314
315 loop {
316 match process_commands_and_await_signal(&mut run_ctxt, error_callback) {
317 Some(ControlFlow::Break) => break,
318 Some(ControlFlow::Continue) => continue,
319 None => (),
320 }
321 let capture_client = match run_ctxt.stream.client_flow {
322 AudioClientFlow::Capture { ref capture_client } => capture_client.clone(),
323 _ => unreachable!(),
324 };
325 match process_input(
326 &run_ctxt.stream,
327 capture_client,
328 data_callback,
329 error_callback,
330 ) {
331 ControlFlow::Break => break,
332 ControlFlow::Continue => continue,
333 }
334 }
335}
336
337fn run_output(
338 mut run_ctxt: RunContext,
339 data_callback: &mut dyn FnMut(&mut Data, &OutputCallbackInfo),
340 error_callback: &mut dyn FnMut(StreamError),
341) {
342 boost_current_thread_priority(
343 run_ctxt.stream.config.buffer_size,
344 run_ctxt.stream.config.sample_rate,
345 );
346
347 loop {
348 match process_commands_and_await_signal(&mut run_ctxt, error_callback) {
349 Some(ControlFlow::Break) => break,
350 Some(ControlFlow::Continue) => continue,
351 None => (),
352 }
353 let render_client = match run_ctxt.stream.client_flow {
354 AudioClientFlow::Render { ref render_client } => render_client.clone(),
355 _ => unreachable!(),
356 };
357 match process_output(
358 &run_ctxt.stream,
359 render_client,
360 data_callback,
361 error_callback,
362 ) {
363 ControlFlow::Break => break,
364 ControlFlow::Continue => continue,
365 }
366 }
367}
368
369#[cfg(feature = "audio_thread_priority")]
370fn boost_current_thread_priority(buffer_size: BufferSize, sample_rate: SampleRate) {
371 use audio_thread_priority::promote_current_thread_to_real_time;
372
373 let buffer_size = if let BufferSize::Fixed(buffer_size) = buffer_size {
374 buffer_size
375 } else {
376 // if the buffer size isn't fixed, let audio_thread_priority choose a sensible default value
377 0
378 };
379
380 if let Err(err) = promote_current_thread_to_real_time(buffer_size, sample_rate) {
381 eprintln!("Failed to promote audio thread to real-time priority: {err}");
382 }
383}
384
385#[cfg(not(feature = "audio_thread_priority"))]
386fn boost_current_thread_priority(_: BufferSize, _: SampleRate) {
387 unsafe {
388 let thread_handle = Threading::GetCurrentThread();
389
390 let _ =
391 Threading::SetThreadPriority(thread_handle, Threading::THREAD_PRIORITY_TIME_CRITICAL);
392 }
393}
394
395enum ControlFlow {
396 Break,
397 Continue,
398}
399
400fn process_commands_and_await_signal(
401 run_context: &mut RunContext,
402 error_callback: &mut dyn FnMut(StreamError),
403) -> Option<ControlFlow> {
404 // Process queued commands.
405 match process_commands(run_context) {
406 Ok(true) => (),
407 Ok(false) => return Some(ControlFlow::Break),
408 Err(err) => {
409 error_callback(err);
410 return Some(ControlFlow::Break);
411 }
412 };
413
414 // Wait for any of the handles to be signalled.
415 let handle_idx = match wait_for_handle_signal(&run_context.handles) {
416 Ok(idx) => idx,
417 Err(err) => {
418 error_callback(err.into());
419 return Some(ControlFlow::Break);
420 }
421 };
422
423 // If `handle_idx` is 0, then it's `pending_scheduled_event` that was signalled in
424 // order for us to pick up the pending commands. Otherwise, a stream needs data.
425 if handle_idx == 0 {
426 return Some(ControlFlow::Continue);
427 }
428
429 None
430}
431
432// The loop for processing pending input data.
433fn process_input(
434 stream: &StreamInner,
435 capture_client: Audio::IAudioCaptureClient,
436 data_callback: &mut dyn FnMut(&Data, &InputCallbackInfo),
437 error_callback: &mut dyn FnMut(StreamError),
438) -> ControlFlow {
439 unsafe {
440 // Get the available data in the shared buffer.
441 let mut buffer: *mut u8 = ptr::null_mut();
442 let mut flags = mem::MaybeUninit::uninit();
443 loop {
444 let mut frames_available = match capture_client.GetNextPacketSize() {
445 Ok(0) => return ControlFlow::Continue,
446 Ok(f) => f,
447 Err(err) => {
448 error_callback(windows_err_to_cpal_err(err));
449 return ControlFlow::Break;
450 }
451 };
452 let mut qpc_position: u64 = 0;
453 let result = capture_client.GetBuffer(
454 &mut buffer,
455 &mut frames_available,
456 flags.as_mut_ptr(),
457 None,
458 Some(&mut qpc_position),
459 );
460
461 match result {
462 // TODO: Can this happen?
463 Err(e) if e.code() == Audio::AUDCLNT_S_BUFFER_EMPTY => continue,
464 Err(e) => {
465 error_callback(windows_err_to_cpal_err(e));
466 return ControlFlow::Break;
467 }
468 Ok(_) => (),
469 }
470
471 debug_assert!(!buffer.is_null());
472
473 let data = buffer as *mut ();
474 let len = frames_available as usize * stream.bytes_per_frame as usize
475 / stream.sample_format.sample_size();
476 let data = Data::from_parts(data, len, stream.sample_format);
477
478 // The `qpc_position` is in 100 nanosecond units. Convert it to nanoseconds.
479 let timestamp = match input_timestamp(stream, qpc_position) {
480 Ok(ts) => ts,
481 Err(err) => {
482 error_callback(err);
483 return ControlFlow::Break;
484 }
485 };
486 let info = InputCallbackInfo { timestamp };
487 data_callback(&data, &info);
488
489 // Release the buffer.
490 let result = capture_client
491 .ReleaseBuffer(frames_available)
492 .map_err(windows_err_to_cpal_err);
493 if let Err(err) = result {
494 error_callback(err);
495 return ControlFlow::Break;
496 }
497 }
498 }
499}
500
501// The loop for writing output data.
502fn process_output(
503 stream: &StreamInner,
504 render_client: Audio::IAudioRenderClient,
505 data_callback: &mut dyn FnMut(&mut Data, &OutputCallbackInfo),
506 error_callback: &mut dyn FnMut(StreamError),
507) -> ControlFlow {
508 // The number of frames available for writing.
509 let frames_available = match get_available_frames(stream) {
510 Ok(0) => return ControlFlow::Continue, // TODO: Can this happen?
511 Ok(n) => n,
512 Err(err) => {
513 error_callback(err);
514 return ControlFlow::Break;
515 }
516 };
517
518 unsafe {
519 let buffer = match render_client.GetBuffer(frames_available) {
520 Ok(b) => b,
521 Err(e) => {
522 error_callback(windows_err_to_cpal_err(e));
523 return ControlFlow::Break;
524 }
525 };
526
527 debug_assert!(!buffer.is_null());
528
529 let data = buffer as *mut ();
530 let len = frames_available as usize * stream.bytes_per_frame as usize
531 / stream.sample_format.sample_size();
532 let mut data = Data::from_parts(data, len, stream.sample_format);
533 let sample_rate = stream.config.sample_rate;
534 let timestamp = match output_timestamp(stream, frames_available, sample_rate) {
535 Ok(ts) => ts,
536 Err(err) => {
537 error_callback(err);
538 return ControlFlow::Break;
539 }
540 };
541 let info = OutputCallbackInfo { timestamp };
542 data_callback(&mut data, &info);
543
544 if let Err(err) = render_client.ReleaseBuffer(frames_available, 0) {
545 error_callback(windows_err_to_cpal_err(err));
546 return ControlFlow::Break;
547 }
548 }
549
550 ControlFlow::Continue
551}
552
553/// Convert the given duration in frames at the given sample rate to a `Duration`.
554fn frames_to_duration(frames: FrameCount, rate: SampleRate) -> Duration {
555 let secsf = frames as f64 / rate as f64;
556 let secs = secsf as u64;
557 let nanos = ((secsf - secs as f64) * 1_000_000_000.0) as u32;
558 Duration::new(secs, nanos)
559}
560
561/// Use the stream's `IAudioClock` to produce the current stream instant.
562///
563/// Uses the QPC position produced via the `GetPosition` method.
564fn stream_instant(stream: &StreamInner) -> Result<crate::StreamInstant, StreamError> {
565 let mut position: u64 = 0;
566 let mut qpc_position: u64 = 0;
567 unsafe {
568 stream
569 .audio_clock
570 .GetPosition(&mut position, Some(&mut qpc_position))
571 .map_err(windows_err_to_cpal_err::<StreamError>)?;
572 };
573 // The `qpc_position` is in 100 nanosecond units. Convert it to nanoseconds.
574 let qpc_nanos = qpc_position as i128 * 100;
575 let instant = crate::StreamInstant::from_nanos_i128(qpc_nanos)
576 .expect("performance counter out of range of `StreamInstant` representation");
577 Ok(instant)
578}
579
580/// Produce the input stream timestamp.
581///
582/// `buffer_qpc_position` is the `qpc_position` returned via the `GetBuffer` call on the capture
583/// client. It represents the instant at which the first sample of the retrieved buffer was
584/// captured.
585fn input_timestamp(
586 stream: &StreamInner,
587 buffer_qpc_position: u64,
588) -> Result<crate::InputStreamTimestamp, StreamError> {
589 // The `qpc_position` is in 100 nanosecond units. Convert it to nanoseconds.
590 let qpc_nanos = buffer_qpc_position as i128 * 100;
591 let capture = crate::StreamInstant::from_nanos_i128(qpc_nanos)
592 .expect("performance counter out of range of `StreamInstant` representation");
593 let callback = stream_instant(stream)?;
594 Ok(crate::InputStreamTimestamp { capture, callback })
595}
596
597/// Produce the output stream timestamp.
598///
599/// `frames_available` is the number of frames available for writing as reported by subtracting the
600/// result of `GetCurrentPadding` from the maximum buffer size.
601///
602/// `sample_rate` is the rate at which audio frames are processed by the device.
603fn output_timestamp(
604 stream: &StreamInner,
605 frames_available: FrameCount,
606 sample_rate: SampleRate,
607) -> Result<crate::OutputStreamTimestamp, StreamError> {
608 let callback = stream_instant(stream)?;
609 // `padding` is the number of frames already queued in the endpoint buffer ahead of the
610 // frames we are about to write. Those frames must drain before ours are heard.
611 let padding = stream.max_frames_in_buffer - frames_available;
612 let playback = callback
613 .add(frames_to_duration(padding, sample_rate) + stream.stream_latency)
614 .expect("`playback` occurs beyond representation supported by `StreamInstant`");
615 Ok(crate::OutputStreamTimestamp { callback, playback })
616}