worker.rs

use std::collections::BTreeMap;
use std::collections::HashMap;
use std::fs::File;
use std::io::Write;
use std::os::unix::prelude::AsRawFd;
use std::path::PathBuf;
use std::sync::Arc;

use anyhow::Context;
use anyhow::Result;
use base::trace;
use base::warn;
use base::Descriptor;
use base::Event;
use base::EventToken;
use base::EventType;
use base::Protection;
use base::SafeDescriptor;
use base::WaitContext;
use v4l2r::bindings::v4l2_decoder_cmd;
use v4l2r::bindings::v4l2_encoder_cmd;
use v4l2r::bindings::v4l2_event;
use v4l2r::bindings::v4l2_event_subscription;
use v4l2r::bindings::v4l2_fmtdesc;
use v4l2r::bindings::v4l2_format;
use v4l2r::bindings::v4l2_input;
use v4l2r::bindings::v4l2_output;
use v4l2r::bindings::v4l2_query_ext_ctrl;
use v4l2r::bindings::v4l2_queryctrl;
use v4l2r::bindings::v4l2_rect;
use v4l2r::bindings::v4l2_requestbuffers;
use v4l2r::bindings::v4l2_selection;
use v4l2r::device::DeviceConfig;
use v4l2r::device::DeviceOpenError;
use v4l2r::ioctl::BufferFlags;
use v4l2r::ioctl::DqBufError;
use v4l2r::ioctl::DqEventError;
use v4l2r::ioctl::EventType as V4l2EventType;
use v4l2r::ioctl::ExpbufFlags;
use v4l2r::ioctl::IntoErrno;
use v4l2r::ioctl::QueryCapError;
use v4l2r::ioctl::SelectionFlags;
use v4l2r::ioctl::SelectionTarget;
use v4l2r::ioctl::SelectionType;
use v4l2r::ioctl::SubscribeEventFlags;
use v4l2r::ioctl::V4l2Buffer;
use v4l2r::memory::MemoryType;
use v4l2r::QueueType;
use vm_control::VmMemorySource;
use vm_memory::GuestAddress;
use vm_memory::GuestMemory;
use zerocopy::AsBytes;
use zerocopy::FromBytes;

use crate::virtio::v4l2::protocol::*;
use crate::virtio::DescriptorChain;
use crate::virtio::Queue;
use crate::virtio::Reader;
use crate::virtio::SharedMemoryMapper;
use crate::virtio::SignalableInterrupt;
use crate::virtio::Writer;
use base::error;
use v4l2r::device::Device as V4l2Device;

/// Hack type to allow the use of zerocopy on external types that do not implement it. Used to
/// easily deserialize V4L2 structures from the virtio queues.
#[repr(transparent)]
struct ForceZeroCopy<T: Sized>(T);

unsafe impl<T: Sized> FromBytes for ForceZeroCopy<T> {
    fn only_derive_is_allowed_to_implement_this_trait()
    where
        Self: Sized,
    {
    }
}

unsafe impl<T: Sized> AsBytes for ForceZeroCopy<T> {
    fn only_derive_is_allowed_to_implement_this_trait()
    where
        Self: Sized,
    {
    }
}

struct V4l2Session {
    id: u32,
    device: Arc<V4l2Device>,

    capture_streaming: bool,
    capture_num_queued: usize,

    output_streaming: bool,
    output_num_allocated: usize,
    output_num_queued: usize,
}

impl V4l2Session {
    fn new(id: u32, device: Arc<V4l2Device>) -> Self {
        Self {
            id,
            device,
            capture_streaming: false,
            capture_num_queued: 0,
            output_streaming: false,
            output_num_allocated: 0,
            output_num_queued: 0,
        }
    }

    /// Returns whether this session should be polling for CAPTURE buffers in its current state.
    fn should_poll_capture(&self) -> bool {
        self.capture_streaming && self.capture_num_queued > 0
    }

    /// Enable polling for CAPTURE buffers for this session.
    fn enable_capture_polling(&self, wait_ctx: &mut WaitContext<Token>) -> Result<()> {
        trace!("enabling capture buffer polling");
        wait_ctx
            .modify(
                &Descriptor(self.device.as_raw_fd()),
                EventType::EventRead,
                Token::V4l2Session(self.id),
            )
            .context("while polling for READ events for V4L2 session")
    }

    /// Disable polling for CAPTURE buffers for this session if all conditions are met.
    fn disable_capture_polling(&self, wait_ctx: &mut WaitContext<Token>) -> Result<()> {
        trace!("disabling capture buffer polling");
        wait_ctx
            .modify(
                &Descriptor(self.device.as_raw_fd()),
                EventType::Event,
                Token::V4l2Session(self.id),
            )
            .context("while disabling polling for READ events for V4L2 session")
    }
}

/// Information about a given MMAP memory plane.
///
/// We keep these around indexed by the memory offset in order to service MMAP commands. Only used
/// if the memory type of the queue is MMAP.
struct V4l2MmapPlaneInfo {
    // Queue the buffer belongs to.
    queue: QueueType,
    // Buffer index.
    index: u8,
    // Plane index.
    plane: u8,
    // Length of the plane in bytes.
    length: u32,
    // Number of times this buffer has been mapped.
    map_count: usize,
}

pub struct Worker<I: SignalableInterrupt> {
    mem: GuestMemory,
    shm_addr: GuestAddress,
    shm_mapper: Box<dyn SharedMemoryMapper>,
    cmd_queue: Queue,
    evt_queue: Queue,
    evt_evt: Event,
    cmd_interrupt: I,
    evt_interrupt: I,

    device_path: PathBuf,
    /// Type of the capture queue, if one has been set up.
    capture_queue_type: Option<QueueType>,
    /// Type of the output queue, if one has been set up.
    output_queue_type: Option<QueueType>,

    /// Hashmap of opened sessions, indexed by the FD of their opened device.
    ///
    /// Leaking a host FD to the guest is probably not recommended, but it will do for the purpose
    /// of demonstration.
    sessions: HashMap<u32, V4l2Session>,

    /// Map of memory offsets to detailed buffer information. Only used for queues which memory
    /// type is MMAP.
    mmap_buffers: BTreeMap<u64, V4l2MmapPlaneInfo>,
}

#[derive(EventToken, Debug)]
pub enum Token {
    CommandQueue,
    V4l2Session(u32),
    Kill,
}

impl<I: SignalableInterrupt> Worker<I> {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        mem: GuestMemory,
        shm_addr: GuestAddress,
        shm_mapper: Box<dyn SharedMemoryMapper>,
        cmd_queue: Queue,
        evt_queue: Queue,
        evt_evt: Event,
        cmd_interrupt: I,
        evt_interrupt: I,
        device_path: PathBuf,
    ) -> Self {
        Self {
            mem,
            shm_addr,
            shm_mapper,
            cmd_queue,
            evt_queue,
            evt_evt,
            cmd_interrupt,
            evt_interrupt,
            device_path,
            capture_queue_type: None,
            output_queue_type: None,
            sessions: Default::default(),
            mmap_buffers: Default::default(),
        }
    }

    pub fn run(&mut self, cmd_evt: Event, kill_evt: &Event) -> Result<()> {
        let mut wait_ctx: WaitContext<Token> =
            WaitContext::build_with(&[(&cmd_evt, Token::CommandQueue), (kill_evt, Token::Kill)])
                .context("Wait context creation failed")?;

        loop {
            let wait_events = wait_ctx.wait().context("Wait error")?;

            for wait_event in wait_events.iter() {
                match wait_event.token {
                    Token::CommandQueue => {
                        let _ = cmd_evt.wait();
                        while let Some(desc) = self.cmd_queue.pop(&self.mem) {
                            self.handle_command(&mut wait_ctx, desc)?;
                        }
                    }
                    Token::V4l2Session(stream_id) => {
                        if wait_event.is_pri {
                            self.dequeue_events(stream_id)?;
                        }
                        if wait_event.is_readable {
                            self.dequeue_capture_buffer(stream_id, &mut wait_ctx)?;
                            // Release OUTPUT buffers while we are at it.
                            self.dequeue_output_buffers(stream_id)?;
                        }
                    }
                    Token::Kill => {
                        return Ok(());
                    }
                }
            }
        }
    }

    /// Send a command response to the guest using `desc`.
    fn send_response<T: AsBytes>(&mut self, desc: DescriptorChain, response: T) -> Result<()> {
        let desc_index = desc.index;
        let mut writer = Writer::new(self.mem.clone(), desc)?;
        if let Err(e) = writer.write_obj(response) {
            error!("failed to write response: {}", e);
        }
        self.cmd_queue
            .add_used(&self.mem, desc_index, writer.bytes_written() as u32);
        self.cmd_queue
            .trigger_interrupt(&self.mem, &self.cmd_interrupt);
        Ok(())
    }

    /// Send an ioctl command response to the guest using `desc`.
    fn send_ioctl_response<T: Sized>(&mut self, desc: DescriptorChain, response: &T) -> Result<()> {
        let slice = unsafe {
            std::slice::from_raw_parts(
                (response) as *const _ as *const u8,
                std::mem::size_of_val(response),
            )
        };

        // Write the response.
        let desc_index = desc.index;
        let mut writer = Writer::new(self.mem.clone(), desc)?;
        if let Err(e) = writer.write_obj(IoctlResp::ok()) {
            error!("failed to write response: {}", e);
        }
        if let Err(e) = writer.write_all(slice) {
            error!("failed to write response: {}", e);
        }
        self.cmd_queue
            .add_used(&self.mem, desc_index, writer.bytes_written() as u32);
        self.cmd_queue
            .trigger_interrupt(&self.mem, &self.cmd_interrupt);

        Ok(())
    }

    /// Send `code` as the error code of an error response using `desc`.
    fn send_err_response(&mut self, desc: DescriptorChain, code: libc::c_int) -> Result<()> {
        self.send_response(desc, RespHeader::err(code))
    }

    /// Wait until an event descriptor becomes available and send `event` to the guest.
    fn send_event(&mut self, event: V4l2Event) -> Result<()> {
        self.evt_evt.wait()?;
        let desc = self.evt_queue.pop(&self.mem).unwrap();

        let slice = unsafe {
            match event {
                V4l2Event::Error(event) => std::slice::from_raw_parts(
                    (&event) as *const _ as *const u8,
                    std::mem::size_of_val(&event),
                ),
                V4l2Event::DequeueBuffer(event) => std::slice::from_raw_parts(
                    (&event) as *const _ as *const u8,
                    std::mem::size_of_val(&event),
                ),
                V4l2Event::Event(event) => std::slice::from_raw_parts(
                    (&event) as *const _ as *const u8,
                    std::mem::size_of_val(&event),
                ),
            }
        };

        let desc_index = desc.index;
        let mut writer = Writer::new(self.mem.clone(), desc)?;
        if let Err(e) = writer.write_all(slice) {
            error!("failed to write event: {}", e);
        }
        self.evt_queue
            .add_used(&self.mem, desc_index, writer.bytes_written() as u32);
        self.evt_queue
            .trigger_interrupt(&self.mem, &self.evt_interrupt);

        Ok(())
    }

    fn handle_command(
        &mut self,
        wait_ctx: &mut WaitContext<Token>,
        desc: DescriptorChain,
    ) -> Result<()> {
        let mut reader =
            Reader::new(self.mem.clone(), desc.clone()).context("invalid descriptor chain")?;

        let hdr: CmdHeader = reader.read_obj().context("while reading command header")?;

        match hdr.cmd {
            VIRTIO_V4L2_CMD_OPEN => self.do_open(wait_ctx, desc),
            VIRTIO_V4L2_CMD_CLOSE => {
                let CloseCmd { stream_id, .. } =
                    reader.read_obj().context("while reading CLOSE cmd")?;
                self.do_close(stream_id, wait_ctx, desc)
            }
            VIRTIO_V4L2_CMD_IOCTL => {
                let IoctlCmd { stream_id, code } =
                    reader.read_obj().context("while reading IOCTL cmd")?;
                self.do_ioctl(stream_id, code, reader, wait_ctx, desc)
            }
            VIRTIO_V4L2_CMD_MMAP => {
                let MmapCmd {
                    stream_id,
                    flags,
                    offset,
                } = reader.read_obj().context("while reading MMAP cmd")?;

                self.do_mmap(stream_id, flags, offset, desc)
            }
            VIRTIO_V4L2_CMD_MUNMAP => {
                let MunmapCmd { offset } = reader.read_obj().context("while reading UNMMAP cmd")?;

                self.do_munmap(offset, desc)
            }
            _ => self.send_err_response(desc, libc::ENOTTY),
        }
    }

    fn do_open(&mut self, wait_ctx: &mut WaitContext<Token>, desc: DescriptorChain) -> Result<()> {
        match V4l2Device::open(&self.device_path, DeviceConfig::new().non_blocking_dqbuf()) {
            Ok(device) => {
                // TODO use a proper ID generator to avoid leaking FDs to the guest...
                let stream_id = device.as_raw_fd() as u32;

                wait_ctx
                    .add_for_event(
                        &Descriptor(device.as_raw_fd()),
                        EventType::Event,
                        Token::V4l2Session(stream_id),
                    )
                    .context("while enabling polling for V4L2 events")?;

                self.sessions
                    .insert(stream_id, V4l2Session::new(stream_id, Arc::new(device)));

                self.send_response(desc, OpenResp::ok(stream_id))
            }
            Err(DeviceOpenError::OpenError(e)) => self.send_err_response(desc, e as i32),
            Err(DeviceOpenError::QueryCapError(QueryCapError::IoctlError(e))) => {
                self.send_err_response(desc, e as i32)
            }
        }
    }

    fn do_close(
        &mut self,
        stream_id: u32,
        wait_ctx: &mut WaitContext<Token>,
        desc: DescriptorChain,
    ) -> Result<()> {
        match self.sessions.remove(&stream_id) {
            Some(session) => {
                wait_ctx
                    .delete(&Descriptor(session.device.as_raw_fd()))
                    .context("while removing poll FD for closing session")?;

                self.send_response(desc, CloseResp::ok())
            }
            None => self.send_err_response(desc, libc::EINVAL),
        }
    }

    fn do_mmap(
        &mut self,
        stream_id: u32,
        flags: u32,
        offset: u64,
        desc: DescriptorChain,
    ) -> Result<()> {
        let session = match self.sessions.get_mut(&stream_id) {
            Some(session) => session,
            None => {
                // This can happen if we got a CLOSE command and an event for the same session was pending.
                warn!("session {} does not exist", stream_id);
                return self.send_err_response(desc, libc::EINVAL);
            }
        };

        let rw = (flags & VIRTIO_V4L2_MMAP_FLAG_RW) != 0;

        let plane_info = match self.mmap_buffers.get_mut(&offset) {
            Some(plane_info) => plane_info,
            None => return self.send_err_response(desc, libc::EINVAL),
        };

        if plane_info.map_count == 0 {
            let buf_desc: SafeDescriptor = match v4l2r::ioctl::expbuf::<File>(
                &session.device,
                plane_info.queue,
                plane_info.index as usize,
                plane_info.plane as usize,
                if rw {
                    ExpbufFlags::RDWR
                } else {
                    ExpbufFlags::RDONLY
                },
            ) {
                Ok(desc) => desc.into(),
                Err(e) => return self.send_err_response(desc, e.into_errno()),
            };

            match self.shm_mapper.add_mapping(
                VmMemorySource::Descriptor {
                    descriptor: buf_desc,
                    offset: 0,
                    size: plane_info.length as u64,
                },
                offset,
                if rw {
                    Protection::read_write()
                } else {
                    Protection::read()
                },
            ) {
                Ok(()) => (),
                Err(e) => {
                    warn!(
                        "cannot map host buffer into device shared memory region: {:#}",
                        e
                    );
                    return self.send_err_response(desc, libc::EINVAL);
                }
            }
        }

        plane_info.map_count += 1;
        let length = plane_info.length as u64;

        self.send_response(desc, MmapResp::ok(self.shm_addr.0 + offset, length))
    }

    fn do_munmap(&mut self, offset: u64, desc: DescriptorChain) -> Result<()> {
        let plane_info = match self.mmap_buffers.get_mut(&offset) {
            Some(plane_info) => plane_info,
            None => return self.send_err_response(desc, libc::EINVAL),
        };

        plane_info.map_count -= 1;

        if plane_info.map_count == 0 {
            match self.shm_mapper.remove_mapping(offset) {
                Ok(()) => (),
                Err(e) => {
                    warn!(
                        "could not unmap host buffer with offset 0x{:x}: {:#}",
                        offset, e
                    );
                    return self.send_err_response(desc, libc::EINVAL);
                }
            }
        }

        self.send_response(desc, MunmapResp::ok())
    }

    fn do_ioctl(
        &mut self,
        stream_id: u32,
        code: u32,
        mut reader: Reader,
        wait_ctx: &mut WaitContext<Token>,
        desc: DescriptorChain,
    ) -> Result<()> {
        trace!("V4L2 device do_ioctl {:x} {}", stream_id, code);

        let session = match self.sessions.get_mut(&stream_id) {
            Some(session) => session,
            None => {
                // This can happen if we got a CLOSE command and an event for the same session was pending.
                warn!("session {} does not exist", stream_id);
                return self.send_err_response(desc, libc::EINVAL);
            }
        };

        match code {
            VIRTIO_V4L2_VIDIOC_ENUM_FMT => {
                let enum_fmt = reader
                    .read_obj::<ForceZeroCopy<v4l2_fmtdesc>>()
                    .context("while reading ENUM_FMT data")?
                    .0;

                let queue_type = match QueueType::n(enum_fmt.type_) {
                    Some(queue_type) => queue_type,
                    None => {
                        error!("unrecognized queue type specified: {}", enum_fmt.type_);
                        return self.send_err_response(desc, libc::EINVAL);
                    }
                };

                let enum_fmt: v4l2_fmtdesc =
                    match v4l2r::ioctl::enum_fmt(&session.device, queue_type, enum_fmt.index) {
                        Ok(enum_fmt) => enum_fmt,
                        Err(e) => return self.send_err_response(desc, e.into_errno()),
                    };

                self.send_ioctl_response(desc, &enum_fmt)
            }
            VIRTIO_V4L2_VIDIOC_G_FMT => {
                let format = reader
                    .read_obj::<ForceZeroCopy<v4l2_format>>()
                    .context("while reading G_FMT data")?
                    .0;

                let queue_type = match QueueType::n(format.type_) {
                    Some(queue_type) => queue_type,
                    None => {
                        error!("unrecognized queue type specified: {}", format.type_);
                        return self.send_err_response(desc, libc::EINVAL);
                    }
                };

                let format: v4l2_format = match v4l2r::ioctl::g_fmt(&session.device, queue_type) {
                    Ok(format) => format,
                    Err(e) => return self.send_err_response(desc, e.into_errno()),
                };

                self.send_ioctl_response(desc, &format)
            }
            VIRTIO_V4L2_VIDIOC_S_FMT => {
                let format = reader
                    .read_obj::<ForceZeroCopy<v4l2_format>>()
                    .context("while reading S_FMT data")?
                    .0;

                let format: v4l2_format = match v4l2r::ioctl::s_fmt(&mut session.device, format) {
                    Ok(format) => format,
                    Err(e) => return self.send_err_response(desc, e.into_errno()),
                };

                self.send_ioctl_response(desc, &format)
            }
            VIRTIO_V4L2_VIDIOC_REQBUFS => {
                let reqbufs = reader
                    .read_obj::<ForceZeroCopy<v4l2_requestbuffers>>()
                    .context("while reading REQBUFS data")?
                    .0;

                let queue = QueueType::n(reqbufs.type_).unwrap();
                let memory = MemoryType::n(reqbufs.memory).unwrap();

                let reqbufs: v4l2_requestbuffers =
                    match v4l2r::ioctl::reqbufs(&session.device, queue, memory, reqbufs.count) {
                        Ok(reqbufs) => reqbufs,
                        Err(e) => return self.send_err_response(desc, e.into_errno()),
                    };

                // Clear all the previous buffers for this queue, and insert new information if the
                // memory type is MMAP.
                self.mmap_buffers.retain(|_, b| b.queue != queue);
                if memory == MemoryType::Mmap {
                    for i in 0..reqbufs.count {
                        let buffer = match v4l2r::ioctl::querybuf::<V4l2Buffer>(
                            &session.device,
                            queue,
                            i as usize,
                        ) {
                            Ok(buffer) => buffer,
                            Err(e) => {
                                warn!("failed to query newly allocated buffer: {:#}", e);
                                continue;
                            }
                        };
                        // TODO(v4l2r) have an iterator instead?
                        for j in 0..buffer.num_planes() {
                            let plane = buffer.get_plane(j).unwrap();
                            let offset = match plane.mem_offset() {
                                Some(offset) => offset,
                                None => {
                                    warn!("expected a plane memory offset but none present");
                                    continue;
                                }
                            };

                            self.mmap_buffers.insert(
                                offset as u64,
                                V4l2MmapPlaneInfo {
                                    queue,
                                    index: buffer.index() as u8,
                                    plane: j as u8,
                                    length: plane.length(),
                                    map_count: 0,
                                },
                            );
                        }
                    }
                }

                // If we allocated on the capture or output queue successfully, remember its type.
                match queue {
                    QueueType::VideoCapture | QueueType::VideoCaptureMplane => {
                        self.capture_queue_type = Some(queue);

                        // REQBUFS(0) is an implicit STREAMOFF.
                        if reqbufs.count == 0 {
                            let was_polling_capture = session.should_poll_capture();
                            session.capture_streaming = false;
                            session.capture_num_queued = 0;
                            if was_polling_capture {
                                session.disable_capture_polling(wait_ctx)?;
                            }
                        }
                    }
                    QueueType::VideoOutput | QueueType::VideoOutputMplane => {
                        self.output_queue_type = Some(queue);
                        session.output_num_allocated = reqbufs.count as usize;
                        //
                        // REQBUFS(0) is an implicit STREAMOFF.
                        if reqbufs.count == 0 {
                            session.output_streaming = false;
                            session.output_num_queued = 0;
                        }
                    }
                    _ => (),
                }

                self.send_ioctl_response(desc, &reqbufs)
            }
            VIRTIO_V4L2_VIDIOC_QUERYBUF => {
                let buffer = reader
                    .read_obj::<ForceZeroCopy<V4l2Buffer>>()
                    .context("while reading QUERYBUF data")?
                    .0;

                let queue = QueueType::n(buffer.queue_type()).unwrap();
                let index = buffer.index() as usize;

                match v4l2r::ioctl::querybuf::<V4l2Buffer>(&session.device, queue, index) {
                    Ok(buffer) => self.send_ioctl_response(desc, &buffer),
                    Err(e) => self.send_err_response(desc, e.into_errno()),
                }
            }
            VIRTIO_V4L2_VIDIOC_QBUF => {
                let buffer = reader
                    .read_obj::<ForceZeroCopy<V4l2Buffer>>()
                    .context("while reading QBUF data")?
                    .0;

                let queue = QueueType::n(buffer.queue_type()).unwrap();
                let index = buffer.index() as usize;

                let buffer = match v4l2r::ioctl::qbuf::<_, V4l2Buffer>(
                    &session.device,
                    queue,
                    index,
                    buffer,
                ) {
                    Ok(buffer) => buffer,
                    Err(e) => return self.send_err_response(desc, e.into_errno()),
                };

                match queue {
                    QueueType::VideoCapture | QueueType::VideoCaptureMplane => {
                        let was_polling_capture = session.should_poll_capture();
                        session.capture_num_queued += 1;
                        if !was_polling_capture && session.should_poll_capture() {
                            session.enable_capture_polling(wait_ctx)?;
                        }
                    }
                    QueueType::VideoOutput | QueueType::VideoOutputMplane => {
                        session.output_num_queued += 1;
                        // Proactively dequeue output buffers we are done with.
                        self.dequeue_output_buffers(stream_id)?;
                    }
                    _ => (),
                }

                self.send_ioctl_response(desc, &buffer)
            }
            VIRTIO_V4L2_VIDIOC_ENUMINPUT => {
                let input = reader
                    .read_obj::<ForceZeroCopy<v4l2_input>>()
                    .context("while reading ENUMINPUT data")?
                    .0;

                match v4l2r::ioctl::enuminput::<v4l2_input>(&session.device, input.index as usize) {
                    Ok(input) => self.send_ioctl_response(desc, &input),
                    Err(e) => self.send_err_response(desc, e.into_errno()),
                }
            }
            VIRTIO_V4L2_VIDIOC_G_INPUT => match v4l2r::ioctl::g_input(&session.device) {
                Ok(index) => self.send_ioctl_response(desc, &(index as u32)),
                Err(e) => self.send_err_response(desc, e.into_errno()),
            },
            VIRTIO_V4L2_VIDIOC_S_INPUT => {
                let index = reader
                    .read_obj::<ForceZeroCopy<u32>>()
                    .context("while reading S_INPUT data")?
                    .0;

                match v4l2r::ioctl::s_input(&session.device, index as usize) {
                    Ok(()) => self.send_response(desc, IoctlResp::ok()),
                    Err(e) => self.send_err_response(desc, e.into_errno()),
                }
            }
            VIRTIO_V4L2_VIDIOC_ENUMOUTPUT => {
                let output = reader
                    .read_obj::<ForceZeroCopy<v4l2_output>>()
                    .context("while reading ENUMOUTPUT data")?
                    .0;

                match v4l2r::ioctl::enumoutput::<v4l2_output>(
                    &session.device,
                    output.index as usize,
                ) {
                    Ok(output) => self.send_ioctl_response(desc, &output),
                    Err(e) => self.send_err_response(desc, e.into_errno()),
                }
            }
            VIRTIO_V4L2_VIDIOC_G_OUTPUT => match v4l2r::ioctl::g_output(&session.device) {
                Ok(index) => self.send_ioctl_response(desc, &(index as u32)),
                Err(e) => self.send_err_response(desc, e.into_errno()),
            },
            VIRTIO_V4L2_VIDIOC_S_OUTPUT => {
                let index = reader
                    .read_obj::<ForceZeroCopy<u32>>()
                    .context("while reading S_OUTPUT data")?
                    .0;

                match v4l2r::ioctl::s_output(&session.device, index as usize) {
                    Ok(()) => self.send_response(desc, IoctlResp::ok()),
                    Err(e) => self.send_err_response(desc, e.into_errno()),
                }
            }
            VIRTIO_V4L2_VIDIOC_STREAMON => {
                let queue_id = reader
                    .read_obj::<ForceZeroCopy<u32>>()
                    .context("while reading STREAMON data")?
                    .0;

                let queue = QueueType::n(queue_id).unwrap();
                match v4l2r::ioctl::streamon(&session.device, queue) {
                    Ok(()) => (),
                    Err(e) => return self.send_err_response(desc, e.into_errno()),
                }

                match queue {
                    QueueType::VideoCapture | QueueType::VideoCaptureMplane
                        if !session.capture_streaming =>
                    {
                        session.capture_streaming = true;
                        if session.should_poll_capture() {
                            session.enable_capture_polling(wait_ctx)?;
                        }
                    }
                    QueueType::VideoOutput | QueueType::VideoOutputMplane
                        if !session.output_streaming =>
                    {
                        session.output_streaming = true;
                    }
                    _ => (),
                }

                self.send_response(desc, IoctlResp::ok())
            }
            VIRTIO_V4L2_VIDIOC_STREAMOFF => {
                let queue_id = reader
                    .read_obj::<ForceZeroCopy<u32>>()
                    .context("while reading STREAMON data")?
                    .0;

                let queue = QueueType::n(queue_id).unwrap();
                match v4l2r::ioctl::streamoff(&session.device, queue) {
                    Ok(()) => (),
                    Err(e) => return self.send_err_response(desc, e.into_errno()),
                }

                match queue {
                    QueueType::VideoCapture | QueueType::VideoCaptureMplane => {
                        let was_polling_capture = session.should_poll_capture();
                        session.capture_streaming = false;
                        session.capture_num_queued = 0;
                        if was_polling_capture {
                            session.disable_capture_polling(wait_ctx)?;
                        }
                    }
                    QueueType::VideoOutput | QueueType::VideoOutputMplane => {
                        session.output_streaming = false;
                        session.output_num_queued = 0;
                    }
                    _ => (),
                }

                self.send_response(desc, IoctlResp::ok())
            }
            VIRTIO_V4L2_VIDIOC_QUERYCTRL => {
                let ctrl = reader
                    .read_obj::<ForceZeroCopy<v4l2_queryctrl>>()
                    .context("while reading QUERYCTRL data")?
                    .0;
                let (id, flags) = v4l2r::ioctl::parse_ctrl_id_and_flags(ctrl.id);

                let ctrl: v4l2_queryctrl = match v4l2r::ioctl::queryctrl(&session.device, id, flags)
                {
                    Ok(ctrl) => ctrl,
                    Err(e) => return self.send_err_response(desc, e.into_errno()),
                };

                self.send_ioctl_response(desc, &ctrl)
            }
            VIRTIO_V4L2_VIDIOC_TRY_FMT => {
                let format = reader
                    .read_obj::<ForceZeroCopy<v4l2_format>>()
                    .context("while reading TRY_FMT data")?
                    .0;

                let format: v4l2_format = match v4l2r::ioctl::try_fmt(&session.device, format) {
                    Ok(format) => format,
                    Err(e) => return self.send_err_response(desc, e.into_errno()),
                };

                self.send_ioctl_response(desc, &format)
            }
            VIRTIO_V4L2_VIDIOC_ENCODER_CMD => {
                let cmd = reader
                    .read_obj::<ForceZeroCopy<v4l2_encoder_cmd>>()
                    .context("while reading ENCODER_CMD data")?
                    .0;

                let cmd: v4l2_encoder_cmd = match v4l2r::ioctl::encoder_cmd(&session.device, cmd) {
                    Ok(cmd) => cmd,
                    Err(e) => return self.send_err_response(desc, e.into_errno()),
                };

                self.send_ioctl_response(desc, &cmd)
            }
            VIRTIO_V4L2_VIDIOC_TRY_ENCODER_CMD => {
                let cmd = reader
                    .read_obj::<ForceZeroCopy<v4l2_encoder_cmd>>()
                    .context("while reading TRY_ENCODER_CMD data")?
                    .0;

                let cmd: v4l2_encoder_cmd =
                    match v4l2r::ioctl::try_encoder_cmd(&session.device, cmd) {
                        Ok(cmd) => cmd,
                        Err(e) => return self.send_err_response(desc, e.into_errno()),
                    };

                self.send_ioctl_response(desc, &cmd)
            }
            VIRTIO_V4L2_VIDIOC_SUBSCRIBE_EVENT => {
                let subscription = reader
                    .read_obj::<ForceZeroCopy<v4l2_event_subscription>>()
                    .context("while reading SUBSCRIBE_EVENT data")?
                    .0;

                let event = V4l2EventType::try_from(&subscription).unwrap();
                let flags = SubscribeEventFlags::from_bits(subscription.flags).unwrap();

                match v4l2r::ioctl::subscribe_event(&session.device, event, flags) {
                    Ok(()) => (),
                    Err(e) => return self.send_err_response(desc, e.into_errno()),
                }

                self.send_response(desc, IoctlResp::ok())
            }
            VIRTIO_V4L2_VIDIOC_UNSUBSCRIBE_EVENT => {
                let subscription = reader
                    .read_obj::<ForceZeroCopy<v4l2_event_subscription>>()
                    .context("while reading UNSUBSCRIBE_EVENT data")?
                    .0;

                let err = if subscription.type_ == v4l2r::bindings::V4L2_EVENT_ALL {
                    v4l2r::ioctl::unsubscribe_all_events(&session.device)
                } else {
                    let event = V4l2EventType::try_from(&subscription).unwrap();

                    v4l2r::ioctl::unsubscribe_event(&session.device, event)
                };

                match err {
                    Ok(()) => self.send_response(desc, IoctlResp::ok()),
                    Err(e) => self.send_err_response(desc, e.into_errno()),
                }
            }
            VIRTIO_V4L2_VIDIOC_G_SELECTION => {
                let mut selection = reader
                    .read_obj::<ForceZeroCopy<v4l2_selection>>()
                    .context("while reading G_SELECTION data")?
                    .0;
                let sel_type = match SelectionType::n(selection.type_) {
                    Some(sel_type) => sel_type,
                    None => return self.send_err_response(desc, libc::EINVAL),
                };
                let sel_target = match SelectionTarget::n(selection.target) {
                    Some(sel_target) => sel_target,
                    None => return self.send_err_response(desc, libc::EINVAL),
                };

                let rect: v4l2_rect =
                    match v4l2r::ioctl::g_selection(&session.device, sel_type, sel_target) {
                        Ok(selection) => selection,
                        Err(e) => return self.send_err_response(desc, e.into_errno()),
                    };

                selection.r = rect;

                self.send_ioctl_response(desc, &selection)
            }
            VIRTIO_V4L2_VIDIOC_S_SELECTION => {
                let mut selection = reader
                    .read_obj::<ForceZeroCopy<v4l2_selection>>()
                    .context("while reading G_SELECTION data")?
                    .0;
                let sel_type = match SelectionType::n(selection.type_) {
                    Some(sel_type) => sel_type,
                    None => return self.send_err_response(desc, libc::EINVAL),
                };
                let sel_target = match SelectionTarget::n(selection.target) {
                    Some(sel_target) => sel_target,
                    None => return self.send_err_response(desc, libc::EINVAL),
                };
                let sel_flags = match SelectionFlags::from_bits(selection.flags) {
                    Some(sel_flags) => sel_flags,
                    None => return self.send_err_response(desc, libc::EINVAL),
                };

                let rect: v4l2_rect = match v4l2r::ioctl::s_selection(
                    &session.device,
                    sel_type,
                    sel_target,
                    selection.r,
                    sel_flags,
                ) {
                    Ok(selection) => selection,
                    Err(e) => return self.send_err_response(desc, e.into_errno()),
                };

                selection.r = rect;

                self.send_ioctl_response(desc, &selection)
            }
            VIRTIO_V4L2_VIDIOC_DECODER_CMD => {
                let cmd = reader
                    .read_obj::<ForceZeroCopy<v4l2_decoder_cmd>>()
                    .context("while reading DECODER_CMD data")?
                    .0;

                let cmd: v4l2_decoder_cmd = match v4l2r::ioctl::decoder_cmd(&session.device, cmd) {
                    Ok(cmd) => cmd,
                    Err(e) => return self.send_err_response(desc, e.into_errno()),
                };

                self.send_ioctl_response(desc, &cmd)
            }
            VIRTIO_V4L2_VIDIOC_TRY_DECODER_CMD => {
                let cmd = reader
                    .read_obj::<ForceZeroCopy<v4l2_decoder_cmd>>()
                    .context("while reading TRY_DECODER_CMD data")?
                    .0;

                let cmd: v4l2_decoder_cmd =
                    match v4l2r::ioctl::try_decoder_cmd(&session.device, cmd) {
                        Ok(cmd) => cmd,
                        Err(e) => return self.send_err_response(desc, e.into_errno()),
                    };

                self.send_ioctl_response(desc, &cmd)
            }
            VIRTIO_V4L2_VIDIOC_QUERY_EXT_CTRL => {
                let ctrl = reader
                    .read_obj::<ForceZeroCopy<v4l2_query_ext_ctrl>>()
                    .context("while reading QUERY_EXT_CTRL data")?
                    .0;
                let (id, flags) = v4l2r::ioctl::parse_ctrl_id_and_flags(ctrl.id);

                let ctrl: v4l2_query_ext_ctrl =
                    match v4l2r::ioctl::query_ext_ctrl(&session.device, id, flags) {
                        Ok(ctrl) => ctrl,
                        Err(e) => return self.send_err_response(desc, e.into_errno()),
                    };

                self.send_ioctl_response(desc, &ctrl)
            }
            _ => unimplemented!(),
        }
    }

    fn dequeue_capture_buffer(
        &mut self,
        stream_id: u32,
        wait_ctx: &mut WaitContext<Token>,
    ) -> Result<()> {
        let capture_queue_type = match self.capture_queue_type {
            Some(queue_type) => queue_type,
            None => {
                warn!("request to dequeue a capture buffer while none has been allocated");
                return Ok(());
            }
        };

        let session = match self.sessions.get_mut(&stream_id) {
            Some(session) => session,
            None => {
                // This can happen if we got a CLOSE command and an event for the same session was pending.
                warn!(
                    "session {} does not exist, cancelling capture buffer dequeue",
                    stream_id
                );
                return Ok(());
            }
        };

        let v4l2_buffer =
            match v4l2r::ioctl::dqbuf::<V4l2Buffer>(&session.device, capture_queue_type) {
                Ok(buffer) => buffer,
                Err(DqBufError::Eos) => return Ok(()),
                Err(DqBufError::NotReady) => return Ok(()),
                Err(e) => {
                    return self
                        .send_event(V4l2Event::Error(ErrorEvent::new(stream_id, e.into_errno())));
                }
            };

        trace!("dequeued capture buffer {} on host", v4l2_buffer.index());

        // This may or may not be needed...
        if v4l2_buffer.flags().contains(BufferFlags::LAST) {
            session.disable_capture_polling(wait_ctx)?;
        }

        let capture_polling_active = session.should_poll_capture();
        session.capture_num_queued -= 1;
        if capture_polling_active && session.capture_num_queued == 0 {
            session.disable_capture_polling(wait_ctx)?;
        }

        self.send_event(V4l2Event::DequeueBuffer(DequeueBufferEvent::new(
            stream_id,
            v4l2_buffer,
        )))
    }

    fn dequeue_output_buffers(&mut self, stream_id: u32) -> Result<()> {
        let output_queue_type = match self.output_queue_type {
            Some(queue_type) => queue_type,
            None => return Ok(()),
        };

        let session = match self.sessions.get_mut(&stream_id) {
            Some(session) => session,
            None => {
                // This can happen if we got a CLOSE command and an event for the same session was pending.
                return Ok(());
            }
        };

        if !session.output_streaming || session.output_num_queued == 0 {
            return Ok(());
        }

        let session_id = session.id;
        // Unfortunate but required to make the borrow checker happy, and cheap anyway.
        let device = Arc::clone(&session.device);

        loop {
            match v4l2r::ioctl::dqbuf::<V4l2Buffer>(&device, output_queue_type) {
                Ok(buffer) => self.send_event(V4l2Event::DequeueBuffer(
                    DequeueBufferEvent::new(session_id, buffer),
                ))?,
                Err(DqBufError::Eos) | Err(DqBufError::NotReady) => return Ok(()),
                Err(e) => {
                    return self.send_event(V4l2Event::Error(ErrorEvent::new(
                        session_id,
                        e.into_errno(),
                    )));
                }
            };
        }
    }

    fn dequeue_events(&mut self, stream_id: u32) -> Result<()> {
        let session = match self.sessions.get_mut(&stream_id) {
            Some(session) => session,
            None => {
                // This can happen if we got a CLOSE command and an event for the same session was pending.
                return Ok(());
            }
        };

        let session_id = session.id;
        // Unfortunate but required to make the borrow checker happy, and cheap anyway.
        let device = Arc::clone(&session.device);

        loop {
            match v4l2r::ioctl::dqevent::<v4l2_event>(&device) {
                Ok(event) => {
                    self.send_event(V4l2Event::Event(StreamEvent::new(session_id, event)))?
                }
                Err(DqEventError::NotReady) => break,
                Err(e) => {
                    return self.send_event(V4l2Event::Error(ErrorEvent::new(
                        session_id,
                        e.into_errno(),
                    )));
                }
            }
        }

        Ok(())
    }
}