Tagged release 0.16.6.

configure.ac

@@ -1,6 +1,6 @@
 m4_pattern_allow([PKG_CHECK_EXISTS])
 
-AC_INIT([rtorrent],[0.16.4],[sundell.software@gmail.com])
+AC_INIT([rtorrent],[0.16.6],[sundell.software@gmail.com])
 
 AC_CONFIG_HEADERS([config.h])
 AC_CONFIG_MACRO_DIRS([scripts])
@@ -48,7 +48,7 @@ if test "x$ax_cv_ncursesw" != xyes && test "x$ax_cv_ncurses" != xyes; then
 fi
 
 PKG_CHECK_MODULES([CPPUNIT], [cppunit],, [no_cppunit="yes"])
-PKG_CHECK_MODULES([DEPENDENCIES], [libtorrent >= 0.16.4])
+PKG_CHECK_MODULES([DEPENDENCIES], [libtorrent >= 0.16.6])
 
 AC_LANG_PUSH(C++)
 TORRENT_WITH_XMLRPC_C

doc/valgrind.commands

@@ -0,0 +1 @@
+valgrind --leak-check=full --track-origins=yes --show-reachable=yes --suppressions=/Users/rakshasa/projects/rtorrent/doc/valgrind.suppression --gen-suppressions=all --log-file=valgrind.log /usr/local/bin/rtorrent

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 \
@@ -125,6 +123,16 @@ libsub_root_a_SOURCES = \
 	rpc/tinyxml2/tinyxml2.cc \
 	rpc/nlohmann/json.h \
 	\
+	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 \
+	session/thread_session.h \
+	\
 	ui/download.cc \
 	ui/download.h \
 	ui/download_list.cc \
@@ -190,9 +198,7 @@ libsub_root_a_SOURCES = \
 	option_parser.cc \
 	option_parser.h \
 	signal_handler.cc \
-	signal_handler.h \
-	thread_worker.cc \
-	thread_worker.h
+	signal_handler.h
 
 
 AM_CPPFLAGS = -I$(srcdir) -I$(top_srcdir) -DPACKAGE_DATADIR=\"$(pkgdatadir)\"

src/command_download.cc

@@ -25,9 +25,9 @@
 #include <torrent/utils/option_strings.h>
 
 #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"
@@ -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) {
@@ -495,6 +495,19 @@ download_get_variable(core::Download* download, const char* first_key, const cha
   return download->bencode()->get_key(first_key).get_key(second_key);
 }
 
+torrent::Object
+download_get_value_or_zero(core::Download* download, const char* first_key, const char* second_key = NULL) {
+  auto object = download_get_variable(download, first_key, second_key);
+
+  if (object.is_empty())
+    return int64_t(0);
+
+  if (!object.is_value())
+    throw torrent::bencode_error("Download variable is not a value.");
+
+  return object;
+}
+
 torrent::Object
 download_set_variable(core::Download* download, const torrent::Object& rawArgs, const char* first_key, const char* second_key = NULL) {
   if (second_key == NULL)
@@ -600,12 +613,16 @@ d_list_remove(core::Download* download, const torrent::Object& rawArgs, const ch
 
 #define CMD2_DL_TIMESTAMP(key, first_key, second_key)                   \
   CMD2_DL(key, std::bind(&download_get_variable, std::placeholders::_1, first_key, second_key)); \
-  CMD2_DL_VALUE_P(key ".set", std::bind(&download_set_variable_value, \
-                                             std::placeholders::_1, std::placeholders::_2, \
-                                             first_key, second_key)); \
+  CMD2_DL_VALUE_P(key ".set", std::bind(&download_set_variable_value,   \
+                                        std::placeholders::_1, std::placeholders::_2, \
+                                        first_key, second_key));        \
   CMD2_DL_VALUE_P(key ".set_if_z", std::bind(&download_set_variable_value_ifz, \
                                              std::placeholders::_1, std::placeholders::_2, \
                                              first_key, second_key));   \
+  CMD2_DL(key ".or_zero", [](core::Download* download, auto) {          \
+      return download_get_value_or_zero(download, first_key, second_key); }); \
+  CMD2_DL_VALUE(key ".elapsed", [](core::Download* download, auto value) { \
+      return download_get_value_or_zero(download, first_key, second_key).as_value() > value; });
 
 #define CMD2_DL_VAR_STRING(key, first_key, second_key)                   \
   CMD2_DL(key, std::bind(&download_get_variable, std::placeholders::_1, first_key, second_key)); \
@@ -683,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));
 

src/command_local.cc

@@ -17,11 +17,11 @@
 
 #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"
 #include "rpc/scgi.h"
+#include "session/session_manager.h"
 #include "utils/file_status_cache.h"
 
 #include "globals.h"
@@ -184,7 +184,6 @@ cmd_file_append(const torrent::Object::list_type& args) {
 void
 initialize_command_local() {
   core::DownloadList*    dList = control->core()->download_list();
-  core::DownloadStore*   dStore = control->core()->download_store();
   torrent::ChunkManager* chunkManager = torrent::chunk_manager();
   torrent::FileManager*  fileManager = torrent::file_manager();
 
@@ -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));
@@ -270,22 +270,25 @@ 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",                    [dList](auto, auto)   { return dList->session_save(); });
 
-  CMD2_ANY_V       ("session.save",            std::bind(&core::DownloadList::session_save, dList));
+  CMD2_ANY         ("magnet.path",                     [](auto, auto)        { return control->core()->magnet_path(); });
+  CMD2_ANY_STRING_V("magnet.path.set",                 [](auto, auto& str)   { return control->core()->set_magnet_path(str); });
 
 #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)                                        \

src/command_logging.cc

@@ -8,16 +8,15 @@
 #include <torrent/utils/log.h>
 #include <torrent/utils/option_strings.h>
 
+#include "globals.h"
+#include "control.h"
+#include "command_helpers.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 +61,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());
 
@@ -135,12 +134,13 @@ initialize_command_logging() {
   CMD2_ANY_LIST    ("log.append_file",      std::bind(&apply_log_open, log_flag_append_file, std::placeholders::_2));
   CMD2_ANY_LIST    ("log.append_gz_file",   std::bind(&apply_log_open, log_flag_append_file, std::placeholders::_2));
 
-  CMD2_ANY_STRING_V("log.close",      std::bind(&torrent::log_close_output_str, std::placeholders::_2));
+  CMD2_ANY_STRING_V("log.close",            std::bind(&torrent::log_close_output_str, std::placeholders::_2));
 
-  CMD2_ANY_LIST    ("log.add_output", std::bind(&apply_log_add_output, std::placeholders::_2));
+  CMD2_ANY_LIST    ("log.add_output",       std::bind(&apply_log_add_output, std::placeholders::_2));
 
-  CMD2_ANY_STRING  ("log.execute",    std::bind(&apply_log, std::placeholders::_2, 0));
-  CMD2_ANY_STRING  ("log.vmmap.dump", std::bind(&log_vmmap_dump, std::placeholders::_2));
-  CMD2_ANY_STRING_V("log.rpc",        std::bind(&ThreadWorker::set_rpc_log, worker_thread, std::placeholders::_2));
-  CMD2_REDIRECT    ("log.xmlrpc",     "log.rpc"); // For backwards compatibility
+  CMD2_ANY_STRING  ("log.execute",          std::bind(&apply_log, std::placeholders::_2, 0));
+  CMD2_ANY_STRING  ("log.vmmap.dump",       std::bind(&log_vmmap_dump, std::placeholders::_2));
+  CMD2_ANY_STRING_V("log.rpc",              [](const auto&, const auto& str) { scgi_thread::set_rpc_log(str); });
+
+  CMD2_REDIRECT    ("log.xmlrpc", "log.rpc"); // For backwards compatibility
 }

src/command_network.cc

@@ -17,6 +17,9 @@
 #include <torrent/utils/log.h>
 #include <torrent/utils/option_strings.h>
 
+#include "globals.h"
+#include "control.h"
+#include "command_helpers.h"
 #include "core/download.h"
 #include "core/manager.h"
 #include "rpc/scgi.h"
@@ -24,10 +27,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;
@@ -79,7 +78,7 @@ initialize_rpc_handlers() {
 
 torrent::Object
 apply_scgi(const std::string& arg, int type) {
-  if (worker_thread->scgi() != NULL)
+  if (scgi_thread::scgi() != nullptr)
     throw torrent::input_error("SCGI already enabled.");
 
   initialize_rpc_handlers();
@@ -142,7 +141,7 @@ apply_scgi(const std::string& arg, int type) {
     throw torrent::input_error(e.what());
   }
 
-  worker_thread->set_scgi(scgi);
+  scgi_thread::set_scgi(scgi);
   return torrent::Object();
 }
 

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);
 

src/control.cc

@@ -9,7 +9,6 @@
 #include <torrent/utils/directory_events.h>
 
 #include "core/dht_manager.h"
-#include "core/download_store.h"
 #include "core/http_queue.h"
 #include "core/manager.h"
 #include "core/view_manager.h"
@@ -26,6 +25,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,7 +58,8 @@ Control::~Control() {
 
 void
 Control::initialize() {
-  worker_thread->start_thread();
+  session_thread::thread()->start_thread();
+  scgi_thread::thread()->start_thread();
 
   display::Canvas::initialize();
   display::Window::slot_schedule([this](display::Window* w, std::chrono::microseconds t) { m_display->schedule(w, t); });
@@ -68,7 +69,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 +86,12 @@ Control::cleanup() {
   if(!display::Canvas::daemon())
     m_inputStdin->remove(torrent::this_thread::poll());
 
-  m_core->download_store()->disable();
+  if (scgi_thread::thread()->is_active())
+    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();
   m_core->cleanup();
@@ -104,7 +109,7 @@ Control::cleanup_exception() {
 
 bool
 Control::is_shutdown_completed() {
-  if (!m_shutdownQuick || worker_thread->is_active())
+  if (!m_shutdownQuick)
     return false;
 
   // Tracker requests can be disowned, so wait for these to
@@ -123,10 +128,10 @@ void
 Control::handle_shutdown() {
   rpc::commands.call_catch("event.system.shutdown", rpc::make_target(), "shutdown", "System shutdown event action failed: ");
 
-  if (!m_shutdownQuick) {
-    if (worker_thread->is_active())
-      worker_thread->stop_thread_wait();
+  if (scgi_thread::thread()->is_active())
+    scgi_thread::thread()->stop_thread_wait();
 
+  if (!m_shutdownQuick) {
     torrent::runtime::network_manager()->listen_close();
 
     m_directory_events->close();
@@ -136,9 +141,6 @@ Control::handle_shutdown() {
       torrent::this_thread::scheduler()->wait_for_ceil_seconds(&m_task_shutdown, 5s);
 
   } else {
-    if (worker_thread->is_active())
-      worker_thread->stop_thread_wait();
-
     m_core->shutdown(true);
   }
 

src/core/dht_manager.cc

@@ -13,10 +13,10 @@
 
 #include "control.h"
 #include "download.h"
-#include "download_store.h"
 #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 +35,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 +115,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;

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"
@@ -255,11 +254,21 @@ DownloadFactory::receive_success() {
                             rpc::call_command_value("system.file.split_size"),
                             rpc::call_command_string("system.file.split_suffix"));
 
-  if (!rtorrent->has_key_string("directory"))
-    rpc::call_command("d.directory.set", m_variables["directory"], rpc::make_target(download));
-  else
+  if (rtorrent->has_key_string("directory")) {
     rpc::call_command("d.directory_base.set", rtorrent->get_key("directory"), rpc::make_target(download));
 
+  } else if (download->download()->info()->is_meta_download()) {
+    auto& metadata_path = control->core()->magnet_path();
+
+    if (!metadata_path.empty())
+      rpc::call_command("d.directory.set", metadata_path, rpc::make_target(download));
+    else
+      rpc::call_command("d.directory.set", session_thread::session_path(), rpc::make_target(download));
+
+  } else {
+    rpc::call_command("d.directory.set", m_variables["directory"], rpc::make_target(download));
+  }
+
   if (!m_session && m_variables["tied_to_file"].as_value())
     rpc::call_command("d.tied_to_file.set", m_uri.empty() ? m_variables["tied_file"] : m_uri, rpc::make_target(download));
 

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().");
 

src/core/download_store.cc

@@ -1,205 +0,0 @@
-// DownloadStore handles the saving and listing of session torrents.
-
-#include "config.h"
-
-#include <fstream>
-#include <stdio.h>
-#include <fcntl.h>
-#include <unistd.h>
-#include <rak/error_number.h>
-#include <rak/path.h>
-#include <rak/string_manip.h>
-#include <torrent/utils/resume.h>
-#include <torrent/object.h>
-#include <torrent/exceptions.h>
-#include <torrent/torrent.h>
-#include <torrent/rate.h>
-#include <torrent/object_stream.h>
-
-#include "utils/directory.h"
-
-#include "download.h"
-#include "download_store.h"
-#include "rpc/parse_commands.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())
-    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);
-
-  std::string base_filename = create_filename(d);
-
-  if (!write_bencode(base_filename + ".libtorrent_resume.new", *resume_base, 0) ||
-      !write_bencode(base_filename + ".rtorrent.new", *rtorrent_base, 0))
-    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());
-
-  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());
-
-  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());
-}
-
-// 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 (!is_enabled())
-    return utils::Directory();
-
-  utils::Directory d(m_path);
-
-  if (!d.update(utils::Directory::update_hide_dot))
-    throw torrent::storage_error("core::DownloadStore::update() could not open directory \"" + m_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 m_path + rak::transform_hex(d->info()->hash().begin(), d->info()->hash().end()) + ".torrent";
-}
-
-}

src/core/download_store.h

@@ -1,49 +0,0 @@
-#ifndef RTORRENT_CORE_DOWNLOAD_STORE_H
-#define RTORRENT_CORE_DOWNLOAD_STORE_H
-
-#include <string>
-
-#include "utils/lockfile.h"
-
-namespace utils {
-  class Directory;
-}
-
-namespace core {
-
-class Download;
-
-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); }
-  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);
-
-  bool                write_bencode(const std::string& filename, const torrent::Object& obj, uint32_t skip_mask);
-
-  std::string         m_path;
-  utils::Lockfile     m_lockfile;
-};
-
-}
-
-#endif

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<DownloadStore>();
   m_download_list     = std::make_unique<DownloadList>();
   m_file_status_cache = std::make_unique<FileStatusCache>();
   m_http_queue        = std::make_unique<HttpQueue>();
@@ -134,6 +132,16 @@ Manager::retrieve_throttle_value(const torrent::Object::string_type& name, bool
   }
 }
 
+void
+Manager::set_magnet_path(const std::string& path) {
+  if (path.empty())
+    m_magnet_path.clear();
+  else if (path.back() == '/')
+    m_magnet_path = path;
+  else
+    m_magnet_path = path + '/';
+}
+
 void
 Manager::cleanup() {
   // Need to disconnect log signals? Not really since we won't receive

src/core/manager.h

@@ -21,7 +21,6 @@ class FileStatusCache;
 
 namespace core {
 
-class DownloadStore;
 class HttpQueue;
 
 typedef std::map<std::string, torrent::ThrottlePair> 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(); }
@@ -63,6 +61,9 @@ public:
 
   void                set_proxy_address(const std::string& addr);
 
+  const std::string&  magnet_path();
+  void                set_magnet_path(const std::string& path);
+
   void                shutdown(bool force);
 
   void                push_log(const char* msg);
@@ -95,7 +96,6 @@ private:
   void                receive_hashing_changed();
 
   std::unique_ptr<DownloadList>    m_download_list;
-  std::unique_ptr<DownloadStore>   m_download_store;
   std::unique_ptr<FileStatusCache> m_file_status_cache;
   std::unique_ptr<HttpQueue>       m_http_queue;
 
@@ -106,11 +106,15 @@ private:
 
   torrent::log_buffer_ptr m_log_important;
   torrent::log_buffer_ptr m_log_complete;
+
+  std::string         m_magnet_path;
 };
 
 // Meh, cleanup.
 extern void receive_tracker_dump(const std::string& url, const char* data, size_t size);
 
+inline const std::string& Manager::magnet_path() { return m_magnet_path; }
+
 }
 
 #endif

src/globals.cc

@@ -5,4 +5,3 @@
 rpc::ip_table_list ip_tables;
 
 Control*           control{};
-ThreadWorker*      worker_thread{};

src/globals.h

@@ -1,14 +1,52 @@
 #ifndef TORRENT_GLOBALS_H
 #define TORRENT_GLOBALS_H
 
-#include "thread_worker.h"
+#include <torrent/common.h>
+
 #include "rpc/ip_table_list.h"
 
 class Control;
 
-extern rpc::ip_table_list          ip_tables;
+extern rpc::ip_table_list ip_tables;
 
 extern Control*      control;
-extern ThreadWorker* worker_thread;
+
+namespace rpc {
+class SCgi;
+}
+
+namespace session {
+class SessionManager;
+}
+
+namespace scgi_thread {
+
+torrent::utils::Thread* thread();
+std::thread::id         thread_id();
+
+void                    callback(void* target, std::function<void ()>&& fn);
+void                    cancel_callback(void* target);
+void                    cancel_callback_and_wait(void* target);
+
+rpc::SCgi*              scgi();
+void                    set_scgi(rpc::SCgi* scgi);
+void                    set_rpc_log(const std::string& filename);
+
+} // namespace torrent::scgi_thread
+
+
+namespace session_thread {
+
+torrent::utils::Thread* thread();
+std::thread::id         thread_id();
+
+void                    callback(void* target, std::function<void ()>&& fn);
+void                    cancel_callback(void* target);
+void                    cancel_callback_and_wait(void* target);
+
+session::SessionManager* manager();
+std::string              session_path();
+
+} // namespace torrent::session_thread
 
 #endif

src/input/input_event.h

@@ -14,12 +14,14 @@ public:
 
   InputEvent(int fd) { m_fileDesc = fd; }
 
+  const char*         type_name() const override { return "input"; }
+
   void                insert(torrent::net::Poll* p);
   void                remove(torrent::net::Poll* p);
 
-  void                event_read();
-  void                event_write();
-  void                event_error();
+  void                event_read() override;
+  void                event_write() override;
+  void                event_error() override;
 
   void                slot_pressed(slot_int s) { m_slotPressed = s; }
 

src/main.cc

@@ -22,17 +22,19 @@
 #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"
 #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 "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"
 
 #include "control.h"
@@ -41,8 +43,6 @@
 #include "signal_handler.h"
 #include "option_parser.h"
 
-#include "thread_worker.h"
-
 #define LT_LOG(log_fmt, ...)                                    \
   lt_log_print(torrent::LOG_SYSTEM, "system: " log_fmt, __VA_ARGS__);
 
@@ -113,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
@@ -215,9 +215,8 @@ main(int argc, char** argv) {
     torrent::initialize();
     torrent::set_main_thread_slots(std::bind(&client_perform));
 
-    // TODO: Move to controller.
-    worker_thread = new ThreadWorker();
-    worker_thread->init_thread();
+    scgi::ThreadScgi::create_thread();
+    session::ThreadSession::create_thread();
 
     // Initialize option handlers after libtorrent to ensure
     // torrent::ConnectionManager* are valid etc.
@@ -530,13 +529,13 @@ main(int argc, char** argv) {
     return -1;
   }
 
-  torrent::log_cleanup();
-
   delete control;
   control = nullptr;
 
-  delete worker_thread;
-  worker_thread = nullptr;
+  scgi::ThreadScgi::destroy_thread();
+  session::ThreadSession::destroy_thread();
+
+  torrent::log_cleanup();
 
   return 0;
 }

src/rpc/parse_commands.cc

@@ -156,9 +156,10 @@ parse_command_file(const std::string& path) {
 
       if (file.gcount() == 0)
         throw torrent::internal_error("parse_command_file(...) file.gcount() == 0.");
+
       int lineLength = file.gcount() - 1;
       // In case we are at the end of the file and the last character is
-      // not a line feed, we'll just increase the read character count so 
+      // not a line feed, we'll just increase the read character count so
       // that the last would also be included in option line.
       if (file.eof() && file.get() != '\n')
         lineLength++;

src/rpc/scgi.cc

@@ -92,7 +92,7 @@ SCgi::open(void* sa, unsigned int length) {
 
 void
 SCgi::activate() {
-  assert(std::this_thread::get_id() == worker_thread->thread_id() && "SCgi::activate() must be called from the worker thread.");
+  assert(torrent::this_thread::thread() == scgi_thread::thread());
 
   torrent::this_thread::poll()->open(this);
   torrent::this_thread::poll()->insert_read(this);
@@ -101,7 +101,7 @@ SCgi::activate() {
 
 void
 SCgi::deactivate() {
-  assert(std::this_thread::get_id() == worker_thread->thread_id() && "SCgi::deactivate() must be called from the worker thread.");
+  assert(torrent::this_thread::thread() == scgi_thread::thread());
 
   torrent::this_thread::poll()->remove_and_close(this);
 }

src/rpc/scgi_task.h

@@ -23,6 +23,8 @@ public:
 
   SCgiTask() { m_fileDesc = -1; }
 
+  const char*         type_name() const override { return "scgi-task"; }
+
   bool                is_open() const      { return m_fileDesc != -1; }
   bool                is_available() const { return m_fileDesc == -1; }
 
@@ -31,9 +33,9 @@ public:
 
   ContentType         content_type() const { return m_content_type; }
 
-  virtual void        event_read();
-  virtual void        event_write();
-  virtual void        event_error();
+  void                event_read() override;
+  void                event_write() override;
+  void                event_error() override;
 
   utils::SocketFd&    get_fd()             { return *reinterpret_cast<utils::SocketFd*>(&m_fileDesc); }
 

src/scgi/thread_scgi.cc

@@ -0,0 +1,138 @@
+#include "config.h"
+
+#include "scgi/thread_scgi.h"
+
+#include <fcntl.h>
+#include <unistd.h>
+#include <rak/path.h>
+#include <torrent/exceptions.h>
+#include <torrent/utils/log.h>
+
+#include "rpc/scgi.h"
+
+namespace scgi {
+
+class ThreadScgiInternal {
+public:
+  static ThreadScgi* thread_scgi() { return ThreadScgi::internal_thread_scgi(); }
+};
+
+ThreadScgi* ThreadScgi::m_thread_scgi{};
+
+void
+ThreadScgi::create_thread() {
+  auto thread = new ThreadScgi;
+
+  m_thread_scgi          = thread;
+  m_thread_scgi->m_state = STATE_INITIALIZED;
+}
+
+void
+ThreadScgi::destroy_thread() {
+  delete m_thread_scgi;
+  m_thread_scgi = nullptr;
+}
+
+ThreadScgi*
+ThreadScgi::thread_scgi() {
+  return m_thread_scgi;
+}
+
+void
+ThreadScgi::cleanup_thread() {
+  if (m_scgi != nullptr)
+    m_scgi.load()->deactivate();
+}
+
+rpc::SCgi*
+ThreadScgi::scgi() {
+  return m_scgi;
+}
+
+bool
+ThreadScgi::set_scgi(rpc::SCgi* scgi) {
+  rpc::SCgi* expected = nullptr;
+
+  if (!m_scgi.compare_exchange_strong(expected, scgi))
+    return false;
+
+  change_rpc_log();
+
+  callback(nullptr, [this]() {
+      if (m_scgi == NULL)
+        throw torrent::internal_error("Tried to start SCGI but object was not present.");
+
+      m_scgi.load()->activate();
+    });
+
+  return true;
+}
+
+void
+ThreadScgi::set_rpc_log(const std::string& filename) {
+  callback(nullptr, [this, filename]() {
+      m_rpc_log_filename = filename;
+      change_rpc_log();
+    });
+}
+
+void
+ThreadScgi::change_rpc_log() {
+  if (scgi() == NULL)
+    return;
+
+  if (scgi()->log_fd() != -1) {
+    ::close(scgi()->log_fd());
+    scgi()->set_log_fd(-1);
+    lt_log_print(torrent::LOG_NOTICE, "Closed RPC log.", 0);
+  }
+
+  if (m_rpc_log_filename.empty())
+    return;
+
+  scgi()->set_log_fd(open(rak::path_expand(m_rpc_log_filename).c_str(), O_WRONLY | O_APPEND | O_CREAT, 0644));
+
+  if (scgi()->log_fd() == -1) {
+    lt_log_print(torrent::LOG_NOTICE, "Could not open RPC log file '%s'.", m_rpc_log_filename.c_str());
+    return;
+  }
+
+  lt_log_print(torrent::LOG_NOTICE, "Logging RPC events to '%s'.", m_rpc_log_filename.c_str());
+}
+
+void
+ThreadScgi::call_events() {
+  if ((m_flags & flag_do_shutdown)) {
+    if ((m_flags & flag_did_shutdown))
+      throw torrent::internal_error("Already trigged shutdown.");
+
+    m_flags |= flag_did_shutdown;
+    throw torrent::shutdown_exception();
+  }
+
+  process_callbacks();
+}
+
+std::chrono::microseconds
+ThreadScgi::next_timeout() {
+  // TODO: This leads to kqueue crash?
+  // return std::chrono::microseconds(1h);
+  return std::chrono::microseconds(10min);
+}
+
+} // namespace scgi
+
+namespace scgi_thread {
+
+torrent::utils::Thread* thread()                         { return scgi::ThreadScgiInternal::thread_scgi(); }
+std::thread::id         thread_id()                      { return scgi::ThreadScgiInternal::thread_scgi()->thread_id(); }
+
+void callback(void* target, std::function<void ()>&& fn) { scgi::ThreadScgiInternal::thread_scgi()->callback(target, std::move(fn)); }
+void cancel_callback(void* target)                       { scgi::ThreadScgiInternal::thread_scgi()->cancel_callback(target); }
+void cancel_callback_and_wait(void* target)              { scgi::ThreadScgiInternal::thread_scgi()->cancel_callback_and_wait(target); }
+
+rpc::SCgi*  scgi()                                       { return scgi::ThreadScgiInternal::thread_scgi()->scgi(); }
+void        set_scgi(rpc::SCgi* scgi)                    { scgi::ThreadScgiInternal::thread_scgi()->set_scgi(scgi); }
+void        set_rpc_log(const std::string& filename)     { scgi::ThreadScgiInternal::thread_scgi()->set_rpc_log(filename); }
+
+} // namespace scgi_thread

src/scgi/thread_scgi.h

@@ -0,0 +1,54 @@
+#ifndef RTORRENT_SCGI_THREAD_SCGI_H
+#define RTORRENT_SCGI_THREAD_SCGI_H
+
+#include <atomic>
+#include <string>
+#include <torrent/utils/thread.h>
+
+namespace rpc {
+class SCgi;
+}
+
+namespace scgi {
+
+class ThreadScgiInternal;
+
+class ThreadScgi : public torrent::utils::Thread {
+public:
+
+  static void         create_thread();
+  static void         destroy_thread();
+  static ThreadScgi*  thread_scgi();
+
+  const char*         name() const override  { return "rtorrent-scgi"; }
+
+  rpc::SCgi*          scgi();
+  bool                set_scgi(rpc::SCgi* scgi);
+
+  void                set_rpc_log(const std::string& filename);
+
+protected:
+  friend class ThreadScgiInternal;
+
+  ThreadScgi() = default;
+
+  static auto         internal_thread_scgi() { return m_thread_scgi; }
+
+  void                cleanup_thread() override;
+
+  void                      call_events() override;
+  std::chrono::microseconds next_timeout() override;
+
+private:
+  void                task_touch_log();
+  void                change_rpc_log();
+
+  static ThreadScgi*  m_thread_scgi;
+
+  std::atomic<rpc::SCgi*> m_scgi{nullptr};
+  std::string             m_rpc_log_filename;
+};
+
+} // namespace scgi
+
+#endif // RTORRENT_SCGI_THREAD_SCGI_H

src/session/download_storer.cc

@@ -0,0 +1,193 @@
+#include "config.h"
+
+#include "download_storer.h"
+
+#include <fcntl.h>
+#include <fstream>
+#include <unistd.h>
+#include <torrent/exceptions.h>
+#include <torrent/object.h>
+#include <torrent/object_stream.h>
+#include <torrent/utils/log.h>
+#include <torrent/utils/resume.h>
+
+#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<std::stringstream>();
+  auto resume_stream   = std::make_unique<std::stringstream>();
+  auto rtorrent_stream = std::make_unique<std::stringstream>();
+
+  if (!skip_static) {
+    torrent_stream = std::make_unique<std::stringstream>();
+    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) + ".torrent";
+}
+
+void
+DownloadStorer::unlink_files(const std::string& session_path) {
+  auto base_path = build_path(session_path);
+
+  auto torrent_path    = base_path;
+  auto libtorrent_path = base_path + ".libtorrent_resume";
+  auto rtorrent_path   = base_path + ".rtorrent";
+  auto metadata_path   = base_path + ".meta";
+
+  ::unlink(libtorrent_path.c_str());
+  ::unlink(rtorrent_path.c_str());
+  ::unlink(torrent_path.c_str());
+  ::unlink(metadata_path.c_str());
+}
+
+namespace {
+
+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;
+}
+
+void
+save_stream(const std::string& path, bool use_fsyncdisk, const std::stringstream& 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())
+    throw torrent::storage_error("failed to open file for writing : " + path);
+
+  output << stream.rdbuf();
+
+  if (!output.good())
+    throw torrent::storage_error("failed to write stream to file : " + path);
+
+  output.close();
+
+  // Ensure that the new file is actually written to the disk
+  int fd = ::open(path.c_str(), O_WRONLY);
+
+  if (fd < 0)
+    throw torrent::storage_error("failed to open file descriptor for fsync : " + path);
+
+  if (use_fsyncdisk) {
+#ifdef __APPLE__
+    ::fsync(fd);
+#else
+    ::fdatasync(fd);
+#endif
+  }
+
+  ::close(fd);
+}
+
+} // namespace anonymous
+
+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) {
+  auto torrent_path    = path;
+  auto libtorrent_path = path + ".libtorrent_resume";
+  auto rtorrent_path   = path + ".rtorrent";
+
+  if (torrent_stream)
+    save_stream(torrent_path + ".new", use_fsyncdisk, *torrent_stream);
+
+  save_stream(libtorrent_path + ".new", use_fsyncdisk, *libtorrent_stream);
+  save_stream(rtorrent_path + ".new", use_fsyncdisk, *rtorrent_stream);
+
+  if (torrent_stream) {
+    if (::rename((torrent_path + ".new").c_str(), torrent_path.c_str()) == -1)
+      throw torrent::storage_error("failed to rename torrent file : " + torrent_path);
+  }
+
+  if (::rename((libtorrent_path + ".new").c_str(), libtorrent_path.c_str()) == -1)
+    throw torrent::storage_error("failed to rename libtorrent resume file : " + libtorrent_path);
+
+  if (::rename((rtorrent_path + ".new").c_str(), rtorrent_path.c_str()) == -1)
+    throw torrent::storage_error("failed to rename rtorrent resume file : " + rtorrent_path);
+}
+
+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

src/session/download_storer.h

@@ -0,0 +1,55 @@
+#ifndef RTORRENT_SESSION_DOWNLOAD_STORER_H
+#define RTORRENT_SESSION_DOWNLOAD_STORER_H
+
+#include <memory>
+#include <sstream>
+#include <string>
+#include <torrent/common.h>
+
+namespace core {
+class Download;
+}
+
+namespace utils {
+class Directory;
+}
+
+namespace session {
+
+class DownloadStorer {
+public:
+  DownloadStorer(core::Download* download);
+
+  core::Download*     download() const { return m_download; }
+
+  std::string         build_path(const std::string& session_path);
+
+  void                build_full_streams()   { build_streams(false); }
+  void                build_resume_streams() { build_streams(true); }
+
+  void                unlink_files(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 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);
+
+  static utils::Directory get_formated_entries(const std::string& session_path);
+
+private:
+  void                build_streams(bool skip_static);
+
+  core::Download*     m_download;
+
+  std::unique_ptr<std::stringstream> m_torrent_stream;
+  std::unique_ptr<std::stringstream> m_rtorrent_stream;
+  std::unique_ptr<std::stringstream> m_libtorrent_stream;
+};
+
+} // namespace session
+
+#endif // RTORRENT_SESSION_DOWNLOAD_STORER_H

src/session/session_manager.cc

@@ -0,0 +1,499 @@
+#include "config.h"
+
+#include "session/session_manager.h"
+
+#include <cassert>
+#include <torrent/exceptions.h>
+#include <torrent/utils/log.h>
+
+#include "globals.h"
+#include "session/download_storer.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 {
+
+SessionManager::SessionManager(torrent::utils::Thread* thread)
+  : m_thread(thread),
+    m_lockfile(std::make_unique<utils::Lockfile>()) {
+}
+
+SessionManager::~SessionManager() = default;
+
+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_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());
+
+  if (m_freeze_info)
+    throw torrent::input_error("Session lock option cannot be changed after startup.");
+
+  m_use_lock = use_lock;
+}
+
+void
+SessionManager::save_resume_download(core::Download* download) {
+  assert(torrent::this_thread::thread() == torrent::main_thread::thread());
+
+  if (m_path.empty())
+    return;
+
+  {
+    std::unique_lock<std::mutex> lock(m_pending_builds_mutex);
+
+    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()) {
+      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);
+
+    LT_LOG("build of resume save data queued : download:%p", download);
+  }
+}
+
+void
+SessionManager::save_full_download(core::Download* download) {
+  assert(torrent::this_thread::thread() == torrent::main_thread::thread());
+
+  if (m_path.empty())
+    return;
+
+  DownloadStorer storer(download);
+
+  storer.build_full_streams();
+
+  auto save_request = SaveRequest{
+    download,
+    storer.build_path(m_path),
+    storer.torrent_stream(),
+    storer.rtorrent_stream(),
+    storer.libtorrent_stream()
+  };
+
+  {
+    std::unique_lock<std::mutex> 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.");
+
+    if (replace_save_request_unsafe(save_request))
+      throw torrent::internal_error("SessionManager::save_full_download() replacing existing save request, not supported?");
+
+    m_save_requests.push_back(std::move(save_request));
+    m_save_request_counter = m_save_requests.size();
+
+    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(); });
+}
+
+void
+SessionManager::remove_download(core::Download* download) {
+  assert(torrent::this_thread::thread() == torrent::main_thread::thread());
+
+  if (m_path.empty())
+    return;
+
+  std::unique_lock<std::mutex> lock(m_mutex);
+
+  if (!m_active)
+    throw torrent::internal_error("SessionManager::remove_download() called while not active.");
+
+  if (remove_completely_unsafe(download, lock))
+    LT_LOG("canceled pending save request : download:%p", download);
+
+  DownloadStorer(download).unlink_files(m_path);
+
+  LT_LOG("removed session files : download:%p", download);
+}
+
+void
+SessionManager::start() {
+  assert(torrent::this_thread::thread() == torrent::main_thread::thread());
+
+  std::unique_lock<std::mutex> lock(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::unique_lock<std::mutex> lock(m_mutex);
+
+  if (!m_active)
+    throw torrent::internal_error("SessionManager::cleanup() called while not active.");
+
+  {
+    std::unique_lock<std::mutex> 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()) {
+    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());
+
+  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);
+  torrent::main_thread::cancel_callback(this);
+}
+
+void
+SessionManager::process_pending_resume_builds(bool is_flushing) {
+  assert(torrent::this_thread::thread() == torrent::main_thread::thread());
+
+  std::vector<SaveRequest> requests;
+
+  // Only main thread is allowed to process pending builds or remove downloads, as such it is safe
+  // to unlock before adding them to save requests.
+  {
+    std::unique_lock<std::mutex> lock(m_pending_builds_mutex);
+
+    while (!m_pending_builds.empty()) {
+      if (!is_flushing && m_save_request_counter + requests.size() >= max_concurrent_requests)
+        break;
+
+      auto* download = m_pending_builds.front();
+      m_pending_builds.pop_front();
+
+      LT_LOG("processing pending resume save : download:%p", download);
+
+      DownloadStorer storer(download);
+
+      storer.build_resume_streams();
+
+      auto save_request = SaveRequest{
+        download,
+        storer.build_path(m_path),
+        nullptr,
+        storer.rtorrent_stream(),
+        storer.libtorrent_stream()
+      };
+
+      requests.push_back(std::move(save_request));
+    }
+  }
+
+  if (requests.empty())
+    return;
+
+  {
+    std::unique_lock<std::mutex> 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 (replace_save_request_unsafe(save_request)) {
+        LT_LOG("updated pending resume save request : download:%p", save_request.download);
+        continue;
+      }
+
+      m_save_requests.push_back(std::move(save_request));
+
+      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(); });
+}
+
+void
+SessionManager::process_save_request() {
+  assert(m_thread == torrent::this_thread::thread());
+
+  if (m_path.empty())
+    throw torrent::internal_error("SessionManager::process_save_request() called with empty path.");
+
+  std::unique_lock<std::mutex> lock(m_mutex);
+
+  if (!m_active)
+    throw torrent::internal_error("SessionManager::process_save_request() called while not active.");
+
+  while (!m_save_requests.empty() && m_processing_saves.size() < max_concurrent_requests)
+    process_next_save_request_unsafe();
+
+  m_processing_saves_callback_scheduled = false;
+}
+
+void
+SessionManager::process_save_request_with_pending_callback() {
+  process_save_request();
+
+  std::unique_lock<std::mutex> lock(m_pending_builds_mutex);
+
+  if (!m_pending_builds.empty())
+    torrent::main_thread::callback(this, [this]() { process_pending_resume_builds(false); });
+}
+
+void
+SessionManager::process_next_save_request_unsafe() {
+  auto request = std::move(m_save_requests.front());
+
+  m_save_requests.pop_front();
+  m_save_request_counter = m_save_requests.size();
+
+  auto itr = m_processing_saves.insert(m_processing_saves.end(), ProcessingSave{});
+
+  itr->second = std::move(request);
+  itr->first = std::async(std::launch::async, [this, itr]() {
+      auto cleanup_fn = [this, itr]() {
+          std::unique_lock<std::mutex> 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_finished_condition.notify_all();
+
+          m_processing_saves.erase(itr);
+        };
+
+      try {
+        DownloadStorer::save_and_move_streams(itr->second.path, m_use_fsyncdisk,
+                                              itr->second.torrent_stream.get(),
+                                              itr->second.rtorrent_stream.get(),
+                                              itr->second.libtorrent_stream.get());
+      } catch (...) {
+        cleanup_fn();
+        throw;
+      }
+
+      cleanup_fn();
+    });
+}
+
+void
+SessionManager::process_finished_saves() {
+  assert(m_thread == torrent::this_thread::thread());
+
+  std::unique_lock<std::mutex> lock(m_mutex);
+
+  if (!m_active)
+    throw torrent::internal_error("SessionManager::process_finished_saves() called while not active.");
+
+  for (auto& request : m_finished_saves) {
+    try {
+      request.first.get();
+
+    } catch (torrent::storage_error& e) {
+      LT_LOG("error saving download : storage error :download:%p path:%s : %s", request.second.download, request.second.path.c_str(), e.what());
+
+      if (m_last_storage_error_message + std::chrono::minutes(5) > torrent::this_thread::cached_time()) {
+        m_ignored_storage_error_count++;
+        continue;
+      }
+
+      lt_log_print(torrent::LOG_ERROR, "Storage errors saving session data for download: ignored:%u : %s", m_ignored_storage_error_count, e.what());
+
+      m_last_storage_error_message = torrent::this_thread::cached_time();
+      m_ignored_storage_error_count = 0;
+      continue;
+
+    } catch (torrent::internal_error& e) {
+      LT_LOG("error saving download : internal error : download:%p path:%s : %s", request.second.download, request.second.path.c_str(), e.what());
+      throw;
+
+    } catch (...) {
+      LT_LOG("error saving download : unknown error : download:%p path:%s", request.second.download, request.second.path.c_str());
+      throw;
+    }
+
+    LT_LOG("finished saving download : download:%p path:%s", request.second.download, request.second.path.c_str());
+  }
+
+  m_finished_saves.clear();
+}
+
+void
+SessionManager::flush_all_and_wait_unsafe(std::unique_lock<std::mutex>& lock) {
+  LT_LOG("flushing all pending saves", 0);
+
+  while (!m_save_requests.empty()) {
+    if (m_processing_saves.size() >= max_cleanup_processing) {
+      m_finished_condition.wait(lock);
+      continue;
+    }
+
+    process_next_save_request_unsafe();
+  }
+
+  while (!m_processing_saves.empty())
+    m_finished_condition.wait(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::replace_save_request_unsafe(SaveRequest& save_request) {
+  // Can be run in any thread.
+
+  auto itr = std::find_if(m_save_requests.begin(), m_save_requests.end(), [download = save_request.download](auto& req) {
+      return req.download == download;
+    });
+
+  if (itr == m_save_requests.end())
+    return false;
+
+  if (itr->path != save_request.path)
+    throw torrent::internal_error("SessionManager::replace_save_request_unsafe() path mismatch on replace: " + itr->path + " != " + save_request.path);
+
+  if (save_request.torrent_stream != nullptr)
+    throw torrent::internal_error("SessionManager::replace_save_request_unsafe() cannot replace full save requests.");
+
+  itr->rtorrent_stream   = std::move(save_request.rtorrent_stream);
+  itr->libtorrent_stream = std::move(save_request.libtorrent_stream);
+
+  return true;
+}
+
+bool
+SessionManager::remove_completely_unsafe(core::Download* download, std::unique_lock<std::mutex>& 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<std::mutex> pending_lock(m_pending_builds_mutex);
+
+    auto itr = std::remove_if(m_pending_builds.begin(), m_pending_builds.end(), [download](auto* req) {
+        return req == download;
+      });
+
+    if (itr == m_pending_builds.end())
+      return false;
+
+    m_pending_builds.erase(itr, m_pending_builds.end());
+    return true;
+  };
+
+  bool removed_something = false;
+
+  if (remove_pending())
+    removed_something = true;
+
+  if (remove_requests())
+    removed_something = true;
+
+  // This may block for a relatively long time if fdatasync is in use, however this is necessary.
+  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;
+
+    m_finished_condition.wait(lock);
+  }
+
+  // Since we're the main thread, no more requests for this download can be added.
+
+  // Checking active after remove_save_request_unsafe to ensure we're calling this after shutdown.
+  if (!m_active)
+    throw torrent::internal_error("SessionManager::remove_completely_unsafe() called while not active.");
+
+  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
+

src/session/session_manager.h

@@ -0,0 +1,126 @@
+#ifndef RTORRENT_SESSION_SESSION_MANAGER_H
+#define RTORRENT_SESSION_SESSION_MANAGER_H
+
+#include <condition_variable>
+#include <deque>
+#include <future>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <sstream>
+#include <string>
+#include <vector>
+#include <torrent/common.h>
+
+class Control;
+
+namespace core {
+class Download;
+}
+
+namespace utils {
+class Lockfile;
+}
+
+namespace session {
+
+class ThreadSession;
+
+struct SaveRequest {
+  core::Download*                    download;
+  std::string                        path;
+  std::unique_ptr<std::stringstream> torrent_stream;
+  std::unique_ptr<std::stringstream> rtorrent_stream;
+  std::unique_ptr<std::stringstream> libtorrent_stream;
+};
+
+class SessionManager {
+public:
+  typedef std::unique_ptr<std::stringstream> stream_ptr;
+
+  // TODO: This should depend on max open sockets / be configurable.
+
+  constexpr static int max_concurrent_requests   = 16;
+  constexpr static int max_concurrent_processing = 16;
+  constexpr static int max_cleanup_processing    = 64;
+
+  SessionManager(torrent::utils::Thread* thread);
+  ~SessionManager();
+
+  bool                is_used() const;
+
+  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                save_full_download(core::Download* download);
+  void                save_resume_download(core::Download* download);
+  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();
+
+  void                flush_all_pending_builds();
+
+private:
+  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();
+  void                process_finished_saves();
+
+  // Requires a higher number of open sockets, and should only be used during shutdown.
+  void                flush_all_and_wait_unsafe(std::unique_lock<std::mutex>& lock);
+
+  bool                replace_save_request_unsafe(SaveRequest& download);
+  bool                remove_completely_unsafe(core::Download* download, std::unique_lock<std::mutex>& lock);
+
+  torrent::utils::Thread* m_thread;
+
+  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{};
+
+  typedef std::pair<std::future<void>, SaveRequest> ProcessingSave;
+
+  std::deque<SaveRequest>     m_save_requests;
+  std::atomic<size_t>         m_save_request_counter{};
+  std::list<ProcessingSave>   m_processing_saves;
+  std::atomic<bool>           m_processing_saves_callback_scheduled{};
+  std::condition_variable     m_finished_condition;
+  std::vector<ProcessingSave> m_finished_saves;
+
+  std::unique_ptr<utils::Lockfile> m_lockfile;
+
+  std::chrono::microseconds   m_last_storage_error_message{};
+  unsigned int                m_ignored_storage_error_count{};
+
+  // Pending builds are only ever locked by main thread.
+  std::mutex                  m_pending_builds_mutex;
+  std::deque<core::Download*> 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 void        SessionManager::flush_all_pending_builds() { process_pending_resume_builds(true); }
+
+} // namespace session
+
+#endif // RTORRENT_SESSION_SESSION_MANAGER_H

src/session/thread_session.cc

@@ -0,0 +1,87 @@
+#include "config.h"
+
+#include "thread_session.h"
+
+#include <torrent/exceptions.h>
+
+#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<SessionManager>(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;
+}
+
+void
+ThreadSession::init_thread_pre_start() {
+  m_manager->start();
+}
+
+// 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.");
+
+  process_callbacks();
+
+  if ((m_flags & flag_do_shutdown)) {
+    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(10min);
+}
+
+} // 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<void ()>&& 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(); }
+std::string              session_path()                  { return session::ThreadSessionInternal::thread_session()->manager()->path(); }
+
+} // namespace session_thread

src/session/thread_session.h

@@ -0,0 +1,43 @@
+#ifndef RTORRENT_SESSION_THREAD_SESSION_H
+#define RTORRENT_SESSION_THREAD_SESSION_H
+
+#include <torrent/utils/thread.h>
+
+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"; }
+
+  SessionManager*     manager() const           { return m_manager.get(); }
+
+protected:
+  friend class ThreadSessionInternal;
+
+  ThreadSession() = default;
+
+  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;
+
+private:
+  static ThreadSession*   m_thread_session;
+
+  std::unique_ptr<SessionManager> m_manager;
+};
+
+} // namespace session
+
+#endif // RTORRENT_SESSION_THREAD_SESSION_H

src/thread_worker.cc

@@ -1,97 +0,0 @@
-#include "config.h"
-
-#include "thread_worker.h"
-
-#include <fcntl.h>
-#include <unistd.h>
-#include <cassert>
-#include <rak/path.h>
-#include <torrent/exceptions.h>
-
-#include "globals.h"
-#include "control.h"
-#include "core/manager.h"
-#include "rpc/scgi.h"
-#include "rpc/parse_commands.h"
-
-ThreadWorker::~ThreadWorker() = default;
-
-void
-ThreadWorker::init_thread() {
-  m_state = STATE_INITIALIZED;
-}
-
-void
-ThreadWorker::cleanup_thread() {
-  if (m_scgi != nullptr)
-    m_scgi.load()->deactivate();
-}
-
-bool
-ThreadWorker::set_scgi(rpc::SCgi* scgi) {
-  rpc::SCgi* expected = nullptr;
-
-  if (!m_scgi.compare_exchange_strong(expected, scgi))
-    return false;
-
-  change_rpc_log();
-
-  callback(nullptr, [this]() {
-      if (m_scgi == NULL)
-        throw torrent::internal_error("Tried to start SCGI but object was not present.");
-
-      m_scgi.load()->activate();
-    });
-
-  return true;
-}
-
-void
-ThreadWorker::set_rpc_log(const std::string& filename) {
-  callback(nullptr, [this, filename]() {
-      m_rpc_log_filename = filename;
-      change_rpc_log();
-    });
-}
-
-void
-ThreadWorker::change_rpc_log() {
-  if (scgi() == NULL)
-    return;
-
-  if (scgi()->log_fd() != -1) {
-    ::close(scgi()->log_fd());
-    scgi()->set_log_fd(-1);
-    control->core()->push_log("Closed RPC log.");
-  }
-
-  if (m_rpc_log_filename.empty())
-    return;
-
-  scgi()->set_log_fd(open(rak::path_expand(m_rpc_log_filename).c_str(), O_WRONLY | O_APPEND | O_CREAT, 0644));
-
-  if (scgi()->log_fd() == -1) {
-    control->core()->push_log_std("Could not open RPC log file '" + m_rpc_log_filename + "'.");
-    return;
-  }
-
-  control->core()->push_log_std("Logging RPC events to '" + m_rpc_log_filename + "'.");
-}
-
-void
-ThreadWorker::call_events() {
-  if ((m_flags & flag_do_shutdown)) {
-    if ((m_flags & flag_did_shutdown))
-      throw torrent::internal_error("Already trigged shutdown.");
-
-    m_flags |= flag_did_shutdown;
-    throw torrent::shutdown_exception();
-  }
-
-  process_callbacks();
-}
-
-std::chrono::microseconds
-ThreadWorker::next_timeout() {
-  return std::chrono::microseconds(10min);
-}

src/thread_worker.h

@@ -1,38 +0,0 @@
-#ifndef RTORRENT_THREAD_WORKER_H
-#define RTORRENT_THREAD_WORKER_H
-
-#include <atomic>
-#include <string>
-#include <torrent/utils/thread.h>
-
-namespace rpc {
-class SCgi;
-}
-
-class ThreadWorker : public torrent::utils::Thread {
-public:
-  ThreadWorker() = default;
-  ~ThreadWorker();
-
-  const char*         name() const override      { return "rtorrent scgi"; }
-
-  void                init_thread() override;
-  void                cleanup_thread() override;
-
-  rpc::SCgi*          scgi()                     { return m_scgi; }
-  bool                set_scgi(rpc::SCgi* scgi);
-
-  void                set_rpc_log(const std::string& filename);
-
-private:
-  void                task_touch_log();
-  void                change_rpc_log();
-
-  void                      call_events() override;
-  std::chrono::microseconds next_timeout() override;
-
-  std::atomic<rpc::SCgi*>   m_scgi{nullptr};
-  std::string               m_rpc_log_filename;
-};
-
-#endif

src/ui/root.cc

@@ -1,5 +1,7 @@
 #include "config.h"
 
+#include "ui/root.h"
+
 #include <fstream>
 #include <stdexcept>
 #include <string.h>
@@ -9,6 +11,8 @@
 #include <torrent/download/resource_manager.h>
 #include <torrent/utils/log.h>
 
+#include "control.h"
+#include "core/download_list.h"
 #include "core/manager.h"
 #include "display/frame.h"
 #include "display/window_http_queue.h"
@@ -18,12 +22,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 +377,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 +445,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 +502,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];
 }

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();

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 <sundell.software@gmail.com>
-
-
 // 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