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;
use rustix::event::{Nsecs, PollFd, PollFlags, Timespec};
use crate::dataplane::{CiDataDevice, TS_PACKET_LEN};
use crate::device::{CaDevice, SlotInfo};
#[allow(clippy::unnecessary_fallible_conversions)]
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: Nsecs::try_from(timeout.subsec_nanos()).unwrap_or(0),
};
rustix::event::poll(&mut fds, Some(&ts))?;
Ok(fds[0].revents().contains(PollFlags::IN))
}
const IOC_NRBITS: u32 = 8;
const IOC_TYPEBITS: u32 = 8;
const IOC_SIZEBITS: u32 = 14;
const IOC_NRSHIFT: u32 = 0;
const IOC_TYPESHIFT: u32 = IOC_NRSHIFT + IOC_NRBITS;
const IOC_SIZESHIFT: u32 = IOC_TYPESHIFT + IOC_TYPEBITS;
const IOC_DIRSHIFT: u32 = IOC_SIZESHIFT + IOC_SIZEBITS;
const IOC_NONE: u32 = 0;
const IOC_READ: u32 = 2;
const fn ioc(dir: u32, typ: u32, nr: u32, size: u32) -> u64 {
((dir << IOC_DIRSHIFT) | (typ << IOC_TYPESHIFT) | (nr << IOC_NRSHIFT) | (size << IOC_SIZESHIFT))
as u64
}
const DVB_CA_MAGIC: u32 = b'o' as u32;
const CA_RESET: u64 = ioc(IOC_NONE, DVB_CA_MAGIC, 128, 0);
const CA_GET_SLOT_INFO: u64 = ioc(
IOC_READ,
DVB_CA_MAGIC,
130,
core::mem::size_of::<CaSlotInfo>() as u32,
);
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 std::io;
use std::os::unix::io::{AsRawFd, BorrowedFd};
pub(super) fn ca_reset(fd: BorrowedFd<'_>) -> io::Result<()> {
let r = unsafe { libc::ioctl(fd.as_raw_fd(), CA_RESET as libc::Ioctl) };
if r < 0 {
return Err(io::Error::last_os_error());
}
Ok(())
}
pub(super) fn ca_get_slot_info(fd: BorrowedFd<'_>, info: &mut CaSlotInfo) -> io::Result<()> {
let r = unsafe {
libc::ioctl(
fd.as_raw_fd(),
CA_GET_SLOT_INFO as libc::Ioctl,
info as *mut CaSlotInfo,
)
};
if r < 0 {
return Err(io::Error::last_os_error());
}
Ok(())
}
}
const RESET_SETTLE: Duration = Duration::from_millis(3000);
#[derive(Debug)]
pub struct LinuxCaDevice {
file: File,
slot: u8,
}
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)?;
Ok(Self { file, slot: 0 })
}
#[must_use]
pub fn from_file(file: File, slot: u8) -> Self {
Self { file, slot }
}
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(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(&[]);
let copy = tpdu.len().min(buf.len());
buf[..copy].copy_from_slice(&tpdu[..copy]);
Ok(copy)
}
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)
}
fn reset(&mut self) -> io::Result<()> {
ioctl::ca_reset(self.file.as_fd())?;
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(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)
}
}
#[derive(Debug)]
pub struct LinuxCiDataDevice {
file: File,
}
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)?;
Ok(Self { file })
}
pub fn open(adapter: u32, ci: u32) -> io::Result<Self> {
let path = format!("/dev/dvb/adapter{adapter}/ci{ci}");
Self::open_path(Path::new(&path))
}
#[must_use]
pub fn from_file(file: File) -> Self {
Self { file }
}
}
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",
));
}
self.file.write_all(ts)
}
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",
));
}
match self.file.read(buf) {
Ok(n) => Ok(n),
Err(e) if e.kind() == io::ErrorKind::WouldBlock => Ok(0),
Err(e) => Err(e),
}
}
fn poll(&mut self, timeout: Duration) -> io::Result<bool> {
poll_readable(&self.file, timeout)
}
}
#[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);
}
#[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_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
);
}
}