rtorrent/src/session/session_manager.h

#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::system::Thread* thread);
  ~SessionManager();

  bool                is_used() const;

  std::string         path() const;
  void                set_path(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                callback_pending_builds();
  void                callback_save_request();
  void                callback_finished_saves();

  void                process_pending_builds(bool is_flushing);
  void                process_save_request();
  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);

  using ProcessingSave = std::pair<std::future<void>, SaveRequest>;

  torrent::system::Thread*     m_thread;
  torrent::system::callback_id m_callback_id;

  bool                         m_freeze_info{};
  std::string                  m_path;
  bool                         m_use_fsyncdisk{true};
  bool                         m_use_lock{true};

  align_cacheline std::mutex   m_mutex;

  bool                         m_active{};

  std::deque<SaveRequest>      m_save_requests;
  std::atomic<size_t>          m_save_request_counter{};
  std::list<ProcessingSave>    m_processing_saves;
  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{};
  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_builds(true); }

} // namespace session

#endif // RTORRENT_SESSION_SESSION_MANAGER_H