Session saving of resume data is added to a pre-queue.

src/session/download_storer.cc

@@ -2,6 +2,9 @@
 
 #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>
@@ -79,9 +82,24 @@ DownloadStorer::build_path(const std::string& 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);
+  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";
+
+  ::unlink(libtorrent_path.c_str());
+  ::unlink(rtorrent_path.c_str());
+  ::unlink(torrent_path.c_str());
+}
+
+namespace {
+
 bool
 is_correct_format(const std::string& f) {
   if (f.size() != 48 || f.substr(40) != ".torrent")
@@ -95,6 +113,87 @@ is_correct_format(const std::string& f) {
   return true;
 }
 
+bool
+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()) {
+    // LT_LOG("failed to open file for writing : path:%s", path.c_str());
+    return false;
+  }
+
+  output << stream.rdbuf();
+
+  if (!output.good()) {
+    // LT_LOG("failed to write stream to file : path:%s", path.c_str());
+    return false;
+  }
+
+  output.close();
+
+  // Ensure that the new file is actually written to the disk
+  int fd = ::open(path.c_str(), O_WRONLY);
+
+  if (fd < 0) {
+    // LT_LOG("failed to open file descriptor for fdatasync : path:%s", path.c_str());
+    return false;
+  }
+
+  if (use_fsyncdisk) {
+#ifdef __APPLE__
+    ::fsync(fd);
+#else
+    ::fdatasync(fd);
+#endif
+  }
+
+  ::close(fd);
+  return true;
+}
+
+} // 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) {
+  // LT_LOG("saving download : download:%p path:%s", download, path.c_str());
+
+  auto torrent_path    = path;
+  auto libtorrent_path = path + ".libtorrent_resume";
+  auto rtorrent_path   = path + ".rtorrent";
+
+  if (torrent_stream) {
+    if (!save_stream(torrent_path + ".new", use_fsyncdisk, torrent_stream))
+      return;
+  }
+
+  if (!save_stream(libtorrent_path + ".new", use_fsyncdisk, libtorrent_stream))
+    return;
+
+  if (!save_stream(rtorrent_path + ".new", use_fsyncdisk, rtorrent_stream))
+    return;
+
+  if (torrent_stream) {
+    if (::rename((torrent_path + ".new").c_str(), torrent_path.c_str()) == -1) {
+      // LT_LOG("failed to rename torrent file : %s", torrent_path.c_str());
+      return;
+    }
+  }
+
+  if (::rename((libtorrent_path + ".new").c_str(), libtorrent_path.c_str()) == -1) {
+    // LT_LOG("failed to rename libtorrent resume file : %s", libtorrent_path.c_str());
+    return;
+  }
+
+  if (::rename((rtorrent_path + ".new").c_str(), rtorrent_path.c_str()) == -1) {
+    // LT_LOG("failed to rename rtorrent resume file : %s", rtorrent_path.c_str());
+    return;
+  }
+}
+
 utils::Directory
 DownloadStorer::get_formated_entries(const std::string& session_path) {
   if (session_path.empty())

src/session/download_storer.h

@@ -11,7 +11,7 @@ class Download;
 }
 
 namespace utils {
-  class Directory;
+class Directory;
 }
 
 namespace session {
@@ -22,16 +22,27 @@ public:
 
   core::Download*     download() const { return m_download; }
 
-  void                build_streams(bool skip_static);
   std::string         build_path(const std::string& session_path);
 
-  auto                torrent_stream()    { return std::move(m_torrent_stream); }
-  auto                rtorrent_stream()   { return std::move(m_rtorrent_stream); }
-  auto                libtorrent_stream() { return std::move(m_libtorrent_stream); }
+  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;

src/session/session_manager.cc

@@ -3,10 +3,6 @@
 #include "session/session_manager.h"
 
 #include <cassert>
-#include <cerrno>
-#include <fstream>
-#include <fcntl.h>
-#include <unistd.h>
 #include <torrent/exceptions.h>
 #include <torrent/utils/log.h>
 
@@ -19,8 +15,6 @@
 
 namespace session {
 
-// TODO: Add session save scheduler that runs in main thread and passes download one-by-one to session manager.
-
 SessionManager::SessionManager(torrent::utils::Thread* thread)
   : m_thread(thread),
     m_lockfile(std::make_unique<utils::Lockfile>()) {
@@ -61,10 +55,35 @@ SessionManager::set_use_lock(bool use_lock) {
   m_use_lock = use_lock;
 }
 
-// TODO: Add separate mutex for queuing save requests in session save.
+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);
+
+    LT_LOG("requesting resume save : download:%p", download);
+
+    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 pending resume save : download:%p", download);
+      return;
+    }
+
+    if (m_pending_builds.empty())
+      torrent::main_thread::callback(this, [this]() { process_pending_resume_builds(); });
+
+    m_pending_builds.push_back(download);
+  }
+}
 
 void
-SessionManager::save_download(core::Download* download, bool skip_static) {
+SessionManager::save_full_download(core::Download* download) {
   assert(torrent::this_thread::thread() == torrent::main_thread::thread());
 
   if (m_path.empty())
@@ -72,7 +91,7 @@ SessionManager::save_download(core::Download* download, bool skip_static) {
 
   DownloadStorer storer(download);
 
-  storer.build_streams(skip_static);
+  storer.build_full_streams();
 
   auto save_request = SaveRequest{
     download,
@@ -90,17 +109,19 @@ SessionManager::save_download(core::Download* download, bool skip_static) {
     if (!m_active)
       throw torrent::internal_error("SessionManager::save_download() called while not active.");
 
-    if (remove_save_request_unsafe(download, lock))
-      LT_LOG("replacing pending save request : download:%p", download);
+    if (remove_or_replace_unsafe(save_request)) {
+      LT_LOG("updated pending full save request : download:%p", download);
+      throw torrent::internal_error("SessionManager::save_full_download() replacing existing save request, not supported?");
 
-    m_save_requests.push_back(std::move(save_request));
+    } else {
+      LT_LOG("queued new full save request : download:%p", download);
+      m_save_requests.push_back(std::move(save_request));
+    }
   }
 
   session_thread::callback(this, [this]() { process_save_request(); });
 }
 
-// TODO: Move various low-level stuff to DownloadStorer.
-
 void
 SessionManager::remove_download(core::Download* download) {
   assert(torrent::this_thread::thread() == torrent::main_thread::thread());
@@ -108,23 +129,15 @@ SessionManager::remove_download(core::Download* download) {
   if (m_path.empty())
     return;
 
-  auto base_path = DownloadStorer(download).build_path(m_path);
-
   std::unique_lock<std::mutex> lock(m_mutex);
 
   if (!m_active)
     throw torrent::internal_error("SessionManager::remove_download() called while not active.");
 
-  if (remove_save_request_unsafe(download, lock))
+  if (remove_completely_unsafe(download, lock))
     LT_LOG("canceled pending save request : download:%p", download);
 
-  auto torrent_path    = base_path;
-  auto libtorrent_path = base_path + ".libtorrent_resume";
-  auto rtorrent_path   = base_path + ".rtorrent";
-
-  ::unlink(libtorrent_path.c_str());
-  ::unlink(rtorrent_path.c_str());
-  ::unlink(torrent_path.c_str());
+  DownloadStorer(download).unlink_files(m_path);
 }
 
 void
@@ -190,6 +203,52 @@ SessionManager::cleanup() {
   LT_LOG("session manager cleaned up", 0);
 
   session_thread::cancel_callback(this);
+  torrent::main_thread::cancel_callback(this);
+}
+
+void
+SessionManager::process_pending_resume_builds() {
+  assert(torrent::this_thread::thread() == torrent::main_thread::thread());
+
+  std::unique_lock<std::mutex> lock(m_pending_builds_mutex);
+
+  if (!m_active)
+    throw torrent::internal_error("SessionManager::process_pending_builds() called while not active.");
+
+  while (!m_pending_builds.empty() && m_pending_builds.size() < max_concurrent_requests) {
+    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()
+    };
+
+    {
+      std::unique_lock<std::mutex> save_lock(m_mutex);
+
+      if (!m_active)
+        throw torrent::internal_error("SessionManager::process_pending_builds() called while not active.");
+
+      if (remove_or_replace_unsafe(save_request)) {
+        LT_LOG("updated pending resume save request : download:%p", download);
+      } else {
+        LT_LOG("queued new resume save request : download:%p", download);
+        m_save_requests.push_back(std::move(save_request));
+      }
+    }
+
+    session_thread::callback(this, [this]() { process_save_request_with_pending_callback(); });
+  }
 }
 
 void
@@ -197,7 +256,7 @@ SessionManager::process_save_request() {
   assert(m_thread == torrent::this_thread::thread());
 
   if (m_path.empty())
-    return;
+    throw torrent::internal_error("SessionManager::process_save_request() called with empty path.");
 
   std::unique_lock<std::mutex> lock(m_mutex);
 
@@ -213,6 +272,16 @@ SessionManager::process_save_request() {
     session_thread::callback(this, [this]() { process_save_request(); });
 }
 
+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(); });
+}
+
 void
 SessionManager::process_next_save_request_unsafe() {
   auto request = std::move(m_save_requests.front());
@@ -225,7 +294,10 @@ SessionManager::process_next_save_request_unsafe() {
       // TODO: Properly handle errors here, and report back to session thread.
       // TODO: Consider adding a failed_saves with error info.
 
-      save_download_unsafe(itr->second);
+      DownloadStorer::save_and_move_streams(itr->second.path, m_use_fsyncdisk,
+                                            *itr->second.torrent_stream,
+                                            *itr->second.rtorrent_stream,
+                                            *itr->second.libtorrent_stream);
 
       {
         std::unique_lock<std::mutex> lock(m_mutex);
@@ -291,10 +363,35 @@ SessionManager::flush_all_and_wait_unsafe(std::unique_lock<std::mutex>& lock) {
 }
 
 bool
-SessionManager::remove_save_request_unsafe(core::Download* download, std::unique_lock<std::mutex>& lock) {
+SessionManager::remove_or_replace_unsafe(SaveRequest& save_request) {
   // Can be run in any thread.
 
-  {
+  auto itr = std::remove_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::remove_or_replace_unsafe() path mismatch on replace.");
+
+  // If this is a full save, we replace just the resume data.
+  itr->rtorrent_stream   = std::move(save_request.rtorrent_stream);
+  itr->libtorrent_stream = std::move(save_request.libtorrent_stream);
+
+  // Checking active after remove_save_request_unsafe to ensure we're calling this after shutdown.
+  if (!m_active)
+    throw torrent::internal_error("SessionManager::remove_or_replace_unsafe() called while not active.");
+
+  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;
       });
@@ -303,9 +400,31 @@ SessionManager::remove_save_request_unsafe(core::Download* download, std::unique
       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;
 
-  // check if in processing saves, then wait for it to finish using wait_for_one_save_unsafe()
   while (true) {
     auto itr = std::find_if(m_processing_saves.begin(), m_processing_saves.end(), [download](auto& req) {
         return req.second.download == download;
@@ -317,98 +436,20 @@ SessionManager::remove_save_request_unsafe(core::Download* download, std::unique
     wait_for_one_save_unsafe(lock);
   }
 
+  // TODO: Do we need this?
+  if (remove_requests())
+    removed_something = true;
+
   // Checking active after remove_save_request_unsafe to ensure we're calling this after shutdown.
   if (!m_active)
-    throw torrent::internal_error("SessionManager::remove_save_request_unsafe() called while not active");
+    throw torrent::internal_error("SessionManager::remove_completely_unsafe() called while not active.");
 
-  return true;
+  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 nad try again.
 
-void
-SessionManager::save_download_unsafe(const SaveRequest& request) {
-  // LT_LOG("saving download : download:%p path:%s", request.download, request.path.c_str());
-
-  if (m_path.empty())
-    throw torrent::internal_error("SessionManager::save_download_unsafe() called with empty session path.");
-
-  auto torrent_path    = request.path;
-  auto libtorrent_path = request.path + ".libtorrent_resume";
-  auto rtorrent_path   = request.path + ".rtorrent";
-
-  if (request.torrent_stream) {
-    if (!save_download_stream_unsafe(torrent_path + ".new", request.torrent_stream))
-      return;
-  }
-
-  if (!save_download_stream_unsafe(libtorrent_path + ".new", request.libtorrent_stream))
-    return;
-
-  if (!save_download_stream_unsafe(rtorrent_path + ".new", request.rtorrent_stream))
-    return;
-
-  if (request.torrent_stream) {
-    if (::rename((torrent_path + ".new").c_str(), torrent_path.c_str()) == -1) {
-      // LT_LOG("failed to rename torrent file : %s", torrent_path.c_str());
-      return;
-    }
-  }
-
-  if (::rename((libtorrent_path + ".new").c_str(), libtorrent_path.c_str()) == -1) {
-    // LT_LOG("failed to rename libtorrent resume file : %s", libtorrent_path.c_str());
-    return;
-  }
-
-  if (::rename((rtorrent_path + ".new").c_str(), rtorrent_path.c_str()) == -1) {
-    // LT_LOG("failed to rename rtorrent resume file : %s", rtorrent_path.c_str());
-    return;
-  }
-}
-
-// TODO: Rewrite to be all done in std::async, and from rdbuf directly to fd to avoid re-opening.
-
-bool
-SessionManager::save_download_stream_unsafe(const std::string& path, const std::unique_ptr<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()) {
-    // LT_LOG("failed to open file for writing : path:%s", path.c_str());
-    return false;
-  }
-
-  output << stream->rdbuf();
-
-  if (!output.good()) {
-    // LT_LOG("failed to write stream to file : path:%s", path.c_str());
-    return false;
-  }
-
-  output.close();
-
-  // Ensure that the new file is actually written to the disk
-  int fd = ::open(path.c_str(), O_WRONLY);
-
-  if (fd < 0) {
-    // LT_LOG("failed to open file descriptor for fdatasync : path:%s", path.c_str());
-    return false;
-  }
-
-  if (m_use_fsyncdisk) {
-#ifdef __APPLE__
-    ::fsync(fd);
-#else
-    ::fdatasync(fd);
-#endif
-  }
-
-  ::close(fd);
-  return true;
-}
-
 }  // namespace session
 

src/session/session_manager.h

@@ -37,8 +37,12 @@ public:
   typedef std::unique_ptr<std::stringstream> stream_ptr;
 
   // TODO: This should depend on max open sockets / be configurable.
-  constexpr static int max_concurrent_saves = 16;
-  constexpr static int max_cleanup_saves    = 64;
+
+  // TODO: max_concurrent_requests should be checked before something, review.
+
+  constexpr static int max_concurrent_requests = 16;
+  constexpr static int max_concurrent_saves    = 16;
+  constexpr static int max_cleanup_saves       = 64;
 
   SessionManager(torrent::utils::Thread* thread);
   ~SessionManager();
@@ -54,8 +58,8 @@ public:
   bool                use_lock() const;
   void                set_use_lock(bool use_lock);
 
-  void                save_full_download(core::Download* download)   { save_download(download, false); }
-  void                save_resume_download(core::Download* download) { save_download(download, true);  }
+  void                save_full_download(core::Download* download);
+  void                save_resume_download(core::Download* download);
   void                remove_download(core::Download* download);
 
 protected:
@@ -68,7 +72,9 @@ protected:
   void                cleanup();
 
 private:
+  void                process_pending_resume_builds();
   void                process_save_request();
+  void                process_save_request_with_pending_callback();
   void                process_next_save_request_unsafe();
   void                process_finished_saves();
 
@@ -77,10 +83,8 @@ private:
   // 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                remove_save_request_unsafe(core::Download* download, std::unique_lock<std::mutex>& lock);
-
-  void                save_download_unsafe(const SaveRequest& request);
-  bool                save_download_stream_unsafe(const std::string& path, const std::unique_ptr<std::stringstream>& stream);
+  bool                remove_or_replace_unsafe(SaveRequest& download);
+  bool                remove_completely_unsafe(core::Download* download, std::unique_lock<std::mutex>& lock);
 
   torrent::utils::Thread* m_thread;
 
@@ -94,14 +98,16 @@ private:
 
   typedef std::pair<std::future<void>, SaveRequest> ProcessingSave;
 
-  // TODO: Add deque of DownloadStorers requests that have yet to build streams.
-
   std::deque<SaveRequest>     m_save_requests;
   std::list<ProcessingSave>   m_processing_saves;
   std::condition_variable     m_finished_condition;
   std::vector<ProcessingSave> m_finished_saves;
 
   std::unique_ptr<utils::Lockfile> m_lockfile;
+
+  // 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(); }