From bd53959b40e6a514ef567ef397063362b8eb705f Mon Sep 17 00:00:00 2001 From: Jari Sundell Date: Sat, 11 Apr 2026 09:50:58 +0200 Subject: [PATCH] Fixed session save getting stuck. --- src/session/session_manager.cc | 139 ++++++++++++++++++++++----------- src/session/session_manager.h | 15 +++- 2 files changed, 105 insertions(+), 49 deletions(-) diff --git a/src/session/session_manager.cc b/src/session/session_manager.cc index 2381a164..2eedd4aa 100644 --- a/src/session/session_manager.cc +++ b/src/session/session_manager.cc @@ -14,6 +14,11 @@ #define LT_LOG(log_fmt, ...) \ lt_log_print(torrent::LOG_SESSION_EVENTS, "session-events: " log_fmt, __VA_ARGS__); +// TODO: Properly handle errors. +// +// TODO: If no more sockets can be opened, wait for a job to finish. if all is finished, use a +// timeout and try again. + namespace session { SessionManager::SessionManager(torrent::utils::Thread* thread) @@ -68,19 +73,20 @@ SessionManager::save_resume_download(core::Download* download) { { std::unique_lock lock(m_pending_builds_mutex); + // TODO: This is under the wrong lock. if (!m_active) throw torrent::internal_error("SessionManager::save_resume_download() called while not active."); if (std::find(m_pending_builds.begin(), m_pending_builds.end(), download) != m_pending_builds.end()) { + // TODO: Do we need to make sure we're getting processed? LT_LOG("download already in pending build of resume save : download:%p", download); return; } - if (m_pending_builds.empty()) - torrent::main_thread::callback(this, [this]() { process_pending_resume_builds(false); }); - m_pending_builds.push_back(download); + callback_pending_builds(); + LT_LOG("build of resume save data queued : download:%p", download); } } @@ -123,8 +129,7 @@ SessionManager::save_full_download(core::Download* download) { LT_LOG("queued new full save request : download:%p", download); } - if (!m_processing_saves_callback_scheduled.exchange(true)) - session_thread::callback(this, [this]() { process_save_request(); }); + callback_save_request(); } void @@ -221,7 +226,40 @@ SessionManager::cleanup() { } void -SessionManager::process_pending_resume_builds(bool is_flushing) { +SessionManager::callback_pending_builds() { + if (m_save_request_counter >= max_concurrent_requests) + return; + + if (m_callback_scheduled_process_pending_builds.exchange(true)) + return; + + torrent::main_thread::callback(this, [this]() { process_pending_builds(false); }); +} + +void +SessionManager::callback_save_request() { + if (m_save_request_counter == 0) + return; + + if (m_processing_save_counter >= max_concurrent_processing) + return; + + if (m_callback_scheduled_process_saves_request.exchange(true)) + return; + + session_thread::callback(this, [this]() { process_save_request(); }); +} + +void +SessionManager::callback_finished_saves() { + if (m_callback_scheduled_process_finished_saves.exchange(true)) + return; + + session_thread::callback(this, [this]() { process_finished_saves(); }); +} + +void +SessionManager::process_pending_builds(bool is_flushing) { assert(torrent::this_thread::thread() == torrent::main_thread::thread()); std::vector requests; @@ -231,6 +269,8 @@ SessionManager::process_pending_resume_builds(bool is_flushing) { { std::unique_lock lock(m_pending_builds_mutex); + m_callback_scheduled_process_pending_builds = false; + while (!m_pending_builds.empty()) { if (!is_flushing && m_save_request_counter + requests.size() >= max_concurrent_requests) break; @@ -272,15 +312,14 @@ SessionManager::process_pending_resume_builds(bool is_flushing) { } m_save_requests.push_back(std::move(save_request)); + m_save_request_counter = m_save_requests.size(); LT_LOG("queued new resume save request : download:%p", save_request.download); } - - m_save_request_counter = m_save_requests.size(); } - if (!is_flushing && !m_processing_saves_callback_scheduled.exchange(true)) - session_thread::callback(this, [this]() { process_save_request_with_pending_callback(); }); + if (!is_flushing) + callback_save_request(); } void @@ -290,25 +329,23 @@ SessionManager::process_save_request() { if (m_path.empty()) throw torrent::internal_error("SessionManager::process_save_request() called with empty path."); - std::unique_lock lock(m_mutex); + { + std::unique_lock lock(m_mutex); - if (!m_active) - throw torrent::internal_error("SessionManager::process_save_request() called while not active."); + m_callback_scheduled_process_saves_request = false; - while (!m_save_requests.empty() && m_processing_saves.size() < max_concurrent_requests) - process_next_save_request_unsafe(); + if (!m_active) + throw torrent::internal_error("SessionManager::process_save_request() called while not active."); - m_processing_saves_callback_scheduled = false; -} + while (!m_save_requests.empty()) { + if (m_processing_saves.size() >= max_concurrent_processing) + break; -void -SessionManager::process_save_request_with_pending_callback() { - process_save_request(); + process_next_save_request_unsafe(); + } + } - std::unique_lock lock(m_pending_builds_mutex); - - if (!m_pending_builds.empty()) - torrent::main_thread::callback(this, [this]() { process_pending_resume_builds(false); }); + callback_pending_builds(); } void @@ -319,19 +356,20 @@ SessionManager::process_next_save_request_unsafe() { m_save_request_counter = m_save_requests.size(); auto itr = m_processing_saves.insert(m_processing_saves.end(), ProcessingSave{}); + m_processing_save_counter = m_processing_saves.size(); itr->second = std::move(request); itr->first = std::async(std::launch::async, [this, itr]() { auto cleanup_fn = [this, itr]() { std::unique_lock lock(m_mutex); - if (m_finished_saves.empty()) - session_thread::callback(this, [this]() { process_finished_saves(); }); + callback_finished_saves(); m_finished_saves.push_back(std::move(*itr)); m_finished_condition.notify_all(); m_processing_saves.erase(itr); + m_processing_save_counter = m_processing_saves.size(); }; try { @@ -346,6 +384,8 @@ SessionManager::process_next_save_request_unsafe() { cleanup_fn(); }); + + LT_LOG("started save of download : download:%p path:%s", itr->second.download, itr->second.path.c_str()); } void @@ -354,6 +394,8 @@ SessionManager::process_finished_saves() { std::unique_lock lock(m_mutex); + m_callback_scheduled_process_finished_saves = false; + if (!m_active) throw torrent::internal_error("SessionManager::process_finished_saves() called while not active."); @@ -373,6 +415,7 @@ SessionManager::process_finished_saves() { m_last_storage_error_message = torrent::this_thread::cached_time(); m_ignored_storage_error_count = 0; + continue; } catch (torrent::internal_error& e) { @@ -388,14 +431,18 @@ SessionManager::process_finished_saves() { } m_finished_saves.clear(); + + callback_save_request(); } void SessionManager::flush_all_and_wait_unsafe(std::unique_lock& lock) { LT_LOG("flushing all pending saves", 0); + // Caller already ensured pending builds are empty. + while (!m_save_requests.empty()) { - if (m_processing_saves.size() >= max_cleanup_processing) { + if (m_processing_save_counter >= max_cleanup_processing) { m_finished_condition.wait(lock); continue; } @@ -430,14 +477,17 @@ SessionManager::replace_save_request_unsafe(SaveRequest& save_request) { // If the existing request is a full save (created during torrent initialization), // keep its torrent_stream but update the resume streams with newer data. + // // Otherwise, replace all streams. + if (itr->torrent_stream != nullptr) { // Keep existing full-save torrent stream, only update resume streams if (save_request.rtorrent_stream != nullptr) itr->rtorrent_stream = std::move(save_request.rtorrent_stream); - + if (save_request.libtorrent_stream != nullptr) itr->libtorrent_stream = std::move(save_request.libtorrent_stream); + } else { // No full save pending, replace everything itr->torrent_stream = std::move(save_request.torrent_stream); @@ -452,18 +502,6 @@ bool SessionManager::remove_completely_unsafe(core::Download* download, std::unique_lock& lock) { assert(torrent::this_thread::thread() == torrent::main_thread::thread()); - auto remove_requests = [this, download]() { - auto itr = std::remove_if(m_save_requests.begin(), m_save_requests.end(), [download](auto& req) { - return req.download == download; - }); - - if (itr == m_save_requests.end()) - return false; - - m_save_requests.erase(itr, m_save_requests.end()); - return true; - }; - auto remove_pending = [this, download]() { std::unique_lock pending_lock(m_pending_builds_mutex); @@ -478,13 +516,28 @@ SessionManager::remove_completely_unsafe(core::Download* download, std::unique_l return true; }; + auto remove_requests = [this, download]() { + auto itr = std::remove_if(m_save_requests.begin(), m_save_requests.end(), [download](auto& req) { + return req.download == download; + }); + + if (itr == m_save_requests.end()) + return false; + + m_save_requests.erase(itr, m_save_requests.end()); + m_save_request_counter = m_save_requests.size(); + return true; + }; + bool removed_something = false; if (remove_pending()) removed_something = true; - if (remove_requests()) + if (remove_requests()) { removed_something = true; + callback_pending_builds(); + } // This may block for a relatively long time if fdatasync is in use, however this is necessary. while (true) { @@ -507,9 +560,5 @@ SessionManager::remove_completely_unsafe(core::Download* download, std::unique_l return removed_something; } -// TODO: Properly handle errors. -// TODO: If no more sockets can be opened, wait for a job to finish. if all is finished, use a -// timeout and try again. - } // namespace session diff --git a/src/session/session_manager.h b/src/session/session_manager.h index 1d9eb9d2..7aa53c04 100644 --- a/src/session/session_manager.h +++ b/src/session/session_manager.h @@ -74,9 +74,12 @@ protected: void flush_all_pending_builds(); private: - void process_pending_resume_builds(bool is_flushing); + void callback_pending_builds(); + void callback_save_request(); + void callback_finished_saves(); + + void process_pending_builds(bool is_flushing); void process_save_request(); - void process_save_request_with_pending_callback(); void process_next_save_request_unsafe(); void process_finished_saves(); @@ -101,10 +104,14 @@ private: std::deque m_save_requests; std::atomic m_save_request_counter{}; std::list m_processing_saves; - std::atomic m_processing_saves_callback_scheduled{}; + std::atomic m_processing_save_counter{}; std::condition_variable m_finished_condition; std::vector m_finished_saves; + std::atomic m_callback_scheduled_process_pending_builds{}; + std::atomic m_callback_scheduled_process_saves_request{}; + std::atomic m_callback_scheduled_process_finished_saves{}; + std::unique_ptr m_lockfile; std::chrono::microseconds m_last_storage_error_message{}; @@ -119,7 +126,7 @@ inline bool SessionManager::is_used() const { return !m_path.e inline std::string SessionManager::path() const { return m_path; } inline bool SessionManager::use_fsyncdisk() const { return true; } inline bool SessionManager::use_lock() const { return m_use_lock; } -inline void SessionManager::flush_all_pending_builds() { process_pending_resume_builds(true); } +inline void SessionManager::flush_all_pending_builds() { process_pending_builds(true); } } // namespace session