From 4bdeb58eb606d6ac21f757bb0422138c2004b96d Mon Sep 17 00:00:00 2001 From: Jari Sundell Date: Sun, 21 Dec 2025 22:26:35 +0100 Subject: [PATCH] Run multiple session save requests in parallel. --- src/Makefile.am | 4 +- src/command_download.cc | 6 +- src/command_local.cc | 4 +- src/control.cc | 3 - src/core/dht_manager.cc | 1 - src/core/download_factory.cc | 1 - src/core/download_list.cc | 26 ++-- src/core/download_store.cc | 127 ----------------- src/core/download_store.h | 35 ----- src/core/manager.cc | 6 +- src/core/manager.h | 3 - src/main.cc | 5 +- src/scgi/thread_scgi.cc | 5 - src/scgi/thread_scgi.h | 13 +- src/session/download_storer.cc | 112 +++++++++++++++ src/session/download_storer.h | 44 ++++++ src/session/session_manager.cc | 248 +++++++++++++++++++++++---------- src/session/session_manager.h | 54 +++++-- src/session/thread_session.cc | 13 +- src/session/thread_session.h | 10 +- src/ui/root.cc | 1 - 21 files changed, 404 insertions(+), 317 deletions(-) delete mode 100644 src/core/download_store.cc delete mode 100644 src/core/download_store.h create mode 100644 src/session/download_storer.cc create mode 100644 src/session/download_storer.h diff --git a/src/Makefile.am b/src/Makefile.am index 690631eb..89280276 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -13,8 +13,6 @@ libsub_root_a_SOURCES = \ core/download_factory.h \ core/download_list.cc \ core/download_list.h \ - core/download_store.cc \ - core/download_store.h \ core/http_queue.cc \ core/http_queue.h \ core/manager.cc \ @@ -128,6 +126,8 @@ libsub_root_a_SOURCES = \ scgi/thread_scgi.cc \ scgi/thread_scgi.h \ \ + session/download_storer.cc \ + session/download_storer.h \ session/session_manager.cc \ session/session_manager.h \ session/thread_session.cc \ diff --git a/src/command_download.cc b/src/command_download.cc index f3768c2d..c6b319a8 100644 --- a/src/command_download.cc +++ b/src/command_download.cc @@ -25,9 +25,9 @@ #include #include "core/download.h" -#include "core/download_store.h" #include "core/manager.h" #include "rpc/parse.h" +#include "session/session_manager.h" #include "globals.h" #include "control.h" @@ -700,8 +700,8 @@ initialize_command_download() { CMD2_DL_V ("d.erase", std::bind(&core::DownloadList::erase_ptr, control->core()->download_list(), std::placeholders::_1)); CMD2_DL_V ("d.check_hash", std::bind(&core::DownloadList::check_hash, control->core()->download_list(), std::placeholders::_1)); - CMD2_DL ("d.save_resume", std::bind(&core::DownloadStore::save_resume, control->core()->download_store(), std::placeholders::_1)); - CMD2_DL ("d.save_full_session", std::bind(&core::DownloadStore::save_full, control->core()->download_store(), std::placeholders::_1)); + CMD2_DL_V ("d.save_resume", [](core::Download* download, auto) { session_thread::manager()->save_resume_download(download); }); + CMD2_DL_V ("d.save_full_session", [](core::Download* download, auto) { session_thread::manager()->save_full_download(download); }); CMD2_DL_V ("d.update_priorities", CMD2_ON_DL(update_priorities)); diff --git a/src/command_local.cc b/src/command_local.cc index 736b969f..fc24cda5 100644 --- a/src/command_local.cc +++ b/src/command_local.cc @@ -17,7 +17,6 @@ #include "core/download.h" #include "core/download_list.h" -#include "core/download_store.h" #include "core/manager.h" #include "rak/string_manip.h" #include "rpc/parse_commands.h" @@ -212,7 +211,8 @@ initialize_command_local() { CMD2_ANY_VALUE_V ("system.files.advise_random.set", std::bind(&FM_t::set_advise_random, fileManager, std::placeholders::_2)); CMD2_ANY ("system.files.advise_random.hashing", std::bind(&FM_t::advise_random_hashing, fileManager)); CMD2_ANY_VALUE_V ("system.files.advise_random.hashing.set", std::bind(&FM_t::set_advise_random_hashing, fileManager, std::placeholders::_2)); - CMD2_VAR_BOOL ("system.files.session.fdatasync", true); + CMD2_ANY ("system.files.session.fdatasync", [](auto, auto) { return session_thread::manager()->use_fsyncdisk(); }); + CMD2_ANY_VALUE_V ("system.files.session.fdatasync.set", [](auto, auto& value) { return session_thread::manager()->set_use_fsyncdisk(value); }); CMD2_ANY ("system.files.opened_counter", std::bind(&FM_t::files_opened_counter, fileManager)); CMD2_ANY ("system.files.closed_counter", std::bind(&FM_t::files_closed_counter, fileManager)); diff --git a/src/control.cc b/src/control.cc index 1542018c..92229c73 100644 --- a/src/control.cc +++ b/src/control.cc @@ -9,7 +9,6 @@ #include #include "core/dht_manager.h" -#include "core/download_store.h" #include "core/http_queue.h" #include "core/manager.h" #include "core/view_manager.h" @@ -59,9 +58,7 @@ Control::~Control() { void Control::initialize() { - session_thread::manager()->start(); session_thread::thread()->start_thread(); - scgi_thread::thread()->start_thread(); display::Canvas::initialize(); diff --git a/src/core/dht_manager.cc b/src/core/dht_manager.cc index 3ab551b4..3a094936 100644 --- a/src/core/dht_manager.cc +++ b/src/core/dht_manager.cc @@ -13,7 +13,6 @@ #include "control.h" #include "download.h" -#include "download_store.h" #include "globals.h" #include "manager.h" #include "rpc/parse_commands.h" diff --git a/src/core/download_factory.cc b/src/core/download_factory.cc index 8811e0a1..c47fac91 100644 --- a/src/core/download_factory.cc +++ b/src/core/download_factory.cc @@ -20,7 +20,6 @@ #include "control.h" #include "globals.h" #include "core/download.h" -#include "core/download_store.h" #include "core/http_queue.h" #include "core/manager.h" #include "rpc/parse_commands.h" diff --git a/src/core/download_list.cc b/src/core/download_list.cc index 5c024e6a..f92e694d 100644 --- a/src/core/download_list.cc +++ b/src/core/download_list.cc @@ -22,10 +22,10 @@ #include "view.h" #include "view_manager.h" -#include "dht_manager.h" -#include "download.h" -#include "download_list.h" -#include "download_store.h" +#include "core/dht_manager.h" +#include "core/download.h" +#include "core/download_list.h" +#include "session/session_manager.h" #include "ui/root.h" #define DL_TRIGGER_EVENT(download, event_name) \ @@ -68,10 +68,8 @@ DownloadList::clear() { void DownloadList::session_save() { - unsigned int c = std::count_if(begin(), end(), [&](Download* d) { return control->core()->download_store()->save_resume(d); }); - - if (c != size()) - lt_log_print(torrent::LOG_ERROR, "Failed to save session torrents."); + for (auto& download : *this) + session_thread::manager()->save_resume_download(download); control->dht_manager()->save_dht_cache(); control->ui()->save_input_history(); @@ -199,10 +197,10 @@ DownloadList::erase(iterator itr) { (*itr)->set_hash_failed(true); close(*itr); - - control->core()->download_store()->remove(*itr); + session_thread::manager()->remove_download(*itr); DL_TRIGGER_EVENT(*itr, "event.download.erased"); + for (auto v : *control->view_manager()) v->erase(*itr); @@ -639,10 +637,8 @@ DownloadList::confirm_finished(Download* download) { // the download. // // Obsolete. - if (!download->is_active() && rpc::call_command_value("session.on_completion") != 0) { - // torrent::resume_save_progress(*download->download(), download->download()->bencode()->get_key("libtorrent_resume")); - control->core()->download_store()->save_resume(download); - } + if (!download->is_active() && rpc::call_command_value("session.on_completion") != 0) + session_thread::manager()->save_resume_download(download); // Send the completed request before resuming so we don't reset the // up/downloaded baseline. @@ -655,7 +651,7 @@ DownloadList::confirm_finished(Download* download) { if (find(infohash) == end()) return; - + // if (download->resume_flags() != ~uint32_t()) // throw torrent::internal_error("DownloadList::confirm_finished(...) download->resume_flags() != ~uint32_t()."); diff --git a/src/core/download_store.cc b/src/core/download_store.cc deleted file mode 100644 index d368314b..00000000 --- a/src/core/download_store.cc +++ /dev/null @@ -1,127 +0,0 @@ -// DownloadStore handles the saving and listing of session torrents. - -#include "config.h" - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include "download.h" -#include "download_store.h" -#include "rpc/parse_commands.h" -#include "session/session_manager.h" -#include "utils/directory.h" - -namespace core { - -bool -DownloadStore::save(Download* d, int flags) { - if (!session_thread::manager()->is_used()) - return true; - - torrent::Object* resume_base = &d->download()->bencode()->get_key("libtorrent_resume"); - torrent::Object* rtorrent_base = &d->download()->bencode()->get_key("rtorrent"); - - // Move this somewhere else? - rtorrent_base->insert_key("chunks_done", d->download()->file_list()->completed_chunks()); - rtorrent_base->insert_key("chunks_wanted", d->download()->data()->wanted_chunks()); - rtorrent_base->insert_key("total_uploaded", d->info()->up_rate()->total()); - rtorrent_base->insert_key("total_downloaded", d->info()->down_rate()->total()); - - // Don't save for completed torrents when we've cleared the uncertain_pieces. - torrent::resume_save_progress(*d->download(), *resume_base); - torrent::resume_save_uncertain_pieces(*d->download(), *resume_base); - - torrent::resume_save_addresses(*d->download(), *resume_base); - torrent::resume_save_file_priorities(*d->download(), *resume_base); - torrent::resume_save_tracker_settings(*d->download(), *resume_base); - - // Temp fixing of all flags, move to a better place: - resume_base->set_flags(torrent::Object::flag_session_data); - rtorrent_base->set_flags(torrent::Object::flag_session_data); - - auto download_stream = std::unique_ptr(); - auto resume_stream = std::make_unique(); - auto rtorrent_stream = std::make_unique(); - - if (!(flags & flag_skip_static)) { - download_stream = std::make_unique(); - torrent::object_write_bencode(&*download_stream, d->bencode(), torrent::Object::flag_session_data); - - if (!download_stream->good()) - return false; - } - - torrent::object_write_bencode(&*resume_stream, resume_base, 0); - - // TODO: Add logging. - if (!resume_stream->good()) - return false; - - torrent::object_write_bencode(&*rtorrent_stream, rtorrent_base, 0); - - if (!rtorrent_stream->good()) - return false; - - auto base_filename = create_filename(d); - - session_thread::manager()->save_download(d, base_filename, std::move(download_stream), std::move(resume_stream), std::move(rtorrent_stream)); - return true; -} - -void -DownloadStore::remove(Download* d) { - session_thread::manager()->remove_download(d, create_filename(d)); -} - -// This also needs to check that it isn't a directory. -bool -not_correct_format(const utils::directory_entry& entry) { - return !DownloadStore::is_correct_format(entry.s_name); -} - -utils::Directory -DownloadStore::get_formated_entries() { - if (!session_thread::manager()->is_used()) - return utils::Directory(); - - utils::Directory d(session_thread::manager()->path()); - - if (!d.update(utils::Directory::update_hide_dot)) - throw torrent::storage_error("core::DownloadStore::update() could not open session directory: " + session_thread::manager()->path()); - - d.erase(std::remove_if(d.begin(), d.end(), [&](const utils::directory_entry& entry) { return not_correct_format(entry); }), d.end()); - - return d; -} - -bool -DownloadStore::is_correct_format(const std::string& f) { - if (f.size() != 48 || f.substr(40) != ".torrent") - return false; - - for (std::string::const_iterator itr = f.begin(); itr != f.end() - 8; ++itr) - if (!(*itr >= '0' && *itr <= '9') && - !(*itr >= 'A' && *itr <= 'F')) - return false; - - return true; -} - -std::string -DownloadStore::create_filename(Download* d) { - return session_thread::manager()->path() + rak::transform_hex(d->info()->hash().begin(), d->info()->hash().end()) + ".torrent"; -} - -} diff --git a/src/core/download_store.h b/src/core/download_store.h deleted file mode 100644 index 19e6eb41..00000000 --- a/src/core/download_store.h +++ /dev/null @@ -1,35 +0,0 @@ -#ifndef RTORRENT_CORE_DOWNLOAD_STORE_H -#define RTORRENT_CORE_DOWNLOAD_STORE_H - -#include -#include - -namespace utils { - class Directory; -} - -namespace core { - -class Download; - -class DownloadStore { -public: - static const int flag_skip_static = 0x1; - - bool save(Download* d, int flags); - bool save_full(Download* d) { return save(d, 0); } - bool save_resume(Download* d) { return save(d, flag_skip_static); } - void remove(Download* d); - - // Currently shows all entries in the correct format. - utils::Directory get_formated_entries(); - - static bool is_correct_format(const std::string& f); - -private: - std::string create_filename(Download* d); -}; - -} - -#endif diff --git a/src/core/manager.cc b/src/core/manager.cc index a50ce4c6..07edbf07 100644 --- a/src/core/manager.cc +++ b/src/core/manager.cc @@ -32,7 +32,6 @@ #include "control.h" #include "core/download.h" #include "core/download_factory.h" -#include "core/download_store.h" #include "core/http_queue.h" #include "core/manager.h" #include "core/view.h" @@ -50,11 +49,10 @@ Manager::push_log(const char* msg) { m_log_complete->lock_and_push_log(msg, strlen(msg), 0); } -Manager::Manager() : - m_log_important(torrent::log_open_log_buffer("important")), +Manager::Manager() + : m_log_important(torrent::log_open_log_buffer("important")), m_log_complete(torrent::log_open_log_buffer("complete")) { - m_download_store = std::make_unique(); m_download_list = std::make_unique(); m_file_status_cache = std::make_unique(); m_http_queue = std::make_unique(); diff --git a/src/core/manager.h b/src/core/manager.h index 89c2bcdc..07820f67 100644 --- a/src/core/manager.h +++ b/src/core/manager.h @@ -21,7 +21,6 @@ class FileStatusCache; namespace core { -class DownloadStore; class HttpQueue; typedef std::map ThrottleMap; @@ -37,7 +36,6 @@ public: ~Manager(); DownloadList* download_list() { return m_download_list.get(); } - DownloadStore* download_store() { return m_download_store.get(); } FileStatusCache* file_status_cache() { return m_file_status_cache.get(); } HttpQueue* http_queue() { return m_http_queue.get(); } @@ -95,7 +93,6 @@ private: void receive_hashing_changed(); std::unique_ptr m_download_list; - std::unique_ptr m_download_store; std::unique_ptr m_file_status_cache; std::unique_ptr m_http_queue; diff --git a/src/main.cc b/src/main.cc index 4c9893c0..19b57920 100644 --- a/src/main.cc +++ b/src/main.cc @@ -22,7 +22,6 @@ #include "core/dht_manager.h" #include "core/download.h" #include "core/download_factory.h" -#include "core/download_store.h" #include "core/manager.h" #include "display/canvas.h" #include "display/window.h" @@ -32,7 +31,9 @@ #include "rpc/command_scheduler_item.h" #include "rpc/parse_commands.h" #include "scgi/thread_scgi.h" +#include "session/download_storer.h" #include "session/thread_session.h" +#include "session/session_manager.h" #include "ui/root.h" #include "utils/directory.h" @@ -112,7 +113,7 @@ initialize_rpc_slots() { void load_session_torrents() { - utils::Directory entries = control->core()->download_store()->get_formated_entries(); + utils::Directory entries = session::DownloadStorer::get_formated_entries(session_thread::manager()->path()); for (const auto& entry : entries) { // We don't really support session torrents that are links. These diff --git a/src/scgi/thread_scgi.cc b/src/scgi/thread_scgi.cc index cf01e35a..edb3aa97 100644 --- a/src/scgi/thread_scgi.cc +++ b/src/scgi/thread_scgi.cc @@ -38,11 +38,6 @@ ThreadScgi::thread_scgi() { return m_thread_scgi; } -// TODO: Remove '= 0'. -void -ThreadScgi::init_thread() { -} - void ThreadScgi::cleanup_thread() { if (m_scgi != nullptr) diff --git a/src/scgi/thread_scgi.h b/src/scgi/thread_scgi.h index 66f7a083..a7173fa2 100644 --- a/src/scgi/thread_scgi.h +++ b/src/scgi/thread_scgi.h @@ -20,10 +20,7 @@ public: static void destroy_thread(); static ThreadScgi* thread_scgi(); - const char* name() const override { return "rtorrent-scgi"; } - - void init_thread() override; - void cleanup_thread() override; + const char* name() const override { return "rtorrent-scgi"; } rpc::SCgi* scgi(); bool set_scgi(rpc::SCgi* scgi); @@ -37,6 +34,8 @@ protected: static auto internal_thread_scgi() { return m_thread_scgi; } + void cleanup_thread() override; + void call_events() override; std::chrono::microseconds next_timeout() override; @@ -46,10 +45,8 @@ private: static ThreadScgi* m_thread_scgi; - std::atomic m_scgi{nullptr}; - std::string m_rpc_log_filename; - - // std::unique_ptr m_manager; + std::atomic m_scgi{nullptr}; + std::string m_rpc_log_filename; }; } // namespace scgi diff --git a/src/session/download_storer.cc b/src/session/download_storer.cc new file mode 100644 index 00000000..e7328c46 --- /dev/null +++ b/src/session/download_storer.cc @@ -0,0 +1,112 @@ +#include "config.h" + +#include "download_storer.h" + +#include +#include +#include +#include +#include + +#include "globals.h" +#include "core/download.h" +#include "utils/directory.h" + +namespace session { + +DownloadStorer::DownloadStorer(core::Download* download) + : m_download(download) { +} + +void +DownloadStorer::build_streams(bool skip_static) { + auto* download = m_download->download(); + + auto& resume_base = download->bencode()->get_key("libtorrent_resume"); + auto& rtorrent_base = download->bencode()->get_key("rtorrent"); + + rtorrent_base.insert_key("chunks_done", download->file_list()->completed_chunks()); + rtorrent_base.insert_key("chunks_wanted", download->data()->wanted_chunks()); + rtorrent_base.insert_key("total_uploaded", m_download->info()->up_rate()->total()); + rtorrent_base.insert_key("total_downloaded", m_download->info()->down_rate()->total()); + + // Don't save for completed torrents when we've cleared the uncertain_pieces. + torrent::resume_save_progress(*download, resume_base); + torrent::resume_save_uncertain_pieces(*download, resume_base); + + torrent::resume_save_addresses(*download, resume_base); + torrent::resume_save_file_priorities(*download, resume_base); + torrent::resume_save_tracker_settings(*download, resume_base); + + // Temp fixing of all flags, move to a better place: + resume_base.set_flags(torrent::Object::flag_session_data); + rtorrent_base.set_flags(torrent::Object::flag_session_data); + + auto torrent_stream = std::unique_ptr(); + auto resume_stream = std::make_unique(); + auto rtorrent_stream = std::make_unique(); + + if (!skip_static) { + torrent_stream = std::make_unique(); + torrent::object_write_bencode(&*torrent_stream, m_download->bencode(), torrent::Object::flag_session_data); + + if (!torrent_stream->good()) + throw torrent::internal_error("DownloadStorer::build_streams() failed to write torrent stream."); + } + + torrent::object_write_bencode(&*resume_stream, &resume_base, 0); + + if (!resume_stream->good()) + throw torrent::internal_error("DownloadStorer::build_streams() failed to write resume stream."); + + torrent::object_write_bencode(&*rtorrent_stream, &rtorrent_base, 0); + + if (!rtorrent_stream->good()) + throw torrent::internal_error("DownloadStorer::build_streams() failed to write rtorrent stream."); + + m_torrent_stream = std::move(torrent_stream); + m_rtorrent_stream = std::move(rtorrent_stream); + m_libtorrent_stream = std::move(resume_stream); +} + +std::string +DownloadStorer::build_path(const std::string& session_path) { + auto info_hash = m_download->info()->info_hash(); + + if (session_path.empty()) + throw torrent::internal_error("DownloadStorer::build_path() called with empty session path."); + + if (session_path.back() != '/') + throw torrent::internal_error("DownloadStorer::build_path() session path missing trailing slash."); + + return session_path + torrent::hash_string_to_hex_str(info_hash); +} + +bool +is_correct_format(const std::string& f) { + if (f.size() != 48 || f.substr(40) != ".torrent") + return false; + + for (std::string::const_iterator itr = f.begin(); itr != f.end() - 8; ++itr) + if (!(*itr >= '0' && *itr <= '9') && + !(*itr >= 'A' && *itr <= 'F')) + return false; + + return true; +} + +utils::Directory +DownloadStorer::get_formated_entries(const std::string& session_path) { + if (session_path.empty()) + return utils::Directory(); + + utils::Directory d(session_path); + + if (!d.update(utils::Directory::update_hide_dot)) + throw torrent::storage_error("session::DownloadStorer::update() could not open session directory: " + session_path); + + d.erase(std::remove_if(d.begin(), d.end(), [](auto& entry) { return !is_correct_format(entry.s_name); }), d.end()); + return d; +} + +} // namespace session diff --git a/src/session/download_storer.h b/src/session/download_storer.h new file mode 100644 index 00000000..01d975c9 --- /dev/null +++ b/src/session/download_storer.h @@ -0,0 +1,44 @@ +#ifndef RTORRENT_SESSION_DOWNLOAD_STORER_H +#define RTORRENT_SESSION_DOWNLOAD_STORER_H + +#include +#include +#include +#include + +namespace core { +class Download; +} + +namespace utils { + class Directory; +} + +namespace session { + +class DownloadStorer { +public: + DownloadStorer(core::Download* download); + + core::Download* download() const { return m_download; } + + void build_streams(bool skip_static); + std::string build_path(const std::string& session_path); + + auto torrent_stream() { return std::move(m_torrent_stream); } + auto rtorrent_stream() { return std::move(m_rtorrent_stream); } + auto libtorrent_stream() { return std::move(m_libtorrent_stream); } + + static utils::Directory get_formated_entries(const std::string& session_path); + +private: + core::Download* m_download; + + std::unique_ptr m_torrent_stream; + std::unique_ptr m_rtorrent_stream; + std::unique_ptr m_libtorrent_stream; +}; + +} // namespace session + +#endif // RTORRENT_SESSION_DOWNLOAD_STORER_H diff --git a/src/session/session_manager.cc b/src/session/session_manager.cc index 47735591..0c9f3754 100644 --- a/src/session/session_manager.cc +++ b/src/session/session_manager.cc @@ -1,6 +1,6 @@ #include "config.h" -#include "session_manager.h" +#include "session/session_manager.h" #include #include @@ -11,6 +11,7 @@ #include #include "globals.h" +#include "session/download_storer.h" #include "utils/lockfile.h" #define LT_LOG(log_fmt, ...) \ @@ -27,18 +28,6 @@ SessionManager::SessionManager(torrent::utils::Thread* thread) SessionManager::~SessionManager() = default; -// TODO: -// * Lock session directory. -// * On shutdown, wait for all saves to finish. -// * Add is_empty_and_done() that also include async fdisksync tasks. -// * Then unlock session directory. - -bool -SessionManager::is_empty() { - std::lock_guard guard(m_mutex); - return m_save_requests.empty(); -} - void SessionManager::set_path(const std::string& path) { assert(torrent::this_thread::thread() == torrent::main_thread::thread()); @@ -52,6 +41,16 @@ SessionManager::set_path(const std::string& path) { m_path = path + '/'; } +void +SessionManager::set_use_fsyncdisk(bool use_fsyncdisk) { + assert(torrent::this_thread::thread() == torrent::main_thread::thread()); + + if (m_freeze_info) + throw torrent::input_error("Session fsyncdisk option cannot be changed after startup."); + + m_use_fsyncdisk = use_fsyncdisk; +} + void SessionManager::set_use_lock(bool use_lock) { assert(torrent::this_thread::thread() == torrent::main_thread::thread()); @@ -62,65 +61,62 @@ SessionManager::set_use_lock(bool use_lock) { m_use_lock = use_lock; } -// TODO: Derive path from download info hash. -// TODO: Generate streams here, not in download store. +// TODO: Add separate mutex for queuing save requests in session save. + void -SessionManager::save_download(core::Download* download, std::string path, stream_ptr torrent_stream, stream_ptr rtorrent_stream, stream_ptr libtorrent_stream) { +SessionManager::save_download(core::Download* download, bool skip_static) { assert(torrent::this_thread::thread() == torrent::main_thread::thread()); if (m_path.empty()) return; - { - std::lock_guard guard(m_mutex); + DownloadStorer storer(download); - LT_LOG("requesting save : download:%p path:%s", download, path.c_str()); + storer.build_streams(skip_static); + + auto save_request = SaveRequest{ + download, + storer.build_path(m_path), + storer.torrent_stream(), + storer.rtorrent_stream(), + storer.libtorrent_stream() + }; + + { + std::unique_lock lock(m_mutex); + + LT_LOG("requesting save : download:%p path:%s", download, save_request.path.c_str()); if (!m_active) throw torrent::internal_error("SessionManager::save_download() called while not active."); - // TODO: Remove is already queued entries. + if (remove_save_request_unsafe(download, lock)) + LT_LOG("replacing pending save request : download:%p", download); - // TODO: Add these to a temp structure - // TODO: When a download already exists, replace it. - - m_save_requests.push_back(SaveRequest{ - download, - std::move(path), - std::move(torrent_stream), - std::move(rtorrent_stream), - std::move(libtorrent_stream) - }); + m_save_requests.push_back(std::move(save_request)); } - session_thread::callback(nullptr, [this]() { process_save_request(); }); + session_thread::callback(this, [this]() { process_save_request(); }); } +// TODO: Move various low-level stuff to DownloadStorer. + void -SessionManager::remove_download(core::Download* download, std::string base_path) { +SessionManager::remove_download(core::Download* download) { assert(torrent::this_thread::thread() == torrent::main_thread::thread()); if (m_path.empty()) return; - std::lock_guard guard(m_mutex); + auto base_path = DownloadStorer(download).build_path(m_path); + + std::unique_lock lock(m_mutex); if (!m_active) throw torrent::internal_error("SessionManager::remove_download() called while not active."); - // TODO: Add these to a temp structure, and remove from to-be-added temp struct. - - 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()) { - LT_LOG("canceling save request : download:%p", download); - m_save_requests.erase(itr, m_save_requests.end()); - } - - // TODO: Use atomic download ptr to check if we're currently processing this download - // TODO: If so, use a lock to wait for save to finish before returning + if (remove_save_request_unsafe(download, lock)) + LT_LOG("canceled pending save request : download:%p", download); auto torrent_path = base_path; auto libtorrent_path = base_path + ".libtorrent_resume"; @@ -135,7 +131,7 @@ void SessionManager::start() { assert(torrent::this_thread::thread() == torrent::main_thread::thread()); - std::lock_guard guard(m_mutex); + std::unique_lock lock(m_mutex); if (m_active || m_freeze_info) throw torrent::internal_error("SessionManager::start() called while already started."); @@ -168,7 +164,7 @@ void SessionManager::cleanup() { assert(m_thread == torrent::this_thread::thread()); - std::lock_guard guard(m_mutex); + std::unique_lock lock(m_mutex); if (!m_active) throw torrent::internal_error("SessionManager::cleanup() called while not active."); @@ -182,12 +178,18 @@ SessionManager::cleanup() { LT_LOG("cleaning up session manager with path: %s", m_path.c_str()); + flush_all_and_wait_unsafe(lock); + if (m_use_lock) { if (!m_lockfile->unlock()) LT_LOG("could not unlock session directory: %s", m_path.c_str()); LT_LOG("unlocked session directory: %s", m_path.c_str()); } + + LT_LOG("session manager cleaned up", 0); + + session_thread::cancel_callback(this); } void @@ -197,36 +199,138 @@ SessionManager::process_save_request() { if (m_path.empty()) return; - std::lock_guard guard(m_mutex); + std::unique_lock lock(m_mutex); if (!m_active) throw torrent::internal_error("SessionManager::process_save_request() called while not active."); - if (m_save_requests.empty()) + if (m_save_requests.empty() || m_processing_saves.size() >= max_concurrent_saves) return; - // pick first request - // process it - // add us back to callbacks? we need to disable shutdown while processing all saves... do we do it at thread cleanup? + process_next_save_request_unsafe(); + if (!m_save_requests.empty()) + session_thread::callback(this, [this]() { process_save_request(); }); +} + +void +SessionManager::process_next_save_request_unsafe() { auto request = std::move(m_save_requests.front()); m_save_requests.pop_front(); - // Keep lock while processing to ensure cancellations do not interfere. + auto itr = m_processing_saves.insert(m_processing_saves.end(), ProcessingSave{}); - save_download_unsafe(request); + itr->second = std::move(request); + itr->first = std::async(std::launch::async, [this, itr]() { + // TODO: Properly handle errors here, and report back to session thread. + // TODO: Consider adding a failed_saves with error info. - if (!m_save_requests.empty()) - session_thread::callback(nullptr, [this]() { process_save_request(); }); + save_download_unsafe(itr->second); + + { + std::unique_lock lock(m_mutex); + + if (m_finished_saves.empty()) + session_thread::callback(this, [this]() { process_finished_saves(); }); + + m_finished_saves.push_back(std::move(*itr)); + m_processing_saves.erase(itr); + + m_finished_condition.notify_all(); + } + }); } -// TODO: Add threads/tasklets that calls fdisksync on shutdown. -// TODO: Parallelize saves. +void +SessionManager::process_finished_saves() { + assert(m_thread == torrent::this_thread::thread()); + + std::unique_lock lock(m_mutex); + + if (!m_active) + throw torrent::internal_error("SessionManager::process_finished_saves() called while not active."); + + for (auto& request : m_finished_saves) + LT_LOG("finished saving download : download:%p path:%s", request.second.download, request.second.path.c_str()); + + m_finished_saves.clear(); +} + +void +SessionManager::wait_for_one_save_unsafe(std::unique_lock& lock) { + assert(m_thread == torrent::this_thread::thread()); + + if (m_processing_saves.empty()) + return; + + m_finished_condition.wait(lock); +} + +void +SessionManager::flush_all_and_wait_unsafe(std::unique_lock& lock) { + LT_LOG("flushing all pending saves", 0); + + while (!m_save_requests.empty()) { + if (m_processing_saves.size() >= max_cleanup_saves) { + wait_for_one_save_unsafe(lock); + continue; + } + + process_next_save_request_unsafe(); + } + + while (!m_processing_saves.empty()) + wait_for_one_save_unsafe(lock); + + for (auto& request : m_finished_saves) + LT_LOG("finished saving download : download:%p path:%s", request.second.download, request.second.path.c_str()); + + m_finished_saves.clear(); + + LT_LOG("flushed all pending saves", 0); +} + +bool +SessionManager::remove_save_request_unsafe(core::Download* download, std::unique_lock& lock) { + // Can be run in any thread. + + { + 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()); + } + + // check if in processing saves, then wait for it to finish using wait_for_one_save_unsafe() + while (true) { + auto itr = std::find_if(m_processing_saves.begin(), m_processing_saves.end(), [download](auto& req) { + return req.second.download == download; + }); + + if (itr == m_processing_saves.end()) + break; + + wait_for_one_save_unsafe(lock); + } + + // Checking active after remove_save_request_unsafe to ensure we're calling this after shutdown. + if (!m_active) + throw torrent::internal_error("SessionManager::remove_save_request_unsafe() called while not active"); + + return true; +} // 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 nad try again. + void SessionManager::save_download_unsafe(const SaveRequest& request) { - LT_LOG("saving download : download:%p path:%s", request.download, request.path.c_str()); + // LT_LOG("saving download : download:%p path:%s", request.download, request.path.c_str()); if (m_path.empty()) throw torrent::internal_error("SessionManager::save_download_unsafe() called with empty session path."); @@ -248,18 +352,18 @@ SessionManager::save_download_unsafe(const SaveRequest& request) { if (request.torrent_stream) { if (::rename((torrent_path + ".new").c_str(), torrent_path.c_str()) == -1) { - LT_LOG("failed to rename torrent file : %s", torrent_path.c_str()); + // LT_LOG("failed to rename torrent file : %s", torrent_path.c_str()); return; } } if (::rename((libtorrent_path + ".new").c_str(), libtorrent_path.c_str()) == -1) { - LT_LOG("failed to rename libtorrent resume file : %s", libtorrent_path.c_str()); + // LT_LOG("failed to rename libtorrent resume file : %s", libtorrent_path.c_str()); return; } if (::rename((rtorrent_path + ".new").c_str(), rtorrent_path.c_str()) == -1) { - LT_LOG("failed to rename rtorrent resume file : %s", rtorrent_path.c_str()); + // LT_LOG("failed to rename rtorrent resume file : %s", rtorrent_path.c_str()); return; } } @@ -270,15 +374,17 @@ bool SessionManager::save_download_stream_unsafe(const std::string& path, const std::unique_ptr& stream) { std::fstream output(path.c_str(), std::ios::out | std::ios::trunc); + // TODO: If we cannot open more files, wait for some to finish and try again. + if (!output.is_open()) { - LT_LOG("failed to open file for writing : path:%s", path.c_str()); + // LT_LOG("failed to open file for writing : path:%s", path.c_str()); return false; } output << stream->rdbuf(); if (!output.good()) { - LT_LOG("failed to write stream to file : path:%s", path.c_str()); + // LT_LOG("failed to write stream to file : path:%s", path.c_str()); return false; } @@ -288,19 +394,11 @@ SessionManager::save_download_stream_unsafe(const std::string& path, const std:: int fd = ::open(path.c_str(), O_WRONLY); if (fd < 0) { - LT_LOG("failed to open file descriptor for fdatasync : path:%s", path.c_str()); + // LT_LOG("failed to open file descriptor for fdatasync : path:%s", path.c_str()); return false; } - // We don't care about cancelation here, as the underlying file gets deleted / replaced anyway. - - // TODO: We can use std::async for these, and only wait for them if we're shutting down. - - // TODO: Use an atomic counter / conditional variable to keep track of async operations count, and - // wait for finished operations on shutdown. - - // if (rpc::call_command_value("system.files.session.fdatasync")) { - if (true) { + if (m_use_fsyncdisk) { #ifdef __APPLE__ ::fsync(fd); #else diff --git a/src/session/session_manager.h b/src/session/session_manager.h index 5e79a400..63c0b5f8 100644 --- a/src/session/session_manager.h +++ b/src/session/session_manager.h @@ -1,11 +1,15 @@ #ifndef RTORRENT_SESSION_SESSION_MANAGER_H #define RTORRENT_SESSION_SESSION_MANAGER_H +#include #include +#include +#include #include #include #include #include +#include #include namespace core { @@ -32,36 +36,48 @@ class SessionManager { public: typedef std::unique_ptr stream_ptr; + // TODO: This should depend on max open sockets / be configurable. + constexpr static int max_concurrent_saves = 16; + constexpr static int max_cleanup_saves = 64; + SessionManager(torrent::utils::Thread* thread); ~SessionManager(); bool is_used() const; - // TODO: Replace with protected `bool shutdown_if_done()`. - bool is_empty(); - // bool is_empty_and_done(); - - void start(); - std::string path() const; void set_path(const std::string& path); + bool use_fsyncdisk() const; + void set_use_fsyncdisk(bool use_fsyncdisk); + bool use_lock() const; void set_use_lock(bool use_lock); - void freeze_info(); - - void save_download(core::Download* download, std::string path, stream_ptr torrent_stream, stream_ptr rtorrent_stream, stream_ptr libtorrent_stream); - void remove_download(core::Download* download, std::string path); + void save_full_download(core::Download* download) { save_download(download, false); } + void save_resume_download(core::Download* download) { save_download(download, true); } + void remove_download(core::Download* download); protected: friend class Control; friend class ThreadSession; + void save_download(core::Download* download, bool skip_static); + + void start(); void cleanup(); private: void process_save_request(); + void process_next_save_request_unsafe(); + void process_finished_saves(); + + void wait_for_one_save_unsafe(std::unique_lock& lock); + + // Requires a higher number of open sockets, and should only be used during shutdown. + void flush_all_and_wait_unsafe(std::unique_lock& lock); + + bool remove_save_request_unsafe(core::Download* download, std::unique_lock& lock); void save_download_unsafe(const SaveRequest& request); bool save_download_stream_unsafe(const std::string& path, const std::unique_ptr& stream); @@ -70,18 +86,28 @@ private: bool m_freeze_info{}; std::string m_path; + bool m_use_fsyncdisk{true}; bool m_use_lock{true}; std::mutex m_mutex; bool m_active{}; - std::deque m_save_requests; + typedef std::pair, SaveRequest> ProcessingSave; + + // TODO: Add deque of DownloadStorers requests that have yet to build streams. + + std::deque m_save_requests; + std::list m_processing_saves; + std::condition_variable m_finished_condition; + std::vector m_finished_saves; + std::unique_ptr m_lockfile; }; -inline bool SessionManager::is_used() const { return !m_path.empty(); } -inline std::string SessionManager::path() const { return m_path; } -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; } } // namespace session diff --git a/src/session/thread_session.cc b/src/session/thread_session.cc index 5215e4e2..3937151d 100644 --- a/src/session/thread_session.cc +++ b/src/session/thread_session.cc @@ -36,9 +36,9 @@ ThreadSession::thread_session() { return m_thread_session; } -// TODO: Remove '= 0'. void -ThreadSession::init_thread() { +ThreadSession::init_thread_pre_start() { + m_manager->start(); } // TODO: Make sure we trigger session save before main thread exits, that it adds all required @@ -52,18 +52,9 @@ void ThreadSession::call_events() { // lt_log_print_locked(torrent::LOG_THREAD_NOTICE, "Got thread_disk tick."); - // TODO: Wait with shutdown until all session data is saved. - process_callbacks(); if ((m_flags & flag_do_shutdown)) { - if (!m_manager->is_empty()) { - // TODO: Figure out a better way to wait for session save to complete. - // TODO: Sanity check to avoid getting stuck not shutting down. - // TODO: Should we depend on next_timeout() instead of callbacks? - return; - } - if ((m_flags & flag_did_shutdown)) throw torrent::internal_error("Already trigged shutdown."); diff --git a/src/session/thread_session.h b/src/session/thread_session.h index 00ed59b2..173d737c 100644 --- a/src/session/thread_session.h +++ b/src/session/thread_session.h @@ -15,12 +15,9 @@ public: static void destroy_thread(); static ThreadSession* thread_session(); - const char* name() const override { return "rtorrent-session"; } + const char* name() const override { return "rtorrent-session"; } - void init_thread() override; - void cleanup_thread() override; - - SessionManager* manager() const { return m_manager.get(); } + SessionManager* manager() const { return m_manager.get(); } protected: friend class ThreadSessionInternal; @@ -29,6 +26,9 @@ protected: static auto internal_thread_session() { return m_thread_session; } + void init_thread_pre_start() override; + void cleanup_thread() override; + void call_events() override; std::chrono::microseconds next_timeout() override; diff --git a/src/ui/root.cc b/src/ui/root.cc index cc05194d..05f8eb9b 100644 --- a/src/ui/root.cc +++ b/src/ui/root.cc @@ -13,7 +13,6 @@ #include "control.h" #include "core/download_list.h" -#include "core/download_store.h" #include "core/manager.h" #include "display/frame.h" #include "display/window_http_queue.h"