perf(audio,hid): 减少 PCM 复制和鼠标热路径分配与轮询

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;
+                if frame_tx.receiver_count() > 0 {
                     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_tx.send(frame);
                 }
             }

src/audio/capture_windows.rs

@@ -97,12 +97,12 @@ pub(super) fn run_capture(
                 if samples.is_empty() {
                     continue;
                 }
+                if frame_tx.receiver_count() > 0 {
                     let frame = AudioFrame::new_interleaved(
                         Bytes::copy_from_slice(bytemuck::cast_slice(&samples)),
                         2,
                         48_000,
                     );
-                if frame_tx.receiver_count() > 0 {
                     if let Err(e) = frame_tx.send(frame) {
                         debug!("No audio receivers: {}", e);
                     }

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<i16>,
+    samples: &'audio [i16],
+}
+
+impl<'audio> PcmFrames<'audio> {
+    fn new(pending: &'audio mut Vec<i16>, 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() {
+                    let mut frames = PcmFrames::new(&mut pending, samples);
-                        pending.extend_from_slice(samples);
+                    while let Some(samples) = frames.next_frame() {
-                    }
-
-                    while pending.len() >= OPUS_STEREO_SAMPLES {
                         let opus_result = {
                             let mut enc_guard = encoder.lock().await;
-                            (*enc_guard)
+                            (*enc_guard).as_mut().map(|enc| enc.encode(samples))
-                                .as_mut()
-                                .map(|enc| enc.encode(&pending[..OPUS_STEREO_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<i16> = (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();

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;

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<MouseReport>,
 }
 
 impl MacosDrag {
@@ -37,8 +38,13 @@ impl MacosDrag {
         event: MouseEvent,
         buttons: u8,
         extent: (u32, u32),
-    ) -> (u8, Vec<MouseReport>) {
+    ) -> (u8, &[MouseReport]) {
-        let mut reports = Vec::new();
+        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);

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)?

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,11 +22,13 @@ impl OtgDeviceIo {
         data: &[u8],
         timeout_ms: i32,
     ) -> std::io::Result<bool> {
+        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_ms as u16)) {
+            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)
+                        if revents.contains(PollFlags::POLLERR)
+                            || revents.contains(PollFlags::POLLHUP)
                         {
                             return Err(std::io::Error::new(
                                 std::io::ErrorKind::BrokenPipe,
@@ -32,7 +36,6 @@ impl OtgDeviceIo {
                             ));
                         }
                     }
-                file.write_all(data)?;
                     Ok(true)
                 }
                 Ok(0) => {
@@ -42,5 +45,119 @@ impl OtgDeviceIo {
                 Ok(_) => Ok(false),
                 Err(e) => Err(std::io::Error::other(e)),
             }
+        })
+    }
+}
+
+#[cfg(unix)]
+fn write_with_wait<Writer: Write>(
+    writer: &mut Writer,
+    mut data: &[u8],
+    timeout_ms: u16,
+    mut wait: impl FnMut(&Writer, u16) -> std::io::Result<bool>,
+) -> std::io::Result<bool> {
+    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<Result<usize>>,
+        data: Vec<u8>,
+    }
+
+    impl Write for ScriptedWriter {
+        fn write(&mut self, input: &[u8]) -> Result<usize> {
+            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
+        );
     }
 }