pub mod buffer;
pub mod direction;
pub mod dqbuf;
pub mod generic;
pub mod handles_provider;
pub mod qbuf;
use super::{AllocatedQueue, Device, FreeBuffersResult, Stream, TryDequeue};
use crate::ioctl::{DqBufResult, MemoryConsistency, QueryBufError, V4l2BufferFromError};
use crate::{bindings, memory::*};
use crate::{
ioctl::{
self, GFmtError, QueryBuffer, ReqbufsError, SFmtError, SelectionTarget, SelectionType,
StreamOffError, StreamOnError, TryFmtError,
},
PlaneLayout, Rect,
};
use crate::{Format, PixelFormat, QueueType};
use buffer::*;
use direction::*;
use dqbuf::*;
use log::debug;
use qbuf::*;
use std::convert::{Infallible, TryFrom};
use std::os::unix::io::{AsRawFd, RawFd};
use std::sync::{Arc, Weak};
use thiserror::Error;
struct QueueBase {
device: Arc<Device>,
type_: QueueType,
capabilities: ioctl::BufferCapabilities,
}
impl AsRawFd for QueueBase {
fn as_raw_fd(&self) -> RawFd {
self.device.as_raw_fd()
}
}
impl Drop for QueueBase {
fn drop(&mut self) {
assert!(self.device.used_queues.lock().unwrap().remove(&self.type_));
}
}
pub trait QueueState {}
pub struct Queue<D, S>
where
D: Direction,
S: QueueState,
{
inner: QueueBase,
_d: std::marker::PhantomData<D>,
state: S,
}
impl<D, S> Queue<D, S>
where
D: Direction,
S: QueueState,
{
pub fn get_capabilities(&self) -> ioctl::BufferCapabilities {
self.inner.capabilities
}
pub fn get_type(&self) -> QueueType {
self.inner.type_
}
pub fn get_format<T: TryFrom<bindings::v4l2_format>>(&self) -> Result<T, GFmtError> {
ioctl::g_fmt(&self.inner, self.inner.type_)
}
pub fn set_format(&mut self, format: Format) -> Result<Format, SFmtError> {
let type_ = self.inner.type_;
ioctl::s_fmt(&mut self.inner, (type_, &format))
}
pub fn try_format(&self, format: Format) -> Result<Format, TryFmtError> {
ioctl::try_fmt(&self.inner, (self.inner.type_, &format))
}
pub fn change_format(&mut self) -> Result<FormatBuilder, GFmtError> {
FormatBuilder::new(&mut self.inner)
}
pub fn format_iter(&self) -> ioctl::FormatIterator<Device> {
ioctl::FormatIterator::new(self.inner.device.as_ref(), self.inner.type_)
}
pub fn get_selection(&self, target: SelectionTarget) -> Result<Rect, ioctl::GSelectionError> {
let selection = match self.get_type() {
QueueType::VideoCapture | QueueType::VideoCaptureMplane => SelectionType::Capture,
QueueType::VideoOutput | QueueType::VideoOutputMplane => SelectionType::Output,
_ => return Err(ioctl::GSelectionError::Invalid),
};
ioctl::g_selection(&self.inner, selection, target)
}
}
pub struct FormatBuilder<'a> {
queue: &'a mut QueueBase,
format: Format,
}
impl<'a> FormatBuilder<'a> {
fn new(queue: &'a mut QueueBase) -> Result<Self, GFmtError> {
let format = ioctl::g_fmt(queue, queue.type_)?;
Ok(Self { queue, format })
}
pub fn format(&self) -> &Format {
&self.format
}
pub fn set_size(mut self, width: usize, height: usize) -> Self {
self.format.width = width as u32;
self.format.height = height as u32;
self
}
pub fn set_pixelformat(mut self, pixel_format: impl Into<PixelFormat>) -> Self {
self.format.pixelformat = pixel_format.into();
self
}
pub fn set_planes_layout<P: IntoIterator<Item = PlaneLayout>>(mut self, planes: P) -> Self {
self.format.plane_fmt = planes.into_iter().collect();
self
}
pub fn apply<O: TryFrom<bindings::v4l2_format>>(self) -> Result<O, SFmtError> {
ioctl::s_fmt(self.queue, (self.queue.type_, &self.format))
}
pub fn try_apply(&mut self) -> Result<(), TryFmtError> {
let new_format = ioctl::try_fmt(self.queue, (self.queue.type_, &self.format))?;
self.format = new_format;
Ok(())
}
}
pub struct QueueInit;
impl QueueState for QueueInit {}
#[derive(Debug, Error)]
pub enum CreateQueueError {
#[error("queue is already in use")]
AlreadyBorrowed,
#[error("error while querying queue capabilities")]
ReqbufsError(#[from] ioctl::ReqbufsError),
}
#[derive(Debug, Error)]
pub enum RequestBuffersError {
#[error("error while requesting buffers")]
ReqbufsError(#[from] ioctl::ReqbufsError),
#[error("error while querying buffer")]
QueryBufferError(#[from] QueryBufError<Infallible>),
}
impl<D: Direction> Queue<D, QueueInit> {
fn create(
device: Arc<Device>,
queue_type: QueueType,
) -> Result<Queue<D, QueueInit>, CreateQueueError> {
let mut used_queues = device.used_queues.lock().unwrap();
if used_queues.contains(&queue_type) {
return Err(CreateQueueError::AlreadyBorrowed);
}
let capabilities: ioctl::BufferCapabilities = ioctl::reqbufs(
&*device,
queue_type,
MemoryType::Mmap,
0,
MemoryConsistency::empty(),
)
.or_else(|e| match e {
ReqbufsError::InvalidBufferType(_, _) => ioctl::reqbufs(
&*device,
queue_type,
MemoryType::DmaBuf,
0,
MemoryConsistency::empty(),
),
_ => Err(e),
})?;
used_queues.insert(queue_type);
drop(used_queues);
Ok(Queue::<D, QueueInit> {
inner: QueueBase {
device,
type_: queue_type,
capabilities,
},
_d: std::marker::PhantomData,
state: QueueInit {},
})
}
pub fn request_buffers_generic_with_flags<P: BufferHandles>(
self,
memory_type: P::SupportedMemoryType,
count: u32,
flags: MemoryConsistency,
) -> Result<Queue<D, BuffersAllocated<P>>, RequestBuffersError> {
let type_ = self.inner.type_;
let num_buffers: usize =
ioctl::reqbufs(&self.inner, type_, memory_type.into(), count, flags)?;
debug!(
"Requested {} buffers on {} queue, obtained {}",
count, type_, num_buffers
);
let mut buffer_features = Vec::new();
for i in 0..num_buffers {
buffer_features.push(ioctl::querybuf(&self.inner, self.inner.type_, i)?);
}
let buffer_stats = Arc::new(BufferStats::new());
let buffer_info = buffer_features
.into_iter()
.map(|features: QueryBuffer| {
Arc::new(BufferInfo::new(features, Arc::clone(&buffer_stats)))
})
.collect();
Ok(Queue {
inner: self.inner,
_d: std::marker::PhantomData,
state: BuffersAllocated {
memory_type,
buffer_info,
buffer_stats,
},
})
}
pub fn request_buffers_generic<P: BufferHandles>(
self,
memory_type: P::SupportedMemoryType,
count: u32,
) -> Result<Queue<D, BuffersAllocated<P>>, RequestBuffersError> {
self.request_buffers_generic_with_flags(memory_type, count, MemoryConsistency::empty())
}
pub fn request_buffers<P: PrimitiveBufferHandles>(
self,
count: u32,
) -> Result<Queue<D, BuffersAllocated<P>>, RequestBuffersError> {
self.request_buffers_generic_with_flags(P::MEMORY_TYPE, count, MemoryConsistency::empty())
}
}
impl Queue<Output, QueueInit> {
pub fn get_output_queue(device: Arc<Device>) -> Result<Self, CreateQueueError> {
Queue::<Output, QueueInit>::create(device, QueueType::VideoOutput)
}
pub fn get_output_mplane_queue(device: Arc<Device>) -> Result<Self, CreateQueueError> {
Queue::<Output, QueueInit>::create(device, QueueType::VideoOutputMplane)
}
}
impl Queue<Capture, QueueInit> {
pub fn get_capture_queue(device: Arc<Device>) -> Result<Self, CreateQueueError> {
Queue::<Capture, QueueInit>::create(device, QueueType::VideoCapture)
}
pub fn get_capture_mplane_queue(device: Arc<Device>) -> Result<Self, CreateQueueError> {
Queue::<Capture, QueueInit>::create(device, QueueType::VideoCaptureMplane)
}
}
pub struct BuffersAllocated<P: BufferHandles> {
memory_type: P::SupportedMemoryType,
buffer_info: Vec<Arc<BufferInfo<P>>>,
buffer_stats: Arc<BufferStats>,
}
impl<P: BufferHandles> QueueState for BuffersAllocated<P> {}
impl<D: Direction, P: BufferHandles> Queue<D, BuffersAllocated<P>> {
fn cancel_queued_buffers(&self) -> Vec<CanceledBuffer<P>> {
let canceled_buffers: Vec<_> = self
.state
.buffer_info
.iter()
.filter_map(|buffer_info| {
let plane_handles = buffer_info.update_state(|state| {
match *state {
BufferState::Queued(_) => {
match std::mem::replace(state, BufferState::Free) {
BufferState::Queued(handles) => Some(handles),
_ => unreachable!(),
}
}
_ => None,
}
})?;
Some(CanceledBuffer {
index: buffer_info.features.index as u32,
plane_handles,
})
})
.collect();
debug!(
"{} buffers canceled on {} queue",
canceled_buffers.len(),
self.get_type()
);
assert_eq!(self.state.buffer_stats.num_queued(), 0);
canceled_buffers
}
fn try_obtain_buffer(&self, index: usize) -> Result<&Arc<BufferInfo<P>>, TryGetBufferError> {
let buffer_info = self
.state
.buffer_info
.get(index)
.ok_or(TryGetBufferError::InvalidIndex(index))?;
buffer_info.update_state(|state| match *state {
BufferState::Free => {
*state = BufferState::PreQueue;
Ok(())
}
_ => Err(TryGetBufferError::AlreadyUsed),
})?;
Ok(buffer_info)
}
}
impl<'a, D: Direction, P: BufferHandles + 'a> AllocatedQueue<'a, D>
for Queue<D, BuffersAllocated<P>>
{
fn num_buffers(&self) -> usize {
self.state.buffer_info.len()
}
fn num_queued_buffers(&self) -> usize {
self.state.buffer_stats.num_queued()
}
fn num_free_buffers(&self) -> usize {
self.state.buffer_stats.num_free()
}
fn free_buffers(self) -> Result<FreeBuffersResult<D, Self>, ioctl::ReqbufsError> {
let type_ = self.inner.type_;
ioctl::reqbufs::<()>(
&self.inner,
type_,
self.state.memory_type.into(),
0,
MemoryConsistency::empty(),
)?;
debug!("Freed all buffers on {} queue", type_);
let canceled_buffers = self.cancel_queued_buffers();
Ok(FreeBuffersResult {
queue: Queue {
inner: self.inner,
_d: std::marker::PhantomData,
state: QueueInit {},
},
canceled_buffers,
})
}
}
pub struct CanceledBuffer<P: BufferHandles> {
pub index: u32,
pub plane_handles: P,
}
impl<D: Direction, P: BufferHandles> Stream for Queue<D, BuffersAllocated<P>> {
type Canceled = CanceledBuffer<P>;
fn stream_on(&self) -> Result<(), StreamOnError> {
debug!("{} queue streaming on", self.get_type());
let type_ = self.inner.type_;
ioctl::streamon(&self.inner, type_)
}
fn stream_off(&self) -> Result<Vec<Self::Canceled>, StreamOffError> {
debug!("{} queue streaming off", self.get_type());
let type_ = self.inner.type_;
ioctl::streamoff(&self.inner, type_)?;
Ok(self.cancel_queued_buffers())
}
}
impl<D: Direction, P: BufferHandles> TryDequeue for Queue<D, BuffersAllocated<P>> {
type Dequeued = DqBuffer<D, P>;
fn try_dequeue(&self) -> DqBufResult<Self::Dequeued, V4l2BufferFromError> {
let dqbuf: ioctl::V4l2Buffer = ioctl::dqbuf(&self.inner, self.inner.type_)?;
let id = dqbuf.index() as usize;
let buffer_info = self
.state
.buffer_info
.get(id)
.expect("Inconsistent buffer state!");
let plane_handles = buffer_info.update_state(|state| match *state {
BufferState::Queued(_) => {
match std::mem::replace(state, BufferState::Dequeued) {
BufferState::Queued(handles) => handles,
_ => unreachable!(),
}
}
_ => unreachable!("Inconsistent buffer state!"),
});
let fuse = BufferStateFuse::new(Arc::downgrade(buffer_info));
let dqbuffer = DqBuffer::new(self, buffer_info, plane_handles, dqbuf, fuse);
Ok(dqbuffer)
}
}
#[derive(Debug, Error)]
pub enum TryGetBufferError {
#[error("buffer with provided index {0} does not exist")]
InvalidIndex(usize),
#[error("buffer is already in use")]
AlreadyUsed,
}
#[derive(Debug, Error)]
pub enum GetFreeBufferError {
#[error("all buffers are currently being used")]
NoFreeBuffer,
}
mod private {
use std::ops::Deref;
use super::*;
pub trait QueueableProvider<'a> {
type Queueable;
}
pub trait GetBufferByIndex<'a>: QueueableProvider<'a> {
fn try_get_buffer(&'a self, index: usize) -> Result<Self::Queueable, TryGetBufferError>;
}
pub trait GetFreeBuffer<'a>: QueueableProvider<'a> {
fn try_get_free_buffer(&'a self) -> Result<Self::Queueable, GetFreeBufferError>;
}
impl<'a, D: Direction, P: PrimitiveBufferHandles> QueueableProvider<'a>
for Queue<D, BuffersAllocated<P>>
{
type Queueable = QBuffer<D, P, P, &'a Queue<D, BuffersAllocated<P>>>;
}
impl<'a, D: Direction, P: PrimitiveBufferHandles> GetBufferByIndex<'a>
for Queue<D, BuffersAllocated<P>>
{
fn try_get_buffer(&'a self, index: usize) -> Result<Self::Queueable, TryGetBufferError> {
Ok(QBuffer::new(self, self.try_obtain_buffer(index)?))
}
}
impl<'a, D, P, Q> QueueableProvider<'a> for Q
where
D: Direction,
P: PrimitiveBufferHandles,
Q: Deref<Target = Queue<D, BuffersAllocated<P>>> + Clone,
{
type Queueable = QBuffer<D, P, P, Q>;
}
impl<'a, D, P, Q> GetBufferByIndex<'a> for Q
where
D: Direction,
P: PrimitiveBufferHandles,
Q: Deref<Target = Queue<D, BuffersAllocated<P>>> + Clone,
{
fn try_get_buffer(&'a self, index: usize) -> Result<Self::Queueable, TryGetBufferError> {
Ok(QBuffer::new(self.clone(), self.try_obtain_buffer(index)?))
}
}
impl<'a, D, P> GetFreeBuffer<'a> for Queue<D, BuffersAllocated<P>>
where
D: Direction,
P: BufferHandles,
Self: GetBufferByIndex<'a>,
{
fn try_get_free_buffer(&'a self) -> Result<Self::Queueable, GetFreeBufferError> {
self.state
.buffer_info
.iter()
.enumerate()
.find(|(_, s)| s.do_with_state(|s| matches!(s, BufferState::Free)))
.ok_or(GetFreeBufferError::NoFreeBuffer)
.map(|(i, _)| self.try_get_buffer(i).unwrap())
}
}
impl<'a, D, P, Q> GetFreeBuffer<'a> for Q
where
D: Direction,
P: PrimitiveBufferHandles,
Q: Deref<Target = Queue<D, BuffersAllocated<P>>> + Clone,
{
fn try_get_free_buffer(&'a self) -> Result<Self::Queueable, GetFreeBufferError> {
self.state
.buffer_info
.iter()
.enumerate()
.find(|(_, s)| s.do_with_state(|s| matches!(s, BufferState::Free)))
.ok_or(GetFreeBufferError::NoFreeBuffer)
.map(|(i, _)| self.try_get_buffer(i).unwrap())
}
}
}
pub trait CaptureQueueable<B: BufferHandles> {
fn queue_with_handles(self, handles: B) -> QueueResult<(), B>;
}
pub trait OutputQueueable<B: BufferHandles> {
fn queue_with_handles(self, handles: B, bytes_used: &[usize]) -> QueueResult<(), B>;
}
pub trait CaptureQueueableProvider<'a, B: BufferHandles> {
type Queueable: CaptureQueueable<B>;
}
impl<'a, B, Q> CaptureQueueableProvider<'a, B> for Q
where
B: BufferHandles,
Q: private::QueueableProvider<'a>,
Q::Queueable: CaptureQueueable<B>,
{
type Queueable = <Self as private::QueueableProvider<'a>>::Queueable;
}
pub trait OutputQueueableProvider<'a, B: BufferHandles> {
type Queueable: OutputQueueable<B>;
}
impl<'a, B, Q> OutputQueueableProvider<'a, B> for Q
where
B: BufferHandles,
Q: private::QueueableProvider<'a>,
Q::Queueable: OutputQueueable<B>,
{
type Queueable = <Self as private::QueueableProvider<'a>>::Queueable;
}
pub trait GetOutputBufferByIndex<'a, B, ErrorType = TryGetBufferError>
where
B: BufferHandles,
Self: OutputQueueableProvider<'a, B>,
{
fn try_get_buffer(&'a self, index: usize) -> Result<Self::Queueable, ErrorType>;
}
impl<'a, B: BufferHandles> GetOutputBufferByIndex<'a, B> for Queue<Output, BuffersAllocated<B>>
where
Self: private::GetBufferByIndex<'a>,
<Self as private::QueueableProvider<'a>>::Queueable: OutputQueueable<B>,
{
fn try_get_buffer(&'a self, index: usize) -> Result<Self::Queueable, TryGetBufferError> {
<Self as private::GetBufferByIndex<'a>>::try_get_buffer(self, index)
}
}
pub trait GetCaptureBufferByIndex<'a, P: BufferHandles, ErrorType = TryGetBufferError>
where
Self: CaptureQueueableProvider<'a, P>,
{
fn try_get_buffer(&'a self, index: usize) -> Result<Self::Queueable, ErrorType>;
}
impl<'a, P: BufferHandles> GetCaptureBufferByIndex<'a, P> for Queue<Capture, BuffersAllocated<P>>
where
Self: private::GetBufferByIndex<'a>,
<Self as private::QueueableProvider<'a>>::Queueable: CaptureQueueable<P>,
{
fn try_get_buffer(&'a self, index: usize) -> Result<Self::Queueable, TryGetBufferError> {
<Self as private::GetBufferByIndex<'a>>::try_get_buffer(self, index)
}
}
pub trait GetFreeOutputBuffer<'a, P: BufferHandles, ErrorType = GetFreeBufferError>
where
Self: OutputQueueableProvider<'a, P>,
{
fn try_get_free_buffer(&'a self) -> Result<Self::Queueable, ErrorType>;
}
impl<'a, P: BufferHandles, Q> GetFreeOutputBuffer<'a, P> for Q
where
Self: private::GetFreeBuffer<'a>,
<Self as private::QueueableProvider<'a>>::Queueable: OutputQueueable<P>,
{
fn try_get_free_buffer(&'a self) -> Result<Self::Queueable, GetFreeBufferError> {
<Self as private::GetFreeBuffer<'a>>::try_get_free_buffer(self)
}
}
pub trait GetFreeCaptureBuffer<'a, P: BufferHandles, ErrorType = GetFreeBufferError>
where
Self: CaptureQueueableProvider<'a, P>,
{
fn try_get_free_buffer(&'a self) -> Result<Self::Queueable, ErrorType>;
}
impl<'a, P: BufferHandles, Q> GetFreeCaptureBuffer<'a, P> for Q
where
Self: private::GetFreeBuffer<'a>,
<Self as private::QueueableProvider<'a>>::Queueable: CaptureQueueable<P>,
{
fn try_get_free_buffer(&'a self) -> Result<Self::Queueable, GetFreeBufferError> {
<Self as private::GetFreeBuffer<'a>>::try_get_free_buffer(self)
}
}
struct BufferStateFuse<P: BufferHandles> {
buffer_info: Weak<BufferInfo<P>>,
}
impl<P: BufferHandles> BufferStateFuse<P> {
fn new(buffer_info: Weak<BufferInfo<P>>) -> Self {
BufferStateFuse { buffer_info }
}
fn disarm(&mut self) {
self.buffer_info = Weak::new();
}
fn trigger(&mut self) {
match self.buffer_info.upgrade() {
None => (),
Some(buffer_info) => {
buffer_info.update_state(|state| *state = BufferState::Free);
self.disarm();
}
};
}
}
impl<P: BufferHandles> Drop for BufferStateFuse<P> {
fn drop(&mut self) {
self.trigger();
}
}