Fixed session save getting stuck.

src/session/session_manager.cc

@@ -14,6 +14,11 @@
 #define LT_LOG(log_fmt, ...)                                            \
   lt_log_print(torrent::LOG_SESSION_EVENTS, "session-events: " log_fmt, __VA_ARGS__);
 
+// 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 {
 
 SessionManager::SessionManager(torrent::utils::Thread* thread)
@@ -68,19 +73,20 @@ SessionManager::save_resume_download(core::Download* download) {
   {
     std::unique_lock<std::mutex> lock(m_pending_builds_mutex);
 
+    // TODO: This is under the wrong lock.
     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()) {
+      // TODO: Do we need to make sure we're getting processed?
       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);
 
+    callback_pending_builds();
+
     LT_LOG("build of resume save data queued : download:%p", download);
   }
 }
@@ -123,8 +129,7 @@ SessionManager::save_full_download(core::Download* download) {
     LT_LOG("queued new full save request : download:%p", download);
   }
 
-  if (!m_processing_saves_callback_scheduled.exchange(true))
+  callback_save_request();
-    session_thread::callback(this, [this]() { process_save_request(); });
 }
 
 void
@@ -221,7 +226,40 @@ SessionManager::cleanup() {
 }
 
 void
-SessionManager::process_pending_resume_builds(bool is_flushing) {
+SessionManager::callback_pending_builds() {
+  if (m_save_request_counter >= max_concurrent_requests)
+    return;
+
+  if (m_callback_scheduled_process_pending_builds.exchange(true))
+    return;
+
+  torrent::main_thread::callback(this, [this]() { process_pending_builds(false); });
+}
+
+void
+SessionManager::callback_save_request() {
+  if (m_save_request_counter == 0)
+    return;
+
+  if (m_processing_save_counter >= max_concurrent_processing)
+    return;
+
+  if (m_callback_scheduled_process_saves_request.exchange(true))
+    return;
+
+  session_thread::callback(this, [this]() { process_save_request(); });
+}
+
+void
+SessionManager::callback_finished_saves() {
+  if (m_callback_scheduled_process_finished_saves.exchange(true))
+    return;
+
+  session_thread::callback(this, [this]() { process_finished_saves(); });
+}
+
+void
+SessionManager::process_pending_builds(bool is_flushing) {
   assert(torrent::this_thread::thread() == torrent::main_thread::thread());
 
   std::vector<SaveRequest> requests;
@@ -231,6 +269,8 @@ SessionManager::process_pending_resume_builds(bool is_flushing) {
   {
     std::unique_lock<std::mutex> lock(m_pending_builds_mutex);
 
+    m_callback_scheduled_process_pending_builds = false;
+
     while (!m_pending_builds.empty()) {
       if (!is_flushing && m_save_request_counter + requests.size() >= max_concurrent_requests)
         break;
@@ -272,15 +312,14 @@ SessionManager::process_pending_resume_builds(bool is_flushing) {
       }
 
       m_save_requests.push_back(std::move(save_request));
+      m_save_request_counter = m_save_requests.size();
 
       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))
+  if (!is_flushing)
-    session_thread::callback(this, [this]() { process_save_request_with_pending_callback(); });
+    callback_save_request();
 }
 
 void
@@ -290,25 +329,23 @@ SessionManager::process_save_request() {
   if (m_path.empty())
     throw torrent::internal_error("SessionManager::process_save_request() called with empty path.");
 
-  std::unique_lock<std::mutex> lock(m_mutex);
+  {
+    std::unique_lock<std::mutex> lock(m_mutex);
 
-  if (!m_active)
+    m_callback_scheduled_process_saves_request = false;
-    throw torrent::internal_error("SessionManager::process_save_request() called while not active.");
 
-  while (!m_save_requests.empty() && m_processing_saves.size() < max_concurrent_requests)
+    if (!m_active)
-    process_next_save_request_unsafe();
+      throw torrent::internal_error("SessionManager::process_save_request() called while not active.");
 
-  m_processing_saves_callback_scheduled = false;
+    while (!m_save_requests.empty()) {
-}
+      if (m_processing_saves.size() >= max_concurrent_processing)
+        break;
 
-void
+      process_next_save_request_unsafe();
-SessionManager::process_save_request_with_pending_callback() {
+    }
-  process_save_request();
+  }
 
-  std::unique_lock<std::mutex> lock(m_pending_builds_mutex);
+  callback_pending_builds();
-
-  if (!m_pending_builds.empty())
-    torrent::main_thread::callback(this, [this]() { process_pending_resume_builds(false); });
 }
 
 void
@@ -319,19 +356,20 @@ SessionManager::process_next_save_request_unsafe() {
   m_save_request_counter = m_save_requests.size();
 
   auto itr = m_processing_saves.insert(m_processing_saves.end(), ProcessingSave{});
+  m_processing_save_counter = m_processing_saves.size();
 
   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())
+          callback_finished_saves();
-            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);
+          m_processing_save_counter = m_processing_saves.size();
         };
 
       try {
@@ -346,6 +384,8 @@ SessionManager::process_next_save_request_unsafe() {
 
       cleanup_fn();
     });
+
+  LT_LOG("started save of download : download:%p path:%s", itr->second.download, itr->second.path.c_str());
 }
 
 void
@@ -354,6 +394,8 @@ SessionManager::process_finished_saves() {
 
   std::unique_lock<std::mutex> lock(m_mutex);
 
+  m_callback_scheduled_process_finished_saves = false;
+
   if (!m_active)
     throw torrent::internal_error("SessionManager::process_finished_saves() called while not active.");
 
@@ -373,6 +415,7 @@ SessionManager::process_finished_saves() {
 
       m_last_storage_error_message = torrent::this_thread::cached_time();
       m_ignored_storage_error_count = 0;
+
       continue;
 
     } catch (torrent::internal_error& e) {
@@ -388,14 +431,18 @@ SessionManager::process_finished_saves() {
   }
 
   m_finished_saves.clear();
+
+  callback_save_request();
 }
 
 void
 SessionManager::flush_all_and_wait_unsafe(std::unique_lock<std::mutex>& lock) {
   LT_LOG("flushing all pending saves", 0);
 
+  // Caller already ensured pending builds are empty.
+
   while (!m_save_requests.empty()) {
-    if (m_processing_saves.size() >= max_cleanup_processing) {
+    if (m_processing_save_counter >= max_cleanup_processing) {
       m_finished_condition.wait(lock);
       continue;
     }
@@ -430,14 +477,17 @@ SessionManager::replace_save_request_unsafe(SaveRequest& save_request) {
 
   // If the existing request is a full save (created during torrent initialization),
   // keep its torrent_stream but update the resume streams with newer data.
+  //
   // Otherwise, replace all streams.
+
   if (itr->torrent_stream != nullptr) {
     // Keep existing full-save torrent stream, only update resume streams
     if (save_request.rtorrent_stream != nullptr)
       itr->rtorrent_stream = std::move(save_request.rtorrent_stream);
-    
+
     if (save_request.libtorrent_stream != nullptr)
       itr->libtorrent_stream = std::move(save_request.libtorrent_stream);
+
   } else {
     // No full save pending, replace everything
     itr->torrent_stream    = std::move(save_request.torrent_stream);
@@ -452,18 +502,6 @@ 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);
 
@@ -478,13 +516,28 @@ SessionManager::remove_completely_unsafe(core::Download* download, std::unique_l
     return true;
   };
 
+  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());
+    m_save_request_counter = m_save_requests.size();
+    return true;
+  };
+
   bool removed_something = false;
 
   if (remove_pending())
     removed_something = true;
 
-  if (remove_requests())
+  if (remove_requests()) {
     removed_something = true;
+    callback_pending_builds();
+  }
 
   // This may block for a relatively long time if fdatasync is in use, however this is necessary.
   while (true) {
@@ -507,9 +560,5 @@ SessionManager::remove_completely_unsafe(core::Download* download, std::unique_l
   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

@@ -74,9 +74,12 @@ protected:
   void                flush_all_pending_builds();
 
 private:
-  void                process_pending_resume_builds(bool is_flushing);
+  void                callback_pending_builds();
+  void                callback_save_request();
+  void                callback_finished_saves();
+
+  void                process_pending_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();
 
@@ -101,10 +104,14 @@ private:
   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::atomic<size_t>         m_processing_save_counter{};
   std::condition_variable     m_finished_condition;
   std::vector<ProcessingSave> m_finished_saves;
 
+  std::atomic<bool>           m_callback_scheduled_process_pending_builds{};
+  std::atomic<bool>           m_callback_scheduled_process_saves_request{};
+  std::atomic<bool>           m_callback_scheduled_process_finished_saves{};
+
   std::unique_ptr<utils::Lockfile> m_lockfile;
 
   std::chrono::microseconds   m_last_storage_error_message{};
@@ -119,7 +126,7 @@ inline bool        SessionManager::is_used() const            { return !m_path.e
 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); }
+inline void        SessionManager::flush_all_pending_builds() { process_pending_builds(true); }
 
 } // namespace session