Cleaned up changes to session manager.

Persist resume data on the first shutdown pass; consume futures with one error policy.

doc/rtorrent.rc

@@ -52,11 +52,11 @@
 # watch directory.
 #
 #directory.watch.ready = ./watch, load.start
-#schedule = watch_directory,5,5,load.start=./watch/*.torrent
+#schedule2 = watch_directory,5,5,load.start=./watch/*.torrent
 
 # Close torrents when disk-space is low.
 #
-#schedule = low_diskspace,5,60,close_low_diskspace=100M
+#schedule2 = low_diskspace,5,60,close_low_diskspace=100M
 
 # The IP address reported to the tracker.
 #
@@ -100,8 +100,8 @@
 
 # Alternative calls to bind and IP that should handle dynamic IP's.
 #
-#schedule = ip_tick,0,1800,ip=rakshasa
-#schedule = bind_tick,0,1800,bind=rakshasa
+#schedule2 = ip_tick,0,1800,ip=rakshasa
+#schedule2 = bind_tick,0,1800,bind=rakshasa
 
 # Encryption options, set to none (default) or any combination of the following:
 # allow_incoming, try_outgoing, require, require_RC4, enable_retry, prefer_plaintext
@@ -145,4 +145,4 @@
 #
 #network.scgi.open_port = "127.0.0.1:5000"
 #network.scgi.open_local = (cat,(session.path),/rpc.sock)
-#schedule = socket_chmod, 0, 0, "execute.nothrow=chmod,770,(cat,(session.path),/rpc.sock)"
+#schedule2 = socket_chmod, 0, 0, "execute.nothrow=chmod,770,(cat,(session.path),/rpc.sock)"

doc/rtorrent.rc-example

@@ -30,6 +30,7 @@ network.listen.port.random.set = no
 # (conservative settings for 'private' trackers, change for 'public')
 dht.mode.set = disable
 protocol.pex.set = no
+trackers.use_udp.set = no
 
 # Peer settings
 throttle.max_uploads.set = 100
@@ -41,22 +42,14 @@ throttle.min_peers.seed.set = 30
 throttle.max_peers.seed.set = 80
 trackers.numwant.set = 80
 
-# Select encryption for handshake and connection:
-#
-# protocol.encryption.set = {deny,allow,prefer,require}
-# protocol.encryption.set = handshake_{deny,allow,prefer,require}, stream_{deny,allow,prefer,require}
-#
-#protocol.encryption.set = deny
-#protocol.encryption.set = require
-#protocol.encryption.set = handshake_prefer, stream_deny
-#protocol.encryption.set = handshake_require, stream_prefer
+protocol.encryption.set = allow_incoming,try_outgoing,enable_retry
 
 # Limits for file handle resources, this is optimized for
 # an `ulimit` of 1024 (a common default). You MUST leave
 # a ceiling of handles reserved for rTorrent's internal needs!
-
-# Deprecated, see: https://github.com/rakshasa/rtorrent/wiki/Socket-Manager-and-Resource-Allocation
-
+network.http.max_open.set = 50
+network.max_open_files.set = 600
+network.max_open_sockets.set = 300
 
 # Memory resource usage (increase if you have a large number of items loaded,
 # and/or the available resources to spend)
@@ -72,10 +65,11 @@ execute.nothrow = sh, -c, (cat, "echo >",\
     (session.path), "rtorrent.pid", " ", (system.pid))
 
 # Other operational settings (check & adapt)
+encoding.add = utf8
 system.umask.set = 0027
 system.cwd.set = (directory.default)
 network.http.dns_cache_timeout.set = 25
-schedule = monitor_diskspace, 15, 60, ((close_low_diskspace, 1000M))
+schedule2 = monitor_diskspace, 15, 60, ((close_low_diskspace, 1000M))
 ##pieces.hash.on_completion.set = no
 ##view.sort_current = seeding, greater=d.ratio=
 ##keys.layout.set = qwerty
@@ -94,8 +88,8 @@ method.insert = d.data_path, simple,\
 method.insert = d.session_file, simple, "cat=(session.path), (d.hash), .torrent"
 
 # Watch directories (add more as you like, but use unique schedule names)
-schedule = watch_start, 10, 10, ((load.start_verbose, (cat, (cfg.watch), "start/*.torrent")))
-schedule = watch_load, 11, 10, ((load.verbose, (cat, (cfg.watch), "load/*.torrent")))
+schedule2 = watch_start, 10, 10, ((load.start_verbose, (cat, (cfg.watch), "start/*.torrent")))
+schedule2 = watch_load, 11, 10, ((load.verbose, (cat, (cfg.watch), "load/*.torrent")))
 
 # Run the rTorrent process as a daemon in the background
 # (and control via XMLRPC sockets)

doc/rtorrent.rc.lua-example

@@ -60,6 +60,7 @@ rc.network.listen.port.random = false
 -- (conservative settings for 'private' trackers, change for 'public')
 rc.dht.mode = 'disable'
 rc.protocol.pex = false
+rc.trackers.use_udp = false
 
 -- Peer settings
 rc.throttle.max_uploads = 100
@@ -71,11 +72,14 @@ rc.throttle.min_peers.seed = 30
 rc.throttle.max_peers.seed = 80
 rc.trackers.numwant = 80
 
+rc.protocol.encryption.set('allow_incoming', 'try_outgoing', 'enable_retry')
+
 -- Limits for file handle resources, this is optimized for
 -- an `ulimit` of 1024 (a common default). You MUST leave
 -- a ceiling of handles reserved for rTorrent's internal needs!
-
--- Deprecated, see: https://github.com/rakshasa/rtorrent/wiki/Socket-Manager-and-Resource-Allocation
+rc.network.http.max_open = 50
+rc.network.max_open_files = 600
+rc.network.max_open_sockets = 300
 
 -- Memory resource usage (increase if you have a large number of items loaded,
 -- and/or the available resources to spend)
@@ -93,10 +97,11 @@ rc.execute.nothrow(
 }))
 
 -- Other operational settings (check & adapt)
+rc.encoding.add('utf8')
 rc.system.umask = 0027
 rc.system.cwd = rc.directory.default()
 rc.network.http.dns_cache_timeout = 25
-rc.schedule('monitor_diskspace', '15', '60', 'close_low_diskspace=1000M')
+rc.schedule2('monitor_diskspace', '15', '60', 'close_low_diskspace=1000M')
 --rc.pieces.hash.on_completion = false
 --rc.view.sort_current('seeding', 'greater=d.ratio=')
 --rc.keys.layout = 'qwerty'
@@ -116,8 +121,8 @@ rc.method.insert('d.data_path', 'simple',
 rc.method.insert('d.session_file', 'simple', 'cat=(session.path), (d.hash), .torrent')
 
 -- Watch directories (add more as you like, but use unique schedule names)
-rc.schedule('watch_start', '10', '10', 'load.start_verbose=(cat, (cfg.watch), "start/*.torrent")')
-rc.schedule('watch_load', '11', '10', 'load.verbose=(cat, (cfg.watch), "load/*.torrent")')
+rc.schedule2('watch_start', '10', '10', 'load.start_verbose=(cat, (cfg.watch), "start/*.torrent")')
+rc.schedule2('watch_load', '11', '10', 'load.verbose=(cat, (cfg.watch), "load/*.torrent")')
 
 -- Run the rTorrent process as a daemon in the background
 -- (and control via XMLRPC sockets)

src/control.cc

@@ -10,6 +10,7 @@
 #include <torrent/utils/directory_events.h>
 
 #include "core/dht_manager.h"
+#include "core/download_list.h"
 #include "core/http_queue.h"
 #include "core/manager.h"
 #include "core/view_manager.h"
@@ -137,6 +138,15 @@ Control::handle_shutdown() {
   if (scgi_thread::thread()->is_active())
     scgi_thread::thread()->stop_thread_wait();
 
+  if (!m_shutdown_session_saved) {
+    // Flush the latest state before stopped announces can consume the timeout. A shutdown hook may
+    // already have queued these resume builds.
+    m_core->download_list()->session_save();
+    session_thread::manager()->flush_shutdown_saves();
+
+    m_shutdown_session_saved = true;
+  }
+
   if (!m_shutdown_quick) {
     torrent::runtime::network_manager()->listen_close();
     torrent::runtime::shutdown();

src/control.h

@@ -112,6 +112,8 @@ private:
 
   int                 m_clear_requests_count{};
 
+  bool                m_shutdown_session_saved{};
+
   align_cacheline
 
   std::atomic<bool>   m_shutdown_received{};

src/session/session_manager.cc

@@ -405,42 +405,67 @@ SessionManager::process_finished_saves() {
   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_ERRORS, "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());
-  }
+  for (auto& request : m_finished_saves)
+    consume_finished_save_unsafe(request);
 
   m_finished_saves.clear();
 
   callback_save_request();
 }
 
+// Returns true when a storage error was suppressed.
+bool
+SessionManager::report_storage_error_unsafe(const std::string& msg) {
+  if (m_last_storage_error_message + std::chrono::minutes(5) > torrent::this_thread::cached_time()) {
+    m_ignored_storage_error_count++;
+    return true;
+  }
+
+  lt_log_print(torrent::LOG_ERRORS, "Storage errors saving session data for download: ignored:%u : %s", m_ignored_storage_error_count, msg.c_str());
+
+  m_last_storage_error_message  = torrent::this_thread::cached_time();
+  m_ignored_storage_error_count = 0;
+
+  return false;
+}
+
+// The one policy for a finished save, on the session thread and in shutdown flushes: storage errors
+// go to report_storage_error_unsafe(), internal and unknown errors propagate.
+//
+// Returns true when a storage error was suppressed.
+bool
+SessionManager::consume_finished_save_unsafe(ProcessingSave& request) {
+  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());
+    return report_storage_error_unsafe(e.what());
+
+  } 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());
+  return false;
+}
+
+void
+SessionManager::flush_shutdown_saves() {
+  assert(torrent::this_thread::thread() == torrent::main_thread::thread());
+
+  flush_all_pending_builds();
+
+  std::unique_lock<std::mutex> lock(m_mutex);
+
+  flush_all_and_wait_unsafe(lock);
+}
+
 void
 SessionManager::flush_all_and_wait_unsafe(std::unique_lock<std::mutex>& lock) {
   LT_LOG("flushing all pending saves", 0);
@@ -459,11 +484,20 @@ SessionManager::flush_all_and_wait_unsafe(std::unique_lock<std::mutex>& lock) {
   while (!m_processing_saves.empty())
     m_finished_condition.wait(lock);
 
+  unsigned int suppressed = 0;
+
   for (auto& request : m_finished_saves)
-    LT_LOG("finished saving download : download:%p path:%s", request.second.download, request.second.path.c_str());
+    suppressed += consume_finished_save_unsafe(request) ? 1 : 0;
 
   m_finished_saves.clear();
 
+  if (m_ignored_storage_error_count != 0) {
+    lt_log_print(torrent::LOG_ERRORS, "Storage errors flushing session data: ignored:%u previous:%u",
+                 suppressed, m_ignored_storage_error_count - suppressed);
+
+    m_ignored_storage_error_count = 0;
+  }
+
   LT_LOG("flushed all pending saves", 0);
 }
 

src/session/session_manager.h

@@ -73,8 +73,10 @@ protected:
   void                cleanup();
 
   void                flush_all_pending_builds();
+  void                flush_shutdown_saves();
 
 private:
+
   void                callback_pending_builds();
   void                callback_save_request();
   void                callback_finished_saves();
@@ -92,6 +94,9 @@ private:
 
   using ProcessingSave = std::pair<std::future<void>, SaveRequest>;
 
+  bool                consume_finished_save_unsafe(ProcessingSave& request);
+  bool                report_storage_error_unsafe(const std::string& msg);
+
   torrent::system::Thread*     m_thread;
   torrent::system::callback_id m_callback_id;