use std::fs::{File, OpenOptions};
use std::io::{self, Read, Write};
use std::os::unix::fs::OpenOptionsExt;
use std::os::unix::io::AsFd;
use std::path::Path;
use std::time::{Duration, Instant};
use rustix::event::{Nsecs, PollFd, PollFlags, Timespec};
use crate::dataplane::{CiDataDevice, PartialWrite, TS_PACKET_LEN};
use crate::device::{CaDevice, DeviceOpError, SlotInfo};
fn poll_readable(fd: &impl AsFd, timeout: Duration) -> io::Result<bool> {
let mut fds = [PollFd::new(fd, PollFlags::IN)];
let ts = Timespec {
tv_sec: i64::try_from(timeout.as_secs()).unwrap_or(i64::MAX),
tv_nsec: timeout.subsec_nanos() as Nsecs,
};
rustix::event::poll(&mut fds, Some(&ts))?;
Ok(fds[0].revents().contains(PollFlags::IN))
}
fn poll_writable(fd: &impl AsFd, timeout: Duration) -> io::Result<bool> {
let mut fds = [PollFd::new(fd, PollFlags::OUT)];
let ts = Timespec {
tv_sec: i64::try_from(timeout.as_secs()).unwrap_or(i64::MAX),
tv_nsec: timeout.subsec_nanos() as Nsecs,
};
rustix::event::poll(&mut fds, Some(&ts))?;
Ok(fds[0].revents().contains(PollFlags::OUT))
}
const WRITE_CHUNK_PACKETS: usize = 21;
const WRITE_STALL_TIMEOUT: Duration = Duration::from_millis(50);
const TORN_PACKET_TIMEOUT: Duration = Duration::from_secs(2);
fn write_whole_packets(file: &mut File, chunk: &[u8]) -> io::Result<usize> {
let mut n = 0;
while n < chunk.len() {
match file.write(&chunk[n..]) {
Ok(0) => {
return Err(io::Error::new(
io::ErrorKind::WriteZero,
"CI data device accepted 0 bytes",
));
}
Ok(k) => n += k,
Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
if n.is_multiple_of(TS_PACKET_LEN) {
return Ok(n);
}
if !poll_writable(file, TORN_PACKET_TIMEOUT)? {
return Err(io::Error::new(
io::ErrorKind::TimedOut,
"CI data device stayed full with a TS packet half written",
));
}
}
Err(e) => return Err(e),
}
}
Ok(n)
}
use rustix::ioctl::{Opcode, opcode};
const DVB_CA_MAGIC: u8 = b'o';
const CA_RESET: Opcode = opcode::none(DVB_CA_MAGIC, 128);
const CA_GET_SLOT_INFO: Opcode = opcode::read::<CaSlotInfo>(DVB_CA_MAGIC, 130);
const CA_CI_MODULE_PRESENT: u32 = 1;
const CA_CI_MODULE_READY: u32 = 2;
#[repr(C)]
struct CaSlotInfo {
num: i32,
typ: i32,
flags: u32,
}
#[cfg(test)]
const KERNEL_CA_SLOT_INFO_SIZE: usize = 12;
#[cfg(test)]
const KERNEL_CA_SLOT_INFO_ALIGN: usize = 4;
#[allow(unsafe_code)]
mod ioctl {
use super::{CA_GET_SLOT_INFO, CA_RESET, CaSlotInfo};
use rustix::ioctl::{NoArg, Updater, ioctl};
use std::io;
use std::os::unix::io::BorrowedFd;
pub(super) fn ca_reset(fd: BorrowedFd<'_>) -> io::Result<()> {
unsafe { ioctl(fd, NoArg::<CA_RESET>::new()) }.map_err(io::Error::from)
}
pub(super) fn ca_get_slot_info(fd: BorrowedFd<'_>, info: &mut CaSlotInfo) -> io::Result<()> {
unsafe { ioctl(fd, Updater::<CA_GET_SLOT_INFO, CaSlotInfo>::new(info)) }
.map_err(io::Error::from)
}
}
const RESET_SETTLE: Duration = Duration::from_millis(3000);
#[derive(Debug)]
pub struct LinuxCaDevice {
file: File,
slot: u8,
label: String,
}
impl LinuxCaDevice {
pub fn open(adapter: u32, ca: u32) -> io::Result<Self> {
let path = format!("/dev/dvb/adapter{adapter}/ca{ca}");
let file = OpenOptions::new()
.read(true)
.write(true)
.open(&path)
.map_err(|e| DeviceOpError::wrap("open (O_RDWR)", &path, e))?;
Ok(Self {
file,
slot: 0,
label: path,
})
}
#[must_use]
pub fn from_file(file: File, slot: u8) -> Self {
Self {
file,
slot,
label: format!("CA device (slot {slot}, from_file)"),
}
}
fn connection_id(tpdu: &[u8]) -> u8 {
dvb_ci::length::decode(tpdu.get(1..).unwrap_or(&[]))
.ok()
.and_then(|(_, hdr)| tpdu.get(1 + hdr).copied())
.unwrap_or(1)
}
}
const MAX_CA_FRAME: usize = 2 + 1 + 3 + 65_535;
impl CaDevice for LinuxCaDevice {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let mut frame = [0u8; MAX_CA_FRAME];
let n = match self.file.read(&mut frame) {
Ok(n) => n,
Err(e) if e.kind() == io::ErrorKind::WouldBlock => return Ok(0),
Err(e) => {
return Err(DeviceOpError::wrap(
format!("read (<= {} byte frame)", frame.len()),
&self.label,
e,
));
}
};
if n == frame.len() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"CA device frame filled the maximum-size scratch buffer; \
it may have been truncated",
));
}
let tpdu = frame.get(2..n).unwrap_or(&[]);
if tpdu.len() > buf.len() {
return Err(DeviceOpError::wrap(
format!(
"read ({} byte TPDU does not fit the caller's {} byte buffer)",
tpdu.len(),
buf.len()
),
&self.label,
io::Error::new(
io::ErrorKind::InvalidData,
"TPDU larger than the read buffer; dropped, not truncated",
),
));
}
buf[..tpdu.len()].copy_from_slice(tpdu);
Ok(tpdu.len())
}
fn write(&mut self, buf: &[u8]) -> io::Result<()> {
let mut frame = Vec::with_capacity(buf.len() + 2);
frame.push(self.slot);
frame.push(Self::connection_id(buf));
frame.extend_from_slice(buf);
self.file.write_all(&frame).map_err(|e| {
DeviceOpError::wrap(
format!(
"write ({} byte frame: slot {}, connection {})",
frame.len(),
frame[0],
frame[1]
),
&self.label,
e,
)
})
}
fn reset(&mut self) -> io::Result<()> {
ioctl::ca_reset(self.file.as_fd())
.map_err(|e| DeviceOpError::wrap("ioctl CA_RESET", &self.label, e))?;
Ok(())
}
fn reset_settle(&self) -> Duration {
RESET_SETTLE
}
fn slot_info(&mut self) -> io::Result<SlotInfo> {
let mut si = CaSlotInfo {
num: i32::from(self.slot),
typ: 0,
flags: 0,
};
if let Err(err) = ioctl::ca_get_slot_info(self.file.as_fd(), &mut si) {
return match err.raw_os_error() {
Some(libc::EINVAL) | Some(libc::ENOTTY) => Ok(SlotInfo {
num: self.slot,
module_ready: true,
module_present: true,
}),
_ => Err(DeviceOpError::wrap(
format!("ioctl CA_GET_SLOT_INFO (slot {})", self.slot),
&self.label,
err,
)),
};
}
Ok(SlotInfo {
num: si.num as u8,
module_ready: si.flags & CA_CI_MODULE_READY != 0,
module_present: si.flags & CA_CI_MODULE_PRESENT != 0,
})
}
fn poll(&mut self, timeout: Duration) -> io::Result<bool> {
poll_readable(&self.file, timeout)
.map_err(|e| DeviceOpError::wrap(format!("poll ({timeout:?})"), &self.label, e))
}
}
#[derive(Debug)]
enum CiEndpoints {
Duplex(File),
Split { write: File, read: File },
}
#[derive(Debug)]
pub struct LinuxCiDataDevice {
io: CiEndpoints,
label: String,
}
impl LinuxCiDataDevice {
fn open_path(path: &Path) -> io::Result<Self> {
let file = OpenOptions::new()
.read(true)
.write(true)
.custom_flags(libc::O_NONBLOCK)
.open(path)
.map_err(|e| DeviceOpError::wrap("open (O_RDWR|O_NONBLOCK)", &path_label(path), e))?;
Ok(Self {
io: CiEndpoints::Duplex(file),
label: path_label(path),
})
}
fn open_split_path(path: &Path) -> io::Result<Self> {
let label = path_label(path);
let read = OpenOptions::new()
.read(true)
.custom_flags(libc::O_NONBLOCK)
.open(path)
.map_err(|e| DeviceOpError::wrap("open (O_RDONLY|O_NONBLOCK)", &label, e))?;
let write = OpenOptions::new()
.write(true)
.open(path)
.map_err(|e| DeviceOpError::wrap("open (O_WRONLY)", &label, e))?;
Ok(Self {
io: CiEndpoints::Split { write, read },
label,
})
}
fn open_with(
path: &Path,
open_duplex: impl FnOnce(&Path) -> io::Result<Self>,
) -> io::Result<Self> {
match open_duplex(path) {
Ok(dev) => Ok(dev),
Err(e) if DeviceOpError::errno_of(&e) == Some(libc::EINVAL) => {
Self::open_split_path(path)
}
Err(e) => Err(e),
}
}
pub fn open(adapter: u32, ci: u32) -> io::Result<Self> {
let path = format!("/dev/dvb/adapter{adapter}/ci{ci}");
Self::open_with(Path::new(&path), Self::open_path)
}
pub fn open_split(adapter: u32, ci: u32) -> io::Result<Self> {
let path = format!("/dev/dvb/adapter{adapter}/ci{ci}");
Self::open_split_path(Path::new(&path))
}
#[must_use]
pub fn from_file(file: File) -> Self {
Self {
io: CiEndpoints::Duplex(file),
label: "CI data device (from_file)".to_owned(),
}
}
#[must_use]
pub fn from_split(write: File, read: File) -> Self {
Self {
io: CiEndpoints::Split { write, read },
label: "CI data device (from_split)".to_owned(),
}
}
fn write_file(&mut self) -> &mut File {
match &mut self.io {
CiEndpoints::Duplex(f) => f,
CiEndpoints::Split { write, .. } => write,
}
}
fn read_file(&mut self) -> &mut File {
match &mut self.io {
CiEndpoints::Duplex(f) => f,
CiEndpoints::Split { read, .. } => read,
}
}
}
fn path_label(path: &Path) -> String {
path.display().to_string()
}
impl CiDataDevice for LinuxCiDataDevice {
fn write(&mut self, ts: &[u8]) -> io::Result<()> {
if !ts.len().is_multiple_of(TS_PACKET_LEN) {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"write not a multiple of 188 bytes",
));
}
let label = self.label.clone();
let mut done = 0;
let mut deadline = None;
while done < ts.len() {
let end = ts.len().min(done + WRITE_CHUNK_PACKETS * TS_PACKET_LEN);
let n = write_whole_packets(self.write_file(), &ts[done..end]).map_err(|e| {
DeviceOpError::wrap(format!("write ({} of {} bytes)", done, ts.len()), &label, e)
})?;
let want = end - done;
done += n;
if n > 0 {
deadline = None;
}
if n == want {
continue;
}
let limit = *deadline.get_or_insert_with(|| Instant::now() + WRITE_STALL_TIMEOUT);
let left = limit.saturating_duration_since(Instant::now());
let writable = !left.is_zero()
&& poll_writable(self.write_file(), left).map_err(|e| {
DeviceOpError::wrap(format!("poll write ({left:?})"), &label, e)
})?;
if !writable {
return Err(io::Error::new(
io::ErrorKind::WouldBlock,
PartialWrite {
written: done,
requested: ts.len(),
device: label,
},
));
}
}
Ok(())
}
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
if !buf.len().is_multiple_of(TS_PACKET_LEN) {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"read buffer not a multiple of 188 bytes",
));
}
let cap = buf.len();
match self.read_file().read(buf) {
Ok(n) => Ok(n),
Err(e) if e.kind() == io::ErrorKind::WouldBlock => Ok(0),
Err(e) => {
let label = self.label.clone();
Err(DeviceOpError::wrap(
format!("read (<= {cap} bytes)"),
&label,
e,
))
}
}
}
fn poll(&mut self, timeout: Duration) -> io::Result<bool> {
let label = self.label.clone();
poll_readable(self.read_file(), timeout)
.map_err(|e| DeviceOpError::wrap(format!("poll ({timeout:?})"), &label, e))
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::mpsc;
fn make_fifo() -> std::path::PathBuf {
let path = std::env::temp_dir().join(format!(
"dvb-ci-runtime-c1-1066-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("system clock before UNIX_EPOCH")
.as_nanos()
));
rustix::fs::mkfifoat(
rustix::fs::CWD,
&path,
rustix::fs::Mode::RUSR | rustix::fs::Mode::WUSR,
)
.expect("mkfifoat failed");
path
}
#[test]
fn read_on_empty_device_returns_immediately_not_blocking() {
let path = make_fifo();
let mut dev = LinuxCiDataDevice::open_path(&path).expect("open_path");
let mut writer = OpenOptions::new()
.write(true)
.open(&path)
.expect("open a second, writer-only handle on the same FIFO");
let (tx, rx) = mpsc::channel();
let (tx_go, rx_go) = mpsc::channel::<()>();
std::thread::spawn(move || {
let mut buf = [0u8; TS_PACKET_LEN];
let r0 = dev.read(&mut buf).map(|n| (n, buf));
if tx.send(r0).is_err() {
return;
}
if rx_go.recv().is_err() {
return;
}
let r1 = dev.read(&mut buf).map(|n| (n, buf));
if tx.send(r1).is_err() {
return;
}
if rx_go.recv().is_err() {
return;
}
let r2 = dev.read(&mut buf).map(|n| (n, buf));
let _ = tx.send(r2);
});
let first = rx.recv_timeout(Duration::from_secs(1));
let Ok(Ok((n0, _))) = first else {
let _ = writer.write_all(&[0u8; TS_PACKET_LEN]);
panic!(
"empty-device read did not return promptly (blocked past 1s) — \
got {first:?}; this is the #1066 hang O_NONBLOCK fixes"
);
};
assert_eq!(n0, 0, "empty device must read 0 bytes");
let mut packet = [0xAAu8; TS_PACKET_LEN];
packet[0] = 0x47; writer.write_all(&packet).expect("write one packet");
tx_go.send(()).expect("reader thread must still be alive");
let (n1, buf1) = rx
.recv_timeout(Duration::from_secs(1))
.expect("reader thread must respond promptly")
.expect("read must not error");
assert_eq!(n1, TS_PACKET_LEN, "must read exactly one packet's worth");
assert_eq!(
&buf1[..n1],
&packet[..],
"must read back the exact bytes written"
);
tx_go.send(()).expect("reader thread must still be alive");
let (n2, _) = rx
.recv_timeout(Duration::from_secs(1))
.expect("empty-again read did not return promptly (blocked past 1s)")
.expect("read must not error");
assert_eq!(n2, 0, "device must read 0 again once drained");
let _ = std::fs::remove_file(&path);
}
fn packet(fill: u8) -> [u8; TS_PACKET_LEN] {
let mut p = [fill; TS_PACKET_LEN];
p[0] = 0x47;
p
}
fn within<T: Send + 'static>(
limit: Duration,
unstick: impl FnOnce(),
f: impl FnOnce() -> T + Send + 'static,
) -> T {
let (tx, rx) = mpsc::channel();
std::thread::spawn(move || {
let _ = tx.send(f());
});
match rx.recv_timeout(limit) {
Ok(v) => v,
Err(_) => {
unstick();
panic!("operation did not return within {limit:?} (blocked)");
}
}
}
#[test]
fn split_read_side_is_nonblocking_and_a_drain_loop_terminates() {
let path = make_fifo();
let open_path = path.clone();
let mut dev = within(
Duration::from_secs(2),
|| {},
move || LinuxCiDataDevice::open_split_path(&open_path).expect("split open"),
);
dev.write(&packet(0xA1)).expect("write packet 1");
dev.write(&packet(0xB2)).expect("write packet 2");
let drained = within(
Duration::from_secs(2),
|| {},
move || {
let mut got = Vec::new();
let mut buf = [0u8; TS_PACKET_LEN];
loop {
let n = dev.read(&mut buf).expect("read");
if n == 0 {
break;
}
assert_eq!(n, TS_PACKET_LEN);
got.push(buf);
}
assert_eq!(dev.read(&mut buf).expect("read"), 0);
got
},
);
assert_eq!(drained, vec![packet(0xA1), packet(0xB2)]);
let _ = std::fs::remove_file(&path);
}
#[test]
fn split_with_a_blocking_read_fd_would_hang_the_drain_loop() {
let path = make_fifo();
let write = OpenOptions::new()
.read(true)
.write(true)
.open(&path)
.expect("write open");
let read = OpenOptions::new()
.read(true)
.open(&path)
.expect("blocking read-only open");
let unblock = write.try_clone().expect("clone write fd");
let mut dev = LinuxCiDataDevice::from_split(write, read);
dev.write(&packet(0xC3)).expect("write");
let (tx, rx) = mpsc::channel();
std::thread::spawn(move || {
let mut buf = [0u8; TS_PACKET_LEN];
let first = dev.read(&mut buf); let _ = tx.send(first);
let second = dev.read(&mut buf); let _ = tx.send(second);
});
assert_eq!(
rx.recv_timeout(Duration::from_secs(2)).unwrap().unwrap(),
TS_PACKET_LEN
);
assert!(
rx.recv_timeout(Duration::from_millis(400)).is_err(),
"a blocking read fd must NOT return on an empty FIFO - the mock is not blocking-faithful"
);
let mut unblock = unblock;
let _ = unblock.write_all(&packet(0xD4));
let _ = std::fs::remove_file(&path);
}
#[test]
fn open_falls_back_to_split_on_einval_only() {
let path = make_fifo();
let mut dev = LinuxCiDataDevice::open_with(&path, |_| {
Err(io::Error::from_raw_os_error(libc::EINVAL))
})
.expect("EINVAL on O_RDWR must fall back to the split open");
assert!(matches!(dev.io, CiEndpoints::Split { .. }));
dev.write(&packet(0xE5)).expect("write");
let mut buf = [0u8; TS_PACKET_LEN];
assert_eq!(dev.read(&mut buf).expect("read"), TS_PACKET_LEN);
assert_eq!(buf, packet(0xE5));
let wrapped = LinuxCiDataDevice::open_with(&path, |p| {
Err(DeviceOpError::wrap(
"open (O_RDWR|O_NONBLOCK)",
&path_label(p),
io::Error::from_raw_os_error(libc::EINVAL),
))
})
.expect("wrapped EINVAL must fall back too");
assert!(matches!(wrapped.io, CiEndpoints::Split { .. }));
for errno in [libc::ENOENT, libc::EACCES, libc::EBUSY, libc::ENODEV] {
let err =
LinuxCiDataDevice::open_with(&path, |_| Err(io::Error::from_raw_os_error(errno)))
.expect_err("a non-EINVAL open error must propagate");
assert_eq!(DeviceOpError::errno_of(&err), Some(errno));
}
let _ = std::fs::remove_file(&path);
}
#[test]
fn open_keeps_the_duplex_fd_when_rdwr_works() {
let path = make_fifo();
let dev =
LinuxCiDataDevice::open_with(&path, LinuxCiDataDevice::open_path).expect("duplex open");
assert!(matches!(dev.io, CiEndpoints::Duplex(_)));
let _ = std::fs::remove_file(&path);
}
#[test]
fn device_errors_carry_operation_device_and_errno() {
let path = std::env::temp_dir().join(format!(
"dvb-ci-runtime-h1-ctx-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("system clock before UNIX_EPOCH")
.as_nanos()
));
std::fs::write(&path, b"x").expect("stage file");
let file = OpenOptions::new()
.read(true)
.open(&path)
.expect("open read-only");
let mut dev = LinuxCaDevice::from_file(file, 2);
let err = dev
.write(&[0xA0, 0x01, 0x01])
.expect_err("write to a read-only fd must fail");
let text = err.to_string();
assert!(
text.contains("write (5 byte frame: slot 2, connection 1)"),
"{text}"
);
assert!(text.contains("CA device (slot 2, from_file)"), "{text}");
assert!(text.contains("os error"), "{text}");
assert_eq!(DeviceOpError::errno_of(&err), Some(libc::EBADF));
assert_eq!(err.kind(), io::Error::from_raw_os_error(libc::EBADF).kind());
let _ = std::fs::remove_file(&path);
}
#[test]
fn read_rejects_a_frame_that_fills_the_scratch_buffer() {
let path = std::env::temp_dir().join(format!(
"dvb-ci-runtime-w2-fullframe-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("system clock before UNIX_EPOCH")
.as_nanos()
));
std::fs::write(&path, vec![0xAAu8; MAX_CA_FRAME]).expect("stage a MAX_CA_FRAME-byte file");
let file = OpenOptions::new()
.read(true)
.write(true)
.open(&path)
.expect("open the staged file");
let mut dev = LinuxCaDevice::from_file(file, 0);
let mut buf = [0u8; 8192];
let err = dev
.read(&mut buf)
.expect_err("a full-buffer read must error");
assert_eq!(err.kind(), io::ErrorKind::InvalidData);
let _ = std::fs::remove_file(&path);
}
#[test]
fn read_does_not_silently_truncate_a_tpdu_larger_than_the_callers_buffer() {
let path = std::env::temp_dir().join(format!(
"dvb-ci-runtime-review-trunc-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("system clock before UNIX_EPOCH")
.as_nanos()
));
let mut frame = vec![0x00, 0x01];
frame.extend(std::iter::repeat_n(0xAB, 5000));
std::fs::write(&path, &frame).expect("stage a 5000-byte TPDU frame");
let file = OpenOptions::new()
.read(true)
.write(true)
.open(&path)
.expect("open the staged file");
let mut dev = LinuxCaDevice::from_file(file, 0);
let mut buf = [0u8; 4096]; let r = dev.read(&mut buf);
let _ = std::fs::remove_file(&path);
assert!(
r.is_err(),
"a 5000-byte TPDU read into a 4096-byte buffer must not be returned as Ok({:?})",
r.ok()
);
}
#[test]
fn duplex_write_never_leaves_a_partial_ts_packet_in_the_device() {
let path = make_fifo();
let mut dev = LinuxCiDataDevice::open_path(&path).expect("open_path");
let mut reader = OpenOptions::new()
.read(true)
.custom_flags(libc::O_NONBLOCK)
.open(&path)
.expect("second reader");
let burst: Vec<u8> = (0..400).flat_map(|i| packet(i as u8)).collect();
let res = dev.write(&burst);
let mut got = Vec::new();
let mut chunk = [0u8; 4096];
loop {
match reader.read(&mut chunk) {
Ok(0) => break,
Ok(n) => got.extend_from_slice(&chunk[..n]),
Err(e) if e.kind() == io::ErrorKind::WouldBlock => break,
Err(e) => panic!("{e}"),
}
}
let _ = std::fs::remove_file(&path);
assert_eq!(
got.len() % TS_PACKET_LEN,
0,
"write returned {res:?} after putting {} bytes (not whole packets) on the device",
got.len()
);
}
#[test]
fn read_still_returns_short_frames_normally() {
let path = std::env::temp_dir().join(format!(
"dvb-ci-runtime-w2-shortframe-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("system clock before UNIX_EPOCH")
.as_nanos()
));
std::fs::write(&path, [0x00, 0x01, 0xDE, 0xAD, 0xBE]).expect("stage a short file");
let file = OpenOptions::new()
.read(true)
.write(true)
.open(&path)
.expect("open the staged file");
let mut dev = LinuxCaDevice::from_file(file, 0);
let mut buf = [0u8; 16];
let n = dev.read(&mut buf).expect("short read must succeed");
assert_eq!(n, 3);
assert_eq!(&buf[..3], &[0xDE, 0xAD, 0xBE]);
let _ = std::fs::remove_file(&path);
}
#[test]
fn slot_info_falls_back_on_enotty() {
let path = std::env::temp_dir().join(format!(
"dvb-ci-runtime-w3-slotinfo-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("system clock before UNIX_EPOCH")
.as_nanos()
));
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(true)
.open(&path)
.expect("create a regular file to stand in for the CA device");
let mut dev = LinuxCaDevice::from_file(file, 3);
let info = dev.slot_info().expect("ENOTTY must fall back, not error");
assert_eq!(info.num, 3);
assert!(info.module_present);
assert!(info.module_ready);
let _ = std::fs::remove_file(&path);
}
fn poll_file(file: &File, timeout: Duration) -> io::Result<bool> {
poll_readable(file, timeout)
}
#[test]
fn poll_readable_reports_data_and_honours_the_timeout() {
let path = make_fifo();
let mut dev = LinuxCiDataDevice::open_path(&path).expect("open_path");
let mut writer = OpenOptions::new().write(true).open(&path).expect("writer");
let start = std::time::Instant::now();
assert!(
!dev.poll(Duration::from_millis(80)).unwrap(),
"empty FIFO is not readable"
);
assert!(
start.elapsed() >= Duration::from_millis(70),
"poll returned before its timeout"
);
writer.write_all(&[0x47; TS_PACKET_LEN]).unwrap();
let start = std::time::Instant::now();
assert!(
dev.poll(Duration::from_secs(5)).unwrap(),
"data written: readable"
);
assert!(
start.elapsed() < Duration::from_secs(4),
"must return as soon as readable"
);
let _ = std::fs::remove_file(&path);
}
#[test]
fn poll_readable_zero_timeout_does_not_block() {
let path = make_fifo();
let mut dev = LinuxCiDataDevice::open_path(&path).expect("open_path");
let _writer = OpenOptions::new().write(true).open(&path).expect("writer");
let start = std::time::Instant::now();
assert!(!dev.poll(Duration::ZERO).unwrap());
assert!(start.elapsed() < Duration::from_millis(500));
let _ = std::fs::remove_file(&path);
}
#[test]
fn poll_readable_sub_millisecond_timeout_keeps_its_duration() {
let path = make_fifo();
let mut dev = LinuxCiDataDevice::open_path(&path).expect("open_path");
let _writer = OpenOptions::new().write(true).open(&path).expect("writer");
let start = std::time::Instant::now();
assert!(!dev.poll(Duration::from_micros(300)).unwrap());
assert!(
start.elapsed() >= Duration::from_micros(250),
"a 300 us timeout was truncated: returned after {:?}",
start.elapsed()
);
let _ = std::fs::remove_file(&path);
}
#[test]
fn poll_readable_ignores_hangup_without_data() {
let path = make_fifo();
let reader = OpenOptions::new()
.read(true)
.custom_flags(libc::O_NONBLOCK)
.open(&path)
.expect("read-only O_NONBLOCK open");
let writer = OpenOptions::new().write(true).open(&path).expect("writer");
drop(writer);
let mut raw = [PollFd::new(&reader, PollFlags::IN)];
let zero = Timespec {
tv_sec: 0,
tv_nsec: 0,
};
let r = rustix::event::poll(&mut raw, Some(&zero)).expect("poll");
assert_eq!(r, 1, "kernel must report an event");
let revents = raw[0].revents();
assert!(
revents.contains(PollFlags::HUP) && !revents.contains(PollFlags::IN),
"precondition: POLLHUP without POLLIN, got revents={revents:?}"
);
assert!(!poll_file(&reader, Duration::from_millis(50)).unwrap());
let _ = std::fs::remove_file(&path);
}
#[test]
fn ca_slot_info_matches_kernel_layout() {
assert_eq!(core::mem::size_of::<CaSlotInfo>(), KERNEL_CA_SLOT_INFO_SIZE);
assert_eq!(
core::mem::align_of::<CaSlotInfo>(),
KERNEL_CA_SLOT_INFO_ALIGN
);
}
#[cfg(not(any(
target_arch = "mips",
target_arch = "mips32r6",
target_arch = "mips64",
target_arch = "mips64r6",
target_arch = "powerpc",
target_arch = "powerpc64",
target_arch = "sparc",
target_arch = "sparc64"
)))]
const _: () = {
assert!(CA_RESET == 0x0000_6F80);
assert!(CA_GET_SLOT_INFO == 0x800C_6F82);
};
#[cfg(any(
target_arch = "mips",
target_arch = "mips32r6",
target_arch = "mips64",
target_arch = "mips64r6",
target_arch = "powerpc",
target_arch = "powerpc64",
target_arch = "sparc",
target_arch = "sparc64"
))]
const _: () = {
assert!(CA_RESET == 0x2000_6F80);
assert!(CA_GET_SLOT_INFO == 0x400C_6F82);
};
}