diff --git a/apps/desktop/src/common/task_queue_lock_free.cpp b/apps/desktop/src/common/task_queue_lock_free.cpp new file mode 100644 index 00000000..273c4a31 --- /dev/null +++ b/apps/desktop/src/common/task_queue_lock_free.cpp @@ -0,0 +1,79 @@ +#include "task_queue_lock_free.h" + +#include +#include +#include +#include + +namespace crossdesk { + +TaskQueueLockFree::TaskQueueLockFree() + : producer_(task_queue_), worker_([this] { WorkerThread(); }) {} + +TaskQueueLockFree::~TaskQueueLockFree() { Stop(); } + +std::future TaskQueueLockFree::PostTask(std::function task) { + TaskItem item; + item.run = std::move(task); + auto result = item.completion.get_future(); + { + std::lock_guard lock(mutex_); + if (stopping_) { + throw std::runtime_error("Cannot post to a stopped task queue"); + } + + // Count before publishing: the worker may dequeue as soon as enqueue returns. + ++pending_tasks_; + if (!task_queue_.enqueue(producer_, std::move(item))) { + --pending_tasks_; + throw std::bad_alloc(); + } + } + wake_.notify_one(); + return result; +} + +void TaskQueueLockFree::Stop() { + // Concurrent Stop callers all wait for the same worker shutdown to finish. + std::call_once(stop_once_, [this] { + { + std::lock_guard lock(mutex_); + stopping_ = true; + } + wake_.notify_all(); + worker_.join(); + }); +} + +void TaskQueueLockFree::WorkerThread() { + while (true) { + TaskItem item; + while (task_queue_.try_dequeue(item)) { + --pending_tasks_; + + std::exception_ptr error; + try { + item.run(); + } catch (...) { + error = std::current_exception(); + } + + // Release captured resources before marking the task complete. + item.run = nullptr; + if (error) { + item.completion.set_exception(error); + } else { + item.completion.set_value(); + } + } + + std::unique_lock lock(mutex_); + wake_.wait(lock, [this] { return stopping_ || pending_tasks_ > 0; }); + // A stop request must not discard tasks submitted while the worker was idle. + if (stopping_ && pending_tasks_ == 0) { + return; + } + } +} + +} // namespace crossdesk diff --git a/apps/desktop/src/common/task_queue_lock_free.h b/apps/desktop/src/common/task_queue_lock_free.h new file mode 100644 index 00000000..f8fa8788 --- /dev/null +++ b/apps/desktop/src/common/task_queue_lock_free.h @@ -0,0 +1,57 @@ +/* + * @Author: DI JUNKUN + * @Date: 2026-09-07 + * Copyright (c) 2026 by DI JUNKUN, All Rights Reserved. + */ + +#ifndef _TASK_QUEUE_LOCK_FREE_H_ +#define _TASK_QUEUE_LOCK_FREE_H_ + +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace crossdesk { + +// A single worker executes tasks in submission order. The concurrent queue +// stores tasks; a mutex coordinates submissions, worker wakeups and shutdown. +class TaskQueueLockFree { + public: + TaskQueueLockFree(); + ~TaskQueueLockFree(); + + // Returns completion or the task's exception. Throws if the queue is stopped. + std::future PostTask(std::function task); + + // Rejects new tasks, finishes accepted tasks and joins the worker. + // May be called repeatedly, but never from the queue's own tasks. + void Stop(); + + private: + struct TaskItem { + std::function run; + std::promise completion; + }; + + void WorkerThread(); + + moodycamel::ConcurrentQueue task_queue_; + // All submissions share one producer so different caller threads keep FIFO. + moodycamel::ProducerToken producer_; + std::atomic pending_tasks_{0}; + std::mutex mutex_; + std::condition_variable wake_; + bool stopping_ = false; + std::once_flag stop_once_; + std::thread worker_; +}; + +} // namespace crossdesk + +#endif \ No newline at end of file diff --git a/apps/desktop/src/gui/application/gui_application.cpp b/apps/desktop/src/gui/application/gui_application.cpp index e47648a4..882a9d4a 100644 --- a/apps/desktop/src/gui/application/gui_application.cpp +++ b/apps/desktop/src/gui/application/gui_application.cpp @@ -1738,7 +1738,7 @@ void GuiApplication::BindServerCallbacks() { }); server->on_disconnect_controller([this] { if (peer_ && !selected_server_remote_id_.empty()) { - LeaveConnection(peer_, selected_server_remote_id_.c_str()); + CloseServerController(selected_server_remote_id_); } }); } @@ -1748,6 +1748,7 @@ void GuiApplication::Tick() { slint::quit_event_loop(); return; } + HandleSessionCleanup(); HandlePasswordChangeResult(); HandleCredentialRecovery(); if (!peer_) { @@ -2008,8 +2009,7 @@ void GuiApplication::HandlePasswordChangeResult() { } show_offline_warning_window_ = true; if (uncertain && peer_) { - LeaveConnection(peer_, client_id_); - DestroyPeer(&peer_); + CloseConnectionPeer(); } return; } @@ -2022,8 +2022,7 @@ void GuiApplication::HandlePasswordChangeResult() { LOG_INFO("Password changed successfully for [{}]", client_id_); if (peer_) { - LeaveConnection(peer_, client_id_); - DestroyPeer(&peer_); + CloseConnectionPeer(); } } @@ -2047,7 +2046,7 @@ void GuiApplication::HandleCredentialRecovery() { if (retry_active) { LOG_INFO("Pending credential was not accepted; retrying active credential"); if (peer_) { - DestroyPeer(&peer_); + CloseConnectionPeer(); } return; } @@ -3083,9 +3082,6 @@ void GuiApplication::CloseStreamTab(const std::string& remote_id) { } keyboard_.ReleaseRemotePressedKeys(remote_id, "stream_tab_closed"); CloseRemoteSession(props); - WaitForThumbnailSaveTasks(); - reload_recent_connections_ = true; - recent_connection_image_save_time_ = 0; ResetRemoteSessionResources(props); { std::unique_lock lock(remote_sessions_mutex_); @@ -3290,7 +3286,7 @@ void GuiApplication::Cleanup() { devices_.DestroyDevices(); devices_.DestroyFactories(); CloseAllRemoteSessions(); - WaitForThumbnailSaveTasks(); + WaitForSessionCleanup(); devices_.DestroyAudioOutput(); if (ui_->stream) { #if defined(__APPLE__) diff --git a/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.cpp b/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.cpp index 3d8d6ceb..0ec2daee 100644 --- a/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.cpp +++ b/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.cpp @@ -20,6 +20,16 @@ namespace crossdesk { FileTransferManager::FileTransferManager(GuiRuntime &owner) : owner_(owner) {} +FileTransferManager::~FileTransferManager() { + std::vector> workers; + { + std::lock_guard lock(workers_mutex_); + stopping_ = true; + workers.swap(workers_); + } + for (auto& worker : workers) worker.wait(); +} + FileTransferManager::FileTransferState &FileTransferManager::global_state() { return global_state_; } @@ -92,8 +102,16 @@ void FileTransferManager::Start( std::shared_ptr props, const std::filesystem::path &file_path, const std::string &file_label, const std::string &remote_id) { + if (props && props->closing_) return; const bool is_global = !props; - PeerPtr *peer = is_global ? owner_.peer_ : props->peer_; + PeerPtr* peer = nullptr; + std::shared_ptr peer_events; + { + std::shared_lock lock(owner_.remote_sessions_mutex_); + if (props && props->closing_) return; + peer = is_global ? owner_.peer_ : props->peer_; + peer_events = is_global ? owner_.peer_events_ : props->peer_events_; + } if (!peer) { LOG_ERROR("StartFileTransfer: invalid peer"); return; @@ -108,7 +126,14 @@ void FileTransferManager::Start( } const auto props_weak = std::weak_ptr(props); - std::thread([this, peer, file_path, file_label, props_weak, remote_id, + std::lock_guard workers_lock(workers_mutex_); + if (stopping_) return; + // Reap finished senders without blocking the UI. + std::erase_if(workers_, [](auto& worker) { + return worker.wait_for(std::chrono::seconds(0)) == std::future_status::ready; + }); + workers_.push_back(std::async(std::launch::async, + [this, peer, peer_events, file_path, file_label, props_weak, remote_id, is_global]() { auto props_locked = props_weak.lock(); FileTransferState *state = nullptr; @@ -166,7 +191,11 @@ void FileTransferManager::Start( const int ret = sender.SendFile( file_path, file_path.filename().string(), - [peer, file_label, remote_id](const char *buffer, size_t size) { + [peer, peer_events, file_label, remote_id, props_locked](const char *buffer, size_t size) { + // Close drains each in-flight send before detaching the peer. + if (props_locked && props_locked->closing_) return -1; + auto callback = peer_events->EnterCallback(); + if (!callback) return -1; if (remote_id.empty()) { return SendReliableDataFrame(peer, buffer, size, file_label.c_str()); @@ -200,12 +229,13 @@ void FileTransferManager::Start( } LOG_ERROR("FileSender::SendFile failed for [{}], ret={}", file_path.string().c_str(), ret); - ProcessQueue(props_locked); - }).detach(); + if (peer_events->IsActive()) ProcessQueue(props_locked); + })); } void FileTransferManager::ProcessQueue( std::shared_ptr props) { + if (props && props->closing_) return; FileTransferState &state = state_for(props); if (state.file_sending_.load()) { return; diff --git a/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h b/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h index 748d02da..fda31dca 100644 --- a/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h +++ b/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h @@ -4,6 +4,9 @@ #include #include #include +#include +#include +#include #include #include #include @@ -21,6 +24,7 @@ public: using RemoteSession = gui_detail::RemoteSession; explicit FileTransferManager(GuiRuntime &owner); + ~FileTransferManager(); FileTransferState &global_state(); FileTransferState & @@ -40,6 +44,9 @@ private: void ProcessQueue(std::shared_ptr props); void Unregister(uint32_t file_id, bool per_peer); + std::mutex workers_mutex_; + std::vector> workers_; + bool stopping_ = false; GuiRuntime &owner_; FileTransferState global_state_; std::unordered_map> diff --git a/apps/desktop/src/gui/features/input/keyboard_controller.cpp b/apps/desktop/src/gui/features/input/keyboard_controller.cpp index 0e5548c7..ec49c5b3 100644 --- a/apps/desktop/src/gui/features/input/keyboard_controller.cpp +++ b/apps/desktop/src/gui/features/input/keyboard_controller.cpp @@ -150,6 +150,7 @@ void KeyboardController::SendHeartbeat(bool force) { const std::string target_id = owner_.controlled_remote_id_.empty() ? owner_.focused_remote_id_ : owner_.controlled_remote_id_; + std::shared_lock sessions_lock(owner_.remote_sessions_mutex_); const auto props_it = owner_.remote_sessions_.find(target_id); if (target_id.empty() || props_it == owner_.remote_sessions_.end() || props_it->second->connection_status_.load() != @@ -189,6 +190,7 @@ int KeyboardController::SendKeyCommand(int key_code, bool is_down, const std::string target_id = owner_.controlled_remote_id_.empty() ? owner_.focused_remote_id_ : owner_.controlled_remote_id_; + std::shared_lock sessions_lock(owner_.remote_sessions_mutex_); const auto props_it = owner_.remote_sessions_.find(target_id); if (!target_id.empty() && props_it != owner_.remote_sessions_.end() && props_it->second->connection_status_.load() == diff --git a/apps/desktop/src/gui/runtime/connection_runtime.cpp b/apps/desktop/src/gui/runtime/connection_runtime.cpp index 9f043d10..d354cc29 100644 --- a/apps/desktop/src/gui/runtime/connection_runtime.cpp +++ b/apps/desktop/src/gui/runtime/connection_runtime.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include @@ -220,6 +221,11 @@ int GuiRuntime::RequestSingleDevicePresence(const std::string& remote_id, } void GuiRuntime::CloseRemoteSession(std::shared_ptr props) { + if (!props || props->closing_.exchange(true)) return; + if (props->peer_events_) props->peer_events_->Deactivate(); + props->connection_established_ = false; + props->streaming_ = false; + props->connection_status_.store(ConnectionStatus::Closed); std::shared_ptr> frame_snapshot; int video_width = 0; int video_height = 0; @@ -245,85 +251,160 @@ void GuiRuntime::CloseRemoteSession(std::shared_ptr props) { } } - if (frame_snapshot && !frame_snapshot->empty() && video_width > 0 && - video_height > 0) { - std::vector buffer_copy(*frame_snapshot); - std::string remote_id = props->remote_id_; - std::string remote_host_name = props->remote_host_name_; - std::string password = - props->remember_password_ ? props->remote_password_ : ""; - - std::thread save_thread([buffer_copy, video_width, video_height, remote_id, - remote_host_name, password, - thumbnail = thumbnail_]() { - thumbnail->SaveToThumbnail((char*)buffer_copy.data(), video_width, - video_height, remote_id, remote_host_name, - password); - }); - - { - std::lock_guard lock(thumbnail_save_threads_mutex_); - thumbnail_save_threads_.emplace_back(std::move(save_thread)); - } - } - if (native_renderer) { native_renderer->DiscardStream(props->remote_id_); video_frame_dirty_.store(true, std::memory_order_release); } - if (props->peer_) { - LOG_INFO("[{}] Leave connection [{}]", props->local_id_, props->remote_id_); - LeaveConnection(props->peer_, props->remote_id_.c_str()); - LOG_INFO("Destroy peer [{}]", props->local_id_); - DestroyPeer(&props->peer_); + PeerPtr* peer = nullptr; + { + // Audio/clipboard senders hold a shared map lock while using this pointer. + std::unique_lock lock(remote_sessions_mutex_); + peer = std::exchange(props->peer_, nullptr); } + const auto queued_at = std::chrono::steady_clock::now(); + session_cleanup_tasks_[props->remote_id_] = session_cleanup_queue_.PostTask( + [props, peer, frame_snapshot, video_width, video_height, + thumbnail = thumbnail_, queued_at]() mutable { + if (peer) { + LOG_INFO("[{}] Background leave connection [{}]", props->local_id_, + props->remote_id_); + LeaveConnection(peer, props->remote_id_.c_str()); + DestroyPeer(&peer); + } + if (thumbnail && frame_snapshot && !frame_snapshot->empty() && + video_width > 0 && video_height > 0) { + thumbnail->SaveToThumbnail( + reinterpret_cast(frame_snapshot->data()), video_width, + video_height, props->remote_id_, props->remote_host_name_, + props->remember_password_ ? props->remote_password_ : ""); + } + const auto elapsed = std::chrono::duration_cast( + std::chrono::steady_clock::now() - queued_at).count(); + LOG_INFO("[{}] Session cleanup completed in {} ms", props->remote_id_, + elapsed); + }); +} + +void GuiRuntime::CloseServerController(const std::string& remote_id) { + if (!peer_ || remote_id.empty() || controller_cleanup_tasks_.count(remote_id)) + return; + controller_cleanup_tasks_[remote_id] = session_cleanup_queue_.PostTask( + [peer = peer_, remote_id] { LeaveConnection(peer, remote_id.c_str()); }); +} + +void GuiRuntime::CloseConnectionPeer() { + if (!peer_) return; + peer_events_->Deactivate(); + std::vector controllers; + { + std::shared_lock lock(connection_status_mutex_); + for (const auto& [id, status] : connection_status_) controllers.push_back(id); + } + for (const auto& id : controllers) + HandleServerControllerDisconnected(id, "peer_closed"); + devices_.StopMouseController(); + devices_.StopScreenCapturer(); + devices_.StopSpeakerCapturer(); + devices_.StopKeyboardCapturer(); + signal_connected_ = false; + signal_status_ = SignalStatus::SignalClosed; + PeerPtr* peer; + { + std::unique_lock lock(remote_sessions_mutex_); + peer = std::exchange(peer_, nullptr); + } + // The same serial queue also owns outstanding controller disconnects, so + // these finish before the shared peer is destroyed. + connection_peer_cleanup_ = session_cleanup_queue_.PostTask( + [peer, handler = peer_events_, id = std::string(client_id_)]() mutable { + LOG_INFO("[{}] Background connection peer cleanup", id); + LeaveConnection(peer, id.c_str()); + DestroyPeer(&peer); + }); } void GuiRuntime::CloseAllRemoteSessions() { - if (peer_) { - LOG_INFO("[{}] Leave connection [{}]", client_id_, client_id_); - LeaveConnection(peer_, client_id_); - is_client_mode_ = false; - devices_.StopMouseController(); - devices_.StopScreenCapturer(); - devices_.StopSpeakerCapturer(); - devices_.StopKeyboardCapturer(); - LOG_INFO("Destroy peer [{}]", client_id_); - DestroyPeer(&peer_); - } - + CloseConnectionPeer(); + is_client_mode_ = false; + pending_reconnects_.clear(); + std::vector> sessions; { - // std::shared_lock lock(remote_sessions_mutex_); - for (auto& it : remote_sessions_) { - auto props = it.second; - CloseRemoteSession(props); - } + std::shared_lock lock(remote_sessions_mutex_); + for (const auto& [id, props] : remote_sessions_) sessions.push_back(props); } - + for (const auto& props : sessions) CloseRemoteSession(props); { - // std::unique_lock lock(remote_sessions_mutex_); + std::unique_lock lock(remote_sessions_mutex_); remote_sessions_.clear(); } } -void GuiRuntime::WaitForThumbnailSaveTasks() { - std::vector threads_to_join; - - { - std::lock_guard lock(thumbnail_save_threads_mutex_); - threads_to_join.swap(thumbnail_save_threads_); +void GuiRuntime::HandleSessionCleanup() { + auto completed = [](std::future& task) { + if (!task.valid() || task.wait_for(std::chrono::seconds(0)) != + std::future_status::ready) return false; + try { + task.get(); + } catch (const std::exception& error) { + LOG_ERROR("Connection cleanup failed: {}", error.what()); + } + return true; + }; + completed(connection_peer_cleanup_); + for (auto it = controller_cleanup_tasks_.begin(); + it != controller_cleanup_tasks_.end();) { + if (completed(it->second)) it = controller_cleanup_tasks_.erase(it); + else ++it; } - - if (threads_to_join.empty()) { - return; + std::vector> reconnects; + for (auto it = session_cleanup_tasks_.begin(); + it != session_cleanup_tasks_.end();) { + if (it->second.wait_for(std::chrono::seconds(0)) != + std::future_status::ready) { + ++it; + continue; + } + try { + it->second.get(); + } catch (const std::exception& error) { + LOG_ERROR("[{}] Session cleanup failed: {}", it->first, error.what()); + } + if (auto pending = pending_reconnects_.find(it->first); + pending != pending_reconnects_.end()) { + reconnects.emplace_back(pending->first, std::move(pending->second)); + pending_reconnects_.erase(pending); + } + it = session_cleanup_tasks_.erase(it); + reload_recent_connections_ = true; + recent_connection_image_save_time_ = 0; } + for (const auto& [id, request] : reconnects) { + ConnectTo(id, request.password.c_str(), request.remember_password); + } +} - for (auto& thread : threads_to_join) { - if (thread.joinable()) { - thread.join(); +void GuiRuntime::WaitForSessionCleanup() { + pending_reconnects_.clear(); + auto drain = [](std::future& task) { + if (!task.valid()) return; + try { + task.get(); + } catch (const std::exception& error) { + LOG_ERROR("Connection cleanup failed: {}", error.what()); + } + }; + drain(connection_peer_cleanup_); + for (auto& [id, task] : controller_cleanup_tasks_) drain(task); + controller_cleanup_tasks_.clear(); + for (auto& [id, task] : session_cleanup_tasks_) { + try { + task.get(); + } catch (const std::exception& error) { + LOG_ERROR("[{}] Session cleanup failed: {}", id, error.what()); } } + session_cleanup_tasks_.clear(); } void GuiRuntime::ResetRemoteSessionResources( @@ -365,6 +446,11 @@ std::shared_ptr GuiRuntime::FindRemoteSession( int GuiRuntime::ConnectTo(const std::string& remote_id, const char* password, bool remember_password, bool bypass_presence_check) { + if (session_cleanup_tasks_.count(remote_id)) { + pending_reconnects_[remote_id] = {password ? password : "", remember_password}; + LOG_INFO("[{}] Reconnect queued until previous peer is destroyed", remote_id); + return 0; + } if (!bypass_presence_check && !device_presence_cache_.IsOnline(remote_id)) { int ret = RequestSingleDevicePresence(remote_id, password, remember_password); @@ -382,16 +468,14 @@ int GuiRuntime::ConnectTo(const std::string& remote_id, const char* password, LOG_INFO("Connect to [{}]", remote_id); focused_remote_id_ = remote_id; - // std::shared_lock shared_lock(remote_sessions_mutex_); - bool exists = (remote_sessions_.find(remote_id) != remote_sessions_.end()); - // shared_lock.unlock(); + bool exists = FindRemoteSession(remote_id) != nullptr; if (!exists) { PeerPtr* peer_to_init = nullptr; std::string local_id; { - // std::unique_lock unique_lock(remote_sessions_mutex_); + std::unique_lock unique_lock(remote_sessions_mutex_); if (remote_sessions_.find(remote_id) == remote_sessions_.end()) { remote_sessions_[remote_id] = std::make_shared(); auto props = remote_sessions_[remote_id]; @@ -399,6 +483,8 @@ int GuiRuntime::ConnectTo(const std::string& remote_id, const char* password, props->remote_id_ = remote_id; memcpy(&props->params_, ¶ms_, sizeof(Params)); props->params_.user_id = props->local_id_.c_str(); + props->peer_events_ = std::make_shared(*this); + props->params_.user_data = props->peer_events_.get(); props->peer_ = CreatePeer(&props->params_); props->control_window_width_ = title_bar_height_ * 10.0f; @@ -445,8 +531,8 @@ int GuiRuntime::ConnectTo(const std::string& remote_id, const char* password, } int ret = -1; - // std::shared_lock read_lock(remote_sessions_mutex_); - auto props = remote_sessions_[remote_id]; + auto props = FindRemoteSession(remote_id); + if (!props || props->closing_) return -1; if (!props->connection_established_) { props->connection_status_.store(ConnectionStatus::Connecting); show_connection_status_window_ = true; @@ -470,7 +556,6 @@ int GuiRuntime::ConnectTo(const std::string& remote_id, const char* password, } } } - // read_lock.unlock(); return 0; } diff --git a/apps/desktop/src/gui/runtime/gui_runtime.cpp b/apps/desktop/src/gui/runtime/gui_runtime.cpp index cc159071..ac9f1b13 100644 --- a/apps/desktop/src/gui/runtime/gui_runtime.cpp +++ b/apps/desktop/src/gui/runtime/gui_runtime.cpp @@ -19,11 +19,17 @@ namespace crossdesk { GuiRuntime::GuiRuntime() : clipboard_(*this), devices_(*this), transfers_(*this), settings_(*this), - keyboard_(*this), peer_events_(*this) {} + keyboard_(*this) {} -GuiRuntime::~GuiRuntime() = default; +GuiRuntime::~GuiRuntime() { WaitForSessionCleanup(); } int GuiRuntime::CreateConnectionPeer() { + // Tick retries after the old peer's leave notification and socket shutdown. + if (connection_peer_cleanup_.valid()) return 0; + { + std::unique_lock lock(remote_sessions_mutex_); + peer_events_ = std::make_shared(*this); + } params_.use_cfg_file = false; std::string signal_server_ip; @@ -189,14 +195,17 @@ int GuiRuntime::CreateConnectionPeer() { params_.on_connection_status = PeerEventHandler::OnConnectionStatus; params_.on_net_status_report = PeerEventHandler::OnNetStatusReport; - params_.user_data = &peer_events_; + params_.user_data = peer_events_.get(); // The previous peer may have left a terminal status behind. Reset it before // Init() starts emitting callbacks for the newly selected server. signal_connected_ = false; signal_status_ = SignalStatus::SignalConnecting; - peer_ = CreatePeer(¶ms_); + { + std::unique_lock lock(remote_sessions_mutex_); + peer_ = CreatePeer(¶ms_); + } if (peer_) { LOG_INFO("Create peer instance [{}] successful", client_id_); Init(peer_); @@ -240,7 +249,7 @@ void GuiRuntime::UpdateLabels() { void GuiRuntime::HandleRecentConnections() { - if (reload_recent_connections_ && thumbnail_) { + if (session_cleanup_tasks_.empty() && reload_recent_connections_ && thumbnail_) { uint32_t now_time = SDL_GetTicks(); if (now_time - recent_connection_image_save_time_ >= 50) { int ret = thumbnail_->LoadThumbnail(recent_connections_, diff --git a/apps/desktop/src/gui/runtime/gui_runtime.h b/apps/desktop/src/gui/runtime/gui_runtime.h index b1da3415..22812f0d 100644 --- a/apps/desktop/src/gui/runtime/gui_runtime.h +++ b/apps/desktop/src/gui/runtime/gui_runtime.h @@ -61,8 +61,11 @@ class GuiRuntime : protected gui_detail::GuiState { void CloseRemoteSession(std::shared_ptr props); void CloseAllRemoteSessions(); + void CloseConnectionPeer(); + void CloseServerController(const std::string& remote_id); void ResetRemoteSessionResources(std::shared_ptr props); - void WaitForThumbnailSaveTasks(); + void HandleSessionCleanup(); + void WaitForSessionCleanup(); std::shared_ptr FindRemoteSession( const std::string& remote_id); @@ -89,7 +92,7 @@ class GuiRuntime : protected gui_detail::GuiState { FileTransferManager transfers_; SettingsManager settings_; KeyboardController keyboard_; - PeerEventHandler peer_events_; + std::shared_ptr peer_events_; std::atomic video_frame_dirty_{false}; std::unique_ptr video_renderer_; @@ -104,4 +107,4 @@ class GuiRuntime : protected gui_detail::GuiState { } // namespace crossdesk -#endif \ No newline at end of file +#endif diff --git a/apps/desktop/src/gui/runtime/peer_data_callbacks.cpp b/apps/desktop/src/gui/runtime/peer_data_callbacks.cpp index d90ab2cb..b839e50a 100644 --- a/apps/desktop/src/gui/runtime/peer_data_callbacks.cpp +++ b/apps/desktop/src/gui/runtime/peer_data_callbacks.cpp @@ -79,6 +79,9 @@ void PeerEventHandler::OnReceiveDataBuffer( const char *data, size_t size, const char *user_id, size_t user_id_size, const char *src_id, size_t src_id_size, void *user_data) { auto *handler = static_cast(user_data); + if (!handler) return; + auto callback = handler->EnterCallback(); + if (!callback) return; GuiRuntime *runtime = handler ? &handler->owner_ : nullptr; if (!runtime) { return; @@ -145,9 +148,8 @@ void PeerEventHandler::OnReceiveDataBuffer( std::string remote_id(user_id, user_id_size); if (remote_action.type == ControlType::service_status) { - auto props_it = runtime->remote_sessions_.find(remote_id); - if (props_it != runtime->remote_sessions_.end()) { - runtime->ApplyRemoteServiceStatus(*props_it->second, remote_action.ss); + if (auto props = runtime->FindRemoteSession(remote_id)) { + runtime->ApplyRemoteServiceStatus(*props, remote_action.ss); } return; } diff --git a/apps/desktop/src/gui/runtime/peer_event_handler.cpp b/apps/desktop/src/gui/runtime/peer_event_handler.cpp index 4c68fe73..af9f92b3 100644 --- a/apps/desktop/src/gui/runtime/peer_event_handler.cpp +++ b/apps/desktop/src/gui/runtime/peer_event_handler.cpp @@ -60,6 +60,9 @@ PeerEventHandler::PeerEventHandler(GuiRuntime& owner) : owner_(owner) {} void PeerEventHandler::OnSignalMessage(const char* message, size_t size, void* user_data) { auto* handler = static_cast(user_data); + if (!handler) return; + auto callback = handler->EnterCallback(); + if (!callback) return; GuiRuntime* runtime = handler ? &handler->owner_ : nullptr; if (!runtime || !message || size == 0) { return; @@ -153,6 +156,9 @@ void PeerEventHandler::OnSignalMessage(const char* message, size_t size, void PeerEventHandler::OnSignalStatus(SignalStatus status, const char* user_id, size_t user_id_size, void* user_data) { auto* handler = static_cast(user_data); + if (!handler) return; + auto callback = handler->EnterCallback(); + if (!callback) return; GuiRuntime* runtime = handler ? &handler->owner_ : nullptr; if (!runtime) { return; @@ -198,12 +204,8 @@ void PeerEventHandler::OnSignalStatus(SignalStatus status, const char* user_id, } std::string remote_id(client_id.begin() + 2, client_id.end()); - // std::shared_lock lock(runtime->remote_sessions_mutex_); - if (runtime->remote_sessions_.find(remote_id) == - runtime->remote_sessions_.end()) { - return; - } - auto props = runtime->remote_sessions_.find(remote_id)->second; + auto props = runtime->FindRemoteSession(remote_id); + if (!props) return; props->signal_status_ = status; if (SignalStatus::SignalConnecting == status) { props->signal_connected_ = false; @@ -230,6 +232,9 @@ void PeerEventHandler::OnConnectionStatus(ConnectionStatus status, const size_t user_id_size, void* user_data) { auto* handler = static_cast(user_data); + if (!handler) return; + auto callback = handler->EnterCallback(); + if (!callback) return; GuiRuntime* runtime = handler ? &handler->owner_ : nullptr; if (!runtime) return; @@ -304,8 +309,7 @@ void PeerEventHandler::OnConnectionStatus(ConnectionStatus status, remote_action_codec::Free(remote_action); } - if (!runtime->need_to_create_stream_window_ && - !runtime->remote_sessions_.empty()) { + if (!runtime->need_to_create_stream_window_) { runtime->need_to_create_stream_window_ = true; } props->connection_established_ = true; @@ -496,6 +500,9 @@ void PeerEventHandler::OnNetStatusReport( const MiniRtcNetTrafficStats* net_traffic_stats, const char* user_id, const size_t user_id_size, void* user_data) { auto* handler = static_cast(user_data); + if (!handler) return; + auto callback = handler->EnterCallback(); + if (!callback) return; GuiRuntime* runtime = handler ? &handler->owner_ : nullptr; if (!runtime) { return; @@ -556,12 +563,8 @@ void PeerEventHandler::OnNetStatusReport( } std::string remote_id(user_id, user_id_size); - // std::shared_lock lock(runtime->remote_sessions_mutex_); - if (runtime->remote_sessions_.find(remote_id) == - runtime->remote_sessions_.end()) { - return; - } - auto props = runtime->remote_sessions_.find(remote_id)->second; + auto props = runtime->FindRemoteSession(remote_id); + if (!props) return; if (props->traversal_mode_ != mode) { props->traversal_mode_ = mode; LOG_INFO("Net mode: [{}]", int(props->traversal_mode_)); diff --git a/apps/desktop/src/gui/runtime/peer_event_handler.h b/apps/desktop/src/gui/runtime/peer_event_handler.h index 3458b50f..92650955 100644 --- a/apps/desktop/src/gui/runtime/peer_event_handler.h +++ b/apps/desktop/src/gui/runtime/peer_event_handler.h @@ -11,6 +11,7 @@ #include #include "minirtc.h" +#include "runtime/session_lifecycle.h" namespace crossdesk { @@ -21,6 +22,10 @@ class PeerEventHandler { public: explicit PeerEventHandler(GuiRuntime &owner); + bool IsActive() const { return callbacks_.IsActive(); } + auto EnterCallback() { return callbacks_.Enter(); } + void Deactivate() { callbacks_.Close(); } + static void OnReceiveVideoBuffer(const MiniRtcVideoFrame *video_frame, const char *user_id, size_t user_id_size, const char *src_id, size_t src_id_size, @@ -49,6 +54,7 @@ private: static void SendClientInfo(PeerPtr *peer, const std::string &client_id); GuiRuntime &owner_; + gui_detail::SessionCallbackGate callbacks_; }; } // namespace crossdesk diff --git a/apps/desktop/src/gui/runtime/peer_media_callbacks.cpp b/apps/desktop/src/gui/runtime/peer_media_callbacks.cpp index f23609e8..8ae5f734 100644 --- a/apps/desktop/src/gui/runtime/peer_media_callbacks.cpp +++ b/apps/desktop/src/gui/runtime/peer_media_callbacks.cpp @@ -32,19 +32,17 @@ void PeerEventHandler::OnReceiveVideoBuffer( const MiniRtcVideoFrame* video_frame, const char* user_id, size_t user_id_size, const char* src_id, size_t src_id_size, void* user_data) { auto* handler = static_cast(user_data); + if (!handler) return; + auto callback = handler->EnterCallback(); + if (!callback) return; GuiRuntime* runtime = handler ? &handler->owner_ : nullptr; if (!runtime) { return; } std::string remote_id(user_id, user_id_size); - // std::shared_lock lock(runtime->remote_sessions_mutex_); - if (runtime->remote_sessions_.find(remote_id) == - runtime->remote_sessions_.end()) { - return; - } - GuiRuntime::RemoteSession* props = - runtime->remote_sessions_.find(remote_id)->second.get(); + auto props = runtime->FindRemoteSession(remote_id); + if (!props) return; if (props->connection_established_) { bool background_snapshot_only = false; @@ -174,6 +172,9 @@ void PeerEventHandler::OnReceiveAudioBuffer( const char* data, size_t size, const char* user_id, size_t user_id_size, const char* src_id, size_t src_id_size, void* user_data) { auto* handler = static_cast(user_data); + if (!handler) return; + auto callback = handler->EnterCallback(); + if (!callback) return; GuiRuntime* runtime = handler ? &handler->owner_ : nullptr; if (!runtime) { return; diff --git a/apps/desktop/src/gui/runtime/remote_session.h b/apps/desktop/src/gui/runtime/remote_session.h index 8bddb8f3..24aa275c 100644 --- a/apps/desktop/src/gui/runtime/remote_session.h +++ b/apps/desktop/src/gui/runtime/remote_session.h @@ -24,6 +24,8 @@ #include "display_info.h" #include "minirtc.h" +namespace crossdesk { class PeerEventHandler; } + namespace crossdesk::gui_detail { struct FileTransferState { @@ -67,6 +69,8 @@ struct FileTransferState { struct RemoteSession { Params params_; PeerPtr* peer_ = nullptr; + std::shared_ptr peer_events_; + std::atomic closing_{false}; std::string audio_label_ = kAudioStream; std::string data_label_ = kDataStream; std::string mouse_label_ = kMouseStream; diff --git a/apps/desktop/src/gui/runtime/runtime_state.h b/apps/desktop/src/gui/runtime/runtime_state.h index 229b22f0..5c435db3 100644 --- a/apps/desktop/src/gui/runtime/runtime_state.h +++ b/apps/desktop/src/gui/runtime/runtime_state.h @@ -22,9 +22,11 @@ #include #include "config_center.h" +#include "common/task_queue_lock_free.h" #include "path_manager.h" #include "runtime/device_presence_cache.h" #include "runtime/remote_session.h" +#include "runtime/session_lifecycle.h" #include "thumbnail.h" namespace crossdesk::gui_detail { @@ -184,8 +186,15 @@ struct ConnectionState { std::unordered_map; RemoteSessionMap remote_sessions_; std::shared_mutex remote_sessions_mutex_; - std::vector thumbnail_save_threads_; - std::mutex thumbnail_save_threads_mutex_; + TaskQueueLockFree session_cleanup_queue_; + std::unordered_map> session_cleanup_tasks_; + std::future connection_peer_cleanup_; + std::unordered_map> controller_cleanup_tasks_; + struct PendingReconnect { + std::string password; + bool remember_password = false; + }; + std::unordered_map pending_reconnects_; std::shared_mutex connection_status_mutex_; std::unordered_map connection_status_; diff --git a/apps/desktop/src/gui/runtime/session_lifecycle.cpp b/apps/desktop/src/gui/runtime/session_lifecycle.cpp new file mode 100644 index 00000000..ae7d8203 --- /dev/null +++ b/apps/desktop/src/gui/runtime/session_lifecycle.cpp @@ -0,0 +1,47 @@ +#include "runtime/session_lifecycle.h" + +namespace crossdesk::gui_detail { + +CallbackGuard::CallbackGuard(SessionCallbackGate& callbacks) + : callbacks_(callbacks), accepted_(callbacks.BeginCallback()) {} + +CallbackGuard::~CallbackGuard() { + if (accepted_) { + callbacks_.EndCallback(); + } +} + +bool SessionCallbackGate::IsActive() const { + std::lock_guard lock(mutex_); + return accepting_callbacks_; +} + +CallbackGuard SessionCallbackGate::Enter() { + return CallbackGuard(*this); +} + +bool SessionCallbackGate::BeginCallback() { + std::lock_guard lock(mutex_); + if (!accepting_callbacks_) { + return false; + } + ++running_callbacks_; + return true; +} + +void SessionCallbackGate::EndCallback() { + std::lock_guard lock(mutex_); + --running_callbacks_; + if (running_callbacks_ == 0) { + callbacks_finished_.notify_all(); + } +} + +void SessionCallbackGate::Close() { + std::unique_lock lock(mutex_); + accepting_callbacks_ = false; + // Waiting releases the lock so running callbacks can call EndCallback(). + callbacks_finished_.wait(lock, [this] { return running_callbacks_ == 0; }); +} + +} // namespace crossdesk::gui_detail diff --git a/apps/desktop/src/gui/runtime/session_lifecycle.h b/apps/desktop/src/gui/runtime/session_lifecycle.h new file mode 100644 index 00000000..bb64495c --- /dev/null +++ b/apps/desktop/src/gui/runtime/session_lifecycle.h @@ -0,0 +1,58 @@ +/* + * @Author: DI JUNKUN + * @Date: 2026-09-07 + * Copyright (c) 2026 by DI JUNKUN, All Rights Reserved. + */ + +#ifndef _SESSION_LIFECYCLE_H_ +#define _SESSION_LIFECYCLE_H_ + +#include +#include +#include + +namespace crossdesk::gui_detail { + +class SessionCallbackGate; + +// Decrements the count on scope exit, including early returns and exceptions. +class CallbackGuard { + public: + explicit CallbackGuard(SessionCallbackGate& callbacks); + ~CallbackGuard(); + + CallbackGuard(const CallbackGuard&) = delete; + CallbackGuard& operator=(const CallbackGuard&) = delete; + + explicit operator bool() const { return accepted_; } + + private: + SessionCallbackGate& callbacks_; + bool accepted_; +}; + +// Each peer counts callbacks as they enter and leave. +// Closing rejects new callbacks, then waits for running callbacks to finish. +class SessionCallbackGate { + public: + bool IsActive() const; + CallbackGuard Enter(); + + // Called by the session owner, never from within this session's callbacks. + void Close(); + + private: + friend class CallbackGuard; + + bool BeginCallback(); + void EndCallback(); + + mutable std::mutex mutex_; + std::condition_variable callbacks_finished_; + bool accepting_callbacks_ = true; + std::size_t running_callbacks_ = 0; +}; + +} // namespace crossdesk::gui_detail + +#endif diff --git a/apps/desktop/xmake/targets.lua b/apps/desktop/xmake/targets.lua index 5a241f32..89f3d48b 100644 --- a/apps/desktop/xmake/targets.lua +++ b/apps/desktop/xmake/targets.lua @@ -34,6 +34,7 @@ function setup_targets() set_kind("object") add_deps("rd_log", "crossdesk_wire") add_packages("libyuv") + add_packages("concurrentqueue", {public = true}) add_files("apps/desktop/src/common/*.cpp") remove_files("apps/desktop/src/common/rounded_corner_button.cpp") if is_os("windows") then diff --git a/deps/submodules/minirtc b/deps/submodules/minirtc index 5d8657e6..64547718 160000 --- a/deps/submodules/minirtc +++ b/deps/submodules/minirtc @@ -1 +1 @@ -Subproject commit 5d8657e63ef851a34a33e6f0012083ca61c3be9a +Subproject commit 645477187a3658ba1986c3cf57daa01940faecf2