From e746c93624582f3affb8461921e39eebc950872c Mon Sep 17 00:00:00 2001 From: Morgan Caron Date: Sun, 20 Sep 2026 05:10:58 +0200 Subject: [PATCH] feat(filesystem): make Watcher passive, rearmable, movable and embed as direct member --- modules/FileSystem/Watcher.mpp | 138 +++++++++++++++++++++++++--- modules/Language/XML/Flow.mpp | 14 +-- modules/Language/XML/FlowRunner.mpp | 8 +- tests/FileSystem/Watcher.mpp | 47 ++++++++++ 4 files changed, 180 insertions(+), 27 deletions(-) diff --git a/modules/FileSystem/Watcher.mpp b/modules/FileSystem/Watcher.mpp index 5d97df7f..ea01539c 100644 --- a/modules/FileSystem/Watcher.mpp +++ b/modules/FileSystem/Watcher.mpp @@ -41,14 +41,15 @@ export namespace CppUtils::FileSystem public: using SubscribeFunction = std::function; -#if not defined(OS_MACOS) +#if defined(OS_MACOS) inline Watcher(): - m_threadLoop { std::bind(&Watcher::listener, this), std::bind(&Watcher::interruptFunction, this) } + m_dispatchQueue{dispatch_queue_create("CppUtils.FileSystem.Watcher", nullptr)} + {} #else - inline Watcher() -#endif + inline Watcher(): + m_threadLoop{[this] { listener(); }, [this] { interruptFunction(); }} { -#if defined(OS_LINUX) +# if defined(OS_LINUX) m_eventPoll = epoll_create1(0); { m_inotify = inotify_init1(IN_NONBLOCK); @@ -65,13 +66,90 @@ export namespace CppUtils::FileSystem epoll_ctl(m_eventPoll, EPOLL_CTL_ADD, m_stopPipe.read, std::addressof(m_stopEvent)), "Can't add pipe file descriptor to epoll"); } -#elif defined(OS_WINDOWS) +# elif defined(OS_WINDOWS) m_stopEvent = CreateEventW(nullptr, true, false, nullptr); m_wakeEvent = CreateEventW(nullptr, true, false, nullptr); +# endif + } +#endif + + Watcher(const Watcher&) = delete; + auto operator=(const Watcher&) -> Watcher& = delete; + +#if defined(OS_MACOS) + inline Watcher(Watcher&& other) noexcept + { + const auto wasRunning = other.isRunning(); + other.stop(); + m_dispatchQueue = std::exchange(other.m_dispatchQueue, nullptr); + m_isRunning.store(other.m_isRunning.load()); + m_subscribedFunctions = std::move(other.m_subscribedFunctions); + m_watchedItems = std::move(other.m_watchedItems); + + if (wasRunning) + start(); + } +#else + inline Watcher(Watcher&& other) noexcept: + m_threadLoop{[this] { listener(); }, [this] { interruptFunction(); }} + { + const auto wasRunning = other.isRunning(); + other.stop(); +# if defined(OS_LINUX) + m_eventPoll = std::move(other.m_eventPoll); + m_inotify = std::move(other.m_inotify); + m_stopPipe = std::move(other.m_stopPipe); + m_inotifyEpollEvent = other.m_inotifyEpollEvent; + m_stopEvent = other.m_stopEvent; + m_events = other.m_events; +# elif defined(OS_WINDOWS) + m_stopEvent = std::exchange(other.m_stopEvent, nullptr); + m_wakeEvent = std::exchange(other.m_wakeEvent, nullptr); +# endif + m_subscribedFunctions = std::move(other.m_subscribedFunctions); + m_watchedItems = std::move(other.m_watchedItems); + + if (wasRunning) + start(); + } +#endif + + inline auto operator=(Watcher&& other) noexcept -> Watcher& + { + if (this == std::addressof(other)) + return *this; + + stop(); + const auto wasRunning = other.isRunning(); + other.stop(); + +#if defined(OS_LINUX) + m_eventPoll = std::move(other.m_eventPoll); + m_inotify = std::move(other.m_inotify); + m_stopPipe = std::move(other.m_stopPipe); + m_inotifyEpollEvent = other.m_inotifyEpollEvent; + m_stopEvent = other.m_stopEvent; + m_events = other.m_events; +#elif defined(OS_WINDOWS) + if (m_stopEvent != nullptr) + CloseHandle(m_stopEvent); + if (m_wakeEvent != nullptr) + CloseHandle(m_wakeEvent); + m_stopEvent = std::exchange(other.m_stopEvent, nullptr); + m_wakeEvent = std::exchange(other.m_wakeEvent, nullptr); #elif defined(OS_MACOS) - m_dispatchQueue = dispatch_queue_create("CppUtils.FileSystem.Watcher", nullptr); + if (m_dispatchQueue != nullptr) + dispatch_release(m_dispatchQueue); + m_dispatchQueue = std::exchange(other.m_dispatchQueue, nullptr); + m_isRunning.store(other.m_isRunning.load()); #endif - start(); + m_subscribedFunctions = std::move(other.m_subscribedFunctions); + m_watchedItems = std::move(other.m_watchedItems); + + if (wasRunning) + start(); + + return *this; } ~Watcher() @@ -105,10 +183,15 @@ export namespace CppUtils::FileSystem inline auto start() -> void { -#if defined(OS_MACOS) - m_isRunning.store(true); -#else +#if defined(OS_LINUX) + drainStopPipe(); + m_threadLoop.start(); +#elif defined(OS_WINDOWS) + if (m_stopEvent != nullptr) + ResetEvent(m_stopEvent); m_threadLoop.start(); +#elif defined(OS_MACOS) + m_isRunning.store(true); #endif } @@ -118,9 +201,22 @@ export namespace CppUtils::FileSystem m_isRunning.store(false); #else m_threadLoop.stop(); +# if defined(OS_LINUX) + drainStopPipe(); +# elif defined(OS_WINDOWS) + if (m_stopEvent != nullptr) + ResetEvent(m_stopEvent); +# endif #endif } + inline auto reset() -> void + { + stop(); + unwatchAll(); + m_subscribedFunctions.access()->clear(); + } + #if not defined(OS_MACOS) inline auto listener() -> void { @@ -232,7 +328,10 @@ export namespace CppUtils::FileSystem { auto eventFileDescriptor = m_events[i].data.fd; if (eventFileDescriptor == m_stopPipe.read) + { + drainStopPipe(); continue; + } length = read(eventFileDescriptor, std::ranges::data(buffer), std::ranges::size(buffer)); if (length == -1 and errno == EINTR) break; @@ -276,6 +375,9 @@ export namespace CppUtils::FileSystem if (not std::filesystem::exists(path)) return; + if (not isRunning()) + start(); + const auto normalizedPath = std::filesystem::weakly_canonical(path); #if defined(OS_LINUX) @@ -432,9 +534,8 @@ export namespace CppUtils::FileSystem auto itemsAccessor = m_watchedItems.access(); #if defined(OS_LINUX) for (const auto& [path, watchDescriptor] : itemsAccessor.value()) - System::throwErrno( - inotify_rm_watch(m_inotify, std::to_underlying(watchDescriptor)), - std::format("Failed to stop watching the file {}", path.string())); + if (inotify_rm_watch(m_inotify, std::to_underlying(watchDescriptor)) == -1 and errno != EINVAL) + System::throwErrno(-1, std::format("Failed to stop watching the file {}", path.string())); #elif defined(OS_WINDOWS) for (auto& item : itemsAccessor.value()) { @@ -477,6 +578,15 @@ export namespace CppUtils::FileSystem SetEvent(m_stopEvent); # endif } + +# if defined(OS_LINUX) + inline auto drainStopPipe() -> void + { + auto buffer = std::array{}; + while (read(m_stopPipe.read, std::ranges::data(buffer), std::ranges::size(buffer)) > 0) + {} + } +# endif #endif #if defined(OS_MACOS) diff --git a/modules/Language/XML/Flow.mpp b/modules/Language/XML/Flow.mpp index 0ff73aa0..2497233f 100644 --- a/modules/Language/XML/Flow.mpp +++ b/modules/Language/XML/Flow.mpp @@ -121,26 +121,22 @@ export namespace CppUtils::Language::Xml auto startWatching() -> void override { - if (not m_sourceDirectory.has_value() or m_watcher) + if (not m_sourceDirectory.has_value() or m_watcher.isRunning()) return; if (m_outputDirectory.has_value()) std::filesystem::create_directories(m_outputDirectory.value()); - m_watcher = std::make_unique(); - m_watcher->onEvent([this](auto, const std::filesystem::path& filePath) { + m_watcher.reset(); + m_watcher.onEvent([this](auto, const std::filesystem::path& filePath) { if (matchesPattern(filePath)) processFileWithStaging(filePath, computeOutputFile(filePath)); }); - m_watcher->watch(m_sourceDirectory.value()); + m_watcher.watch(m_sourceDirectory.value()); } auto stopWatching() -> void override { - if (not m_watcher) - return; - - m_watcher->stop(); m_watcher.reset(); } @@ -294,7 +290,7 @@ export namespace CppUtils::Language::Xml std::reference_wrapper m_threadPool; bool m_watch = false; - std::unique_ptr m_watcher; + FileSystem::Watcher m_watcher; std::unordered_map m_tagRegistry; std::unordered_map m_formatRegistry; std::vector> m_pipelineNodes; diff --git a/modules/Language/XML/FlowRunner.mpp b/modules/Language/XML/FlowRunner.mpp index 1e5315e5..55174666 100644 --- a/modules/Language/XML/FlowRunner.mpp +++ b/modules/Language/XML/FlowRunner.mpp @@ -195,8 +195,8 @@ export namespace CppUtils::Language::Xml for (const auto& entry : m_flows) entry.value->startWatching(); - m_flowsWatcher = std::make_unique(); - m_flowsWatcher->onEvent([this](FileSystem::Event event, const std::filesystem::path& filePath) { + m_flowsWatcher.reset(); + m_flowsWatcher.onEvent([this](FileSystem::Event event, const std::filesystem::path& filePath) { if (filePath.extension() != ".xml") return; @@ -246,7 +246,7 @@ export namespace CppUtils::Language::Xml "Failed to load or reload workflow from '{}': {}", filePath.string(), exception.what()); } }); - m_flowsWatcher->watch(m_flowsDirectory); + m_flowsWatcher.watch(m_flowsDirectory); } auto stopWatching() -> void @@ -261,6 +261,6 @@ export namespace CppUtils::Language::Xml std::filesystem::path m_flowsDirectory; FlowContainer m_flows; std::unordered_map m_factories; - std::unique_ptr m_flowsWatcher; + FileSystem::Watcher m_flowsWatcher; }; } diff --git a/tests/FileSystem/Watcher.mpp b/tests/FileSystem/Watcher.mpp index 67c71c62..9b06d85b 100644 --- a/tests/FileSystem/Watcher.mpp +++ b/tests/FileSystem/Watcher.mpp @@ -174,5 +174,52 @@ namespace CppUtils::UnitTest::FileSystem::Watcher suite.expectWait([&] { return expectDeleteFile.load(); }); }}; }); + + suite.addTest("Passive start and stop", [&] { + auto watcher = CppUtils::FileSystem::Watcher{}; + suite.expect(not watcher.isRunning()); + + watcher.start(); + suite.expect(watcher.isRunning()); + + watcher.stop(); + suite.expect(not watcher.isRunning()); + + watcher.start(); + suite.expect(watcher.isRunning()); + + watcher.stop(); + suite.expect(not watcher.isRunning()); + }); + + suite.addTest("Reset watcher", [&] { + CppUtils::FileSystem::TemporaryDirectory{[&suite](const auto& directory) -> void { + auto watcher = CppUtils::FileSystem::Watcher{}; + suite.expect(not watcher.isRunning()); + + watcher.watch(directory); + suite.expect(watcher.isRunning()); + + watcher.reset(); + suite.expect(not watcher.isRunning()); + }}; + }); + + suite.addTest("Move watcher", [&] { + CppUtils::FileSystem::TemporaryDirectory{[&suite](const auto& directory) -> void { + auto watcher1 = CppUtils::FileSystem::Watcher{}; + suite.expect(not watcher1.isRunning()); + + watcher1.watch(directory); + suite.expect(watcher1.isRunning()); + + auto watcher2 = std::move(watcher1); + suite.expect(watcher2.isRunning()); + suite.expect(not watcher1.isRunning()); + + watcher2.stop(); + suite.expect(not watcher2.isRunning()); + }}; + }); }}; }