From 7074e43f5997eeb0529db13aba4063914f4dd48e Mon Sep 17 00:00:00 2001 From: Jeff Zhang <2854276+ferstar@users.noreply.github.com> Date: Fri, 23 Oct 2026 12:44:26 +0800 Subject: [PATCH 1/5] video: switch the encoder of running desktop streams on SIGUSR2 set-encoder and set-codec notify the running host through host.pid. Running desktop streams restart only the video encoder (first frame IDR, negotiated codec kept, VAAPI drops 4:4:4 to 4:2:0 with CaptureSettingSync). A codec change is logged and applies to the next desktop connection. SIGUSR2 does not re-probe. --- AGENTS.md | 4 +- README.en.md | 8 ++- README.md | 2 +- src/main.rs | 23 +++++++- src/video.rs | 45 +++++++++++++- src/video_select.rs | 126 ++++++++++++++++++++++++++++++++++++++++ src/video_wake_tests.rs | 2 +- src/webrtc.rs | 20 ++++++- 8 files changed, 220 insertions(+), 10 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 10f1dde..ab7adc3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -114,7 +114,7 @@ viewer does not reconnect them automatically. | Item | Meaning | |---|---| | `~/.config/uuremote/config.json` | user_id, device_id, client_id, system_id, machine_guid, token, csrf; optional `sign_key`, `ffmpeg_root`. `$UU_CONFIG` overrides the path | -| `~/.config/uuremote/host_settings.json` | persisted 安全 settings: auto_lock, keep_mute, mute_permanent; optional `video_encoder` (`auto`/`cpu`/`vaapi`/`nvenc`) and `video_codec` (`auto`/`h264`/`h265`), written by `set-encoder` / `set-codec` (0600, merge write). Non-`auto` beats `UU_VIDEO_ENCODER` / `UU_VIDEO_CODEC`. Applied on the next desktop stream | +| `~/.config/uuremote/host_settings.json` | persisted 安全 settings: auto_lock, keep_mute, mute_permanent; optional `video_encoder` (`auto`/`cpu`/`vaapi`/`nvenc`) and `video_codec` (`auto`/`h264`/`h265`), written by `set-encoder` / `set-codec` (0600, merge write). Non-`auto` beats `UU_VIDEO_ENCODER` / `UU_VIDEO_CODEC`. `set-encoder` and `set-codec` both send `SIGUSR2` to the pid in `host.pid` (same binary). Live desktop streams restart the encoder (`EncoderTask::restart`, sequence kept, first frame IDR) without dropping the peer. Codec preference is logged and applies on the next desktop stream. `SIGUSR2` does not re-probe. Hand edits: `kill -USR2 $(cat ~/.config/uuremote/host.pid)` | | `UU_DESKTOP_BACKEND` | `auto` (default) / `x11` / `wayland` / `portal` / `kms` (experimental feature) / `none` (headless: no display; desktop offers ignored; tool peers get a nominal 1920x1080 screen + Allow permissions, which the port-mapping viewer requires) | | `UU_DESKTOP_RESIZE=0` | X11: the viewer's 分辨率 choice scales the encoded stream instead of changing the display mode. Default: `src/desktop_mode.rs` switches the primary output via xrandr (adds a CVT-RB mode when missing, re-applies once if mutter resets it), restores the original mode for resolution_type 3 and 2 s after the last desktop stream ends | | `UU_NVENC_RGB=0` / `=1` | NVENC 4:2:0 without scaling takes packed BGRA/BGR0 directly (no CPU swscale). Default: on, except when a connected DRM connector belongs to an nvidia-bound card (desktop composited on the encoding GPU; #16 measured 82.6 vs 98 fps at 4K). `0` forces CPU conversion, `1` forces the direct path | @@ -165,7 +165,7 @@ for a new QR login. | `src/viewer_main.rs`, `src/viewer/` | experimental viewer (`--features native-viewer,viewer-sdl`): cloud join, WebRTC offer side, SDL desktop window (ffmpeg CLI decode), files/terminal/portmap/power/assist tools, device removal (`POST /api/v1/device/{id}/unbind`, verified live; shared devices use `/api/v1/device/share/{id}/unbind`, strings only), GTK launcher | | `c_engine/viewer_window.c`, `c_engine/viewer_tools.c` | viewer GTK launcher and tool windows (files, port mapping, terminal list) | -Diagnostic subcommands: `doctor` (paste-safe report: session, selected backend, engine plugins, uinput, encoder and codec with source, sign-in/service state; booleans and names only), `set-encoder auto|cpu|vaapi|nvenc`, `set-codec auto|h264|h265` (next desktop stream; illegal values exit 2), `login [--no-gui]`, `unlock-check` (password on stdin, prints only accepted/rejected), +Diagnostic subcommands: `doctor` (paste-safe report: session, selected backend, engine plugins, uinput, encoder and codec with source, sign-in/service state; booleans and names only), `set-encoder auto|cpu|vaapi|nvenc` and `set-codec auto|h264|h265` (both notify a running host with SIGUSR2; live encoder switch, negotiated codec unchanged, codec preference on the next desktop stream; illegal values exit 2), `login [--no-gui]`, `unlock-check` (password on stdin, prints only accepted/rejected), `desktop-probe [new.png]`, `desktop-video-probe` (`UU_PROBE_FPS`, `UU_PROBE_FRAMES`, `UU_PROBE_QUALITY` 1-5 = viewer bitrate preset, 5 auto; `UU_PROBE_CHROMA=444` 真彩; `UU_PROBE_INPUT` fixed video/image sequence, `UU_PROBE_LOOP=1` repeats), `wayland-input-probe`, `clipboard-acceptance`, `portal-acceptance`, `help`. Unknown arguments exit with code 2. Builds up to v0.1.0 did **not** reject them and started a second full host server instead, so check that a subcommand exists in the binary you run. diff --git a/README.en.md b/README.en.md index 926bc02..aa4ec80 100644 --- a/README.en.md +++ b/README.en.md @@ -360,7 +360,13 @@ stays cached until the process exits, except when NVIDIA presence (`/proc/driver or `/dev/nvidia0`) no longer matches the report: the next desktop connection then starts one background re-probe (`warm_auto_probe`) and still uses the cached encoder until the new result is stored. An NVENC failure is kept for 10 minutes, and the next desktop connection after that -starts the same background re-probe while still using the cached encoder. They take effect the next time a desktop connection starts. Explicit `vaapi` with 真彩 stays on VAAPI, +starts the same background re-probe while still using the cached encoder. `set-encoder` and +`set-codec` both send `SIGUSR2` to the host in `host.pid` when that process is this program. +An encoder change switches each live desktop stream without disconnecting and keeps the codec +negotiated for that session. A failed start follows the existing fallback (in-process, then the +FFmpeg process, then CPU). After a hand edit, `kill -USR2 $(cat ~/.config/uuremote/host.pid)` +does the same. `SIGUSR2` does not re-probe. A codec change takes effect the next time a desktop +connection starts; the running host only logs it. Explicit `vaapi` with 真彩 stays on VAAPI, encodes 4:2:0, and reports chroma 4:2:0 in CaptureSettingSync field 3. If the chosen encoder cannot start, the stream falls back to CPU and logs a warning. VAAPI is not an auto candidate until #33. diff --git a/README.md b/README.md index 471edb7..80e8dcf 100644 --- a/README.md +++ b/README.md @@ -317,7 +317,7 @@ systemctl --user restart uuremote-native.service | `UU_WOL_HELPER=1` | 实验性：让这台主机像官方 Windows 被控端一样充当「远程开机」辅助设备——登录后向网易上报本机局域网信息，收到云端唤醒请求时向同一局域网发送唤醒包（未经端到端验证） | | `UU_USE_PROXY=1` | 使用环境变量里的 HTTP/SOCKS 代理；默认忽略代理、直连网易服务器 | -「远程结束后自动锁屏」「静音运行」等设置保存在 `~/.config/uuremote/host_settings.json`。同一文件里的 `video_encoder`（`auto`/`cpu`/`vaapi`/`nvenc`）和 `video_codec`（`auto`/`h264`/`h265`）用 `uuremote-server set-encoder`、`uuremote-server set-codec` 写入（非法参数退出码 2）。非 `auto` 的值优先于 `UU_VIDEO_ENCODER` / `UU_VIDEO_CODEC`；`auto` 或留空则落到环境变量，再落到 auto（有 NVIDIA 时 NVENC 优先，失败再 CPU；没有 NVIDIA 时就是 CPU，不探测；codec 仍是硬件优先 H.265、CPU 用 H.264）。探测在启动时后台进行，不挡住第一路出图。成功结果缓存到进程退出，但 NVIDIA 是否存在（`/proc/driver/nvidia/version` 或 `/dev/nvidia0`）与报告不一致时，下一次桌面连接会在后台再探测一次（`warm_auto_probe`），这一路仍用缓存的编码器，直到新结果写入。NVENC 失败只缓存 10 分钟，到期后的桌面连接同样在后台再探测一次，这一路仍用当前结果。下次建立桌面连接时生效。显式 `vaapi` 请求真彩时仍用 VAAPI、编码 4:2:0，并在 CaptureSettingSync 的 field 3 回报 4:2:0。所选编码器在本机不可用时，建流回退 CPU 并记 warning。VAAPI 进入 auto 要等 #33。 +「远程结束后自动锁屏」「静音运行」等设置保存在 `~/.config/uuremote/host_settings.json`。同一文件里的 `video_encoder`（`auto`/`cpu`/`vaapi`/`nvenc`）和 `video_codec`（`auto`/`h264`/`h265`）用 `uuremote-server set-encoder`、`uuremote-server set-codec` 写入（非法参数退出码 2）。非 `auto` 的值优先于 `UU_VIDEO_ENCODER` / `UU_VIDEO_CODEC`；`auto` 或留空则落到环境变量，再落到 auto（有 NVIDIA 时 NVENC 优先，失败再 CPU；没有 NVIDIA 时就是 CPU，不探测；codec 仍是硬件优先 H.265、CPU 用 H.264）。探测在启动时后台进行，不挡住第一路出图。成功结果缓存到进程退出，但 NVIDIA 是否存在（`/proc/driver/nvidia/version` 或 `/dev/nvidia0`）与报告不一致时，下一次桌面连接会在后台再探测一次（`warm_auto_probe`），这一路仍用缓存的编码器，直到新结果写入。NVENC 失败只缓存 10 分钟，到期后的桌面连接同样在后台再探测一次，这一路仍用当前结果。`set-encoder` 和 `set-codec` 写完后若 `host.pid` 指向正在运行的本程序，都会发 `SIGUSR2`：编码器变化时进行中的桌面流立刻换编码器（不断开，沿用本会话已协商的 codec；新编码器起不来时按进程内 → FFmpeg 进程 → CPU 回退）。手改文件后可 `kill -USR2 $(cat ~/.config/uuremote/host.pid)`。`SIGUSR2` 不重新探测。codec 偏好仍在下次建立桌面连接时生效（运行中的 host 只记日志，不重开编码）。显式 `vaapi` 请求真彩时仍用 VAAPI、编码 4:2:0，并在 CaptureSettingSync 的 field 3 回报 4:2:0。所选编码器在本机不可用时，建流回退 CPU 并记 warning。VAAPI 进入 auto 要等 #33。 自定义码率按客户端选择的 Mbps 生效，例如 1 Mbps 对应 1000 kbit/s 的编码上限。 diff --git a/src/main.rs b/src/main.rs index 6006ffa..d8ef96c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -144,10 +144,26 @@ fn set_video_pref() -> Result<(), Box> { let selection = video_select::selection(); let codec = match selection.codec_pref { video_select::CodecPref::Auto => "auto", video_select::CodecPref::H264 => "h264", video_select::CodecPref::H265 => "h265" }; println!("encoder={:?} source={} codec_pref={codec} source={}", selection.encoder, video_select::encoder_source_label(selection.encoder_source), video_select::source_label(selection.codec_source)); - println!("takes effect the next time a desktop connection starts"); + // Both subcommands wake the host. Encoder changes restart live desktop streams; codec changes are logged only. + match notify_host(libc::SIGUSR2) { + Ok(()) => println!("notified the running host"), + Err(_) => {} + } + if which == "set-codec" { + println!("takes effect the next time a desktop connection starts"); + } Ok(()) } +/// Signal the host recorded in host.pid when its executable matches this one. +fn notify_host(signal: i32) -> Result<(), ()> { + let pid = std::fs::read_to_string(host_pid_path()).ok().and_then(|s| s.trim().parse::().ok()).filter(|p| *p > 1).ok_or(())?; + let exe = std::fs::read_link(format!("/proc/{pid}/exe")).ok(); + let own = std::env::current_exe().ok(); + if exe.as_ref().and_then(|p| p.file_name()) != own.as_ref().and_then(|p| p.file_name()) { return Err(()); } + if unsafe { libc::kill(pid, signal) } == 0 { Ok(()) } else { Err(()) } +} + /// Explicit command words for set-encoder / set-codec. Empty and whitespace are illegal. pub(crate) fn cli_pref_ok(which: &str, value: &str) -> bool { let value = value.trim(); @@ -198,11 +214,14 @@ mod cli_exit_tests { } } /// SIGUSR1 (`uuremote-server show`) opens the host window. +/// SIGUSR2 (set-encoder, set-codec, or `kill -USR2`) re-reads host_settings. Encoder changes switch live desktop streams; codec changes are logged for the next connection. fn present_on_usr1(ui: std::sync::Arc) { - // Install the handler before publishing the pid: SIGUSR1's default action terminates the process. + // Install the handlers before publishing the pid: the default action of either signal terminates the process. let Ok(mut usr1) = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::user_defined1()) else { return }; + let Ok(mut usr2) = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::user_defined2()) else { return }; write_host_pid(); tokio::spawn(async move { while usr1.recv().await.is_some() { ui.present(); } }); + tokio::spawn(async move { while usr2.recv().await.is_some() { crate::video_select::publish_reload(); } }); } fn host_pid_path() -> std::path::PathBuf { config::path().with_file_name("host.pid") } diff --git a/src/video.rs b/src/video.rs index dc01a1a..1dd4e04 100644 --- a/src/video.rs +++ b/src/video.rs @@ -1094,9 +1094,10 @@ fn keep_stream_display_awake() { if result.is_err() { log::warn!("Display wake failed; continuing desktop stream and retrying next interval"); } } +/// `sync_tx` receives a Message 25 payload when a live encoder switch changes the encoded chroma. pub async fn stream_desktop(pc: Arc, track: Arc, mut size: (u32, u32), display: String, portal:Option>, mut settings: tokio::sync::watch::Receiver, applied: Arc>, idr: Arc, - selection: crate::video_select::Selection) -> Result<(), String> { + selection: crate::video_select::Selection, sync_tx: Option>>) -> Result<(), String> { let _desktop_mode = crate::desktop_mode::StreamGuard::new(); loop { if transport_closed(&pc) { return Ok(()); } @@ -1117,7 +1118,13 @@ pub async fn stream_desktop(pc: Arc, track: Arc, track: Arc {} reason={} chroma={chroma}", + format!("{:?}", live_selection.encoder).to_ascii_lowercase(), + format!("{encoder:?}").to_ascii_lowercase(), + crate::video_select::encoder_source_label(fresh.encoder_source)); + live_selection.encoder = encoder; + live_selection.encoder_source = fresh.encoder_source; + seen_codec_pref = fresh.codec_pref; + if plan.chroma_downgraded { + applied.send_modify(|value| value.chroma = plan.chroma); + current.chroma = plan.chroma; + settings.borrow_and_update(); + if let Some(sync) = sync_tx.as_ref() { + let native = x11_resized(&display, portal_stream, size).unwrap_or(size); + let payload = crate::capture_setting::sync_payload(¤t, native); + let _ = sync.send(payload); + } + } + // In-process engines honor this on the first frame; the process pipeline starts on an IDR. + idr.store(true, std::sync::atomic::Ordering::Relaxed); + reader.restart(open(size, live_selection, current)).await; + } else if plan.is_some() { + log::info!("Codec preference changed; takes effect the next time a desktop connection starts"); + seen_codec_pref = fresh.codec_pref; + live_selection.codec_pref = fresh.codec_pref; + live_selection.codec_source = fresh.codec_source; + } + } let next = *settings.borrow_and_update(); if next != current || send_interrupted { let discard = next.codec != current.codec || next.source_generation != current.source_generation || next.source_id != current.source_id || send_interrupted; diff --git a/src/video_select.rs b/src/video_select.rs index c4ee64e..9de461a 100644 --- a/src/video_select.rs +++ b/src/video_select.rs @@ -239,6 +239,87 @@ pub fn clear_probe() { if let Ok(mut slot) = cache().lock() { *slot = None; } } +/// Why a live desktop encoder was asked to change. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum SwitchReason { + /// Effective encoder differs. The log line uses `Selection::encoder_source`, not this variant. + Encoder, + /// Codec preference changed. The negotiated codec stays until the next stream. + CodecDeferred, +} + +/// What SIGUSR2 should do to one live desktop stream. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct SwitchPlan { + /// Restart the encoder. None keeps the running encoder (codec-only or no change). + pub encoder: Option, + pub reason: SwitchReason, + /// Chroma the restarted encoder can actually produce. VAAPI has no 4:4:4. + pub chroma: u64, + /// True when the request was 4:4:4 and the new encoder must report 4:2:0. + pub chroma_downgraded: bool, +} + +/// Compare the selection this stream is using with a freshly read one. +/// Codec preference is the stored preference, not the negotiated m-line codec. +pub fn plan_switch(active_encoder: Encoder, previous_codec: CodecPref, next: &Selection, active_chroma: u64) -> Option { + if next.encoder != active_encoder { + let chroma = if active_chroma == crate::capture_setting::CHROMA_444 && !crate::video::full_chroma_for(next.encoder) { + crate::capture_setting::CHROMA_420 + } else { + active_chroma + }; + return Some(SwitchPlan { + encoder: Some(next.encoder), + reason: SwitchReason::Encoder, + chroma, + chroma_downgraded: chroma != active_chroma, + }); + } + if next.codec_pref != previous_codec { + return Some(SwitchPlan { + encoder: None, + reason: SwitchReason::CodecDeferred, + chroma: active_chroma, + chroma_downgraded: false, + }); + } + None +} + +/// Monotonic generation bumped by SIGUSR2. Streams watch it and re-read `selection()`. +fn reload_generation() -> &'static std::sync::atomic::AtomicU64 { + static GEN: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + &GEN +} + +pub fn reload_generation_value() -> u64 { + reload_generation().load(std::sync::atomic::Ordering::Relaxed) +} + +/// SIGUSR2: ask live desktop streams to re-read host_settings. Does not clear the probe cache. +pub fn notify_reload() { + reload_generation().fetch_add(1, std::sync::atomic::Ordering::Relaxed); +} + +pub fn reload_watch() -> &'static tokio::sync::watch::Sender { + static WATCH: std::sync::OnceLock> = std::sync::OnceLock::new(); + WATCH.get_or_init(|| { + let (tx, _) = tokio::sync::watch::channel(reload_generation_value()); + tx + }) +} + +pub fn subscribe_reload() -> tokio::sync::watch::Receiver { + reload_watch().subscribe() +} + +/// Publish the generation so every subscribed desktop stream wakes. +pub fn publish_reload() { + notify_reload(); + let _ = reload_watch().send(reload_generation_value()); +} + /// Setting (non-auto) beats the environment, which beats the NVIDIA/CPU default. pub fn selection() -> Selection { selection_with(cached_probe().as_ref()) @@ -486,6 +567,51 @@ mod tests { assert_eq!(resolve_pair(None, None, None, Some("h264"), false, Some(&va)).codec_pref, CodecPref::H264); } + fn selection_of(encoder: Encoder, codec: CodecPref) -> Selection { + Selection { encoder, encoder_source: Source::Setting, codec_pref: codec, codec_source: Source::Setting } + } + + #[test] + fn live_switch_restarts_encoder_and_defers_codec() { + let nv = selection_of(Encoder::Nvenc, CodecPref::Auto); + assert!(plan_switch(Encoder::Nvenc, CodecPref::Auto, &nv, crate::capture_setting::CHROMA_420).is_none()); + let to_cpu = plan_switch(Encoder::Nvenc, CodecPref::Auto, &selection_of(Encoder::Cpu, CodecPref::Auto), crate::capture_setting::CHROMA_444).unwrap(); + assert_eq!(to_cpu.encoder, Some(Encoder::Cpu)); + assert_eq!(to_cpu.reason, SwitchReason::Encoder); + assert!(!to_cpu.chroma_downgraded); + assert_eq!(to_cpu.chroma, crate::capture_setting::CHROMA_444); + let to_vaapi = plan_switch(Encoder::Nvenc, CodecPref::H265, &selection_of(Encoder::Vaapi, CodecPref::H265), crate::capture_setting::CHROMA_444).unwrap(); + assert_eq!(to_vaapi.encoder, Some(Encoder::Vaapi)); + assert!(to_vaapi.chroma_downgraded); + assert_eq!(to_vaapi.chroma, crate::capture_setting::CHROMA_420); + let codec = plan_switch(Encoder::Nvenc, CodecPref::Auto, &selection_of(Encoder::Nvenc, CodecPref::H264), crate::capture_setting::CHROMA_420).unwrap(); + assert_eq!(codec.encoder, None); + assert_eq!(codec.reason, SwitchReason::CodecDeferred); + let both = plan_switch(Encoder::Cpu, CodecPref::Auto, &selection_of(Encoder::Nvenc, CodecPref::H264), crate::capture_setting::CHROMA_420).unwrap(); + assert_eq!(both.encoder, Some(Encoder::Nvenc)); + assert_eq!(both.reason, SwitchReason::Encoder); + // The log reason is the source of the new selection (setting / env / auto), not a fixed word. + for (source, label) in [(Source::Setting, "setting"), (Source::Env, "env"), (Source::Auto, "auto")] { + let mut next = selection_of(Encoder::Vaapi, CodecPref::Auto); + next.encoder_source = source; + let plan = plan_switch(Encoder::Nvenc, CodecPref::Auto, &next, crate::capture_setting::CHROMA_420).unwrap(); + assert!(plan.encoder.is_some()); + assert_eq!(encoder_source_label(next.encoder_source), label); + } + } + + #[test] + fn reload_does_not_clear_the_probe_cache() { + let _guard = PROBE_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let report = probe_report(true, &[ProbeOutcome::Ok, ProbeOutcome::Ok]); + store_probe(report.clone()); + let before = reload_generation_value(); + notify_reload(); + assert_eq!(reload_generation_value(), before + 1); + assert_eq!(cached_probe(), Some(report)); + clear_probe(); + } + /// Codec at CONTROL open is `preferred_codec_for` on the session snapshot. /// Auto + CPU uses H.264 when both codecs are negotiated; auto + NVENC keeps H.265. /// Forced codec ignores the encoder. 真彩 does not switch codec (true_color_keeps_codec_and_vaapi_reports_420). diff --git a/src/video_wake_tests.rs b/src/video_wake_tests.rs index 5caf0d3..e14fef4 100644 --- a/src/video_wake_tests.rs +++ b/src/video_wake_tests.rs @@ -78,7 +78,7 @@ async fn exercise(fail_wake: usize, fail_capture: bool) { let (settings_tx, settings_rx) = tokio::sync::watch::channel(settings); let selection = crate::video_select::resolve_pair(Some("cpu"), None, Some("h264"), None, false, None); let mut task = tokio::spawn(stream_desktop(host.clone(), track.clone(), (640, 480), - DISPLAY.into(), None, settings_rx, Arc::new(settings_tx), Arc::new(AtomicBool::new(false)), selection)); + DISPLAY.into(), None, settings_rx, Arc::new(settings_tx), Arc::new(AtomicBool::new(false)), selection, None)); let result = if fail_capture { Some((&mut task).await.unwrap()) } else { diff --git a/src/webrtc.rs b/src/webrtc.rs index fb7fc3b..25105de 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -305,6 +305,10 @@ impl WebRtcSession { let portal=capture.portal.clone(); let stream_desktop = portal_desktop.clone(); let stream_selection = video_selection; + // Live encoder switch (SIGUSR2) sends CaptureSettingSync when chroma is downgraded. + let (chroma_sync_tx, chroma_sync_rx) = tokio::sync::mpsc::unbounded_channel::>(); + let chroma_sync = (desktop && !cfg!(test)).then_some(chroma_sync_tx); + let chroma_sync_rx = Arc::new(std::sync::Mutex::new((!cfg!(test)).then_some(chroma_sync_rx))); // File/terminal/port-mapping connections carry no desktop video. let control_video_track = video_track.clone(); let stream_settings = video_settings.clone(); @@ -330,7 +334,7 @@ impl WebRtcSession { let (capture_size, capture_display, portal) = screen.as_ref().map_or( (capture_size, capture_display, portal), |screen| ((screen.width, screen.height), screen.display.clone(), screen.portal.clone())); - if let Err(e) = crate::video::stream_desktop(streaming_pc, video_track, capture_size, capture_display, portal, video_settings_rx, stream_settings, idr_request, stream_selection).await { + if let Err(e) = crate::video::stream_desktop(streaming_pc, video_track, capture_size, capture_display, portal, video_settings_rx, stream_settings, idr_request, stream_selection, chroma_sync.clone()).await { // A send racing the viewer's close fails with a misleading track error. if matches!(state_pc.connection_state(), RTCPeerConnectionState::Closed | RTCPeerConnectionState::Failed) { log::info!("Live desktop stream ended: peer connection closed ({})", e); @@ -421,6 +425,7 @@ impl WebRtcSession { let peer_video_track = control_video_track.clone(); let peer_actions = peer_actions.clone(); let peer_selection = video_selection; + let chroma_sync_rx = chroma_sync_rx.clone(); if label == "FILE_DATA_CHANNEL" { *file_channel.lock().unwrap() = Some(d.clone()); } Box::pin(async move { @@ -443,6 +448,19 @@ impl WebRtcSession { } // Ordered outbound queue for protocol replies produced outside this callback. let (outbound, mut outbound_rx) = tokio::sync::mpsc::unbounded_channel::>(); + if label == "CONTROL_DATA_CHANNEL" { + if let Some(mut chroma_sync_rx) = chroma_sync_rx.lock().unwrap().take() { + let sync_out = outbound.clone(); + let sync_seq = reply_sequence.clone(); + tokio::spawn(async move { + while let Some(payload) = chroma_sync_rx.recv().await { + let timestamp = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); + let seq = sync_seq.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let _ = sync_out.send(crate::initialization::envelope(seq, timestamp, 25, &payload)); + } + }); + } + } { let out_channel = d.clone(); let out_label = label.clone(); From dd2e24bbfc8af6b47d88727fb1c90a93a52daabc Mon Sep 17 00:00:00 2001 From: Jeff Zhang <2854276+ferstar@users.noreply.github.com> Date: Fri, 23 Oct 2026 15:43:25 +0800 Subject: [PATCH 2/5] video: install SIGUSR2 without remote assist, keep the switched encoder on restarts, consume reload notifications --- src/main.rs | 44 ++++++++++++++++++++++++++++++++++++++------ src/video.rs | 29 +++++++++++++++++++---------- src/video_select.rs | 30 ++++++++++++++++++++++++++++++ src/webrtc.rs | 4 +++- 4 files changed, 90 insertions(+), 17 deletions(-) diff --git a/src/main.rs b/src/main.rs index d8ef96c..b44c70f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -212,16 +212,46 @@ mod cli_exit_tests { assert_eq!(super::reject_cli(Some("set-encoder"), 3, super::cli_pref_ok("set-encoder", "")), Some(2)); assert_eq!(super::reject_cli(Some("set-codec"), 3, super::cli_pref_ok("set-codec", " ")), Some(2)); } + + /// SIGUSR2 is installed with no assist UI. `host.pid` is written only after that handler exists. + #[test] + fn sigusr2_handler_installs_without_assist() { + let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + rt.block_on(async { + let path = super::host_pid_path(); + if let Some(dir) = path.parent() { std::fs::create_dir_all(dir).unwrap(); } + assert!(super::install_host_signals(None)); + let pid = std::fs::read_to_string(&path).unwrap(); + assert_eq!(pid.trim(), std::process::id().to_string()); + }); + } } /// SIGUSR1 (`uuremote-server show`) opens the host window. /// SIGUSR2 (set-encoder, set-codec, or `kill -USR2`) re-reads host_settings. Encoder changes switch live desktop streams; codec changes are logged for the next connection. -fn present_on_usr1(ui: std::sync::Arc) { - // Install the handlers before publishing the pid: the default action of either signal terminates the process. - let Ok(mut usr1) = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::user_defined1()) else { return }; - let Ok(mut usr2) = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::user_defined2()) else { return }; +/// The default action of either signal terminates the process, so both handlers are installed before `host.pid` is written. +/// SIGUSR2 is independent of remote assistance: `UU_REMOTE_ASSIST=0` and a failed assist setup still switch live encoders. +fn install_host_signals(ui: Option>) -> bool { + // Called again when assist attaches: SIGUSR2 is already installed, so only add SIGUSR1. + static USR2: std::sync::Once = std::sync::Once::new(); + let mut usr2_ok = true; + USR2.call_once(|| { + match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::user_defined2()) { + Ok(mut usr2) => { tokio::spawn(async move { while usr2.recv().await.is_some() { crate::video_select::publish_reload(); } }); } + Err(_) => usr2_ok = false, + } + }); + if !usr2_ok { return false; } + if let Some(ui) = ui { + if let Ok(mut usr1) = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::user_defined1()) { + tokio::spawn(async move { while usr1.recv().await.is_some() { ui.present(); } }); + } + } write_host_pid(); - tokio::spawn(async move { while usr1.recv().await.is_some() { ui.present(); } }); - tokio::spawn(async move { while usr2.recv().await.is_some() { crate::video_select::publish_reload(); } }); + true +} + +fn present_on_usr1(ui: std::sync::Arc) { + let _ = install_host_signals(Some(ui)); } fn host_pid_path() -> std::path::PathBuf { config::path().with_file_name("host.pid") } @@ -540,6 +570,8 @@ async fn main() -> Result<(), Box> { virtual_audio::init(); let _audio_cleanup=virtual_audio::Cleanup; log::info!("Starting uu-remote-linux host v{}...", env!("CARGO_PKG_VERSION")); + // Before host.pid, and before assist: a missing or failed assist session must still handle SIGUSR2. + let _ = install_host_signals(None); video::warm_auto_probe(); if !desktop_backend::headless() && desktop_backend::detect()? == desktop_backend::DesktopBackend::X11 && desktop_backend::wayland_session_running() { diff --git a/src/video.rs b/src/video.rs index 1dd4e04..3967139 100644 --- a/src/video.rs +++ b/src/video.rs @@ -1095,9 +1095,15 @@ fn keep_stream_display_awake() { } /// `sync_tx` receives a Message 25 payload when a live encoder switch changes the encoded chroma. -pub async fn stream_desktop(pc: Arc, track: Arc, mut size: (u32, u32), display: String, portal:Option>, - mut settings: tokio::sync::watch::Receiver, applied: Arc>, idr: Arc, +pub async fn stream_desktop(pc: Arc, track: Arc, size: (u32, u32), display: String, portal:Option>, + settings: tokio::sync::watch::Receiver, applied: Arc>, idr: Arc, selection: crate::video_select::Selection, sync_tx: Option>>) -> Result<(), String> { + stream_desktop_at(pc, track, size, display, portal, settings, applied, idr, selection, sync_tx, crate::video_select::reload_generation_value()).await +} + +pub async fn stream_desktop_at(pc: Arc, track: Arc, mut size: (u32, u32), display: String, portal:Option>, + mut settings: tokio::sync::watch::Receiver, applied: Arc>, idr: Arc, + selection: crate::video_select::Selection, sync_tx: Option>>, reload_at_session: u64) -> Result<(), String> { let _desktop_mode = crate::desktop_mode::StreamGuard::new(); loop { if transport_closed(&pc) { return Ok(()); } @@ -1122,9 +1128,10 @@ pub async fn stream_desktop(pc: Arc, track: Arc, track: Arc, track: Arc, track: Arc, track: Arc {}x{}); restarting capture", size.0, size.1, resized.0, resized.1); size = resized; while receiver.try_recv().is_ok() {} - reader.0 = tokio::spawn(open(size, selection, current)); + reader.0 = tokio::spawn(open(size, live_selection, current)); continue; } if let (Err(error), VideoCodec::H265, true) = (&result, current.codec, track.supports(VideoCodec::H264)) { @@ -1217,7 +1226,7 @@ pub async fn stream_desktop(pc: Arc, track: Arc tokio::sync::watch::Receiver { reload_watch().subscribe() } +/// True when SIGUSR2 has bumped the generation since the session snapshot. +/// A background auto probe that finishes during setup does not bump it. +pub fn reload_since_session(at_session: u64, current: u64) -> bool { + current != at_session +} + /// Publish the generation so every subscribed desktop stream wakes. pub fn publish_reload() { notify_reload(); @@ -609,9 +615,33 @@ mod tests { notify_reload(); assert_eq!(reload_generation_value(), before + 1); assert_eq!(cached_probe(), Some(report)); + assert!(reload_since_session(before, reload_generation_value())); + assert!(!reload_since_session(reload_generation_value(), reload_generation_value())); clear_probe(); } + #[test] + fn probe_finish_is_not_a_reload_since_the_session_snapshot() { + let at_session = reload_generation_value(); + // A finished background probe updates the cache only. It does not bump the generation. + assert!(!reload_since_session(at_session, reload_generation_value())); + notify_reload(); + assert!(reload_since_session(at_session, reload_generation_value())); + } + + #[test] + fn mark_reload_seen_clears_has_changed() { + let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + rt.block_on(async { + let mut watch = subscribe_reload(); + let _ = watch.borrow_and_update(); + publish_reload(); + assert!(watch.has_changed().unwrap()); + let _ = watch.borrow_and_update(); + assert!(!watch.has_changed().unwrap()); + }); + } + /// Codec at CONTROL open is `preferred_codec_for` on the session snapshot. /// Auto + CPU uses H.264 when both codecs are negotiated; auto + NVENC keeps H.265. /// Forced codec ignores the encoder. 真彩 does not switch codec (true_color_keeps_codec_and_vaapi_reports_420). diff --git a/src/webrtc.rs b/src/webrtc.rs index 25105de..7583424 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -65,6 +65,8 @@ impl WebRtcSession { // Encoder, codec preference and 4:4:4 stay fixed for this peer (codec is negotiated at CONTROL open). // Read once before the HEVC probe so a concurrent set-encoder/set-codec cannot split capability from CONTROL. + // reload_at_session is the generation at that snapshot. A probe that finishes later is not a reload. + let reload_at_session = crate::video_select::reload_generation_value(); let video_selection = crate::video_select::selection(); let hevc = desktop && crate::video::hevc_available(video_selection.encoder).await; if hevc { crate::video_track::register_hevc(&mut m).map_err(|e| format!("Register HEVC failed: {e}"))?; } @@ -334,7 +336,7 @@ impl WebRtcSession { let (capture_size, capture_display, portal) = screen.as_ref().map_or( (capture_size, capture_display, portal), |screen| ((screen.width, screen.height), screen.display.clone(), screen.portal.clone())); - if let Err(e) = crate::video::stream_desktop(streaming_pc, video_track, capture_size, capture_display, portal, video_settings_rx, stream_settings, idr_request, stream_selection, chroma_sync.clone()).await { + if let Err(e) = crate::video::stream_desktop_at(streaming_pc, video_track, capture_size, capture_display, portal, video_settings_rx, stream_settings, idr_request, stream_selection, chroma_sync.clone(), reload_at_session).await { // A send racing the viewer's close fails with a misleading track error. if matches!(state_pc.connection_state(), RTCPeerConnectionState::Closed | RTCPeerConnectionState::Failed) { log::info!("Live desktop stream ended: peer connection closed ({})", e); From 5911a80c32361a0b30a3607bce37a41ce6b23897 Mon Sep 17 00:00:00 2001 From: Jeff Zhang <2854276+ferstar@users.noreply.github.com> Date: Fri, 23 Oct 2026 16:48:50 +0800 Subject: [PATCH 3/5] Host signals: install before host.pid, exact-binary SIGUSR2, report when the host is not notified --- src/main.rs | 79 +++++++++++++++++++++++++++++++++++++-------- src/video_select.rs | 3 +- 2 files changed, 67 insertions(+), 15 deletions(-) diff --git a/src/main.rs b/src/main.rs index b44c70f..2d9961e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -147,7 +147,9 @@ fn set_video_pref() -> Result<(), Box> { // Both subcommands wake the host. Encoder changes restart live desktop streams; codec changes are logged only. match notify_host(libc::SIGUSR2) { Ok(()) => println!("notified the running host"), - Err(_) => {} + Err(_) => println!( + "host not running, or the running host is not this binary: takes effect the next time a desktop connection starts, or after restarting the service" + ), } if which == "set-codec" { println!("takes effect the next time a desktop connection starts"); @@ -155,13 +157,31 @@ fn set_video_pref() -> Result<(), Box> { Ok(()) } -/// Signal the host recorded in host.pid when its executable matches this one. +/// Signal the host recorded in host.pid only when `/proc//exe` is exactly +/// `canonicalize(current_exe())`. A filename match is not enough: an older host +/// has no SIGUSR2 handler, and the default action terminates the process. +/// A replaced binary reads back as `… (deleted)` and does not match. fn notify_host(signal: i32) -> Result<(), ()> { - let pid = std::fs::read_to_string(host_pid_path()).ok().and_then(|s| s.trim().parse::().ok()).filter(|p| *p > 1).ok_or(())?; - let exe = std::fs::read_link(format!("/proc/{pid}/exe")).ok(); - let own = std::env::current_exe().ok(); - if exe.as_ref().and_then(|p| p.file_name()) != own.as_ref().and_then(|p| p.file_name()) { return Err(()); } - if unsafe { libc::kill(pid, signal) } == 0 { Ok(()) } else { Err(()) } + let pid = std::fs::read_to_string(host_pid_path()) + .ok() + .and_then(|s| s.trim().parse::().ok()) + .filter(|p| *p > 1) + .ok_or(())?; + let exe = std::fs::read_link(format!("/proc/{pid}/exe")).map_err(|_| ())?; + let own = std::fs::canonicalize(std::env::current_exe().map_err(|_| ())?).map_err(|_| ())?; + if !same_host_binary(&exe, &own) { + return Err(()); + } + if unsafe { libc::kill(pid, signal) } == 0 { + Ok(()) + } else { + Err(()) + } +} + +/// `/proc//exe` must equal the canonical path of this binary. A trailing ` (deleted)` does not match. +fn same_host_binary(proc_exe: &std::path::Path, own: &std::path::Path) -> bool { + proc_exe == own } /// Explicit command words for set-encoder / set-codec. Empty and whitespace are illegal. @@ -213,26 +233,54 @@ mod cli_exit_tests { assert_eq!(super::reject_cli(Some("set-codec"), 3, super::cli_pref_ok("set-codec", " ")), Some(2)); } - /// SIGUSR2 is installed with no assist UI. `host.pid` is written only after that handler exists. + /// SIGUSR1 and SIGUSR2 are installed with no assist UI. `host.pid` is written only after both exist. #[test] - fn sigusr2_handler_installs_without_assist() { + fn signal_handlers_install_without_assist() { let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); rt.block_on(async { - let path = super::host_pid_path(); - if let Some(dir) = path.parent() { std::fs::create_dir_all(dir).unwrap(); } + let _guard = crate::config::test_env_lock().lock().unwrap(); + let dir = + std::env::temp_dir().join(format!("uuremote-signal-test-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::env::set_var("UU_CONFIG", dir.join("config.json")); assert!(super::install_host_signals(None)); - let pid = std::fs::read_to_string(&path).unwrap(); + let pid = std::fs::read_to_string(super::host_pid_path()).unwrap(); assert_eq!(pid.trim(), std::process::id().to_string()); + std::env::remove_var("UU_CONFIG"); + let _ = std::fs::remove_dir_all(&dir); }); } + + #[test] + fn sigusr2_requires_the_same_canonical_binary() { + let own = std::path::Path::new("/home/user/.local/bin/uuremote-server"); + assert!(super::same_host_binary(own, own)); + assert!(!super::same_host_binary( + std::path::Path::new("/tmp/unpack/uuremote-server"), + own + )); + assert!(!super::same_host_binary( + std::path::Path::new("/home/user/.local/bin/uuremote-server (deleted)"), + own + )); + } } /// SIGUSR1 (`uuremote-server show`) opens the host window. /// SIGUSR2 (set-encoder, set-codec, or `kill -USR2`) re-reads host_settings. Encoder changes switch live desktop streams; codec changes are logged for the next connection. /// The default action of either signal terminates the process, so both handlers are installed before `host.pid` is written. /// SIGUSR2 is independent of remote assistance: `UU_REMOTE_ASSIST=0` and a failed assist setup still switch live encoders. +/// SIGUSR1 is installed first as a no-op so `show` cannot terminate a host that has no assist window yet. fn install_host_signals(ui: Option>) -> bool { - // Called again when assist attaches: SIGUSR2 is already installed, so only add SIGUSR1. + static USR1: std::sync::Once = std::sync::Once::new(); static USR2: std::sync::Once = std::sync::Once::new(); + let mut usr1_ok = true; + USR1.call_once(|| { + match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::user_defined1()) { + Ok(mut usr1) => { tokio::spawn(async move { while usr1.recv().await.is_some() {} }); } + Err(_) => usr1_ok = false, + } + }); + if !usr1_ok { return false; } let mut usr2_ok = true; USR2.call_once(|| { match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::user_defined2()) { @@ -241,6 +289,7 @@ fn install_host_signals(ui: Option>) -> bool { } }); if !usr2_ok { return false; } + // Assist attaches later and replaces the no-op with the window. A second registration replaces the first. if let Some(ui) = ui { if let Ok(mut usr1) = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::user_defined1()) { tokio::spawn(async move { while usr1.recv().await.is_some() { ui.present(); } }); @@ -259,7 +308,9 @@ fn host_pid_path() -> std::path::PathBuf { config::path().with_file_name("host.p fn write_host_pid() { use std::os::unix::fs::OpenOptionsExt; use std::io::Write; - if let Ok(mut f) = std::fs::OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(host_pid_path()) { + let path = host_pid_path(); + if let Some(dir) = path.parent() { let _ = std::fs::create_dir_all(dir); } + if let Ok(mut f) = std::fs::OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(path) { let _ = write!(f, "{}", std::process::id()); } } diff --git a/src/video_select.rs b/src/video_select.rs index c775034..84876eb 100644 --- a/src/video_select.rs +++ b/src/video_select.rs @@ -323,7 +323,8 @@ pub fn reload_since_session(at_session: u64, current: u64) -> bool { /// Publish the generation so every subscribed desktop stream wakes. pub fn publish_reload() { notify_reload(); - let _ = reload_watch().send(reload_generation_value()); + // send() drops the value when nobody is subscribed (the gap between desktop connections). + reload_watch().send_replace(reload_generation_value()); } /// Setting (non-auto) beats the environment, which beats the NVIDIA/CPU default. From b9de0988e0644717086d6972bd26aa78fd2b071e Mon Sep 17 00:00:00 2001 From: Jeff Zhang <2854276+ferstar@users.noreply.github.com> Date: Fri, 23 Oct 2026 17:14:43 +0800 Subject: [PATCH 4/5] Encoder sync: report the active Wayland output size; serialize reload-generation tests --- src/video.rs | 3 ++- src/video_select.rs | 7 +++++++ src/wayland.rs | 35 +++++++++++++++++++++++++++++++++++ 3 files changed, 44 insertions(+), 1 deletion(-) diff --git a/src/video.rs b/src/video.rs index 70acc9b..e4cb501 100644 --- a/src/video.rs +++ b/src/video.rs @@ -1225,7 +1225,8 @@ pub async fn stream_desktop_at(pc: Arc, track: Arc Option { #[cfg(test)] pub(crate) static PROBE_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); +/// Tests that read or bump the process-global reload generation share this lock. +#[cfg(test)] +pub(crate) static RELOAD_TEST_LOCK:std::sync::Mutex<()> = std::sync::Mutex::new(()); + #[cfg(test)] pub fn clear_probe() { if let Ok(mut slot) = cache().lock() { *slot = None; } @@ -609,6 +613,7 @@ mod tests { #[test] fn reload_does_not_clear_the_probe_cache() { + let _reload = RELOAD_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); let _guard = PROBE_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); let report = probe_report(true, &[ProbeOutcome::Ok, ProbeOutcome::Ok]); store_probe(report.clone()); @@ -623,6 +628,7 @@ mod tests { #[test] fn probe_finish_is_not_a_reload_since_the_session_snapshot() { + let _reload = RELOAD_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); let at_session = reload_generation_value(); // A finished background probe updates the cache only. It does not bump the generation. assert!(!reload_since_session(at_session, reload_generation_value())); @@ -632,6 +638,7 @@ mod tests { #[test] fn mark_reload_seen_clears_has_changed() { + let _reload = RELOAD_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); rt.block_on(async { let mut watch = subscribe_reload(); diff --git a/src/wayland.rs b/src/wayland.rs index a8e8c42..a7a77e4 100644 --- a/src/wayland.rs +++ b/src/wayland.rs @@ -255,6 +255,25 @@ pub fn selection()->(usize,u64) {let s=SELECTION.lock().unwrap();(s.id,s.generat pub fn selected()->usize {selection().0} pub fn source(id:usize)->Option<&'static Output> {outputs().get(id)} pub fn input_size(fallback:(u32,u32))->(u32,u32) {source(selected()).map_or(fallback,|s|(s.width,s.height))} +/// Pixel size of one advertised output. Unknown ids keep `fallback` (connection size). +pub fn source_size(id:usize,fallback:(u32,u32))->(u32,u32) { + #[cfg(test)] + if let Some(outputs)=*test_outputs().lock().unwrap_or_else(|e|e.into_inner()) { + return outputs.get(id).map_or(fallback,|s|(s.width,s.height)); + } + source(id).map_or(fallback,|s|(s.width,s.height)) +} +#[cfg(test)] +fn test_outputs()->&'static std::sync::Mutex> { + static SLOT:std::sync::OnceLock>>=std::sync::OnceLock::new(); + SLOT.get_or_init(||std::sync::Mutex::new(None)) +} +/// Native size written into CaptureSettingSync. Wayland uses the output the encoder is reading +/// (`source_id`), which can differ from the size captured when the connection opened. X11, portal +/// and KMS keep the connection size (X11 mode changes go through `x11_resized`). +pub fn sync_native(display:&str,source_id:usize,connection:(u32,u32))->(u32,u32) { + if display.starts_with("wayland:") {source_size(source_id,connection)} else {connection} +} pub fn input_ready()->bool {outputs().is_empty() || SELECTION.lock().unwrap().ready()} pub fn switch_failed()->bool {SELECTION.lock().unwrap().failed} pub fn presented(id:usize,generation:u64) {SELECTION.lock().unwrap().sent(id,generation);} @@ -353,4 +372,20 @@ mod selection_regressions { assert_eq!(outputs.len(),1);assert_eq!(outputs[0].name,"normal"); assert!(parse_outputs(b"not JSON","").is_err()); } + #[test] + fn sync_native_follows_the_wayland_source_id() { + assert_eq!(sync_native(":0",1,(1920,1080)),(1920,1080)); + assert_eq!(sync_native("kms:/dev/dri/card0",1,(1920,1080)),(1920,1080)); + let inventory:&'static [Output]=Box::leak(vec![ + Output {name:"eDP-1".into(),width:1920,height:1080,fps:60}, + Output {name:"HDMI-A-1".into(),width:2560,height:1440,fps:75}, + ].into_boxed_slice()); + *test_outputs().lock().unwrap()=Some(inventory); + // Connection opened on the first output; the encoder is reading the second. + assert_eq!(sync_native("wayland:wayland-0",1,(1920,1080)),(2560,1440)); + assert_eq!(source_size(0,(800,600)),(1920,1080)); + assert_eq!(sync_native("wayland:wayland-0",9,(800,600)),(800,600)); + assert_eq!(sync_native(":0",1,(1920,1080)),(1920,1080)); + *test_outputs().lock().unwrap()=None; + } } From e3588c4b9aab3cf0827517e68b483b9a79a5d823 Mon Sep 17 00:00:00 2001 From: Jeff Zhang <2854276+ferstar@users.noreply.github.com> Date: Fri, 23 Oct 2026 18:06:25 +0800 Subject: [PATCH 5/5] Control channel: start the outbound drain from on_open so pre-open replies are not dropped --- src/webrtc.rs | 49 +++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 41 insertions(+), 8 deletions(-) diff --git a/src/webrtc.rs b/src/webrtc.rs index 7583424..1cc3c60 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -449,7 +449,13 @@ impl WebRtcSession { })); } // Ordered outbound queue for protocol replies produced outside this callback. - let (outbound, mut outbound_rx) = tokio::sync::mpsc::unbounded_channel::>(); + // webrtc-rs announces the channel before handle_open (ready_state Connecting); + // send() returns ErrClosedPipe until Open. Draining CONTROL here would drop + // Message 25 and every later reply. Start CONTROL from the on_open below + // (one handler; a second on_open replaces it). Receiver taken once. + let mut control_drain: Option std::pin::Pin + Send>>) + Send + Sync>> = None; + let (outbound, outbound_rx) = tokio::sync::mpsc::unbounded_channel::>(); + let outbound_rx = Arc::new(tokio::sync::Mutex::new(Some(outbound_rx))); if label == "CONTROL_DATA_CHANNEL" { if let Some(mut chroma_sync_rx) = chroma_sync_rx.lock().unwrap().take() { let sync_out = outbound.clone(); @@ -466,13 +472,24 @@ impl WebRtcSession { { let out_channel = d.clone(); let out_label = label.clone(); - tokio::spawn(async move { - while let Some(message) = outbound_rx.recv().await { - if let Err(e) = out_channel.send(&bytes::Bytes::from(message)).await { - log::warn!("{} send failed: {}", out_label, e); break; - } - } - }); + let outbound_rx = outbound_rx.clone(); + let start_drain = move || { + let out_channel = out_channel.clone(); + let out_label = out_label.clone(); + let outbound_rx = outbound_rx.clone(); + Box::pin(async move { + let Some(mut outbound_rx) = take_outbound_once(&outbound_rx).await else { return; }; + tokio::spawn(async move { + while let Some(message) = outbound_rx.recv().await { + if let Err(e) = out_channel.send(&bytes::Bytes::from(message)).await { + log::warn!("{} send failed: {}", out_label, e); break; + } + } + }); + }) as std::pin::Pin + Send>> + }; + if label != "CONTROL_DATA_CHANNEL" { tokio::spawn(start_drain()); } + else { control_drain = Some(Box::new(start_drain)); } } let (capture_tx, mut capture_rx) = tokio::sync::mpsc::channel::>(16); if matches!(label.as_str(),"CONTROL_DATA_CHANNEL"|"TEXT_DATA_CHANNEL") { @@ -660,6 +677,7 @@ impl WebRtcSession { let open_screen = screen.clone(); let open_selection = peer_selection; d.on_open(Box::new(move || { + if let Some(start_drain) = control_drain { tokio::spawn(start_drain()); } let init_actions = open_actions.clone(); let init_video = open_video.clone(); let init_track = open_track.clone(); @@ -1442,6 +1460,11 @@ fn negotiated_bundle_media(sdp: &str) -> Option<(String, u16)> { None } +/// First drain owns the outbound receiver; a later open must not start another. +async fn take_outbound_once(slot: &tokio::sync::Mutex>>) -> Option> { + slot.lock().await.take() +} + #[cfg(test)] mod tests { use super::negotiated_bundle_media; @@ -1464,6 +1487,16 @@ mod tests { let sdp = "a=group:BUNDLE data video\nm=video 9 RTP 98\na=mid:video\nm=application 9 SCTP 0\na=mid:data\n"; assert_eq!(negotiated_bundle_media(sdp), Some(("data".into(), 1))); } + #[tokio::test] + async fn outbound_receiver_is_taken_once() { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel::>(); + let slot = std::sync::Arc::new(tokio::sync::Mutex::new(Some(rx))); + let first = super::take_outbound_once(&slot).await; + let second = super::take_outbound_once(&slot).await; + assert!(first.is_some()); + assert!(second.is_none()); + drop(tx); + } } #[cfg(test)]