From 8f644e65dd538c1c3e2f68310c44947d5692d043 Mon Sep 17 00:00:00 2001 From: Jari Sundell Date: Wed, 17 Dec 2025 22:42:44 +0100 Subject: [PATCH] Use separate thread for saving session data. --- src/Makefile.am | 5 + src/command_download.cc | 2 +- src/command_local.cc | 21 +-- src/command_logging.cc | 10 +- src/command_network.cc | 8 +- src/command_tracker.cc | 2 +- src/control.cc | 9 +- src/core/dht_manager.cc | 16 +- src/core/download_store.cc | 138 ++++---------- src/core/download_store.h | 16 +- src/globals.h | 28 ++- src/main.cc | 12 +- src/rpc/scgi.cc | 1 + src/session/session_manager.cc | 316 +++++++++++++++++++++++++++++++++ src/session/session_manager.h | 88 +++++++++ src/session/thread_session.cc | 95 ++++++++++ src/session/thread_session.h | 43 +++++ src/thread_worker.h | 2 +- src/ui/root.cc | 28 +-- src/ui/root.h | 3 +- src/utils/lockfile.h | 34 ---- 21 files changed, 668 insertions(+), 209 deletions(-) create mode 100644 src/session/session_manager.cc create mode 100644 src/session/session_manager.h create mode 100644 src/session/thread_session.cc create mode 100644 src/session/thread_session.h diff --git a/src/Makefile.am b/src/Makefile.am index 01bca664..7552ac9d 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -125,6 +125,11 @@ libsub_root_a_SOURCES = \ rpc/tinyxml2/tinyxml2.cc \ rpc/nlohmann/json.h \ \ + session/session_manager.cc \ + session/session_manager.h \ + session/thread_session.cc \ + session/thread_session.h \ + \ ui/download.cc \ ui/download.h \ ui/download_list.cc \ diff --git a/src/command_download.cc b/src/command_download.cc index 4fc24243..f3768c2d 100644 --- a/src/command_download.cc +++ b/src/command_download.cc @@ -302,7 +302,7 @@ apply_d_add_peer(core::Download* download, const std::string& arg) { if (port < 1 || port > 65535) throw torrent::input_error("Invalid port number."); - assert(std::this_thread::get_id() == torrent::main_thread::thread()->thread_id()); + assert(std::this_thread::get_id() == torrent::main_thread::thread_id()); // Currently discarding SOCK_STREAM. torrent::this_thread::resolver()->resolve_preferred(NULL, host, AF_UNSPEC, AF_INET, [download, port](torrent::c_sa_shared_ptr sa, int err) { diff --git a/src/command_local.cc b/src/command_local.cc index 1968197c..3bc24634 100644 --- a/src/command_local.cc +++ b/src/command_local.cc @@ -22,6 +22,7 @@ #include "rak/string_manip.h" #include "rpc/parse_commands.h" #include "rpc/scgi.h" +#include "session/session_manager.h" #include "utils/file_status_cache.h" #include "globals.h" @@ -270,22 +271,22 @@ initialize_command_local() { CMD2_ANY ("pieces.hash.queue_size", std::bind(&torrent::main_thread::hash_queue_size)); CMD2_VAR_BOOL ("pieces.hash.on_completion", true); - CMD2_VAR_STRING ("directory.default", "./"); + CMD2_VAR_STRING ("directory.default", "./"); - CMD2_VAR_STRING ("session.name", ""); - CMD2_VAR_BOOL ("session.use_lock", true); - CMD2_VAR_BOOL ("session.on_completion", true); + CMD2_VAR_STRING ("session.name", ""); + CMD2_ANY ("session.path", [](auto, auto) { return session_thread::manager()->path(); }); + CMD2_ANY_STRING_V("session.path.set", [](auto, auto& str) { return session_thread::manager()->set_path(str); }); + CMD2_ANY ("session.use_lock", [](auto, auto) { return session_thread::manager()->use_lock(); }); + CMD2_ANY_VALUE_V ("session.use_lock.set", [](auto, auto& value) { return session_thread::manager()->set_use_lock(value); }); + CMD2_VAR_BOOL ("session.on_completion", true); - CMD2_ANY ("session.path", std::bind(&core::DownloadStore::path, dStore)); - CMD2_ANY_STRING_V("session.path.set", std::bind(&core::DownloadStore::set_path, dStore, std::placeholders::_2)); - - CMD2_ANY_V ("session.save", std::bind(&core::DownloadList::session_save, dList)); + CMD2_ANY_V ("session.save", [dList](auto, auto) { return dList->session_save(); }); #ifdef HAVE_LUA rpc::LuaEngine* lua_engine = control->lua_engine(); - CMD2_ANY ("lua.execute", std::bind(&rpc::execute_lua, lua_engine, std::placeholders::_1, std::placeholders::_2, 0)); - CMD2_ANY ("lua.execute.str", std::bind(&rpc::execute_lua, lua_engine, std::placeholders::_1, std::placeholders::_2, rpc::LuaEngine::flag_string)); + CMD2_ANY ("lua.execute", std::bind(&rpc::execute_lua, lua_engine, std::placeholders::_1, std::placeholders::_2, 0)); + CMD2_ANY ("lua.execute.str", std::bind(&rpc::execute_lua, lua_engine, std::placeholders::_1, std::placeholders::_2, rpc::LuaEngine::flag_string)); #endif #define CMD2_EXECUTE(key, flags) \ diff --git a/src/command_logging.cc b/src/command_logging.cc index 9f31ad77..0e22e30d 100644 --- a/src/command_logging.cc +++ b/src/command_logging.cc @@ -8,16 +8,16 @@ #include #include +#include "globals.h" +#include "control.h" +#include "command_helpers.h" +#include "thread_worker.h" #include "core/download.h" #include "core/download_list.h" #include "core/manager.h" #include "rak/path.h" #include "rpc/parse_commands.h" -#include "globals.h" -#include "control.h" -#include "command_helpers.h" - static const int log_flag_use_gz = 0x1; static const int log_flag_append_pid = 0x2; static const int log_flag_append_file = 0x4; @@ -62,7 +62,7 @@ torrent::Object apply_log_add_output(const torrent::Object::list_type& args) { if (args.size() != 2) throw torrent::input_error("Invalid number of arguments."); - + log_add_group_output_str(args.front().as_string().c_str(), args.back().as_string().c_str()); diff --git a/src/command_network.cc b/src/command_network.cc index f977282c..b7ae6126 100644 --- a/src/command_network.cc +++ b/src/command_network.cc @@ -17,6 +17,10 @@ #include #include +#include "globals.h" +#include "control.h" +#include "command_helpers.h" +#include "thread_worker.h" #include "core/download.h" #include "core/manager.h" #include "rpc/scgi.h" @@ -24,10 +28,6 @@ #include "rpc/parse.h" #include "rpc/parse_commands.h" -#include "globals.h" -#include "control.h" -#include "command_helpers.h" - torrent::Object apply_encryption(const torrent::Object::list_type& args) { uint32_t options_mask = torrent::net::NetworkConfig::encryption_none; diff --git a/src/command_tracker.cc b/src/command_tracker.cc index a25771cd..86a870ec 100644 --- a/src/command_tracker.cc +++ b/src/command_tracker.cc @@ -45,7 +45,7 @@ apply_dht_add_node(const std::string& arg) { if (port < 1 || port > 65535) throw torrent::input_error("Invalid port number."); - assert(std::this_thread::get_id() == torrent::main_thread::thread()->thread_id()); + assert(std::this_thread::get_id() == torrent::main_thread::thread_id()); auto host_str = std::string(host); diff --git a/src/control.cc b/src/control.cc index fbfd057e..d33b1f28 100644 --- a/src/control.cc +++ b/src/control.cc @@ -8,6 +8,7 @@ #include #include +#include "thread_worker.h" #include "core/dht_manager.h" #include "core/download_store.h" #include "core/http_queue.h" @@ -26,6 +27,7 @@ #include "rpc/lua.h" #include "rpc/parse_commands.h" #include "rpc/object_storage.h" +#include "session/session_manager.h" #include "ui/root.h" Control::Control() @@ -58,6 +60,9 @@ Control::~Control() { void Control::initialize() { + session_thread::manager()->start(); + session_thread::thread()->start_thread(); + worker_thread->start_thread(); display::Canvas::initialize(); @@ -68,7 +73,6 @@ Control::initialize() { torrent::net_thread::http_stack()->set_user_agent(USER_AGENT); m_core->listen_open(); - m_core->download_store()->enable(rpc::call_command_value("session.use_lock")); m_core->set_hashing_view(*m_view_manager->find_throw("hashing")); m_ui->init(this); @@ -86,7 +90,8 @@ Control::cleanup() { if(!display::Canvas::daemon()) m_inputStdin->remove(torrent::this_thread::poll()); - m_core->download_store()->disable(); + // Wait for all session files to be written. + session_thread::thread()->stop_thread_wait(); m_ui->cleanup(); m_core->cleanup(); diff --git a/src/core/dht_manager.cc b/src/core/dht_manager.cc index 8448e47c..3ab551b4 100644 --- a/src/core/dht_manager.cc +++ b/src/core/dht_manager.cc @@ -17,6 +17,7 @@ #include "globals.h" #include "manager.h" #include "rpc/parse_commands.h" +#include "session/session_manager.h" #define LT_LOG(log_fmt, ...) \ lt_log_print_subsystem(torrent::LOG_DHT_CONTROLLER, "dht_manager", log_fmt, __VA_ARGS__); @@ -35,13 +36,13 @@ DhtManager::~DhtManager() { void DhtManager::load_dht_cache() { - if (m_start == dht_disable || !control->core()->download_store()->is_enabled()) { + if (m_start == dht_disable || !session_thread::manager()->is_used()) { LT_LOG("ignoring cache file", 0); return; } - std::string cache_filename = control->core()->download_store()->path() + "rtorrent.dht_cache"; - std::fstream cache_stream(cache_filename.c_str(), std::ios::in | std::ios::binary); + auto cache_filename = session_thread::manager()->path() + "rtorrent.dht_cache"; + auto cache_stream = std::fstream(cache_filename.c_str(), std::ios::in | std::ios::binary); torrent::Object cache = torrent::Object::create_map(); @@ -115,14 +116,15 @@ DhtManager::stop_dht() { void DhtManager::save_dht_cache() { - if (!control->core()->download_store()->is_enabled()) + if (!session_thread::manager()->is_used()) return; + if (!torrent::runtime::network_manager()->is_dht_valid()) return; - std::string filename = control->core()->download_store()->path() + "rtorrent.dht_cache"; - std::string filename_tmp = filename + ".new"; - std::fstream cache_file(filename_tmp.c_str(), std::ios::out | std::ios::trunc); + auto filename = session_thread::manager()->path() + "rtorrent.dht_cache"; + auto filename_tmp = filename + ".new"; + auto cache_file = std::fstream(filename_tmp.c_str(), std::ios::out | std::ios::trunc); if (!cache_file.is_open()) return; diff --git a/src/core/download_store.cc b/src/core/download_store.cc index 5f857a24..d368314b 100644 --- a/src/core/download_store.cc +++ b/src/core/download_store.cc @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -16,104 +17,17 @@ #include #include -#include "utils/directory.h" - #include "download.h" #include "download_store.h" #include "rpc/parse_commands.h" +#include "session/session_manager.h" +#include "utils/directory.h" namespace core { -void -DownloadStore::enable(bool lock) { - if (is_enabled()) - throw torrent::input_error("Session directory already enabled."); - - if (m_path.empty()) - return; - - if (lock) - m_lockfile.set_path(m_path + "rtorrent.lock"); - else - m_lockfile.set_path(std::string()); - - if (!m_lockfile.try_lock()) { - if (rak::error_number::current().is_bad_path()) - throw torrent::input_error("Could not lock session directory: \"" + m_path + "\", " + rak::error_number::current().c_str()); - else - throw torrent::input_error("Could not lock session directory: \"" + m_path + "\", held by \"" + m_lockfile.locked_by_as_string() + "\"."); - } -} - -void -DownloadStore::disable() { - if (!is_enabled()) - return; - - m_lockfile.unlock(); -} - -void -DownloadStore::set_path(const std::string& path) { - if (is_enabled()) - throw torrent::input_error("Tried to change session directory while it is enabled."); - - if (!path.empty() && *path.rbegin() != '/') - m_path = rak::path_expand(path + '/'); - else - m_path = rak::path_expand(path); -} - -bool -DownloadStore::write_bencode(const std::string& filename, const torrent::Object& obj, uint32_t skip_mask) { - int fd; - torrent::Object tmp; - std::fstream output(filename.c_str(), std::ios::out | std::ios::trunc); - - if (!output.is_open()) - goto download_store_save_error; - - torrent::object_write_bencode(&output, &obj, skip_mask); - - if (!output.good()) - goto download_store_save_error; - - output.close(); - - // Test the new file, to ensure it is a valid bencode string. - output.open(filename.c_str(), std::ios::in); - output >> tmp; - - if (!output.good()) - goto download_store_save_error; - - output.close(); - - // Ensure that the new file is actually written to the disk - fd = ::open(filename.c_str(), O_WRONLY); - if (fd < 0) - goto download_store_save_error; - - if (rpc::call_command_value("system.files.session.fdatasync")) { -#ifdef __APPLE__ - fsync(fd); -#else - fdatasync(fd); -#endif - } - - ::close(fd); - - return true; - -download_store_save_error: - output.close(); - return false; -} - bool DownloadStore::save(Download* d, int flags) { - if (!is_enabled()) + if (!session_thread::manager()->is_used()) return true; torrent::Object* resume_base = &d->download()->bencode()->get_key("libtorrent_resume"); @@ -137,30 +51,38 @@ DownloadStore::save(Download* d, int flags) { resume_base->set_flags(torrent::Object::flag_session_data); rtorrent_base->set_flags(torrent::Object::flag_session_data); - std::string base_filename = create_filename(d); + auto download_stream = std::unique_ptr(); + auto resume_stream = std::make_unique(); + auto rtorrent_stream = std::make_unique(); - if (!write_bencode(base_filename + ".libtorrent_resume.new", *resume_base, 0) || - !write_bencode(base_filename + ".rtorrent.new", *rtorrent_base, 0)) + 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; - ::rename((base_filename + ".libtorrent_resume.new").c_str(), (base_filename + ".libtorrent_resume").c_str()); - ::rename((base_filename + ".rtorrent.new").c_str(), (base_filename + ".rtorrent").c_str()); + torrent::object_write_bencode(&*rtorrent_stream, rtorrent_base, 0); - if (!(flags & flag_skip_static) && - write_bencode(base_filename + ".new", *d->bencode(), torrent::Object::flag_session_data)) - ::rename((base_filename + ".new").c_str(), base_filename.c_str()); + 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) { - if (!is_enabled()) - return; - - ::unlink((create_filename(d) + ".libtorrent_resume").c_str()); - ::unlink((create_filename(d) + ".rtorrent").c_str()); - ::unlink(create_filename(d).c_str()); + session_thread::manager()->remove_download(d, create_filename(d)); } // This also needs to check that it isn't a directory. @@ -171,13 +93,13 @@ not_correct_format(const utils::directory_entry& entry) { utils::Directory DownloadStore::get_formated_entries() { - if (!is_enabled()) + if (!session_thread::manager()->is_used()) return utils::Directory(); - utils::Directory d(m_path); + 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 directory \"" + m_path + "\""); + 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()); @@ -199,7 +121,7 @@ DownloadStore::is_correct_format(const std::string& f) { std::string DownloadStore::create_filename(Download* d) { - return m_path + rak::transform_hex(d->info()->hash().begin(), d->info()->hash().end()) + ".torrent"; + 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 index 2acdd406..19e6eb41 100644 --- a/src/core/download_store.h +++ b/src/core/download_store.h @@ -2,8 +2,7 @@ #define RTORRENT_CORE_DOWNLOAD_STORE_H #include - -#include "utils/lockfile.h" +#include namespace utils { class Directory; @@ -17,14 +16,6 @@ class DownloadStore { public: static const int flag_skip_static = 0x1; - bool is_enabled() { return m_lockfile.is_locked(); } - - void enable(bool lock); - void disable(); - - const std::string& path() const { return m_path; } - void set_path(const std::string& path); - 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); } @@ -37,11 +28,6 @@ public: private: std::string create_filename(Download* d); - - bool write_bencode(const std::string& filename, const torrent::Object& obj, uint32_t skip_mask); - - std::string m_path; - utils::Lockfile m_lockfile; }; } diff --git a/src/globals.h b/src/globals.h index 60ac8519..eda79244 100644 --- a/src/globals.h +++ b/src/globals.h @@ -1,14 +1,38 @@ #ifndef TORRENT_GLOBALS_H #define TORRENT_GLOBALS_H -#include "thread_worker.h" +#include + #include "rpc/ip_table_list.h" class Control; +class ThreadWorker; -extern rpc::ip_table_list ip_tables; +extern rpc::ip_table_list ip_tables; extern Control* control; + +// TODO: Update to new thread model. extern ThreadWorker* worker_thread; +namespace session { + +class SessionManager; + +} // namespace session + + +namespace session_thread { + +torrent::utils::Thread* thread(); +std::thread::id thread_id(); + +void callback(void* target, std::function&& fn); +void cancel_callback(void* target); +void cancel_callback_and_wait(void* target); + +session::SessionManager* manager(); + +} // namespace torrent::session_thread + #endif diff --git a/src/main.cc b/src/main.cc index 961f6d3e..8a98b22f 100644 --- a/src/main.cc +++ b/src/main.cc @@ -28,11 +28,11 @@ #include "display/window.h" #include "display/manager.h" #include "input/bindings.h" -#include "ui/root.h" - #include "rpc/command_scheduler.h" #include "rpc/command_scheduler_item.h" #include "rpc/parse_commands.h" +#include "session/thread_session.h" +#include "ui/root.h" #include "utils/directory.h" #include "control.h" @@ -215,6 +215,8 @@ main(int argc, char** argv) { torrent::initialize(); torrent::set_main_thread_slots(std::bind(&client_perform)); + session::ThreadSession::create_thread(); + // TODO: Move to controller. worker_thread = new ThreadWorker(); worker_thread->init_thread(); @@ -530,14 +532,16 @@ main(int argc, char** argv) { return -1; } - torrent::log_cleanup(); - delete control; control = nullptr; + session::ThreadSession::destroy_thread(); + delete worker_thread; worker_thread = nullptr; + torrent::log_cleanup(); + return 0; } diff --git a/src/rpc/scgi.cc b/src/rpc/scgi.cc index 97e48041..271b18c3 100644 --- a/src/rpc/scgi.cc +++ b/src/rpc/scgi.cc @@ -11,6 +11,7 @@ #include "control.h" #include "globals.h" +#include "thread_worker.h" #include "rpc/scgi_task.h" #include "utils/socket_fd.h" diff --git a/src/session/session_manager.cc b/src/session/session_manager.cc new file mode 100644 index 00000000..47735591 --- /dev/null +++ b/src/session/session_manager.cc @@ -0,0 +1,316 @@ +#include "config.h" + +#include "session_manager.h" + +#include +#include +#include +#include +#include +#include +#include + +#include "globals.h" +#include "utils/lockfile.h" + +#define LT_LOG(log_fmt, ...) \ + lt_log_print(torrent::LOG_SESSION_EVENTS, "session-events: " log_fmt, __VA_ARGS__); + +namespace session { + +// TODO: Add session save scheduler that runs in main thread and passes download one-by-one to session manager. + +SessionManager::SessionManager(torrent::utils::Thread* thread) + : m_thread(thread), + m_lockfile(std::make_unique()) { +} + +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()); + + if (m_freeze_info) + throw torrent::input_error("Session path cannot be changed after startup."); + + if (path.empty() || path.back() == '/') + m_path = path; + else + m_path = path + '/'; +} + +void +SessionManager::set_use_lock(bool use_lock) { + assert(torrent::this_thread::thread() == torrent::main_thread::thread()); + + if (m_freeze_info) + throw torrent::input_error("Session lock option cannot be changed after startup."); + + m_use_lock = use_lock; +} + +// TODO: Derive path from download info hash. +// TODO: Generate streams here, not in download store. +void +SessionManager::save_download(core::Download* download, std::string path, stream_ptr torrent_stream, stream_ptr rtorrent_stream, stream_ptr libtorrent_stream) { + assert(torrent::this_thread::thread() == torrent::main_thread::thread()); + + if (m_path.empty()) + return; + + { + std::lock_guard guard(m_mutex); + + LT_LOG("requesting save : download:%p path:%s", download, path.c_str()); + + if (!m_active) + throw torrent::internal_error("SessionManager::save_download() called while not active."); + + // TODO: Remove is already queued entries. + + // 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) + }); + } + + session_thread::callback(nullptr, [this]() { process_save_request(); }); +} + +void +SessionManager::remove_download(core::Download* download, std::string base_path) { + assert(torrent::this_thread::thread() == torrent::main_thread::thread()); + + if (m_path.empty()) + return; + + std::lock_guard guard(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 + + auto torrent_path = base_path; + auto libtorrent_path = base_path + ".libtorrent_resume"; + auto rtorrent_path = base_path + ".rtorrent"; + + ::unlink(libtorrent_path.c_str()); + ::unlink(rtorrent_path.c_str()); + ::unlink(torrent_path.c_str()); +} + +void +SessionManager::start() { + assert(torrent::this_thread::thread() == torrent::main_thread::thread()); + + std::lock_guard guard(m_mutex); + + if (m_active || m_freeze_info) + throw torrent::internal_error("SessionManager::start() called while already started."); + + m_active = true; + m_freeze_info = true; + + if (m_path.empty()) { + LT_LOG("session manager started with empty path, disabling session management", 0); + return; + } + + LT_LOG("starting session manager with path: %s", m_path.c_str()); + + if (m_use_lock) { + m_lockfile->set_path(m_path + "rtorrent.lock"); + + if (!m_lockfile->try_lock()) { + if (errno == ENOENT || errno == ENOTDIR || errno == EACCES) + throw torrent::input_error("Could not lock session directory: " + std::string(std::strerror(errno)) + " : " + m_path); + else + throw torrent::input_error("Could not lock session directory, held by: " + m_lockfile->locked_by_as_string() + " : " + m_path); + } + + LT_LOG("locked session directory: %s", m_path.c_str()); + } +} + +void +SessionManager::cleanup() { + assert(m_thread == torrent::this_thread::thread()); + + std::lock_guard guard(m_mutex); + + if (!m_active) + throw torrent::internal_error("SessionManager::cleanup() called while not active."); + + m_active = false; + + if (m_path.empty()) { + LT_LOG("session manager cleanup called with empty path, skipping", 0); + return; + } + + LT_LOG("cleaning up session manager with path: %s", m_path.c_str()); + + 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()); + } +} + +void +SessionManager::process_save_request() { + assert(m_thread == torrent::this_thread::thread()); + + if (m_path.empty()) + return; + + std::lock_guard guard(m_mutex); + + if (!m_active) + throw torrent::internal_error("SessionManager::process_save_request() called while not active."); + + if (m_save_requests.empty()) + 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? + + auto request = std::move(m_save_requests.front()); + m_save_requests.pop_front(); + + // Keep lock while processing to ensure cancellations do not interfere. + + save_download_unsafe(request); + + if (!m_save_requests.empty()) + session_thread::callback(nullptr, [this]() { process_save_request(); }); +} + +// TODO: Add threads/tasklets that calls fdisksync on shutdown. +// TODO: Parallelize saves. + +// TODO: Properly handle errors. +void +SessionManager::save_download_unsafe(const SaveRequest& request) { + 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."); + + auto torrent_path = request.path; + auto libtorrent_path = request.path + ".libtorrent_resume"; + auto rtorrent_path = request.path + ".rtorrent"; + + if (request.torrent_stream) { + if (!save_download_stream_unsafe(torrent_path + ".new", request.torrent_stream)) + return; + } + + if (!save_download_stream_unsafe(libtorrent_path + ".new", request.libtorrent_stream)) + return; + + if (!save_download_stream_unsafe(rtorrent_path + ".new", request.rtorrent_stream)) + return; + + 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()); + 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()); + 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()); + return; + } +} + +// TODO: Rewrite to be all done in std::async, and from rdbuf directly to fd to avoid re-opening. + +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); + + if (!output.is_open()) { + 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()); + return false; + } + + output.close(); + + // Ensure that the new file is actually written to the disk + 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()); + 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) { +#ifdef __APPLE__ + ::fsync(fd); +#else + ::fdatasync(fd); +#endif + } + + ::close(fd); + return true; +} + +} // namespace session + diff --git a/src/session/session_manager.h b/src/session/session_manager.h new file mode 100644 index 00000000..5e79a400 --- /dev/null +++ b/src/session/session_manager.h @@ -0,0 +1,88 @@ +#ifndef RTORRENT_SESSION_SESSION_MANAGER_H +#define RTORRENT_SESSION_SESSION_MANAGER_H + +#include +#include +#include +#include +#include +#include + +namespace core { +class Download; +} + +namespace utils { +class Lockfile; +} + +namespace session { + +class ThreadSession; + +struct SaveRequest { + core::Download* download; + std::string path; + std::unique_ptr torrent_stream; + std::unique_ptr rtorrent_stream; + std::unique_ptr libtorrent_stream; +}; + +class SessionManager { +public: + typedef std::unique_ptr stream_ptr; + + 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_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); + +protected: + friend class Control; + friend class ThreadSession; + + void cleanup(); + +private: + void process_save_request(); + + void save_download_unsafe(const SaveRequest& request); + bool save_download_stream_unsafe(const std::string& path, const std::unique_ptr& stream); + + torrent::utils::Thread* m_thread; + + bool m_freeze_info{}; + std::string m_path; + bool m_use_lock{true}; + + std::mutex m_mutex; + bool m_active{}; + + std::deque m_save_requests; + 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; } + +} // namespace session + +#endif // RTORRENT_SESSION_SESSION_MANAGER_H diff --git a/src/session/thread_session.cc b/src/session/thread_session.cc new file mode 100644 index 00000000..a7dee96d --- /dev/null +++ b/src/session/thread_session.cc @@ -0,0 +1,95 @@ +#include "config.h" + +#include "thread_session.h" + +#include + +#include "session/session_manager.h" + +namespace session { + +class ThreadSessionInternal { +public: + static ThreadSession* thread_session() { return ThreadSession::internal_thread_session(); } +}; + +ThreadSession* ThreadSession::m_thread_session{nullptr}; + +void +ThreadSession::create_thread() { + auto thread = new ThreadSession; + + thread->m_manager = std::make_unique(thread); + + m_thread_session = thread; + m_thread_session->m_state = STATE_INITIALIZED; +} + +void +ThreadSession::destroy_thread() { + delete m_thread_session; + m_thread_session = nullptr; +} + +ThreadSession* +ThreadSession::thread_session() { + return m_thread_session; +} + +// TODO: Remove '= 0'. +void +ThreadSession::init_thread() { +} + +// TODO: Make sure we trigger session save before main thread exits, that it adds all required +// downloads to the queue. +void +ThreadSession::cleanup_thread() { + m_manager->cleanup(); +} + +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."); + + m_flags |= flag_did_shutdown; + throw torrent::shutdown_exception(); + } +} + +std::chrono::microseconds +ThreadSession::next_timeout() { + // TODO: This leads to kqueue crash? + // return std::chrono::microseconds(1h); + return std::chrono::microseconds(10s); +} + +} // namespace session + +namespace session_thread { + +torrent::utils::Thread* thread() { return session::ThreadSessionInternal::thread_session(); } +std::thread::id thread_id() { return session::ThreadSessionInternal::thread_session()->thread_id(); } + +void callback(void* target, std::function&& fn) { session::ThreadSessionInternal::thread_session()->callback(target, std::move(fn)); } +void cancel_callback(void* target) { session::ThreadSessionInternal::thread_session()->cancel_callback(target); } +void cancel_callback_and_wait(void* target) { session::ThreadSessionInternal::thread_session()->cancel_callback_and_wait(target); } + +session::SessionManager* manager() { return session::ThreadSessionInternal::thread_session()->manager(); } + +} // namespace session_thread diff --git a/src/session/thread_session.h b/src/session/thread_session.h new file mode 100644 index 00000000..00ed59b2 --- /dev/null +++ b/src/session/thread_session.h @@ -0,0 +1,43 @@ +#ifndef RTORRENT_SESSION_THREAD_SESSION_H +#define RTORRENT_SESSION_THREAD_SESSION_H + +#include + +namespace session { + +class SessionManager; +class ThreadSessionInternal; + +class ThreadSession : public torrent::utils::Thread { +public: + + static void create_thread(); + static void destroy_thread(); + static ThreadSession* thread_session(); + + const char* name() const override { return "rtorrent-session"; } + + void init_thread() override; + void cleanup_thread() override; + + SessionManager* manager() const { return m_manager.get(); } + +protected: + friend class ThreadSessionInternal; + + ThreadSession() = default; + + static auto internal_thread_session() { return m_thread_session; } + + void call_events() override; + std::chrono::microseconds next_timeout() override; + +private: + static ThreadSession* m_thread_session; + + std::unique_ptr m_manager; +}; + +} // namespace session + +#endif // RTORRENT_SESSION_THREAD_SESSION_H diff --git a/src/thread_worker.h b/src/thread_worker.h index 0eb3caf9..736fc060 100644 --- a/src/thread_worker.h +++ b/src/thread_worker.h @@ -35,4 +35,4 @@ private: std::string m_rpc_log_filename; }; -#endif +#endif // RTORRENT_THREAD_WORKER_H diff --git a/src/ui/root.cc b/src/ui/root.cc index 1fdbdb6d..cc05194d 100644 --- a/src/ui/root.cc +++ b/src/ui/root.cc @@ -1,5 +1,7 @@ #include "config.h" +#include "ui/root.h" + #include #include #include @@ -9,6 +11,9 @@ #include #include +#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" @@ -18,12 +23,7 @@ #include "input/manager.h" #include "input/text_input.h" #include "rpc/parse_commands.h" - -#include "control.h" -#include "download_list.h" -#include "core/download_store.h" - -#include "root.h" +#include "session/session_manager.h" namespace ui { @@ -378,13 +378,13 @@ Root::set_input_history_size(int size) { void Root::load_input_history() { - if (m_control == nullptr || !m_control->core()->download_store()->is_enabled()) { + if (m_control == nullptr || !session_thread::manager()->is_used()) { lt_log_print(torrent::LOG_DEBUG, "ignoring input history file"); return; } - std::string history_filename = m_control->core()->download_store()->path() + "rtorrent.input_history"; - std::fstream history_file(history_filename.c_str(), std::ios::in); + auto history_filename = session_thread::manager()->path() + "rtorrent.input_history"; + auto history_file = std::fstream(history_filename.c_str(), std::ios::in); if (history_file.is_open()) { // Create a temp object of the content since size of history categories can be smaller than this. @@ -446,12 +446,12 @@ Root::load_input_history() { void Root::save_input_history() { - if (m_control == nullptr || !m_control->core()->download_store()->is_enabled()) + if (m_control == nullptr || !session_thread::manager()->is_used()) return; - std::string history_filename = m_control->core()->download_store()->path() + "rtorrent.input_history"; - std::string history_filename_tmp = history_filename + ".new"; - std::fstream history_file(history_filename_tmp.c_str(), std::ios::out | std::ios::trunc); + auto history_filename = session_thread::manager()->path() + "rtorrent.input_history"; + auto history_filename_tmp = history_filename + ".new"; + auto history_file = std::fstream(history_filename_tmp.c_str(), std::ios::out | std::ios::trunc); if (!history_file.is_open()) { lt_log_print(torrent::LOG_DEBUG, "could not open input history file for writing (path:%s)", history_filename.c_str()); @@ -503,7 +503,7 @@ Root::set_keymap_style(const std::string& style) { m_keymap_style = style; } -const int +int Root::navigation_key(NavigationKeymap key) { return m_keymap[key]; } diff --git a/src/ui/root.h b/src/ui/root.h index 714c432f..6f4d5717 100644 --- a/src/ui/root.h +++ b/src/ui/root.h @@ -98,7 +98,8 @@ public: const std::string& keymap_style() { return m_keymap_style; } void set_keymap_style(const std::string& style); - const int navigation_key(NavigationKeymap key); + + int navigation_key(NavigationKeymap key); private: void setup_keys(); diff --git a/src/utils/lockfile.h b/src/utils/lockfile.h index fb17abc7..6754cb9b 100644 --- a/src/utils/lockfile.h +++ b/src/utils/lockfile.h @@ -1,37 +1,3 @@ -// rTorrent - BitTorrent client -// Copyright (C) 2005-2011, Jari Sundell -// -// This program is free software; you can redistribute it and/or modify -// it under the terms of the GNU General Public License as published by -// the Free Software Foundation; either version 2 of the License, or -// (at your option) any later version. -// -// This program is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU General Public License for more details. -// -// You should have received a copy of the GNU General Public License -// along with this program; if not, write to the Free Software -// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA -// -// In addition, as a special exception, the copyright holders give -// permission to link the code of portions of this program with the -// OpenSSL library under certain conditions as described in each -// individual source file, and distribute linked combinations -// including the two. -// -// You must obey the GNU General Public License in all respects for -// all of the code used other than OpenSSL. If you modify file(s) -// with this exception, you may extend this exception to your version -// of the file(s), but you are not obligated to do so. If you do not -// wish to do so, delete this exception statement from your version. -// If you delete this exception statement from all source files in the -// program, then also delete it here. -// -// Contact: Jari Sundell - - // A simple, and not guaranteed atomic, lockfile implementation. It // saves the hostname and pid in the lock file, which may be accessed // by Lockfile::locked_by(). If the path is an empty string then no