io: Introduce new Queue type to replace Arena/Stream

The notion of a "queue" is closer to the actual V4L2 streaming I/O
implementation detail. The new type design should allow us to model
planar buffers as well in the future.

Signed-off-by: Christopher N. Hesse <raymanfx@gmail.com>

Author: raymanfx

examples/io_capture.rs

@@ -0,0 +1,76 @@
+use std::io;
+use std::time::Instant;
+
+use v4l::buffer::Type;
+use v4l::io::Queue;
+use v4l::prelude::*;
+use v4l::video::Capture;
+
+fn main() -> io::Result<()> {
+    let path = "/dev/video0";
+    println!("Using device: {}\n", path);
+
+    // Capture 4 frames by default
+    let count = 4;
+
+    let dev = Device::with_path(path)?;
+    let format = dev.format()?;
+    let params = dev.params()?;
+    println!("Active format:\n{}", format);
+    println!("Active parameters:\n{}", params);
+
+    // set up a queue
+    let mut queue = Queue::with_mmap(dev.handle(), Type::VideoCapture, 4)?;
+
+    // enqueue all buffers once
+    for i in 0..queue.len() {
+        let buf = queue.query_buf(i as u32)?;
+        queue.enqueue(&buf)?;
+    }
+
+    // start streaming
+    let mut queue = queue.start_stream()?;
+
+    // warm up
+    let buf = queue.dequeue()?;
+    let _ = queue.enqueue(&buf)?;
+
+    let start = Instant::now();
+    let mut megabytes_ps: f64 = 0.0;
+    for i in 0..count {
+        let t0 = Instant::now();
+
+        // check whether there's something for us to do
+        dev.handle().poll(libc::POLLIN, -1)?;
+
+        // remove a buffer from the drivers' outgoing queue
+        let buf = queue.dequeue()?;
+
+        let duration_us = t0.elapsed().as_micros();
+
+        let cur = buf.bytesused as f64 / 1_048_576.0 * 1_000_000.0 / duration_us as f64;
+        if i == 0 {
+            megabytes_ps = cur;
+        } else {
+            // ignore the first measurement
+            let prev = megabytes_ps * (i as f64 / (i + 1) as f64);
+            let now = cur * (1.0 / (i + 1) as f64);
+            megabytes_ps = prev + now;
+        }
+
+        println!("Buffer");
+        println!("  sequence  : {}", buf.sequence);
+        println!("  timestamp : {}", buf.timestamp);
+        println!("  flags     : {}", buf.flags);
+        println!("  length    : {}", buf.bytesused);
+
+        // once we're done with it, add the buffer to the drivers' incoming queue
+        queue.enqueue(&buf)?;
+    }
+
+    println!();
+    println!("FPS: {}", count as f64 / start.elapsed().as_secs_f64());
+    println!("MB/s: {}", megabytes_ps);
+
+    Ok(())
+}

src/buffer.rs

@@ -1,7 +1,9 @@
 use bitflags::bitflags;
-use std::fmt;
+use std::{convert::TryInto, fmt, mem};
 
-use crate::timestamp::Timestamp;
+use v4l2_sys::v4l2_buffer;
+
+use crate::{memory::Memory, timestamp::Timestamp};
 
 /// Buffer type
 ///
@@ -99,8 +101,12 @@ impl fmt::Display for Flags {
 }
 
 /// Buffer metadata, mostly used not to convolute the main buffer structs
-#[derive(Copy, Clone, Default)]
+#[derive(Copy, Clone)]
 pub struct Metadata {
+    /// Number of the buffer
+    pub index: u32,
+    /// Type of the buffer
+    pub type_: u32,
     /// Number of bytes occupied by the data in the buffer
     pub bytesused: u32,
     /// Buffer flags
@@ -111,4 +117,44 @@ pub struct Metadata {
     pub timestamp: Timestamp,
     /// Sequence number, counting the frames
     pub sequence: u32,
+    /// Memory type, depending on streaming I/O method
+    pub memory: Memory,
+    /// Single-planar API: size of the buffer (not payload!)
+    /// Multi-planar API: number of planes
+    pub length: u32,
+}
+
+impl From<v4l2_buffer> for Metadata {
+    fn from(buf: v4l2_buffer) -> Self {
+        Self {
+            index: buf.index,
+            type_: buf.type_,
+            bytesused: buf.bytesused,
+            flags: buf.flags.into(),
+            field: buf.field,
+            timestamp: buf.timestamp.into(),
+            sequence: buf.sequence,
+            memory: buf.memory.try_into().unwrap(),
+            length: buf.length,
+        }
+    }
+}
+
+impl Into<v4l2_buffer> for Metadata {
+    fn into(self) -> v4l2_buffer {
+        unsafe {
+            v4l2_buffer {
+                index: self.index,
+                type_: self.type_,
+                bytesused: self.bytesused,
+                flags: self.flags.into(),
+                field: self.field,
+                timestamp: self.timestamp.into(),
+                sequence: self.sequence,
+                memory: self.memory as u32,
+                length: self.length,
+                ..mem::zeroed()
+            }
+        }
+    }
 }

src/io/mmap/stream.rs

@@ -52,7 +52,7 @@ impl<'a> Stream<'a> {
         let mut arena = Arena::new(dev.handle(), buf_type);
         let count = arena.allocate(buf_count)?;
         let mut buf_meta = Vec::new();
-        buf_meta.resize(count as usize, Metadata::default());
+        buf_meta.resize(count as usize, unsafe { Metadata { ..mem::zeroed() } });
 
         Ok(Stream {
             handle: dev.handle(),
@@ -175,14 +175,7 @@ impl<'a, 'b> CaptureStream<'b> for Stream<'a> {
             )?;
         }
         self.arena_index = v4l2_buf.index as usize;
-
-        self.buf_meta[self.arena_index] = Metadata {
-            bytesused: v4l2_buf.bytesused,
-            flags: v4l2_buf.flags.into(),
-            field: v4l2_buf.field,
-            timestamp: v4l2_buf.timestamp.into(),
-            sequence: v4l2_buf.sequence,
-        };
+        self.buf_meta[self.arena_index] = Metadata::from(v4l2_buf);
 
         Ok(self.arena_index)
     }
@@ -254,14 +247,7 @@ impl<'a, 'b> OutputStream<'b> for Stream<'a> {
             )?;
         }
         self.arena_index = v4l2_buf.index as usize;
-
-        self.buf_meta[self.arena_index] = Metadata {
-            bytesused: v4l2_buf.bytesused,
-            flags: v4l2_buf.flags.into(),
-            field: v4l2_buf.field,
-            timestamp: v4l2_buf.timestamp.into(),
-            sequence: v4l2_buf.sequence,
-        };
+        self.buf_meta[self.arena_index] = Metadata::from(v4l2_buf);
 
         Ok(self.arena_index)
     }