fix(wayland): a pending portal request no longer stalls the server (#16542)

* fix(wayland): a pending portal request no longer stalls the server

The screencast portal waits for the user while its picker is open, up to
three minutes. check_init waited under the CAP_DISPLAY_INFO write lock on
a tokio worker and get_capturables under RDP_SESSION_INFO. Every desktop
login read the display state on a worker behind them, and the session
close when the last connection went waited for the slot on a worker too.
With the server on 4 cpus a second login was enough: new connections
were accepted by the kernel and not served until the request ended
(#15973).

The locks keep their role: the write lock still serializes inits and
orders them with clear(), and the slot still orders the session close
calls after a pending request. What changes is where the waiting happens.
The part of check_init under the write lock moves, as is, to
init_displays(), run in spawn_blocking. is_inited and the first look in
check_init use try_read; when a writer holds the lock, is_inited answers
not inited yet and check_init redoes its look in spawn_blocking, then
goes on as before. get_displays_and_primary takes its read guard in
spawn_blocking, is_rdp_session_hold uses try_lock, close_session from the
ipc listener runs in spawn_blocking, and try_close_session from
AuthedConnID::drop on its own thread.

* fix(wayland): throttle the log when the session close thread cannot start

A peer can drop connections at any rate, so the error logged when
AuthedConnID::drop cannot spawn the thread for try_close_session goes
through throttled_log!, once a minute, like the auto-login logs of this
file (greptile on #16542).

* fix(wayland): count the remote connections again in the close thread

The session close moved to its own thread in this pr, so a remote that
comes in after the count and before the thread runs could have the
session it reuses closed under it. The thread counts the remote
connections again right before try_close_session, which brings the
window back to the one the close had on the dropping thread (coderabbit
on #16542).

* fix(wayland): one portal session close at a time, for remote and camera connections only

The close that AuthedConnID::drop runs off the dropping thread spawned one
OS thread per drop that left no remote connection, whatever the connection
type, and while a portal request waits for the user each of those threads
waited for the session slot. Now only a remote or view camera connection
leaving asks for the close (their video service is the one that opens the
portal session), no thread starts when the slot is free and empty (drm,
x11, nothing left to close), and one close runs at a time: a close asked
while one runs is left to it. The logic moves to wayland.rs; connection.rs
keeps a hook.

libs/scrap/src/wayland/pipewire.rs

@@ -77,7 +77,9 @@ pub fn close_session() {
 
 #[inline]
 pub fn is_rdp_session_hold() -> bool {
-    RDP_SESSION_INFO.lock().unwrap().is_some()
+    // try_lock: a request waiting for the user holds the slot (no session yet), and so do the short
+    // looks while a session is set up. Busy answers no session rather than wait.
+    matches!(RDP_SESSION_INFO.try_lock(), Ok(info) if info.is_some())
 }
 
 pub fn try_close_session() {

src/ipc.rs

@@ -1192,8 +1192,16 @@ async fn handle(data: Data, stream: &mut Connection) {
                 }
             } else if value == "clear" {
                 set_local_option(key.clone(), "".to_owned());
+                // The session slot is held while a portal request waits for the user, and this ipc
+                // listener runs on one thread: wait for the slot off it, then close what the
+                // request brings back.
                 #[cfg(target_os = "linux")]
-                scrap::wayland::pipewire::close_session();
+                allow_err!(
+                    hbb_common::tokio::task::spawn_blocking(
+                        scrap::wayland::pipewire::close_session
+                    )
+                    .await
+                );
                 Some("".to_owned())
             } else {
                 None

src/server/connection.rs

@@ -7058,7 +7058,9 @@ mod raii {
                 #[cfg(windows)]
                 let _ = virtual_display_manager::reset_all();
                 #[cfg(target_os = "linux")]
-                scrap::wayland::pipewire::try_close_session();
+                if self.1 == AuthConnType::Remote || self.1 == AuthConnType::ViewCamera {
+                    wayland::try_close_portal_session();
+                }
             }
             Self::check_wake_lock();
             #[cfg(not(any(target_os = "android", target_os = "ios")))]

src/server/wayland.rs

@@ -404,7 +404,9 @@ pub(super) fn is_inited() -> Option<Message> {
         if super::drm_capturer::is_available_cached() {
             return None;
         }
-        if CAP_DISPLAY_INFO.read().unwrap().is_empty() {
+        // try_read: an init waiting for the portal's picker holds the write lock, and this runs on
+        // every desktop login. Busy (that init, or clear()) answers not inited yet.
+        if !matches!(CAP_DISPLAY_INFO.try_read(), Ok(map) if !map.is_empty()) {
             let mut msg_out = Message::new();
             let res = MessageBox {
                 msgtype: "nook-nocancel-hasclose".to_owned(),
@@ -423,7 +425,17 @@ pub(super) fn is_inited() -> Option<Message> {
 
 pub(super) async fn check_init() -> ResultType<()> {
     if !is_x11() {
-        if CAP_DISPLAY_INFO.read().unwrap().is_empty() {
+        // try_read for the same reason as is_inited; when busy, the same look waits off the worker.
+        let looked = CAP_DISPLAY_INFO.try_read().map(|map| map.is_empty()).ok();
+        let empty = match looked {
+            Some(empty) => empty,
+            None => hbb_common::tokio::task::spawn_blocking(|| {
+                CAP_DISPLAY_INFO.read().unwrap().is_empty()
+            })
+            .await
+            .map_err(|err| anyhow::anyhow!("Wayland display state task failed: {err}"))?,
+        };
+        if empty {
             if crate::input_service::wayland_use_uinput() {
                 // The cached layout may predate compositor changes made while no session
                 // was active, https://github.com/rustdesk/rustdesk/issues/15601
@@ -509,65 +521,77 @@ pub(super) async fn check_init() -> ResultType<()> {
                 }
             }
 
-            let mut lock = CAP_DISPLAY_INFO.write().unwrap();
-            if lock.is_empty() {
-                // Check if PipeWire is already initialized to prevent duplicate recorder creation
-                if *PIPEWIRE_INITIALIZED.read().unwrap() {
-                    log::warn!("wayland_diag: Preventing duplicate PipeWire initialization");
-                    return Ok(());
-                }
+            // The portal can wait minutes for the user to answer its picker, under the write lock:
+            // it serializes inits and orders them with clear(). Wait on a blocking thread, so no
+            // tokio worker is parked meanwhile.
+            hbb_common::tokio::task::spawn_blocking(init_displays)
+                .await
+                .map_err(|err| anyhow::anyhow!("Wayland display init task failed: {err}"))??;
+        }
+    }
+    Ok(())
+}
 
-                let mut all = Display::all()?;
-                log::debug!("Initializing displays with fill_displays()");
-                {
-                    let temp_mouse_move_handle = input_service::TemporaryMouseMoveHandle::new();
-                    let move_mouse_to = |x, y| temp_mouse_move_handle.move_mouse_to(x, y);
-                    fill_displays(move_mouse_to, crate::get_cursor_pos, &mut all)
-                        .map_err(map_staged_err)?;
-                }
-                log::debug!("Attempting to fix logical size with try_fix_logical_size()");
-                try_fix_logical_size(&mut all);
-                *PIPEWIRE_INITIALIZED.write().unwrap() = true;
-                let num = all.len();
-                let primary = super::display_service::get_primary_2(&all);
-                let mut displays = super::display_service::update_sync_displays(&all);
-                for display in displays.iter_mut() {
-                    display.cursor_embedded = is_cursor_embedded();
-                }
+/// The part of `check_init` under the CAP_DISPLAY_INFO write lock, the wait for the user while the
+/// portal's picker is open included. Blocking.
+fn init_displays() -> ResultType<()> {
+    let mut lock = CAP_DISPLAY_INFO.write().unwrap();
+    if lock.is_empty() {
+        // Check if PipeWire is already initialized to prevent duplicate recorder creation
+        if *PIPEWIRE_INITIALIZED.read().unwrap() {
+            log::warn!("wayland_diag: Preventing duplicate PipeWire initialization");
+            return Ok(());
+        }
 
-                let mut rects: Vec<((i32, i32), usize, usize)> = Vec::new();
-                for d in &all {
-                    rects.push((d.origin(), d.width(), d.height()));
-                }
+        let mut all = Display::all()?;
+        log::debug!("Initializing displays with fill_displays()");
+        {
+            let temp_mouse_move_handle = input_service::TemporaryMouseMoveHandle::new();
+            let move_mouse_to = |x, y| temp_mouse_move_handle.move_mouse_to(x, y);
+            fill_displays(move_mouse_to, crate::get_cursor_pos, &mut all)
+                .map_err(map_staged_err)?;
+        }
+        log::debug!("Attempting to fix logical size with try_fix_logical_size()");
+        try_fix_logical_size(&mut all);
+        *PIPEWIRE_INITIALIZED.write().unwrap() = true;
+        let num = all.len();
+        let primary = super::display_service::get_primary_2(&all);
+        let mut displays = super::display_service::update_sync_displays(&all);
+        for display in displays.iter_mut() {
+            display.cursor_embedded = is_cursor_embedded();
+        }
 
-                log::debug!(
-                    "#displays={}, primary={}, rects: {:?}, cpus={}/{}",
-                    num,
-                    primary,
-                    rects,
-                    num_cpus::get_physical(),
-                    num_cpus::get()
-                );
+        let mut rects: Vec<((i32, i32), usize, usize)> = Vec::new();
+        for d in &all {
+            rects.push((d.origin(), d.width(), d.height()));
+        }
 
-                // Create individual CapDisplayInfo for each display with its own capturer
-                for (idx, display) in all.into_iter().enumerate() {
-                    // No `with_context` here: the peer is shown `format!("{}", err)`, which
-                    // renders only the outermost layer, and the mapped reason is the inner one.
-                    let capturer = Box::into_raw(Box::new(Capturer::new(display)?));
-                    let capturer = CapturerPtr(capturer);
+        log::debug!(
+            "#displays={}, primary={}, rects: {:?}, cpus={}/{}",
+            num,
+            primary,
+            rects,
+            num_cpus::get_physical(),
+            num_cpus::get()
+        );
 
-                    let cap_display_info = Box::into_raw(Box::new(CapDisplayInfo {
-                        rects: rects.clone(),
-                        displays: displays.clone(),
-                        num,
-                        primary,
-                        current: idx,
-                        capturer,
-                    }));
+        // Create individual CapDisplayInfo for each display with its own capturer
+        for (idx, display) in all.into_iter().enumerate() {
+            // No `with_context` here: the peer is shown `format!("{}", err)`, which
+            // renders only the outermost layer, and the mapped reason is the inner one.
+            let capturer = Box::into_raw(Box::new(Capturer::new(display)?));
+            let capturer = CapturerPtr(capturer);
 
-                    lock.insert(idx, cap_display_info as u64);
-                }
-            }
+            let cap_display_info = Box::into_raw(Box::new(CapDisplayInfo {
+                rects: rects.clone(),
+                displays: displays.clone(),
+                num,
+                primary,
+                current: idx,
+                capturer,
+            }));
+
+            lock.insert(idx, cap_display_info as u64);
         }
     }
     Ok(())
@@ -596,17 +620,22 @@ pub(super) async fn get_displays_and_primary() -> ResultType<(Vec<DisplayInfo>,
         }
     }
     check_init().await?;
-    // Keep one read guard so clear/reinitialization cannot split these across cache snapshots.
-    let cap_map = CAP_DISPLAY_INFO.read().unwrap();
-    if let Some(addr) = cap_map.values().next() {
-        let cap_display_info: *const CapDisplayInfo = *addr as _;
-        unsafe {
-            let cap_display_info = &*cap_display_info;
-            Ok((cap_display_info.displays.clone(), cap_display_info.primary))
+    // The read can wait for an init another thread started since: off the worker too.
+    hbb_common::tokio::task::spawn_blocking(|| -> ResultType<(Vec<DisplayInfo>, usize)> {
+        // Keep one read guard so clear/reinitialization cannot split these across cache snapshots.
+        let cap_map = CAP_DISPLAY_INFO.read().unwrap();
+        if let Some(addr) = cap_map.values().next() {
+            let cap_display_info: *const CapDisplayInfo = *addr as _;
+            unsafe {
+                let cap_display_info = &*cap_display_info;
+                Ok((cap_display_info.displays.clone(), cap_display_info.primary))
+            }
+        } else {
+            bail!("Failed to get capturer display info");
         }
-    } else {
-        bail!("Failed to get capturer display info");
-    }
+    })
+    .await
+    .map_err(|err| anyhow::anyhow!("Wayland display info task failed: {err}"))?
 }
 
 pub fn clear() {
@@ -640,6 +669,44 @@ pub fn clear() {
     *PIPEWIRE_INITIALIZED.write().unwrap() = false;
 }
 
+/// The portal session close when the last remote connection goes (`AuthedConnID::drop`). A portal
+/// request waiting for the user holds the session slot for up to 3 minutes, and the drop can run
+/// on a tokio worker: the close waits on its own thread, and one runs at a time.
+pub(super) fn try_close_portal_session() {
+    static CLOSER: Mutex<Option<std::thread::JoinHandle<()>>> = Mutex::new(None);
+    // No session and no request pending (drm, x11, closed already): nothing to wait for.
+    if matches!(scrap::wayland::pipewire::RDP_SESSION_INFO.try_lock(), Ok(info) if info.is_none()) {
+        return;
+    }
+    let mut closer = CLOSER.lock().unwrap();
+    // A close still running covers this one.
+    if matches!(&*closer, Some(close) if !close.is_finished()) {
+        return;
+    }
+    match std::thread::Builder::new()
+        .name("portal-session-close".into())
+        .spawn(|| {
+            // The drop counted before this thread ran: count again.
+            if !AUTHED_CONNS
+                .lock()
+                .unwrap()
+                .iter()
+                .any(|c| c.conn_type == AuthConnType::Remote)
+            {
+                scrap::wayland::pipewire::try_close_session();
+            }
+        }) {
+        Ok(close) => *closer = Some(close),
+        // A peer can drop connections at any rate.
+        Err(err) => hbb_common::throttled_log!(
+            std::time::Duration::from_secs(60),
+            error,
+            "Failed to spawn the portal session close: {}",
+            err
+        ),
+    }
+}
+
 /// Initialize the PipeWire/portal capture path from the plain (sync) video thread, so a DRM display
 /// that cannot be captured can fall through to PipeWire for THAT display. `ensure_inited` short-circuits
 /// to the DRM branch whenever DRM is globally available, so it never runs `check_init`; this helper