diff --git a/src/control.cc b/src/control.cc index 92229c73..87f2e97b 100644 --- a/src/control.cc +++ b/src/control.cc @@ -90,6 +90,7 @@ Control::cleanup() { scgi_thread::thread()->stop_thread_wait(); // Wait for all session files to be written. + session_thread::manager()->flush_all_pending_builds(); session_thread::thread()->stop_thread_wait(); m_ui->cleanup(); diff --git a/src/session/download_storer.cc b/src/session/download_storer.cc index 5a17b04f..e7b016f1 100644 --- a/src/session/download_storer.cc +++ b/src/session/download_storer.cc @@ -156,9 +156,9 @@ save_stream(const std::string& path, bool use_fsyncdisk, const std::stringstream void DownloadStorer::save_and_move_streams(const std::string& path, bool use_fsyncdisk, - const std::stringstream& torrent_stream, - const std::stringstream& rtorrent_stream, - const std::stringstream& libtorrent_stream) { + const std::stringstream* torrent_stream, + const std::stringstream* rtorrent_stream, + const std::stringstream* libtorrent_stream) { // LT_LOG("saving download : download:%p path:%s", download, path.c_str()); auto torrent_path = path; @@ -166,14 +166,14 @@ DownloadStorer::save_and_move_streams(const std::string& path, bool use_fsyncdis auto rtorrent_path = path + ".rtorrent"; if (torrent_stream) { - if (!save_stream(torrent_path + ".new", use_fsyncdisk, torrent_stream)) + if (!save_stream(torrent_path + ".new", use_fsyncdisk, *torrent_stream)) return; } - if (!save_stream(libtorrent_path + ".new", use_fsyncdisk, libtorrent_stream)) + if (!save_stream(libtorrent_path + ".new", use_fsyncdisk, *libtorrent_stream)) return; - if (!save_stream(rtorrent_path + ".new", use_fsyncdisk, rtorrent_stream)) + if (!save_stream(rtorrent_path + ".new", use_fsyncdisk, *rtorrent_stream)) return; if (torrent_stream) { diff --git a/src/session/download_storer.h b/src/session/download_storer.h index 9601b443..9472ce21 100644 --- a/src/session/download_storer.h +++ b/src/session/download_storer.h @@ -34,9 +34,9 @@ public: auto libtorrent_stream() { return std::move(m_libtorrent_stream); } static void save_and_move_streams(const std::string& path, bool use_fsyncdisk, - const std::stringstream& torrent_stream, - const std::stringstream& rtorrent_stream, - const std::stringstream& libtorrent_stream); + const std::stringstream* torrent_stream, + const std::stringstream* rtorrent_stream, + const std::stringstream* libtorrent_stream); static utils::Directory get_formated_entries(const std::string& session_path); diff --git a/src/session/session_manager.cc b/src/session/session_manager.cc index 4c397de8..82bb2b47 100644 --- a/src/session/session_manager.cc +++ b/src/session/session_manager.cc @@ -76,7 +76,7 @@ SessionManager::save_resume_download(core::Download* download) { } if (m_pending_builds.empty()) - torrent::main_thread::callback(this, [this]() { process_pending_resume_builds(); }); + torrent::main_thread::callback(this, [this]() { process_pending_resume_builds(false); }); m_pending_builds.push_back(download); } @@ -182,6 +182,13 @@ SessionManager::cleanup() { if (!m_active) throw torrent::internal_error("SessionManager::cleanup() called while not active."); + { + std::unique_lock pending_lock(m_pending_builds_mutex); + + if (!m_pending_builds.empty()) + throw torrent::internal_error("SessionManager::cleanup() called with pending builds."); + } + m_active = false; if (m_path.empty()) { @@ -207,48 +214,60 @@ SessionManager::cleanup() { } void -SessionManager::process_pending_resume_builds() { +SessionManager::process_pending_resume_builds(bool is_flushing) { assert(torrent::this_thread::thread() == torrent::main_thread::thread()); - std::unique_lock lock(m_pending_builds_mutex); + std::vector requests; - if (!m_active) - throw torrent::internal_error("SessionManager::process_pending_builds() called while not active."); + { + std::unique_lock lock(m_pending_builds_mutex); - while (!m_pending_builds.empty() && m_pending_builds.size() < max_concurrent_requests) { - auto* download = m_pending_builds.front(); - m_pending_builds.pop_front(); + if (m_pending_builds.empty()) + return; - LT_LOG("processing pending resume save : download:%p", download); + while (!m_pending_builds.empty()) { + if (!is_flushing && m_pending_builds.size() >= max_concurrent_requests) + break; - DownloadStorer storer(download); + auto* download = m_pending_builds.front(); + m_pending_builds.pop_front(); - storer.build_resume_streams(); + LT_LOG("processing pending resume save : download:%p", download); - auto save_request = SaveRequest{ - download, - storer.build_path(m_path), - nullptr, - storer.rtorrent_stream(), - storer.libtorrent_stream() - }; + DownloadStorer storer(download); - { - std::unique_lock save_lock(m_mutex); + storer.build_resume_streams(); - if (!m_active) - throw torrent::internal_error("SessionManager::process_pending_builds() called while not active."); + auto save_request = SaveRequest{ + download, + storer.build_path(m_path), + nullptr, + storer.rtorrent_stream(), + storer.libtorrent_stream() + }; + requests.push_back(std::move(save_request)); + } + } + + { + std::unique_lock save_lock(m_mutex); + + if (!m_active) + throw torrent::internal_error("SessionManager::process_pending_builds() called while not active."); + + for (auto& save_request : requests) { if (remove_or_replace_unsafe(save_request)) { - LT_LOG("updated pending resume save request : download:%p", download); + LT_LOG("updated pending resume save request : download:%p", save_request.download); } else { - LT_LOG("queued new resume save request : download:%p", download); + LT_LOG("queued new resume save request : download:%p", save_request.download); m_save_requests.push_back(std::move(save_request)); } } - - session_thread::callback(this, [this]() { process_save_request_with_pending_callback(); }); } + + if (!is_flushing) + session_thread::callback(this, [this]() { process_save_request_with_pending_callback(); }); } void @@ -279,7 +298,7 @@ SessionManager::process_save_request_with_pending_callback() { std::unique_lock lock(m_pending_builds_mutex); if (!m_pending_builds.empty()) - torrent::main_thread::callback(this, [this]() { process_pending_resume_builds(); }); + torrent::main_thread::callback(this, [this]() { process_pending_resume_builds(false); }); } void @@ -295,9 +314,9 @@ SessionManager::process_next_save_request_unsafe() { // TODO: Consider adding a failed_saves with error info. DownloadStorer::save_and_move_streams(itr->second.path, m_use_fsyncdisk, - *itr->second.torrent_stream, - *itr->second.rtorrent_stream, - *itr->second.libtorrent_stream); + itr->second.torrent_stream.get(), + itr->second.rtorrent_stream.get(), + itr->second.libtorrent_stream.get()); { std::unique_lock lock(m_mutex); @@ -330,7 +349,7 @@ SessionManager::process_finished_saves() { void SessionManager::wait_for_one_save_unsafe(std::unique_lock& lock) { - assert(m_thread == torrent::this_thread::thread()); + // Can be called in any thread. if (m_processing_saves.empty()) return; diff --git a/src/session/session_manager.h b/src/session/session_manager.h index 67af1aff..effd834e 100644 --- a/src/session/session_manager.h +++ b/src/session/session_manager.h @@ -12,6 +12,8 @@ #include #include +class Control; + namespace core { class Download; } @@ -63,7 +65,7 @@ public: void remove_download(core::Download* download); protected: - friend class Control; + friend class ::Control; friend class ThreadSession; void save_download(core::Download* download, bool skip_static); @@ -71,8 +73,10 @@ protected: void start(); void cleanup(); + void flush_all_pending_builds(); + private: - void process_pending_resume_builds(); + void process_pending_resume_builds(bool is_flushing); void process_save_request(); void process_save_request_with_pending_callback(); void process_next_save_request_unsafe(); @@ -110,10 +114,11 @@ private: std::deque m_pending_builds; }; -inline bool SessionManager::is_used() const { return !m_path.empty(); } -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 bool SessionManager::is_used() const { return !m_path.empty(); } +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); } } // namespace session