diff --git a/src/audio/capture_linux.rs b/src/audio/capture_linux.rs index af11b15a..7db1b518 100644 --- a/src/audio/capture_linux.rs +++ b/src/audio/capture_linux.rs @@ -79,12 +79,12 @@ pub(super) fn run_capture( Ok(frames_read) => { consecutive_errors = 0; let byte_count = frames_read * config.channels as usize * 2; - let frame = AudioFrame::new_interleaved( - Bytes::copy_from_slice(&buffer[..byte_count]), - config.channels, - config.sample_rate, - ); if frame_tx.receiver_count() > 0 { + let frame = AudioFrame::new_interleaved( + Bytes::copy_from_slice(&buffer[..byte_count]), + config.channels, + config.sample_rate, + ); let _ = frame_tx.send(frame); } } diff --git a/src/audio/capture_windows.rs b/src/audio/capture_windows.rs index 260d6314..db613ee2 100644 --- a/src/audio/capture_windows.rs +++ b/src/audio/capture_windows.rs @@ -97,12 +97,12 @@ pub(super) fn run_capture( if samples.is_empty() { continue; } - let frame = AudioFrame::new_interleaved( - Bytes::copy_from_slice(bytemuck::cast_slice(&samples)), - 2, - 48_000, - ); if frame_tx.receiver_count() > 0 { + let frame = AudioFrame::new_interleaved( + Bytes::copy_from_slice(bytemuck::cast_slice(&samples)), + 2, + 48_000, + ); if let Err(e) = frame_tx.send(frame) { debug!("No audio receivers: {}", e); } diff --git a/src/audio/streamer.rs b/src/audio/streamer.rs index c44f69c8..ca402a3c 100644 --- a/src/audio/streamer.rs +++ b/src/audio/streamer.rs @@ -15,6 +15,39 @@ use crate::error::{AppError, Result}; /// 48 kHz stereo: 20 ms = 960 × 2 samples (S16LE). const OPUS_STEREO_SAMPLES: usize = 960 * 2; +struct PcmFrames<'audio> { + pending: &'audio mut Vec, + samples: &'audio [i16], +} + +impl<'audio> PcmFrames<'audio> { + fn new(pending: &'audio mut Vec, samples: &'audio [i16]) -> Self { + Self { pending, samples } + } + + fn next_frame(&mut self) -> Option<&[i16]> { + if self.pending.len() == OPUS_STEREO_SAMPLES { + self.pending.clear(); + } + if self.pending.is_empty() && self.samples.len() >= OPUS_STEREO_SAMPLES { + let (frame, remaining) = self.samples.split_at(OPUS_STEREO_SAMPLES); + self.samples = remaining; + return Some(frame); + } + if self.samples.is_empty() { + return None; + } + let count = (OPUS_STEREO_SAMPLES - self.pending.len()).min(self.samples.len()); + if self.pending.capacity() < OPUS_STEREO_SAMPLES { + self.pending + .reserve_exact(OPUS_STEREO_SAMPLES - self.pending.len()); + } + self.pending.extend_from_slice(&self.samples[..count]); + self.samples = &self.samples[count..]; + (self.pending.len() == OPUS_STEREO_SAMPLES).then(|| self.pending.as_slice()) + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum AudioStreamState { #[default] @@ -331,18 +364,12 @@ impl AudioStreamer { continue; } }; - if !samples.is_empty() { - pending.extend_from_slice(samples); - } - - while pending.len() >= OPUS_STEREO_SAMPLES { + let mut frames = PcmFrames::new(&mut pending, samples); + while let Some(samples) = frames.next_frame() { let opus_result = { let mut enc_guard = encoder.lock().await; - (*enc_guard) - .as_mut() - .map(|enc| enc.encode(&pending[..OPUS_STEREO_SAMPLES])) + (*enc_guard).as_mut().map(|enc| enc.encode(samples)) }; - pending.drain(..OPUS_STEREO_SAMPLES); match opus_result { Some(Ok(opus_frame)) => { @@ -396,6 +423,53 @@ mod tests { use super::*; use bytes::Bytes; + #[test] + fn aligned_pcm_frames_borrow_input_without_allocating() { + let samples = vec![123; OPUS_STEREO_SAMPLES * 2]; + let mut pending = Vec::new(); + let mut frames = PcmFrames::new(&mut pending, &samples); + assert_eq!(frames.next_frame().unwrap().as_ptr(), samples.as_ptr()); + assert_eq!( + frames.next_frame().unwrap().as_ptr(), + samples[OPUS_STEREO_SAMPLES..].as_ptr() + ); + assert!(frames.next_frame().is_none()); + assert!(pending.is_empty()); + assert_eq!(pending.capacity(), 0); + } + + #[test] + fn fragmented_pcm_preserves_samples_and_opus_packets() { + let samples: Vec = (0..OPUS_STEREO_SAMPLES * 8 + 31) + .map(|index| ((index * 73 % 65536) as i32 - 32768) as i16) + .collect(); + let mut pending = Vec::new(); + let mut actual = Vec::new(); + let mut encoder = OpusEncoder::new(OpusConfig::default()).unwrap(); + let mut reference_encoder = OpusEncoder::new(OpusConfig::default()).unwrap(); + let mut offset = 0; + for size in [1, 317, 4601, 2000, 11, 8470, samples.len()] { + let end = (offset + size).min(samples.len()); + let mut frames = PcmFrames::new(&mut pending, &samples[offset..end]); + while let Some(frame) = frames.next_frame() { + actual.extend_from_slice(frame); + let encoded = encoder.encode(frame).unwrap(); + let start = actual.len() - OPUS_STEREO_SAMPLES; + let reference = reference_encoder + .encode(&samples[start..actual.len()]) + .unwrap(); + assert_eq!(encoded.data, reference.data); + assert_eq!(encoded.sequence, reference.sequence); + assert_eq!(encoded.duration_ms, reference.duration_ms); + } + assert!(pending.len() < OPUS_STEREO_SAMPLES); + assert!(pending.capacity() <= OPUS_STEREO_SAMPLES); + offset = end; + } + assert_eq!(actual, samples[..OPUS_STEREO_SAMPLES * 8]); + assert_eq!(pending, samples[OPUS_STEREO_SAMPLES * 8..]); + } + #[test] fn test_streamer_config_default() { let config = AudioStreamerConfig::default(); diff --git a/src/hid/ch9329.rs b/src/hid/ch9329.rs index b213bcc4..54314930 100644 --- a/src/hid/ch9329.rs +++ b/src/hid/ch9329.rs @@ -1291,7 +1291,7 @@ impl HidBackend for Ch9329Backend { let mut state = self.macos_drag_state.lock(); let (buttons, reports) = state.plan(event, buttons, *self.screen_resolution.read()); self.mouse_buttons.store(buttons, Ordering::Relaxed); - for report in reports { + for report in reports.iter().copied() { match report { MouseReport::Absolute { buttons, x, y } => { let x = (u32::from(x) * CH9329_MOUSE_RESOLUTION / 32768) as u16; diff --git a/src/hid/macos_drag.rs b/src/hid/macos_drag.rs index ce97a1cc..d9efc37c 100644 --- a/src/hid/macos_drag.rs +++ b/src/hid/macos_drag.rs @@ -3,7 +3,7 @@ use super::{MouseEvent, MouseEventType}; -#[derive(Debug, PartialEq)] +#[derive(Clone, Copy, Debug, PartialEq)] pub(super) enum MouseReport { Absolute { buttons: u8, @@ -25,6 +25,7 @@ pub(super) struct MacosDrag { extent: (u32, u32), relative_input: bool, absolute_drag: bool, + reports: Vec, } impl MacosDrag { @@ -37,8 +38,13 @@ impl MacosDrag { event: MouseEvent, buttons: u8, extent: (u32, u32), - ) -> (u8, Vec) { - let mut reports = Vec::new(); + ) -> (u8, &[MouseReport]) { + if self.reports.capacity() > 128 { + self.reports = Vec::new(); + } else { + self.reports.clear(); + } + let reports = &mut self.reports; let mut next_buttons = buttons; if extent != self.extent || buttons == 0 { self.remainder = (0, 0); @@ -56,7 +62,7 @@ impl MacosDrag { Self::delta(position.0, self.position.0, extent.0, &mut self.remainder.0); let dy = Self::delta(position.1, self.position.1, extent.1, &mut self.remainder.1); - Self::motion(&mut reports, buttons, dx, dy); + Self::motion(reports, buttons, dx, dy); } else { reports.push(MouseReport::Absolute { buttons: 0, @@ -69,7 +75,7 @@ impl MacosDrag { MouseEventType::Move => { self.relative_input = true; Self::motion( - &mut reports, + reports, buttons, i64::from(event.x.clamp(-127, 127)), i64::from(event.y.clamp(-127, 127)), @@ -127,7 +133,7 @@ impl MacosDrag { }); } } - (next_buttons, reports) + (next_buttons, reports.as_slice()) } fn delta(current: u16, previous: u16, extent: u32, remainder: &mut i64) -> i64 { @@ -160,13 +166,39 @@ mod tests { use super::*; use crate::hid::MouseButton; + #[test] + fn repeated_motion_reuses_report_storage() { + let mut state = MacosDrag::default(); + let (_, reports) = state.plan(MouseEvent::move_rel(1, 1), 0, (1920, 1080)); + let allocation = reports.as_ptr(); + for _ in 0..1000 { + let (_, reports) = state.plan(MouseEvent::move_rel(1, 1), 0, (1920, 1080)); + assert_eq!(reports.as_ptr(), allocation); + assert_eq!(reports.len(), 1); + } + } + + #[test] + fn oversized_report_storage_is_not_retained_for_small_events() { + let mut state = MacosDrag::default(); + state.reports = Vec::with_capacity(256); + assert!(state.reports.capacity() > 128); + state.plan(MouseEvent::move_rel(1, 1), 0, (1920, 1080)); + assert!(state.reports.capacity() <= 128); + } + fn drag(steps: &[i32], extent: (u32, u32)) -> (i64, i64) { let mut state = MacosDrag::default(); state.plan(MouseEvent::move_abs(8000, 8000), 0, extent); state.plan(MouseEvent::button_down(MouseButton::Left), 0, extent); let mut total = (0, 0); for &x in steps { - for report in state.plan(MouseEvent::move_abs(x, x), 1, extent).1 { + for report in state + .plan(MouseEvent::move_abs(x, x), 1, extent) + .1 + .iter() + .copied() + { if let MouseReport::Relative { dx, dy, .. } = report { total.0 += i64::from(dx); total.1 += i64::from(dy); diff --git a/src/hid/otg.rs b/src/hid/otg.rs index c122e2eb..aa4b84b7 100644 --- a/src/hid/otg.rs +++ b/src/hid/otg.rs @@ -871,7 +871,7 @@ impl HidBackend for OtgBackend { let extent = self.screen_resolution.read().unwrap_or((1920, 1080)); let (buttons, reports) = state.plan(event, buttons, extent); self.mouse_buttons.store(buttons, Ordering::Relaxed); - for report in reports { + for report in reports.iter().copied() { match report { MouseReport::Absolute { buttons, x, y } => { self.send_mouse_report_absolute(buttons, x, y, 0)? diff --git a/src/hid/otg_device.rs b/src/hid/otg_device.rs index 9d47a689..3ac40abc 100644 --- a/src/hid/otg_device.rs +++ b/src/hid/otg_device.rs @@ -4,6 +4,8 @@ use std::fs::File; use std::io::Write; #[cfg(unix)] use std::os::unix::io::AsFd; +#[cfg(unix)] +use std::time::{Duration, Instant}; #[cfg(unix)] use nix::poll::{poll, PollFd, PollFlags, PollTimeout}; @@ -20,27 +22,142 @@ impl OtgDeviceIo { data: &[u8], timeout_ms: i32, ) -> std::io::Result { - let mut pollfd = [PollFd::new(file.as_fd(), PollFlags::POLLOUT)]; - match poll(&mut pollfd, PollTimeout::from(timeout_ms as u16)) { - Ok(1) => { - if let Some(revents) = pollfd[0].revents() { - if revents.contains(PollFlags::POLLERR) || revents.contains(PollFlags::POLLHUP) - { - return Err(std::io::Error::new( - std::io::ErrorKind::BrokenPipe, - "Device error or hangup", - )); + write_with_wait(file, data, timeout_ms as u16, |file, timeout| { + let mut pollfd = [PollFd::new(file.as_fd(), PollFlags::POLLOUT)]; + match poll(&mut pollfd, PollTimeout::from(timeout)) { + Ok(1) => { + if let Some(revents) = pollfd[0].revents() { + if revents.contains(PollFlags::POLLERR) + || revents.contains(PollFlags::POLLHUP) + { + return Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "Device error or hangup", + )); + } } + Ok(true) } - file.write_all(data)?; - Ok(true) + Ok(0) => { + trace!("HID write timeout, dropping data"); + Ok(false) + } + Ok(_) => Ok(false), + Err(e) => Err(std::io::Error::other(e)), } - Ok(0) => { - trace!("HID write timeout, dropping data"); - Ok(false) - } - Ok(_) => Ok(false), - Err(e) => Err(std::io::Error::other(e)), - } + }) + } +} + +#[cfg(unix)] +fn write_with_wait( + writer: &mut Writer, + mut data: &[u8], + timeout_ms: u16, + mut wait: impl FnMut(&Writer, u16) -> std::io::Result, +) -> std::io::Result { + let mut deadline = None; + while !data.is_empty() { + match writer.write(data) { + Ok(0) => return Err(std::io::ErrorKind::WriteZero.into()), + Ok(written) => data = &data[written..], + Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + let deadline = deadline.get_or_insert_with(|| { + Instant::now() + Duration::from_millis(u64::from(timeout_ms)) + }); + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() + || !wait( + writer, + remaining + .as_nanos() + .div_ceil(1_000_000) + .min(u16::MAX as u128) as u16, + )? + { + return Ok(false); + } + } + Err(error) => return Err(error), + } + } + Ok(true) +} + +#[cfg(all(test, unix))] +mod tests { + use super::*; + use std::collections::VecDeque; + use std::io::{ErrorKind, Result}; + + struct ScriptedWriter { + steps: VecDeque>, + data: Vec, + } + + impl Write for ScriptedWriter { + fn write(&mut self, input: &[u8]) -> Result { + let count = self.steps.pop_front().unwrap_or(Ok(input.len()))?; + self.data.extend_from_slice(&input[..count]); + Ok(count) + } + + fn flush(&mut self) -> Result<()> { + Ok(()) + } + } + + #[test] + fn ready_device_skips_poll_and_preserves_report_bytes() { + let mut writer = ScriptedWriter { + steps: VecDeque::new(), + data: Vec::new(), + }; + assert!( + write_with_wait(&mut writer, &[1, 2, 3, 4], 100, |_, _| panic!( + "unexpected poll" + )) + .unwrap() + ); + assert_eq!(writer.data, [1, 2, 3, 4]); + } + + #[test] + fn partial_and_interrupted_writes_do_not_duplicate_bytes() { + let mut writer = ScriptedWriter { + steps: [ + Ok(2), + Err(ErrorKind::Interrupted.into()), + Err(ErrorKind::WouldBlock.into()), + ] + .into(), + data: Vec::new(), + }; + let mut waits = 0; + assert!(write_with_wait(&mut writer, &[1, 2, 3, 4], 100, |_, _| { + waits += 1; + Ok(true) + }) + .unwrap()); + assert_eq!(waits, 1); + assert_eq!(writer.data, [1, 2, 3, 4]); + } + + #[test] + fn blocked_device_retains_bounded_timeout_and_error_handling() { + let mut writer = ScriptedWriter { + steps: [Err(ErrorKind::WouldBlock.into())].into(), + data: Vec::new(), + }; + assert!(!write_with_wait(&mut writer, &[1], 100, |_, _| Ok(false)).unwrap()); + assert!(writer.data.is_empty()); + writer.steps.push_back(Err(ErrorKind::BrokenPipe.into())); + assert_eq!( + write_with_wait(&mut writer, &[1], 100, |_, _| Ok(true)) + .unwrap_err() + .kind(), + ErrorKind::BrokenPipe + ); } }