rtorrent/src/session/session_manager.cc

#include "config.h"

#include "session_manager.h"

#include <cassert>
#include <cerrno>
#include <fstream>
#include <fcntl.h>
#include <unistd.h>
#include <torrent/exceptions.h>
#include <torrent/utils/log.h>

#include "globals.h"
#include "utils/lockfile.h"

#define LT_LOG(log_fmt, ...)                                            \
  lt_log_print(torrent::LOG_SESSION_EVENTS, "session-events: " log_fmt, __VA_ARGS__);

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>()) {
}

SessionManager::~SessionManager() = default;

// TODO:
//  * Lock session directory.
//  * On shutdown, wait for all saves to finish.
//    * Add is_empty_and_done() that also include async fdisksync tasks.
//  * Then unlock session directory.

bool
SessionManager::is_empty() {
  std::lock_guard<std::mutex> guard(m_mutex);
  return m_save_requests.empty();
}

void
SessionManager::set_path(const std::string& path) {
  assert(torrent::this_thread::thread() == torrent::main_thread::thread());

  if (m_freeze_info)
    throw torrent::input_error("Session path cannot be changed after startup.");

  if (path.empty() || path.back() == '/')
    m_path = path;
  else
    m_path = path + '/';
}

void
SessionManager::set_use_lock(bool use_lock) {
  assert(torrent::this_thread::thread() == torrent::main_thread::thread());

  if (m_freeze_info)
    throw torrent::input_error("Session lock option cannot be changed after startup.");

  m_use_lock = use_lock;
}

// TODO: Derive path from download info hash.
// TODO: Generate streams here, not in download store.
void
SessionManager::save_download(core::Download* download, std::string path, stream_ptr torrent_stream, stream_ptr rtorrent_stream, stream_ptr libtorrent_stream) {
  assert(torrent::this_thread::thread() == torrent::main_thread::thread());

  if (m_path.empty())
    return;

  {
    std::lock_guard<std::mutex> guard(m_mutex);

    LT_LOG("requesting save : download:%p path:%s", download, path.c_str());

    if (!m_active)
      throw torrent::internal_error("SessionManager::save_download() called while not active.");

    // TODO: Remove is already queued entries.

    // TODO: Add these to a temp structure
    // TODO: When a download already exists, replace it.

    m_save_requests.push_back(SaveRequest{
        download,
        std::move(path),
        std::move(torrent_stream),
        std::move(rtorrent_stream),
        std::move(libtorrent_stream)
      });
  }

  session_thread::callback(nullptr, [this]() { process_save_request(); });
}

void
SessionManager::remove_download(core::Download* download, std::string base_path) {
  assert(torrent::this_thread::thread() == torrent::main_thread::thread());

  if (m_path.empty())
    return;

  std::lock_guard<std::mutex> guard(m_mutex);

  if (!m_active)
    throw torrent::internal_error("SessionManager::remove_download() called while not active.");

  // TODO: Add these to a temp structure, and remove from to-be-added temp struct.

  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()) {
    LT_LOG("canceling save request : download:%p", download);
    m_save_requests.erase(itr, m_save_requests.end());
  }

  // TODO: Use atomic download ptr to check if we're currently processing this download
  // TODO: If so, use a lock to wait for save to finish before returning

  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());
}

void
SessionManager::start() {
  assert(torrent::this_thread::thread() == torrent::main_thread::thread());

  std::lock_guard<std::mutex> guard(m_mutex);

  if (m_active || m_freeze_info)
    throw torrent::internal_error("SessionManager::start() called while already started.");

  m_active      = true;
  m_freeze_info = true;

  if (m_path.empty()) {
    LT_LOG("session manager started with empty path, disabling session management", 0);
    return;
  }

  LT_LOG("starting session manager with path: %s", m_path.c_str());

  if (m_use_lock) {
    m_lockfile->set_path(m_path + "rtorrent.lock");

    if (!m_lockfile->try_lock()) {
      if (errno == ENOENT || errno == ENOTDIR || errno == EACCES)
        throw torrent::input_error("Could not lock session directory: " + std::string(std::strerror(errno)) + " : " + m_path);
      else
        throw torrent::input_error("Could not lock session directory, held by: " + m_lockfile->locked_by_as_string() + " : " + m_path);
    }

    LT_LOG("locked session directory: %s", m_path.c_str());
  }
}

void
SessionManager::cleanup() {
  assert(m_thread == torrent::this_thread::thread());

  std::lock_guard<std::mutex> guard(m_mutex);

  if (!m_active)
    throw torrent::internal_error("SessionManager::cleanup() called while not active.");

  m_active = false;

  if (m_path.empty()) {
    LT_LOG("session manager cleanup called with empty path, skipping", 0);
    return;
  }

  LT_LOG("cleaning up session manager with path: %s", m_path.c_str());

  if (m_use_lock) {
    if (!m_lockfile->unlock())
      LT_LOG("could not unlock session directory: %s", m_path.c_str());

    LT_LOG("unlocked session directory: %s", m_path.c_str());
  }
}

void
SessionManager::process_save_request() {
  assert(m_thread == torrent::this_thread::thread());

  if (m_path.empty())
    return;

  std::lock_guard<std::mutex> guard(m_mutex);

  if (!m_active)
    throw torrent::internal_error("SessionManager::process_save_request() called while not active.");

  if (m_save_requests.empty())
    return;

  // pick first request
  // process it
  // add us back to callbacks? we need to disable shutdown while processing all saves... do we do it at thread cleanup?

  auto request = std::move(m_save_requests.front());
  m_save_requests.pop_front();

  // Keep lock while processing to ensure cancellations do not interfere.

  save_download_unsafe(request);

  if (!m_save_requests.empty())
    session_thread::callback(nullptr, [this]() { process_save_request(); });
}

// TODO: Add threads/tasklets that calls fdisksync on shutdown.
// TODO: Parallelize saves.

// TODO: Properly handle errors.
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);

  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;
  }

  // We don't care about cancelation here, as the underlying file gets deleted / replaced anyway.

  // TODO: We can use std::async for these, and only wait for them if we're shutting down.

  // TODO: Use an atomic counter / conditional variable to keep track of async operations count, and
  // wait for finished operations on shutdown.

  // if (rpc::call_command_value("system.files.session.fdatasync")) {
  if (true) {
#ifdef __APPLE__
    ::fsync(fd);
#else
    ::fdatasync(fd);
#endif
  }

  ::close(fd);
  return true;
}

}  // namespace session