diff --git a/src/cap_portal.rs b/src/cap_portal.rs index bff3400..2e870c4 100644 --- a/src/cap_portal.rs +++ b/src/cap_portal.rs @@ -1,4 +1,5 @@ use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::thread::{self, JoinHandle}; use anyhow::Result; @@ -33,6 +34,7 @@ pub struct CapPortal { struct PwThreadCtx { frame_tx: Sender, + dropped: AtomicU64, shutdown_read: OwnedFd, pw_fd: OwnedFd, node_id: u32, @@ -69,6 +71,7 @@ impl CapPortal { let ctx = PwThreadCtx { frame_tx, + dropped: AtomicU64::new(0), shutdown_read: unsafe { OwnedFd::from_raw_fd(efd) }, pw_fd, node_id, @@ -178,6 +181,7 @@ fn pipewire_thread(ctx: PwThreadCtx) { let PwThreadCtx { frame_tx, + dropped, shutdown_read, pw_fd, node_id, @@ -269,6 +273,7 @@ fn pipewire_thread(ctx: PwThreadCtx) { .process({ let format_info = format_info.clone(); let frame_tx = frame_tx.clone(); + let dropped = dropped; move |stream, _| { let raw_buf = unsafe { stream.dequeue_raw_buffer() }; if raw_buf.is_null() { @@ -341,7 +346,14 @@ fn pipewire_thread(ctx: PwThreadCtx) { pts, }; - let _ = frame_tx.send(PwEvent::Frame(frame)); + if let Err(crossbeam_channel::TrySendError::Full(_)) = + frame_tx.try_send(PwEvent::Frame(frame)) + { + let prev = dropped.fetch_add(1, Ordering::Relaxed); + if prev > 0 && prev % 30 == 0 { + tracing::warn!("dropped {prev} frames total: encoder backlog"); + } + } unsafe { stream.queue_raw_buffer(raw_buf) }; } })