Add directory.watch.ready for safe torrent watch folders

configure.ac

@@ -51,6 +51,19 @@ PKG_CHECK_MODULES([CPPUNIT], [cppunit],, [no_cppunit="yes"])
 PKG_CHECK_MODULES([ZLIB], [zlib])
 PKG_CHECK_MODULES([DEPENDENCIES], [libtorrent >= 0.16.12])
 
+AC_LANG_PUSH(C++)
+rtorrent_save_CXXFLAGS="$CXXFLAGS"
+CXXFLAGS="$CXXFLAGS $DEPENDENCIES_CFLAGS"
+AC_MSG_CHECKING([for libtorrent directory_events ready watch support])
+AC_COMPILE_IFELSE(
+  [AC_LANG_PROGRAM([[#include <torrent/utils/directory_events.h>]],
+                   [[return torrent::directory_events::flag_on_ready == 0;]])],
+  [AC_MSG_RESULT([yes])],
+  [AC_MSG_RESULT([no])
+   AC_MSG_ERROR([requires libtorrent with directory_events::flag_on_ready])])
+CXXFLAGS="$rtorrent_save_CXXFLAGS"
+AC_LANG_POP(C++)
+
 AC_LANG_PUSH(C++)
 TORRENT_WITH_XMLRPC_C
 AC_LANG_POP(C++)

doc/rtorrent.rc

@@ -46,8 +46,12 @@
 #session.path.set = ./session
 
 # Watch a directory for new torrents, and stop those that have been
-# deleted.
+# deleted. Use directory.watch.ready for network shares or other watch
+# directories where files may be copied or written directly into place.
+# Do not mix directory.watch.ready and directory.watch.added on the same
+# watch directory.
 #
+#directory.watch.ready = ./watch, load.start
 #schedule2 = watch_directory,5,5,load.start=./watch/*.torrent
 
 # Close torrents when disk-space is low.

src/Makefile.am

@@ -174,6 +174,8 @@ libsub_root_a_SOURCES = \
 	utils/list_focus.h \
 	utils/lockfile.cc \
 	utils/lockfile.h \
+	utils/watch_ready_queue.cc \
+	utils/watch_ready_queue.h \
 	\
 	command_download.cc \
 	command_dynamic.cc \

src/command_events.cc

@@ -2,6 +2,8 @@
 
 #include <functional>
 #include <cstdio>
+#include <string>
+#include <vector>
 #include <torrent/rate.h>
 #include <torrent/hash_string.h>
 #include <torrent/utils/log.h>
@@ -19,6 +21,7 @@
 #include "rpc/command_scheduler.h"
 #include "rpc/parse.h"
 #include "rpc/parse_commands.h"
+#include "utils/watch_ready_queue.h"
 
 torrent::Object
 apply_on_ratio(const torrent::Object& rawArgs) {
@@ -291,22 +294,39 @@ call_watch_command(const std::string& command, const std::string& path) {
 }
 
 torrent::Object
-directory_watch_added(const torrent::Object::list_type& args) {
+directory_watch(const torrent::Object::list_type& args, int flags) {
   if (args.size() != 2)
     throw torrent::input_error("Too few arguments.");
 
   auto& path = args.front().as_string();
   auto& command = args.back().as_string();
+  std::string expanded_path = expand_path(path);
 
   if (!control->directory_events()->open())
     throw torrent::input_error("Could not open inotify:" + std::string(std::strerror(errno)));
 
-  control->directory_events()->notify_on(path.c_str(),
-                                         torrent::directory_events::flag_on_added | torrent::directory_events::flag_on_updated,
-                                         std::bind(&call_watch_command, command, std::placeholders::_1));
+  torrent::watch_descriptor::slot_string slot =
+    flags == torrent::directory_events::flag_on_ready ?
+    torrent::watch_descriptor::slot_string([command](const auto& arg) { control->watch_ready_queue()->push(command, arg); }) :
+    torrent::watch_descriptor::slot_string(std::bind(&call_watch_command, command, std::placeholders::_1));
+
+  control->directory_events()->notify_on(expanded_path.c_str(), flags, slot);
   return torrent::Object();
 }
 
+torrent::Object
+directory_watch_added(const torrent::Object::list_type& args) {
+  return directory_watch(args,
+                         torrent::directory_events::flag_on_added |
+                         torrent::directory_events::flag_on_updated);
+}
+
+torrent::Object
+directory_watch_ready(const torrent::Object::list_type& args) {
+  return directory_watch(args,
+                         torrent::directory_events::flag_on_ready);
+}
+
 void
 initialize_command_events() {
   CMD2_ANY_STRING  ("on_ratio",        std::bind(&apply_on_ratio, std::placeholders::_2));
@@ -344,6 +364,7 @@ initialize_command_events() {
   CMD2_ANY_LIST    ("d.multicall.filtered", std::bind(&d_multicall_filtered, std::placeholders::_2));
 
   CMD2_ANY_LIST    ("directory.watch.added", std::bind(&directory_watch_added, std::placeholders::_2));
+  CMD2_ANY_LIST    ("directory.watch.ready", std::bind(&directory_watch_ready, std::placeholders::_2));
 
   rpc::rpc.mark_safe("start_tied");
   rpc::rpc.mark_safe("stop_untied");

src/control.cc

@@ -27,6 +27,7 @@
 #include "rpc/object_storage.h"
 #include "session/session_manager.h"
 #include "ui/root.h"
+#include "utils/watch_ready_queue.h"
 
 Control::Control()
   : m_ui(new ui::Root()),
@@ -36,7 +37,8 @@ Control::Control()
     m_commandScheduler(new rpc::CommandScheduler()),
     m_objectStorage(new rpc::object_storage()),
     m_lua_engine(new rpc::LuaEngine()),
-    m_directory_events(new torrent::directory_events()) {
+    m_directory_events(new torrent::directory_events()),
+    m_watch_ready_queue(new utils::WatchReadyQueue()) {
 
   m_core         = std::make_unique<core::Manager>();
   m_view_manager = std::make_unique<core::ViewManager>();
@@ -126,6 +128,8 @@ Control::is_shutdown_completed() {
 
 void
 Control::handle_shutdown() {
+  m_watch_ready_queue->shutdown();
+
   rpc::commands.call_catch("event.system.shutdown", rpc::make_target(), "shutdown", "System shutdown event action failed: ");
 
   if (scgi_thread::thread()->is_active())
@@ -147,4 +151,3 @@ Control::handle_shutdown() {
   m_shutdownQuick = true;
   m_shutdownReceived = false;
 }
-

src/control.h

@@ -38,6 +38,10 @@ namespace torrent {
   class directory_events;
 }
 
+namespace utils {
+  class WatchReadyQueue;
+}
+
 class Control {
 public:
   Control();
@@ -70,6 +74,7 @@ public:
   rpc::LuaEngine*        lua_engine()               { return m_lua_engine.get(); }
 
   torrent::directory_events* directory_events()     { return m_directory_events.get(); }
+  utils::WatchReadyQueue*    watch_ready_queue()    { return m_watch_ready_queue.get(); }
 
   uint64_t            tick() const                  { return m_tick; }
   void                inc_tick()                    { m_tick++; }
@@ -94,6 +99,7 @@ private:
   std::unique_ptr<rpc::object_storage>       m_objectStorage;
   std::unique_ptr<rpc::LuaEngine>            m_lua_engine;
   std::unique_ptr<torrent::directory_events> m_directory_events;
+  std::unique_ptr<utils::WatchReadyQueue>    m_watch_ready_queue;
 
   uint64_t            m_tick{};
 

src/utils/file_status_cache.cc

@@ -21,13 +21,16 @@ FileStatusCache::insert(const std::string& path) {
 
   std::pair<iterator, bool> result = base_type::insert(value_type(path, file_status()));
 
-  // Return false if the file hasn't been modified since last time. We
-  // use 'equal to' instead of 'greater than' since the file might
-  // have been replaced by another file, and thus should be re-tried.
-  if (!result.second && result.first->second.m_mtime == (uint32_t)fs.modified_time())
+  // Return false if the file hasn't changed since last time. We use
+  // 'equal to' instead of 'greater than' since the file might have
+  // been replaced by another file, and thus should be re-tried.
+  if (!result.second &&
+      result.first->second.m_mtime == (uint32_t)fs.modified_time() &&
+      result.first->second.m_size == (int64_t)fs.size())
     return false;
 
   result.first->second.m_flags = 0;
+  result.first->second.m_size  = (int64_t)fs.size();
   result.first->second.m_mtime = fs.modified_time();
 
   return true;
@@ -41,7 +44,9 @@ FileStatusCache::prune() {
     torrent::utils::FileStat fs;
     iterator tmp = itr++;
 
-    if (!fs.update(expand_path(tmp->first)) || tmp->second.m_mtime != (uint32_t)fs.modified_time())
+    if (!fs.update(expand_path(tmp->first)) ||
+        tmp->second.m_mtime != (uint32_t)fs.modified_time() ||
+        tmp->second.m_size != (int64_t)fs.size())
       base_type::erase(tmp);
   }
 }

src/utils/file_status_cache.h

@@ -9,6 +9,7 @@ namespace utils {
 
 struct file_status {
   int      m_flags;
+  int64_t  m_size;
   uint32_t m_mtime;
 };
 
@@ -32,14 +33,14 @@ public:
 
   using base_type::erase;
 
-  // Insert and return true if the entry does not exist or the new
-  // file's mtime is more recent.
+  // Insert and return true if the entry does not exist or the file's
+  // status has changed.
   bool                insert(const std::string& path);
 
   // Add a function for pruning a sorted list of paths.
 
   // Function for pruning entries that no longer points to a file, or
-  // has a different mtime.
+  // has different status.
   void                prune();
 };
 

src/utils/watch_ready_queue.cc

@@ -0,0 +1,158 @@
+#include "config.h"
+
+#include "utils/watch_ready_queue.h"
+
+#include <algorithm>
+#include <torrent/utils/file_stat.h>
+#include <torrent/utils/log.h>
+
+#include "globals.h"
+#include "rpc/rpc_manager.h"
+
+namespace utils {
+
+namespace {
+
+auto compare_next_time = [](const auto* lhs, const auto* rhs) {
+  return lhs->next_time > rhs->next_time;
+};
+
+}
+
+WatchReadyQueue::WatchReadyQueue() {
+  m_task_process.slot() = [this]() { process(); };
+}
+
+WatchReadyQueue::~WatchReadyQueue() {
+  torrent::this_thread::scheduler()->erase(&m_task_process);
+}
+
+void
+WatchReadyQueue::push(const std::string& command, const std::string& path) {
+  if (!m_active)
+    return;
+
+  auto result = m_entries.emplace(path, Entry());
+  auto& entry = result.first->second;
+
+  if (result.second) {
+    entry.path = path;
+    entry.first_seen = torrent::this_thread::cached_time();
+  }
+
+  entry.command = command;
+  entry.last_changed = torrent::this_thread::cached_time();
+  update_status(&entry);
+
+  if (result.second) {
+    update_next_time(&entry);
+    push_entry(&entry);
+  } else {
+    update_entry(&entry);
+  }
+
+  schedule();
+}
+
+void
+WatchReadyQueue::shutdown() {
+  m_active = false;
+  m_entries.clear();
+  m_entry_queue.clear();
+  torrent::this_thread::scheduler()->erase(&m_task_process);
+}
+
+void
+WatchReadyQueue::push_entry(Entry* entry) {
+  m_entry_queue.push_back(entry);
+  std::push_heap(m_entry_queue.begin(), m_entry_queue.end(), compare_next_time);
+}
+
+void
+WatchReadyQueue::update_entry(Entry* entry) {
+  update_next_time(entry);
+  std::make_heap(m_entry_queue.begin(), m_entry_queue.end(), compare_next_time);
+}
+
+void
+WatchReadyQueue::update_next_time(Entry* entry) {
+  if (entry->regular && entry->size > 0) {
+    entry->next_time = entry->last_changed + quiet_time;
+    return;
+  }
+
+  entry->next_time = std::min(torrent::this_thread::cached_time() + retry_time,
+                              entry->first_seen + stale_time);
+}
+
+void
+WatchReadyQueue::update_status(Entry* entry) {
+  torrent::utils::FileStat fs;
+  bool regular = false;
+  int64_t size = -1;
+  time_t mtime = 0;
+
+  if (fs.update(expand_path(entry->path))) {
+    regular = fs.is_regular();
+    size = fs.size();
+    mtime = fs.modified_time();
+  }
+
+  if (entry->regular != regular || entry->size != size || entry->mtime != mtime) {
+    entry->regular = regular;
+    entry->size = size;
+    entry->mtime = mtime;
+    entry->last_changed = torrent::this_thread::cached_time();
+  }
+}
+
+void
+WatchReadyQueue::process() {
+  ready_list ready;
+
+  while (!m_entry_queue.empty() && m_entry_queue.front()->next_time <= torrent::this_thread::cached_time()) {
+    std::pop_heap(m_entry_queue.begin(), m_entry_queue.end(), compare_next_time);
+    Entry* entry = m_entry_queue.back();
+    m_entry_queue.pop_back();
+
+    update_status(entry);
+
+    bool unchanged_entry = torrent::this_thread::cached_time() - entry->last_changed >= quiet_time;
+    bool stale_entry = torrent::this_thread::cached_time() - entry->first_seen >= stale_time;
+    bool nonempty_regular = entry->regular && entry->size > 0;
+
+    if (nonempty_regular && unchanged_entry) {
+      ready.emplace_back(entry->command, entry->path);
+      std::string path = entry->path;
+      m_entries.erase(path);
+      continue;
+    }
+
+    if (!nonempty_regular && stale_entry) {
+      std::string path = entry->path;
+      lt_log_print(torrent::LOG_SYSTEM, "system: Dropped unready watch file after timeout: \"%s\"", path.c_str());
+      m_entries.erase(path);
+      continue;
+    }
+
+    update_next_time(entry);
+    push_entry(entry);
+  }
+
+  schedule();
+
+  for (const auto& item : ready)
+    rpc::commands.call_catch(item.first.c_str(), rpc::make_target(), item.second);
+}
+
+void
+WatchReadyQueue::schedule() {
+  if (m_entry_queue.empty()) {
+    torrent::this_thread::scheduler()->erase(&m_task_process);
+    return;
+  }
+
+  torrent::this_thread::scheduler()->update_wait_until(&m_task_process, m_entry_queue.front()->next_time);
+}
+
+}

src/utils/watch_ready_queue.h

@@ -0,0 +1,60 @@
+#ifndef RTORRENT_UTILS_WATCH_READY_QUEUE_H
+#define RTORRENT_UTILS_WATCH_READY_QUEUE_H
+
+#include <chrono>
+#include <cstdint>
+#include <ctime>
+#include <map>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <torrent/utils/scheduler.h>
+
+namespace utils {
+
+class WatchReadyQueue {
+public:
+  WatchReadyQueue();
+  ~WatchReadyQueue();
+
+  void                push(const std::string& command, const std::string& path);
+  void                shutdown();
+  bool                empty() const { return m_entries.empty(); }
+
+private:
+  using time_type = std::chrono::microseconds;
+
+  static constexpr auto quiet_time = std::chrono::milliseconds(500);
+  static constexpr auto retry_time = std::chrono::milliseconds(250);
+  static constexpr auto stale_time = std::chrono::seconds(10);
+
+  using ready_list = std::vector<std::pair<std::string, std::string>>;
+
+  struct Entry {
+    std::string command;
+    std::string path;
+    bool        regular{};
+    int64_t     size{-1};
+    time_t      mtime{};
+    time_type   first_seen{};
+    time_type   last_changed{};
+    time_type   next_time{};
+  };
+
+  void               process();
+  void               push_entry(Entry* entry);
+  void               update_entry(Entry* entry);
+  void               update_next_time(Entry* entry);
+  void               update_status(Entry* entry);
+  void               schedule();
+
+  std::map<std::string, Entry>   m_entries;
+  std::vector<Entry*>            m_entry_queue;
+  torrent::utils::SchedulerEntry m_task_process;
+  bool                           m_active{true};
+};
+
+}
+
+#endif

test/Makefile.am

@@ -48,7 +48,9 @@ rtorrent_Test_Rpc_SOURCES = $(rtorrent_Test_Common) \
 
 rtorrent_Test_Src_SOURCES = $(rtorrent_Test_Common) \
 	src/test_command_dynamic.cc \
-	src/test_command_dynamic.h
+	src/test_command_dynamic.h \
+	src/test_watch_ready_queue.cc \
+	src/test_watch_ready_queue.h
 
 rtorrent_Test_Rpc_CXXFLAGS = $(CPPUNIT_CFLAGS)
 rtorrent_Test_Rpc_LDFLAGS = $(CPPUNIT_LIBS) -ldl

test/src/test_watch_ready_queue.cc

@@ -0,0 +1,175 @@
+#include "config.h"
+
+#include "test/src/test_watch_ready_queue.h"
+
+#include <cstdlib>
+#include <fcntl.h>
+#include <string>
+#include <sys/types.h>
+#include <unistd.h>
+#include <vector>
+
+#include "command_helpers.h"
+#include "rpc/command_map.h"
+#include "utils/watch_ready_queue.h"
+
+CPPUNIT_TEST_SUITE_REGISTRATION(TestWatchReadyQueue);
+
+namespace {
+
+const char* test_load_command = "test.watch_ready.load";
+std::vector<std::string> loaded_paths;
+
+torrent::Object
+cmd_watch_ready_load([[maybe_unused]] rpc::target_type target, const std::string& path) {
+  loaded_paths.push_back(path);
+  return torrent::Object();
+}
+
+std::string
+temporary_path() {
+  char path[] = "/tmp/rtorrent-watch-ready-XXXXXX";
+  int fd = ::mkstemp(path);
+
+  CPPUNIT_ASSERT(fd != -1);
+  CPPUNIT_ASSERT(::close(fd) == 0);
+  CPPUNIT_ASSERT(::unlink(path) == 0);
+
+  return path;
+}
+
+void
+write_file(const std::string& path, const std::string& contents) {
+  int fd = ::open(path.c_str(), O_WRONLY | O_CREAT | O_TRUNC, 0600);
+
+  CPPUNIT_ASSERT(fd != -1);
+  CPPUNIT_ASSERT(::write(fd, contents.data(), contents.size()) == static_cast<ssize_t>(contents.size()));
+  CPPUNIT_ASSERT(::close(fd) == 0);
+}
+
+} // namespace
+
+void
+TestWatchReadyQueue::setUp() {
+  TestFixtureWithMainThread::setUp();
+
+  loaded_paths.clear();
+  m_main_thread->test_set_cached_time(std::chrono::seconds(0));
+
+  if (!rpc::commands.has(test_load_command))
+    CMD2_ANY_STRING(test_load_command, &cmd_watch_ready_load);
+}
+
+void
+TestWatchReadyQueue::tearDown() {
+  loaded_paths.clear();
+
+  TestFixtureWithMainThread::tearDown();
+}
+
+void
+TestWatchReadyQueue::test_repeated_events_reset_quiet_time() {
+  auto path = temporary_path();
+  write_file(path, "torrent");
+
+  utils::WatchReadyQueue queue;
+  queue.push(test_load_command, path);
+
+  m_main_thread->test_add_cached_time(std::chrono::milliseconds(300));
+  queue.push(test_load_command, path);
+
+  m_main_thread->test_add_cached_time(std::chrono::milliseconds(300));
+  m_main_thread->test_process_events_without_cached_time();
+  CPPUNIT_ASSERT(loaded_paths.empty());
+
+  m_main_thread->test_add_cached_time(std::chrono::milliseconds(201));
+  m_main_thread->test_process_events_without_cached_time();
+  CPPUNIT_ASSERT(loaded_paths.size() == 1);
+  CPPUNIT_ASSERT(loaded_paths.front() == path);
+
+  CPPUNIT_ASSERT(::unlink(path.c_str()) == 0);
+}
+
+void
+TestWatchReadyQueue::test_retry_restats_zero_length_file_without_new_event() {
+  auto path = temporary_path();
+  write_file(path, "");
+
+  utils::WatchReadyQueue queue;
+  queue.push(test_load_command, path);
+
+  m_main_thread->test_add_cached_time(std::chrono::milliseconds(251));
+  m_main_thread->test_process_events_without_cached_time();
+  CPPUNIT_ASSERT(loaded_paths.empty());
+
+  m_main_thread->test_add_cached_time(std::chrono::milliseconds(10));
+  // No second queue event: retry processing must re-stat the same pending path.
+  write_file(path, "torrent");
+
+  m_main_thread->test_add_cached_time(std::chrono::milliseconds(240));
+  m_main_thread->test_process_events_without_cached_time();
+  CPPUNIT_ASSERT(loaded_paths.empty());
+
+  m_main_thread->test_add_cached_time(std::chrono::milliseconds(500));
+  m_main_thread->test_process_events_without_cached_time();
+  CPPUNIT_ASSERT(loaded_paths.size() == 1);
+  CPPUNIT_ASSERT(loaded_paths.front() == path);
+
+  CPPUNIT_ASSERT(::unlink(path.c_str()) == 0);
+}
+
+void
+TestWatchReadyQueue::test_unrelated_events_do_not_delay_missing_retry() {
+  auto path = temporary_path();
+  auto unrelated_path = temporary_path();
+
+  utils::WatchReadyQueue queue;
+  queue.push(test_load_command, path);
+
+  m_main_thread->test_add_cached_time(std::chrono::milliseconds(100));
+  queue.push(test_load_command, unrelated_path);
+
+  m_main_thread->test_add_cached_time(std::chrono::milliseconds(160));
+  write_file(path, "torrent");
+  m_main_thread->test_process_events_without_cached_time();
+  CPPUNIT_ASSERT(loaded_paths.empty());
+
+  m_main_thread->test_add_cached_time(std::chrono::milliseconds(501));
+  m_main_thread->test_process_events_without_cached_time();
+  CPPUNIT_ASSERT(loaded_paths.size() == 1);
+  CPPUNIT_ASSERT(loaded_paths.front() == path);
+
+  CPPUNIT_ASSERT(::unlink(path.c_str()) == 0);
+}
+
+void
+TestWatchReadyQueue::test_missing_paths_expire_without_dispatch() {
+  auto path = temporary_path();
+
+  utils::WatchReadyQueue queue;
+  queue.push(test_load_command, path);
+
+  m_main_thread->test_add_cached_time(std::chrono::seconds(10) + std::chrono::milliseconds(1));
+  m_main_thread->test_process_events_without_cached_time();
+
+  CPPUNIT_ASSERT(loaded_paths.empty());
+  CPPUNIT_ASSERT(queue.empty());
+}
+
+void
+TestWatchReadyQueue::test_shutdown_discards_pending_loads() {
+  auto path = temporary_path();
+  write_file(path, "torrent");
+
+  utils::WatchReadyQueue queue;
+  queue.push(test_load_command, path);
+  queue.shutdown();
+
+  m_main_thread->test_add_cached_time(std::chrono::seconds(1));
+  m_main_thread->test_process_events_without_cached_time();
+
+  CPPUNIT_ASSERT(loaded_paths.empty());
+  CPPUNIT_ASSERT(queue.empty());
+
+  CPPUNIT_ASSERT(::unlink(path.c_str()) == 0);
+}

test/src/test_watch_ready_queue.h

@@ -0,0 +1,23 @@
+#include "test/helpers/test_main_thread.h"
+
+class TestWatchReadyQueue : public TestFixtureWithMainThread {
+  CPPUNIT_TEST_SUITE(TestWatchReadyQueue);
+
+  CPPUNIT_TEST(test_repeated_events_reset_quiet_time);
+  CPPUNIT_TEST(test_retry_restats_zero_length_file_without_new_event);
+  CPPUNIT_TEST(test_unrelated_events_do_not_delay_missing_retry);
+  CPPUNIT_TEST(test_missing_paths_expire_without_dispatch);
+  CPPUNIT_TEST(test_shutdown_discards_pending_loads);
+
+  CPPUNIT_TEST_SUITE_END();
+
+public:
+  void setUp();
+  void tearDown();
+
+  void test_repeated_events_reset_quiet_time();
+  void test_retry_restats_zero_length_file_without_new_event();
+  void test_unrelated_events_do_not_delay_missing_retry();
+  void test_missing_paths_expire_without_dispatch();
+  void test_shutdown_discards_pending_loads();
+};