294 lines
9.5 KiB
Rust
294 lines
9.5 KiB
Rust
use std::collections::{HashSet, VecDeque};
|
|
use std::time::{Duration, Instant};
|
|
|
|
use smithay::backend::allocator::dmabuf::Dmabuf;
|
|
use smithay::backend::drm::DrmNode;
|
|
use smithay::backend::renderer::ImportDma;
|
|
use smithay::reexports::calloop::RegistrationToken;
|
|
use smithay::reexports::calloop::timer::{TimeoutAction, Timer};
|
|
use smithay::reexports::wayland_server::backend::ClientId;
|
|
use smithay::wayland::dmabuf::{DmabufGlobal, ImportNotifier};
|
|
|
|
use crate::Ferese;
|
|
|
|
const MAX_PENDING: usize = 64;
|
|
const MAX_PER_CLIENT: usize = 16;
|
|
const MAX_WAIT: Duration = Duration::from_secs(5);
|
|
|
|
struct Pending<K, C, T> {
|
|
owner: K,
|
|
client: C,
|
|
deadline: Instant,
|
|
value: T,
|
|
}
|
|
|
|
struct Queue<K, C, T> {
|
|
pending: VecDeque<Pending<K, C, T>>,
|
|
}
|
|
|
|
impl<K, C, T> Default for Queue<K, C, T> {
|
|
fn default() -> Self {
|
|
Self {
|
|
pending: VecDeque::new(),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<K: Eq, C: Eq, T> Queue<K, C, T> {
|
|
fn enqueue(&mut self, owner: K, client: C, value: T, now: Instant) -> Result<(), T> {
|
|
if self.pending.len() >= MAX_PENDING
|
|
|| self.pending.iter().filter(|entry| entry.client == client).count() >= MAX_PER_CLIENT
|
|
{
|
|
return Err(value);
|
|
}
|
|
|
|
self.pending.push_back(Pending {
|
|
owner,
|
|
client,
|
|
value,
|
|
deadline: now + MAX_WAIT,
|
|
});
|
|
Ok(())
|
|
}
|
|
|
|
fn process(&mut self, owner: &K, mut complete: impl FnMut(T)) {
|
|
self.remove_if(|entry| &entry.owner == owner, |entry| complete(entry.value));
|
|
}
|
|
|
|
fn remove_if(
|
|
&mut self,
|
|
mut remove: impl FnMut(&Pending<K, C, T>) -> bool,
|
|
mut complete: impl FnMut(Pending<K, C, T>),
|
|
) {
|
|
for _ in 0..self.pending.len() {
|
|
let entry = self.pending.pop_front().unwrap();
|
|
if remove(&entry) {
|
|
complete(entry);
|
|
} else {
|
|
self.pending.push_back(entry);
|
|
}
|
|
}
|
|
}
|
|
|
|
fn deadline(&self) -> Option<Instant> {
|
|
self.pending.front().map(|entry| entry.deadline)
|
|
}
|
|
}
|
|
|
|
struct Request {
|
|
buffer: Dmabuf,
|
|
notifier: ImportNotifier,
|
|
}
|
|
|
|
#[derive(Default)]
|
|
pub(crate) struct ImportManager {
|
|
globals: HashSet<DmabufGlobal>,
|
|
queue: Queue<DmabufGlobal, ClientId, Request>,
|
|
timer: Option<RegistrationToken>,
|
|
}
|
|
|
|
impl Drop for ImportManager {
|
|
fn drop(&mut self) {
|
|
self.queue.remove_if(|_| true, |entry| entry.value.notifier.failed());
|
|
}
|
|
}
|
|
|
|
impl ImportManager {
|
|
pub(crate) fn register(&mut self, global: DmabufGlobal) {
|
|
self.globals.insert(global);
|
|
}
|
|
|
|
pub(crate) fn remove(&mut self, global: DmabufGlobal) {
|
|
self.globals.remove(&global);
|
|
self.queue.process(&global, |request| request.notifier.failed());
|
|
}
|
|
|
|
pub(crate) fn enqueue(&mut self, global: DmabufGlobal, buffer: Dmabuf, notifier: ImportNotifier, now: Instant) {
|
|
let Some(client) = notifier.client() else {
|
|
notifier.failed();
|
|
return;
|
|
};
|
|
|
|
if !self.globals.contains(&global) {
|
|
notifier.failed();
|
|
return;
|
|
}
|
|
|
|
if let Err(request) = self
|
|
.queue
|
|
.enqueue(global, client.id(), Request { buffer, notifier }, now)
|
|
{
|
|
request.notifier.failed();
|
|
}
|
|
}
|
|
|
|
pub(crate) fn process<R: ImportDma>(&mut self, global: DmabufGlobal, renderer: &mut R, node: Option<DrmNode>) {
|
|
self.process_with(global, node, |buffer| renderer.import_dmabuf(buffer, None).is_ok());
|
|
}
|
|
|
|
pub(crate) fn process_with(
|
|
&mut self,
|
|
global: DmabufGlobal,
|
|
node: Option<DrmNode>,
|
|
mut import: impl FnMut(&Dmabuf) -> bool,
|
|
) {
|
|
self.queue.process(&global, |request| {
|
|
if request.notifier.client().is_none() {
|
|
request.notifier.failed();
|
|
return;
|
|
}
|
|
|
|
if !import(&request.buffer) {
|
|
request.notifier.failed();
|
|
return;
|
|
}
|
|
|
|
if let Some(node) = node {
|
|
request.buffer.set_node(node);
|
|
}
|
|
|
|
if let Err(error) = request.notifier.successful::<Ferese>() {
|
|
tracing::debug!(?error, "dma-buf client disappeared before import completed");
|
|
}
|
|
});
|
|
}
|
|
|
|
pub(crate) fn expire(&mut self, now: Instant) {
|
|
self.queue.remove_if(
|
|
|entry| entry.deadline <= now || entry.value.notifier.client().is_none(),
|
|
|entry| entry.value.notifier.failed(),
|
|
);
|
|
}
|
|
}
|
|
|
|
impl Ferese {
|
|
pub(crate) fn process_pending_dmabuf_imports(&mut self) {
|
|
self.dmabuf_imports.expire(Instant::now());
|
|
if self.dmabuf_imports.queue.deadline().is_some() {
|
|
if let Some(backend) = self.direct_backend.as_mut() {
|
|
backend.process_dmabuf_imports(&mut self.dmabuf_imports);
|
|
} else if let Some(backend) = self.nested_backend.as_ref() {
|
|
let mut backend = backend.borrow_mut();
|
|
// Nested Ferese publishes exactly one DMA-BUF global.
|
|
if let Some(global) = self.dmabuf_imports.globals.iter().next().copied() {
|
|
self.dmabuf_imports.process(global, backend.renderer(), None);
|
|
}
|
|
}
|
|
}
|
|
|
|
if let Some(deadline) = self.dmabuf_imports.queue.deadline() {
|
|
if self.dmabuf_imports.timer.is_none() {
|
|
match self
|
|
.loop_handle
|
|
.insert_source(Timer::from_deadline(deadline), |_, _, state| {
|
|
state.dmabuf_imports.expire(Instant::now());
|
|
if let Some(deadline) = state.dmabuf_imports.queue.deadline() {
|
|
TimeoutAction::ToInstant(deadline)
|
|
} else {
|
|
state.dmabuf_imports.timer = None;
|
|
TimeoutAction::Drop
|
|
}
|
|
}) {
|
|
Ok(token) => self.dmabuf_imports.timer = Some(token),
|
|
Err(error) => {
|
|
tracing::warn!(%error, "could not schedule DMA-BUF import deadline");
|
|
self.dmabuf_imports
|
|
.queue
|
|
.remove_if(|_| true, |entry| entry.value.notifier.failed());
|
|
}
|
|
}
|
|
}
|
|
} else if let Some(token) = self.dmabuf_imports.timer.take() {
|
|
self.loop_handle.remove(token);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use std::os::fd::AsRawFd;
|
|
|
|
#[test]
|
|
fn renderer_order_cannot_consume_another_globals_requests() {
|
|
for order in [[1, 2], [2, 1]] {
|
|
let now = Instant::now();
|
|
let mut queue = Queue::default();
|
|
queue.enqueue(2, 1, "B-only", now).unwrap();
|
|
let mut results = Vec::new();
|
|
for owner in order {
|
|
queue.process(&owner, |buffer| {
|
|
assert_eq!(owner, 2);
|
|
assert_eq!(buffer, "B-only");
|
|
results.push(true);
|
|
});
|
|
}
|
|
|
|
assert_eq!(results, [true]);
|
|
assert!(queue.pending.is_empty());
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn unavailable_imports_are_bounded_and_recover_before_the_deadline() {
|
|
let now = Instant::now();
|
|
let mut queue = Queue::default();
|
|
for index in 0..MAX_PENDING {
|
|
queue.enqueue(1, index / MAX_PER_CLIENT, index, now).unwrap();
|
|
}
|
|
|
|
assert_eq!(queue.enqueue(1, 99, 100, now), Err(100));
|
|
let mut completed = Vec::new();
|
|
queue.remove_if(
|
|
|entry| entry.deadline <= now + MAX_WAIT / 2,
|
|
|_| panic!("expired early"),
|
|
);
|
|
queue.process(&1, |value| completed.push(value));
|
|
assert_eq!(completed, (0..MAX_PENDING).collect::<Vec<_>>());
|
|
assert!(queue.deadline().is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn per_client_admission_leaves_room_for_other_clients() {
|
|
let now = Instant::now();
|
|
let mut queue = Queue::default();
|
|
for index in 0..MAX_PER_CLIENT {
|
|
queue.enqueue(1, 1, index, now).unwrap();
|
|
}
|
|
|
|
assert_eq!(queue.enqueue(2, 1, 100, now), Err(100));
|
|
queue.enqueue(2, 2, 101, now).unwrap();
|
|
}
|
|
|
|
#[test]
|
|
fn expiry_and_device_removal_release_resources_once_without_touching_other_owners() {
|
|
if !crate::startup_tests::private_runtime(
|
|
"dmabuf_imports::tests::expiry_and_device_removal_release_resources_once_without_touching_other_owners",
|
|
) {
|
|
return;
|
|
}
|
|
|
|
let now = Instant::now();
|
|
let mut queue = Queue::default();
|
|
let removed = tempfile::tempfile().unwrap();
|
|
let expired = tempfile::tempfile().unwrap();
|
|
let fd_removed = removed.as_raw_fd();
|
|
let fd_expired = expired.as_raw_fd();
|
|
queue.enqueue(1, 1, removed, now).unwrap();
|
|
queue.enqueue(2, 1, expired, now + Duration::from_secs(1)).unwrap();
|
|
let mut failures = 0;
|
|
queue.process(&1, |_| failures += 1);
|
|
// SAFETY: F_GETFD observes descriptor validity without modifying it.
|
|
assert_eq!(unsafe { libc::fcntl(fd_removed, libc::F_GETFD) }, -1);
|
|
assert_ne!(unsafe { libc::fcntl(fd_expired, libc::F_GETFD) }, -1);
|
|
queue.remove_if(|entry| entry.deadline <= now + MAX_WAIT, |_| panic!("expired early"));
|
|
queue.remove_if(
|
|
|entry| entry.deadline <= now + MAX_WAIT + Duration::from_secs(1),
|
|
|_| failures += 1,
|
|
);
|
|
assert_eq!(unsafe { libc::fcntl(fd_expired, libc::F_GETFD) }, -1);
|
|
assert_eq!(failures, 2);
|
|
assert!(queue.deadline().is_none());
|
|
}
|
|
}
|