v4l/io/dmabuf/

arena.rs

use dma_boom::DmaBuf;
use parking_lot::{ArcRwLockWriteGuard, RwLock, RawRwLock};
use std::fmt;
use std::{io, mem};
use std::ops;
use std::os::fd::{FromRawFd, AsRawFd};
use std::sync::Arc;
use tracing::error;

use crate::buffer;
use crate::device::Handle;
use crate::memory::Memory;
use crate::v4l2;
use crate::v4l_sys::{v4l2_format, v4l2_requestbuffers, v4l2_exportbuffer};

/// Implements all IO operations
pub struct Hardware;

impl Hardware {
    unsafe fn g_fmt(handle: &Handle, v4l2_fmt: &mut v4l2_format) -> Result<(), io::Error> {
        v4l2::ioctl(
            handle.as_raw_fd(),
            v4l2::vidioc::VIDIOC_G_FMT,
            v4l2_fmt as *mut _ as *mut std::os::raw::c_void,
        )
    }
    
    unsafe fn reqbufs(handle: &Handle, v4l2_reqbufs: &mut v4l2_requestbuffers) -> Result<(), io::Error> {
        v4l2::ioctl(
            handle.as_raw_fd(),
            v4l2::vidioc::VIDIOC_REQBUFS,
            v4l2_reqbufs as *mut _ as *mut std::os::raw::c_void,
        )
    }
    unsafe fn expbuf(handle: &Handle, v4l2_exportbuf: &mut v4l2_exportbuffer) -> Result<(), io::Error> {
        v4l2::ioctl(
            handle.as_raw_fd(),
            v4l2::vidioc::VIDIOC_EXPBUF,
            v4l2_exportbuf as *mut _ as *mut std::os::raw::c_void,
        )
    }
}

struct Common {
    handle: Arc<Handle>,
    buf_type: buffer::Type,
    _io: Hardware,
}

impl Common {
    fn requestbuffers_desc(&self) -> v4l2_requestbuffers {
        v4l2_requestbuffers {
            type_: self.buf_type as u32,
            ..unsafe { mem::zeroed() }
        }
    }

    /// Returns the number of buffers present in the arena (same as the number in `self.bufs`).
    fn allocate(&mut self, count: u32) -> io::Result<Vec<DmaBuf>> {
        // we need to get the maximum buffer size from the format first
        let mut v4l2_fmt = v4l2_format {
            type_: self.buf_type as u32,
            ..unsafe { mem::zeroed() }
        };

        unsafe { Hardware::g_fmt(&self.handle, &mut v4l2_fmt) }?;

        let mut v4l2_reqbufs = v4l2_requestbuffers {
            count,
            memory: Memory::Mmap as u32,
            ..self.requestbuffers_desc()
        };
        unsafe { Hardware::reqbufs(&self.handle, &mut v4l2_reqbufs) }?;

        let mut bufs = Vec::new();
        for index in 0..v4l2_reqbufs.count {
            let mut v4l2_exportbuf = v4l2_exportbuffer {
                index,
                type_: self.buf_type as u32,
                flags: libc::O_RDWR as _,
                ..unsafe { mem::zeroed() }
            };
            let buf = unsafe {
                Hardware::expbuf(&self.handle, &mut v4l2_exportbuf)?;
                DmaBuf::from_raw_fd(v4l2_exportbuf.fd)
            };
            bufs.push(buf);
        }
        
        let mut v4l2_reqbufs = v4l2_requestbuffers {
            count,
            memory: Memory::DmaBuf as u32,
            ..self.requestbuffers_desc()
        };
        // FIXME: how do the buffers above need to be cleaned up if this fails?
        unsafe { Hardware::reqbufs(&self.handle, &mut v4l2_reqbufs) }?;
        
        
        Ok(bufs)
    }
    
    fn release(&mut self) -> io::Result<()> {
        // free all buffers by requesting 0
        let mut v4l2_reqbufs = v4l2_requestbuffers {
            count: 0,
            memory: Memory::DmaBuf as u32,
            ..self.requestbuffers_desc()
        };
        unsafe { Hardware::reqbufs(&self.handle, &mut v4l2_reqbufs) }
    }
}

impl Drop for Common {
    fn drop(&mut self) {
        if let Err(e) = self.release() {
            if let Some(code) = e.raw_os_error() {
                // ENODEV means the file descriptor wrapped in the handle became invalid, most
                // likely because the device was unplugged or the connection (USB, PCI, ..)
                // broke down. Handle this case gracefully by ignoring it.
                if code == 19 {
                    /* ignore */
                    return;
                }
            }

            error!("DMABUF leak: {:?}", e);
        }
    }
}

/// Manage dmabuf buffers
///
/// All buffers are released in the Drop impl.
pub struct Arena {
    common: Common,
    bufs: Vec<DmaBuf>,
}

impl Arena {
    /// Returns a new buffer manager instance
    ///
    /// You usually do not need to use this directly.
    /// A UserBufferStream creates its own manager instance by default.
    ///
    /// # Arguments
    ///
    /// * `handle` - Device handle to get its file descriptor
    /// * `buf_type` - Type of the buffers
    pub fn new(handle: Arc<Handle>, buf_type: buffer::Type) -> Self {
        Self {
            common: Common {
                handle,
                buf_type,
                _io: Hardware,
            },
            bufs: Vec::new(),
        }
    }
    
    /// Returns the number of allocated buffers
    pub fn allocate(&mut self, count: u32) -> io::Result<usize> {
        self.bufs = self.common.allocate(count)?;
        Ok(self.bufs.len())
    }
    
    pub fn len(&self) -> usize {
        self.bufs.len()
    }
}


impl Arena {
    /// Returns the buffer for this index
    pub fn get_dmabuf(&mut self, index: usize) -> Result<&DmaBuf, &'static str> {
        self.bufs.get(index)
            .ok_or("Index higher than available buffers")
    }
    
    /// Returns the buffer for this index
    pub fn get_dmabuf_mut(&mut self, index: usize) -> Result<&mut DmaBuf, &'static str> {
        self.bufs.get_mut(index)
            .ok_or("Index higher than available buffers")
    }
}

/// A DMA-BUF buffer which is handed out to the user as if owned, but cannot be accessed while it's enqueued at the device for writing.
///
/// Watch out for deadlocks!
///
/// ```no_run
/// use v4l::io::dmabuf::DmaBufProtected;
/// use v4l::dummy;
///
/// let buf: DmaBufProtected = dummy::new_stub();
/// let guard = buf.read();
/// let mmap = guard.memory_map_ro();
/// ```
pub type DmaBufProtected = BufProtected<DmaBuf>;

pub struct BufProtected<T>(Arc<RwLock<T>>);

impl<T> Clone for BufProtected<T> {
    fn clone(&self) -> Self {
        Self(self.0.clone())
    }
}

impl<T> ops::Deref for BufProtected<T> {
    type Target = Arc<RwLock<T>>;
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl<T> fmt::Debug for BufProtected<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_tuple("BufProtected")
            .field(&self.0.data_ptr())
            .finish()
    }
}

trait UniqueId {
    type Target;
    fn unique_id(&self) -> Self::Target;
}

impl<T> UniqueId for RwLock<T> {
    type Target = *mut T;
    fn unique_id(&self) -> *mut T {
        self.data_ptr()
    }
}

pub struct LockedBuffer<T> {
    /// This serves as a stable identifier of a buffer corresponding to this slot, meaning a buffer with this ID has the same index known by the kernel and the same device.
    id: *mut T,
    /// The buffer which may be handed out to the user after it's ready.
    /// RwLock locked for writing ensures no other reference can try to use it while it's waiting to be filled.
    buf: Option<ArcRwLockWriteGuard<RawRwLock, T>>,
}

impl<T> From<T> for LockedBuffer<T> {
    fn from(value: T) -> Self {
        let buf = Arc::new(RwLock::new(value));
        Self {
            id: buf.data_ptr(),
            buf: Some(buf.write_arc()),
        }
    }
}


impl<T> fmt::Debug for LockedBuffer<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_tuple("LockedBuffer")
            .field(&self.id)
            .field(&self.buf.as_ref().map(|_| ()))
            .finish()
    }
}

/// Doesn't cleanup buffers in any way, so you have to ensure that
struct BufferSwapper<T>(Vec<LockedBuffer<T>>);

impl<T> BufferSwapper<T> {
    fn new(bufs: Vec<T>) ->Self {
        Self(bufs.into_iter().map(LockedBuffer::from).collect())
    }
    
    fn len(&self) -> usize {
        self.0.len()
    }

    fn all_locked(&self) -> bool {
        self.0.iter()
            .find(|LockedBuffer { buf, .. }| buf.is_none())
            .is_none()
    }
    
    /// Finds the kernel index for this buffer and places it there.
    /// Doesn't check if the buffer was there already. Returns the kernel index.
    pub fn replace_buffer(&mut self, buf: BufProtected<T>) -> Result<usize, (BufProtected<T>, &'static str)> {
        let index = self.0.iter().position(|entry| entry.id == buf.0.unique_id());

        match index {
            None => Err((buf, "Buffer not from this stream")),
            Some(index) => {
                self.0[index] = LockedBuffer {
                    id: buf.0.unique_id(),
                    buf: Some(buf.0.write_arc()),
                };
                Ok(index)
            }
        }
    }
    
    /// Removes and returns buffer at this kernel index.
    pub fn take_buffer(&mut self, index: usize) -> Result<BufProtected<T>, &'static str> {
        let rw_buf = self.0.get_mut(index)
            .ok_or("Index higher than available buffers")?
            .buf.take()
            .ok_or("No buffer was stored at this index")?;

        Ok(BufProtected(ArcRwLockWriteGuard::rwlock(&rw_buf).clone()))
    }

    /// Returns a writeably-locked buffer for this index
    pub fn get_buf_mut(&mut self, index: usize) -> Result<&mut T, &'static str> {
        Ok(
            self.0.get_mut(index)
                .ok_or("Index higher than available buffers")?
                .buf.as_mut()
                .ok_or("No buffer was stored at this index")?
        )
    }
}

/// Allows the user to manage buffers manually. Buffers must be explicitly replaced. Release will only succeed if all buffers were returned with replace_buffer.
pub struct ManuallyManaged {
    common: Common,
    bufs: BufferSwapper<DmaBuf>,
}


impl ManuallyManaged {
    /// Returns a new buffer manager instance
    ///
    /// You usually do not need to use this directly.
    /// A UserBufferStream creates its own manager instance by default.
    ///
    /// # Arguments
    ///
    /// * `handle` - Device handle to get its file descriptor
    /// * `buf_type` - Type of the buffers
    pub fn new(handle: Arc<Handle>, buf_type: buffer::Type, count: usize) -> io::Result<Self> {
        let mut common = Common {
            handle,
            buf_type,
            _io: Hardware,
        };
        Ok(Self {
            bufs: BufferSwapper::new(common.allocate(count as u32)?),
            common,
        })
    }
    
    pub fn len(&self) -> usize {
        self.bufs.len()
    }
    
    pub fn take_buffer(&mut self, index:usize)
        -> Result<BufProtected<DmaBuf>, &'static str>
    {
        self.bufs.take_buffer(index)
    }
    
    pub fn replace_buffer(&mut self, buf: BufProtected<DmaBuf>)
        -> Result<usize, (BufProtected<DmaBuf>, &'static str)>
    {
        self.bufs.replace_buffer(buf)
    }
    
    /// Returns a writeably-locked buffer for this index
    pub fn get_buf_mut(&mut self, index: usize) -> Result<&mut DmaBuf, &'static str> {
        self.bufs.get_buf_mut(index)
    }
}

impl ManuallyManaged {
    /// Release buffers without checking if any buffers are gone missing.
    ///
    /// What makes this unsafe is that the buffers out there may become unuseable (TODO: check semantics).
    /// See [V4L2_BUF_CAP_SUPPORTS_ORPHANED_BUFS](https://docs.kernel.org/userspace-api/media/v4l/vidioc-reqbufs.html#description).
    pub unsafe fn force_release(&mut self) -> io::Result<()> {
        self.common.release()
    }

    /// Safely release buffers
    pub fn release(&mut self) -> io::Result<()> {
        if !self.bufs.all_locked() {
            return Err(io::Error::other("Busy: not all buffers were returned"));
        }
        
        unsafe { self.force_release() }
    }
}

impl Drop for ManuallyManaged {
    fn drop(&mut self) {
        // This exists just to give user the error message before giving up
        if let Err(e) = self.release() {
            if let Some(code) = e.raw_os_error() {
                // ENODEV means the file descriptor wrapped in the handle became invalid, most
                // likely because the device was unplugged or the connection (USB, PCI, ..)
                // broke down. Handle this case gracefully by ignoring it.
                if code == 19 {
                    /* ignore */
                    return;
                }
            }

            error!("DMABUF leak: {:?}", e);
        }
    }
}


#[cfg(test)]
mod test {
    use super::*;

    /// The buffer handed out to the user must not be already locked
    #[test]
    fn return_unlocked() {
        let mut arena = BufferSwapper::new((0..4).collect());
        
        let b = arena.take_buffer(0).unwrap();
        assert!(b.is_locked() == false);
        
        arena.replace_buffer(b.clone()).unwrap();
        assert!(b.is_locked() == true);
        let b = arena.take_buffer(1).unwrap();
        assert!(b.is_locked() == false);
        
        arena.replace_buffer(b.clone()).unwrap();
        assert!(b.is_locked() == true);
        let b = arena.take_buffer(2).unwrap();
        assert!(b.is_locked() == false);
        
        arena.replace_buffer(b.clone()).unwrap();
        assert!(b.is_locked() == true);
        let b = arena.take_buffer(3).unwrap();
        assert!(b.is_locked() == false);
    }
}