Merge pull request #32 from MatiasVara/add-delay-reading-from-guest

crates/sound: read buffers from guest memory rather than copying them

crates/sound/src/audio_backends/alsa.rs

@@ -145,13 +145,17 @@ fn write_samples_direct(
         let Some(buffer) = stream.buffers.front_mut() else {
             return Ok(false);
         };
-        let mut iter = buffer.bytes[buffer.pos..].iter().cloned();
+        let mut buf = vec![0; buffer.data_descriptor.len() as usize];
+        let read_bytes = buffer
+            .consume(&mut buf)
+            .expect("failed to read buffer from guest");
+        let mut iter = buf[0..read_bytes as usize].iter().cloned();
         let frames = mmap.write(&mut iter);
         let written_bytes = pcm.frames_to_bytes(frames);
         if let Ok(written_bytes) = usize::try_from(written_bytes) {
             buffer.pos += written_bytes;
         }
-        if buffer.pos >= buffer.bytes.len() {
+        if buffer.pos >= buffer.data_descriptor.len() as usize {
             stream.buffers.pop_front();
         }
     }
@@ -191,7 +195,14 @@ fn write_samples_io(
             let Some(buffer) = stream.buffers.front_mut() else {
                 return 0;
             };
-            let mut iter = buffer.bytes[buffer.pos..].iter().cloned();
+            let mut data = vec![0; buffer.data_descriptor.len() as usize];
+
+            // consume() always reads (buffer.data_descriptor.len() -
+            // buffer.pos) bytes
+            let read_bytes = buffer
+                .consume(&mut data)
+                .expect("failed to read buffer from guest");
+            let mut iter = data[0..read_bytes as usize].iter().cloned();
 
             let mut written_bytes = 0;
             for (sample, byte) in buf.iter_mut().zip(&mut iter) {
@@ -199,7 +210,7 @@ fn write_samples_io(
                 written_bytes += 1;
             }
             buffer.pos += written_bytes as usize;
-            if buffer.pos >= buffer.bytes.len() {
+            if buffer.pos >= buffer.data_descriptor.len() as usize {
                 stream.buffers.pop_front();
             }
             p.bytes_to_frames(written_bytes)

crates/sound/src/audio_backends/pipewire.rs

@@ -329,26 +329,27 @@ impl AudioBackend for PwBackend {
 
                             let mut start = buffer.pos;
 
-                            let avail = (buffer.bytes.len() - start) as i32;
+                            let avail = (buffer.data_descriptor.len() - start as u32) as i32;
 
                             if avail < n_bytes as i32 {
                                 n_bytes = avail.try_into().unwrap();
                             }
-                            let p = &mut slice[buffer.pos..start + n_bytes];
+                            let p = &mut slice[0..n_bytes];
                             if avail <= 0 {
                                 // pad with silence
                                 unsafe {
                                     ptr::write_bytes(p.as_mut_ptr(), 0, n_bytes);
                                 }
                             } else {
-                                let slice = &buffer.bytes[buffer.pos..start + n_bytes];
-                                p.copy_from_slice(slice);
+                                // consume() always reads (buffer.data_descriptor.len() -
+                                // buffer.pos) bytes
+                                buffer.consume(p).expect("failed to read buffer from guest");
 
                                 start += n_bytes;
 
                                 buffer.pos = start;
 
-                                if start >= buffer.bytes.len() {
+                                if start >= buffer.data_descriptor.len() as usize {
                                     streams.buffers.pop_front();
                                 }
                             }

crates/sound/src/device.rs

@@ -469,14 +469,16 @@ impl VhostUserSoundThread {
                         .into());
                     }
                     TxState::WaitingBufferForStreamId(_stream_id) => {
-                        let mut buf = vec![0; descriptor.len() as usize];
-                        let bytes_read = desc_chain
-                            .memory()
-                            .read(&mut buf, descriptor.addr())
-                            .map_err(|_| Error::DescriptorReadFailed)?;
-                        buf.truncate(bytes_read);
-
-                        buffers.push(Buffer::new(buf, Arc::clone(&message)));
+                        /*
+                        Rather than copying the content of a descriptor, buffer keeps a pointer to it.
+                        When we copy just after the request is enqueued, the guest's userspace may or
+                        may not have updated the buffer contents. Guest driver simply moves buffers
+                        from the used ring to the available ring without knowing whether the content
+                        has been updated. The device only reads the buffer from guest memory when the
+                        audio engine requires it, which is about after a period thus ensuring that the
+                        buffer is up-to-date.
+                        */
+                        buffers.push(Buffer::new(*descriptor, Arc::clone(&message)));
                     }
                 }
             }

crates/sound/src/stream.rs

@@ -4,7 +4,7 @@
 use std::{collections::VecDeque, sync::Arc};
 
 use thiserror::Error as ThisError;
-use vm_memory::{Le32, Le64};
+use vm_memory::{Address, Bytes, Le32, Le64};
 
 use crate::{virtio_sound::*, IOMessage, SUPPORTED_FORMATS, SUPPORTED_RATES};
 
@@ -15,6 +15,8 @@ pub enum Error {
     InvalidStateTransition(PCMState, PCMState),
     #[error("Guest requested an invalid stream id: {0}")]
     InvalidStreamId(u32),
+    #[error("Descriptor read failed")]
+    DescriptorReadFailed,
 }
 
 type Result<T> = std::result::Result<T, Error>;
@@ -234,7 +236,8 @@ impl Default for PcmParams {
 }
 
 pub struct Buffer {
-    pub bytes: Vec<u8>,
+    // TODO: to make private and add len usize
+    pub data_descriptor: virtio_queue::Descriptor,
     pub pos: usize,
     pub message: Arc<IOMessage>,
 }
@@ -242,7 +245,6 @@ pub struct Buffer {
 impl std::fmt::Debug for Buffer {
     fn fmt(&self, fmt: &mut std::fmt::Formatter) -> std::fmt::Result {
         fmt.debug_struct(stringify!(Buffer))
-            .field("bytes", &self.bytes.len())
             .field("pos", &self.pos)
             .field("message", &Arc::as_ptr(&self.message))
             .finish()
@@ -250,13 +252,29 @@ impl std::fmt::Debug for Buffer {
 }
 
 impl Buffer {
-    pub fn new(bytes: Vec<u8>, message: Arc<IOMessage>) -> Self {
+    pub fn new(data_descriptor: virtio_queue::Descriptor, message: Arc<IOMessage>) -> Self {
         Self {
-            bytes,
             pos: 0,
+            data_descriptor,
             message,
         }
     }
+
+    pub fn consume(&self, buf: &mut [u8]) -> Result<u32> {
+        let addr = self.data_descriptor.addr();
+        let offset = self.pos as u64;
+        let len = self
+            .message
+            .desc_chain
+            .memory()
+            .read(
+                buf,
+                addr.checked_add(offset)
+                    .expect("invalid guest memory address"),
+            )
+            .map_err(|_| Error::DescriptorReadFailed)?;
+        Ok(len as u32)
+    }
 }
 
 impl Drop for Buffer {