diff --git a/src/session/download_storer.cc b/src/session/download_storer.cc index e7328c46..5a17b04f 100644 --- a/src/session/download_storer.cc +++ b/src/session/download_storer.cc @@ -2,6 +2,9 @@ #include "download_storer.h" +#include +#include +#include #include #include #include @@ -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()) diff --git a/src/session/download_storer.h b/src/session/download_storer.h index 01d975c9..9601b443 100644 --- a/src/session/download_storer.h +++ b/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 m_torrent_stream; diff --git a/src/session/session_manager.cc b/src/session/session_manager.cc index 0c9f3754..4c397de8 100644 --- a/src/session/session_manager.cc +++ b/src/session/session_manager.cc @@ -3,10 +3,6 @@ #include "session/session_manager.h" #include -#include -#include -#include -#include #include #include @@ -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()) { @@ -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 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 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 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 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 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 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 lock(m_mutex); @@ -291,10 +363,35 @@ SessionManager::flush_all_and_wait_unsafe(std::unique_lock& lock) { } bool -SessionManager::remove_save_request_unsafe(core::Download* download, std::unique_lock& 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& 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 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& 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 diff --git a/src/session/session_manager.h b/src/session/session_manager.h index 63c0b5f8..67af1aff 100644 --- a/src/session/session_manager.h +++ b/src/session/session_manager.h @@ -37,8 +37,12 @@ public: typedef std::unique_ptr 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& lock); - bool remove_save_request_unsafe(core::Download* download, std::unique_lock& lock); - - void save_download_unsafe(const SaveRequest& request); - bool save_download_stream_unsafe(const std::string& path, const std::unique_ptr& stream); + bool remove_or_replace_unsafe(SaveRequest& download); + bool remove_completely_unsafe(core::Download* download, std::unique_lock& lock); torrent::utils::Thread* m_thread; @@ -94,14 +98,16 @@ private: typedef std::pair, SaveRequest> ProcessingSave; - // TODO: Add deque of DownloadStorers requests that have yet to build streams. - std::deque m_save_requests; std::list m_processing_saves; std::condition_variable m_finished_condition; std::vector m_finished_saves; std::unique_ptr m_lockfile; + + // Pending builds are only ever locked by main thread. + std::mutex m_pending_builds_mutex; + std::deque m_pending_builds; }; inline bool SessionManager::is_used() const { return !m_path.empty(); }