diff --git a/src/cap_portal.rs b/src/cap_portal.rs index 3ab5cbc..0267aba 100644 --- a/src/cap_portal.rs +++ b/src/cap_portal.rs @@ -832,6 +832,13 @@ fn pipewire_thread(ctx: PwThreadCtx) { fps, } = ctx; + // PipeWire 三件套初始化(典型 PW 客户端架构): + // MainLoop —— 事件循环(epoll 后端),所有回调都在此线程派发。 + // Context —— 加载 PW 模块、管理代理对象的上下文,挂在 MainLoop 上。 + // Core —— 与 PipeWire daemon 的连接(此处用 connect_fd 走 Portal + // 下发的 socket fd 而非默认的 `pipewire-0`)。 + // 任一初始化失败都通过 event_tx 上报 PwCtrlEvent::Error 并退出本线程, + // 让主线程的 select 报告具体阶段错误。 let mainloop = match pw::main_loop::MainLoopBox::new(None) { Ok(ml) => ml, Err(e) => { @@ -893,8 +900,17 @@ fn pipewire_thread(ctx: PwThreadCtx) { } }; + // 共享的可变格式信息容器:Rc>>。 + // - Rc 单线程引用计数(PipeWire 回调全在同一线程),类比 Go 中"通过指针 + // 共享的可变全局变量"但带编译期 Send 约束。 + // - Cell> 提供内部可变性(无需 Mutex),通过 .get()/.set() + // 整体替换值——比 RefCell 更轻,因为这里值是 Copy 的元组。 + // - 类比 Go: var formatInfo = *(u32,u32,u32,u64) // 取地址 + atomic 赋值。 let format_info: Rc>> = Rc::new(Cell::new(None)); + // crossbeam channel 的 Sender 是 Clone + Send,每次 clone 给一个回调 + // 捕获,多回调可并发往同一 channel 投递事件。类比 Go: ch := make(chan T, 8) + // 各 goroutine 持有 ch 共享发送端。 let event_tx_state = event_tx.clone(); let _listener = stream .add_local_listener::<()>() @@ -975,6 +991,13 @@ fn pipewire_thread(ctx: PwThreadCtx) { let frame_tx = frame_tx.clone(); let dropped = dropped; move |stream, _| { + // 以下大量 unsafe 块均为对 PipeWire/libspa C API 的直接访问。 + // pipewire-rs 的 stream 类型只暴露 `dequeue_raw_buffer` / + // `queue_raw_buffer` 这类 unsafe 接口,因为返回的是 C 分配的 + // 裸 `*mut spa_buffer`,其生命周期由 PipeWire 控制(在 + // dequeue 与下一次 queue 之间稳定),Rust 类型系统无法表达。 + // 调用约定:每个 dequeue 必须恰好配一次 queue(包括所有错误 + // 退出路径),否则 PipeWire 会认为该 buffer 仍被使用而耗尽池。 let raw_buf = unsafe { stream.dequeue_raw_buffer() }; if raw_buf.is_null() { tracing::trace!("process: null raw_buf"); @@ -1064,6 +1087,11 @@ fn pipewire_thread(ctx: PwThreadCtx) { } // 构建帧数据对象,所有必要的帧信息已收集完毕 + // unsafe: OwnedFd::from_raw_fd 把刚刚 dup 出的 fd 所有权移交给 + // Rust 的 RAII 包装。此后 dup_fd 的关闭由 PwDmaBufFrame::Drop + // 负责,不能再在外部 close 它。from_raw_fd 之所以 unsafe,是 + // 因为调用方必须保证传入的 fd 此前没有任何 Owner(否则会 double + // close)。这里 libc::dup 刚返回的新 fd 满足该前提。 let frame = PwDmaBufFrame { fd: unsafe { OwnedFd::from_raw_fd(dup_fd) }, offset, @@ -1075,9 +1103,13 @@ fn pipewire_thread(ctx: PwThreadCtx) { pts, }; + // try_send 非阻塞投递;channel 容量=1(见 CapPortal::new), + // 当下游编码器落后时立刻返回 Full。 + // 类比 Go: select { case ch <- frame: default: /* drop */ } match frame_tx.try_send(frame) { Ok(()) => {} Err(crossbeam_channel::TrySendError::Full(_)) => { + // 丢帧计数(Relaxed 序,仅做统计;不要求与其他线程同步)。 dropped.fetch_add(1, Ordering::Relaxed); } Err(crossbeam_channel::TrySendError::Disconnected(_)) => {} @@ -1087,6 +1119,8 @@ fn pipewire_thread(ctx: PwThreadCtx) { }) .register(); + // 空的 SPA POD 参数数组——之前已在 param_changed 回调中接受了 PipeWire + // 推送的格式,这里不需要主动声明格式约束。`&mut [...]` 借用切片给 C API。 let mut params: [&pw::spa::pod::Pod; 0] = []; if let Err(e) = stream.connect( @@ -1113,14 +1147,24 @@ fn pipewire_thread(ctx: PwThreadCtx) { // previous detached helper thread approach. // 保存 mainloop 的原始指针,用于在 shutdown 回调中调用 pw_main_loop_quit // 这是安全的,因为回调只在 mainloop.run() 阻塞期间执行 + // + // `as_raw_ptr()` 返回 `*mut pw_main_loop`(裸指针,不带生命周期), + // 取裸指针本身是 safe 的——风险在使用它。下面 `pw_main_loop_quit` 的 + // unsafe 块依赖"回调仅在 run() 期间触发"这一 PipeWire 协议保证。 let mainloop_ptr = mainloop.as_raw_ptr(); + // 把 shutdown_read 的可读事件注册到 PipeWire loop 的 epoll/win32 等价物。 + // 每次 fd 变可读(CapPortal::drop 写入 8 字节触发),loop 在同一线程 + // 调用此闭包。返回的 _shutdown_source 在 drop 时自动从 loop 注销。 let _shutdown_source = loop_.add_io( shutdown_read, libspa::support::system::IoFlags::IN, move |fd| { // Drain the eventfd so it doesn't re-trigger let mut buf: u64 = 0; + // unsafe: libc::read 是 C 标准库 FFI。eventfd 语义保证 8 字节 + // 整数读,因此 &mut u64 转 *mut void + size_of::() 安全。 + // 返回值忽略——即使读失败也无法在此回调中做有意义处理。 let _ = unsafe { libc::read( fd.as_raw_fd(),