diff --git a/configure.ac b/configure.ac index b3973dbe..c291db54 100644 --- a/configure.ac +++ b/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 ]], + [[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++) diff --git a/doc/rtorrent.rc b/doc/rtorrent.rc index 6d25cefa..08c35267 100644 --- a/doc/rtorrent.rc +++ b/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. diff --git a/src/Makefile.am b/src/Makefile.am index af4b0d1d..76018dbf 100644 --- a/src/Makefile.am +++ b/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 \ diff --git a/src/command_events.cc b/src/command_events.cc index 23e262c4..27ed37a9 100644 --- a/src/command_events.cc +++ b/src/command_events.cc @@ -2,6 +2,8 @@ #include #include +#include +#include #include #include #include @@ -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"); diff --git a/src/control.cc b/src/control.cc index c415e613..b5ffb9b6 100644 --- a/src/control.cc +++ b/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(); m_view_manager = std::make_unique(); @@ -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; } - diff --git a/src/control.h b/src/control.h index 8502a251..69ae6a49 100644 --- a/src/control.h +++ b/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 m_objectStorage; std::unique_ptr m_lua_engine; std::unique_ptr m_directory_events; + std::unique_ptr m_watch_ready_queue; uint64_t m_tick{}; diff --git a/src/utils/file_status_cache.cc b/src/utils/file_status_cache.cc index a32afef8..627fc1eb 100644 --- a/src/utils/file_status_cache.cc +++ b/src/utils/file_status_cache.cc @@ -21,13 +21,16 @@ FileStatusCache::insert(const std::string& path) { std::pair 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); } } diff --git a/src/utils/file_status_cache.h b/src/utils/file_status_cache.h index f3c795a3..cf8536a2 100644 --- a/src/utils/file_status_cache.h +++ b/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(); }; diff --git a/src/utils/watch_ready_queue.cc b/src/utils/watch_ready_queue.cc new file mode 100644 index 00000000..8c36bfdb --- /dev/null +++ b/src/utils/watch_ready_queue.cc @@ -0,0 +1,158 @@ +#include "config.h" + +#include "utils/watch_ready_queue.h" + +#include +#include +#include + +#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); +} + +} diff --git a/src/utils/watch_ready_queue.h b/src/utils/watch_ready_queue.h new file mode 100644 index 00000000..1f0515ec --- /dev/null +++ b/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 +#include +#include +#include +#include +#include +#include + +#include + +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>; + + 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 m_entries; + std::vector m_entry_queue; + torrent::utils::SchedulerEntry m_task_process; + bool m_active{true}; +}; + +} + +#endif diff --git a/test/Makefile.am b/test/Makefile.am index 690d956a..aae18225 100644 --- a/test/Makefile.am +++ b/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 diff --git a/test/src/test_watch_ready_queue.cc b/test/src/test_watch_ready_queue.cc new file mode 100644 index 00000000..852d3a9a --- /dev/null +++ b/test/src/test_watch_ready_queue.cc @@ -0,0 +1,175 @@ +#include "config.h" + +#include "test/src/test_watch_ready_queue.h" + +#include +#include +#include +#include +#include +#include + +#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 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(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); +} diff --git a/test/src/test_watch_ready_queue.h b/test/src/test_watch_ready_queue.h new file mode 100644 index 00000000..ec9e8735 --- /dev/null +++ b/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(); +};