From ce47b227a27300923e914a40fe3ecaf51a3e23bc Mon Sep 17 00:00:00 2001 From: Marco Edoardo Santimaria Date: Wed, 23 Sep 2026 13:55:14 +0200 Subject: [PATCH 1/5] Added commit on close count support --- bindings/python_bindings.cpp | 1 + capiocl/engine.h | 14 +- capiocl/monitor.h | 34 +- doxygen/semantics.md | 17 +- src/Engine.cpp | 89 +++- src/Monitor.cpp | 32 +- src/monitors/FileSystem.cpp | 204 ++++++++- src/monitors/Interface.cpp | 10 + src/monitors/Multicast.cpp | 190 +++++++-- tests/cpp/main.cpp | 3 +- tests/cpp/test_runtime_commit.hpp | 533 ++++++++++++++++++++++++ tests/tomls/runtime_fs_only.toml | 5 + tests/tomls/runtime_mixed_monitors.toml | 14 + tests/tomls/runtime_multicast_only.toml | 14 + 14 files changed, 1096 insertions(+), 64 deletions(-) create mode 100644 tests/cpp/test_runtime_commit.hpp create mode 100644 tests/tomls/runtime_fs_only.toml create mode 100644 tests/tomls/runtime_mixed_monitors.toml create mode 100644 tests/tomls/runtime_multicast_only.toml diff --git a/bindings/python_bindings.cpp b/bindings/python_bindings.cpp index b3c4512..fe96f2e 100644 --- a/bindings/python_bindings.cpp +++ b/bindings/python_bindings.cpp @@ -107,6 +107,7 @@ PYBIND11_MODULE(_py_capio_cl, m) { .def("setWorkflowName", &capiocl::engine::Engine::setWorkflowName, py::arg("name")) .def("setCommitted", &capiocl::engine::Engine::setCommitted, py::arg("path")) .def("isCommitted", &capiocl::engine::Engine::isCommitted, py::arg("path")) + .def("increaseCloseCount", &capiocl::engine::Engine::increaseCloseCount, py::arg("path")) .def("setHomeNode", &capiocl::engine::Engine::setHomeNode, py::arg("path")) .def("getPaths", &capiocl::engine::Engine::getPaths) .def("startApiServer", &capiocl::engine::Engine::startApiServer) diff --git a/capiocl/engine.h b/capiocl/engine.h index 71f6376..1b6437e 100644 --- a/capiocl/engine.h +++ b/capiocl/engine.h @@ -130,6 +130,9 @@ class Engine final { */ void _newFile(const std::filesystem::path &path) const; + /// Materialize and snapshot a normalized concrete runtime entry under one lock. + CapioCLEntry materializeRuntimeEntry(const std::filesystem::path &path) const; + /** * @brief Updates the number of entries in the parent directory of the given path. * @@ -427,6 +430,15 @@ class Engine final { */ bool isCommitted(const std::filesystem::path &path) const; + /** + * @brief Record one qualified logical close for an ON_CLOSE path. + * Filesystem-backed counts greater than one require CAPIO_METADATA_DIR to identify a trusted, + * unique-per-run metadata directory with coherent atomic exclusive create/unlink. Multicast + * counts are volatile and eventual. + * @return true when the path is effectively committed after this call. + */ + bool increaseCloseCount(const std::filesystem::path &path) const; + /** * Set file indicated by path as committed * @param path @@ -466,4 +478,4 @@ class Engine final { } // namespace capiocl::engine -#endif // CAPIO_CL_ENGINE_H \ No newline at end of file +#endif // CAPIO_CL_ENGINE_H diff --git a/capiocl/monitor.h b/capiocl/monitor.h index 32bdc7b..146b0f5 100644 --- a/capiocl/monitor.h +++ b/capiocl/monitor.h @@ -3,8 +3,10 @@ #include #include +#include #include #include +#include #include #include #include @@ -106,6 +108,13 @@ class MonitorInterface { */ virtual void setCommitted(const std::filesystem::path &path) const; + /** + * @brief Atomically record a close and commit when @p threshold is reached. + * @return the committed state, or std::nullopt when this backend has no persistent counter. + */ + virtual std::optional increaseCloseCount(const std::filesystem::path &path, + long threshold) const; + /** * Set the current hostname as the home node for a given path * @param path @@ -149,6 +158,12 @@ class MulticastMonitor final : public MonitorInterface { /// @brief variable to terminate execution std::atomic terminate = false; + /// Per-path maximum close-count snapshot observed from each origin. + mutable std::unordered_map> + close_counts; + mutable std::mutex close_count_lock; + std::string close_count_origin; + /// @brief Multicast poll timeout interval static constexpr int MULTICAST_THREAD_POLL_INTERVAL = 250; @@ -165,18 +180,18 @@ class MulticastMonitor final : public MonitorInterface { /** * @brief Supported network command types for commit messages. */ - typedef enum { SET = '!', GET = '?' } MESSAGE_COMMANDS; + typedef enum { SET = '!', GET = '?', COUNT_SET = 'C', COUNT_GET = 'Q' } MESSAGE_COMMANDS; /** * @brief Send a commit or request message over multicast. * * @param ip_addr Destination multicast address. * @param ip_port Destination multicast port. - * @param path File path associated with the message. + * @param payload Message payload. * @param action The type of message to send (COMMIT or REQUEST). */ - static void _send_message(const std::string &ip_addr, int ip_port, const std::string &path, - MESSAGE_COMMANDS action); + static void _send_message(const std::string &ip_addr, int ip_port, const std::string &payload, + MESSAGE_COMMANDS action); /** * @brief Background thread function to listen for commit messages. @@ -192,6 +207,10 @@ class MulticastMonitor final : public MonitorInterface { * @param terminate Atomic Boolean flag to terminate thread */ static void commit_listener(std::vector &committed_files, std::mutex &lock, + std::unordered_map> + &close_counts, + std::mutex &close_count_lock, const std::string &ip_addr, int ip_port, const std::atomic *terminate); @@ -227,6 +246,8 @@ class MulticastMonitor final : public MonitorInterface { bool isCommitted(const std::filesystem::path &path) const override; void setCommitted(const std::filesystem::path &path) const override; + std::optional increaseCloseCount(const std::filesystem::path &path, + long threshold) const override; void setHomeNode(const std::filesystem::path &path) const override; std::string getHomeNode(const std::filesystem::path &path) const override; }; @@ -286,6 +307,8 @@ class FileSystemMonitor final : public MonitorInterface { bool isCommitted(const std::filesystem::path &path) const override; void setCommitted(const std::filesystem::path &path) const override; + std::optional increaseCloseCount(const std::filesystem::path &path, + long threshold) const override; void setHomeNode(const std::filesystem::path &path) const override; std::string getHomeNode(const std::filesystem::path &path) const override; }; @@ -319,6 +342,9 @@ class Monitor { */ void setCommitted(std::filesystem::path path) const; + /// @brief Record a close independently in every counter-capable backend. + [[nodiscard]] bool increaseCloseCount(const std::filesystem::path &path, long threshold) const; + /** * Add a new backend for monitor. Must be a derived class from MonitorInterface * @param interface diff --git a/doxygen/semantics.md b/doxygen/semantics.md index 6a7c88c..ab03254 100644 --- a/doxygen/semantics.md +++ b/doxygen/semantics.md @@ -24,7 +24,18 @@ commit behaviors: - `Commit on Close (CoC)`: This behavior allows subsequent steps to initiate reading a file as soon as the producer step invokes a close operation on a given file, signaling that all I/O operations on that file are completed. It is also - allowed to consider a file committed after `n` close operations are performed. The following Gant diagram visually + allowed to consider a file committed after `n` close operations are performed. Counts of zero or one commit on the + first close; larger counts require a counter-capable filesystem or multicast monitor. Filesystem counters require + `CAPIO_METADATA_DIR` to name a trusted, non-attacker-writable directory unique to the workflow run. CAPIO-CL owns its + `capiocl` subtree. Multi-process and multi-node producers must share this metadata directory through a distributed + filesystem providing coherent atomic exclusive file creation and unlink. + Transient locks use a private subtree separate from counters. Counter replacement is atomic, but CAPIO-CL does not + claim power-loss durability for the latest close on distributed storage. + A producer crash can leave a stale lock file; CAPIO-CL does not steal or expire such locks automatically. + Multicast counters exchange monotonic per-origin snapshots and are eventual and volatile. UDP loss can defer + commitment, and origin death before propagation can lose a close; there is no durability, consensus, or + authentication. + The following Gant diagram visually explains the CoC semantics: ![The Commit on Close rule](media/coc.png){ width=60% } @@ -33,7 +44,9 @@ commit behaviors: when the number of `open()` and `close()` system calls operations for a given file is not statically known. Instead, we are aware that the I/O operations can be considered concluded on a given file if another file has been committed. This additional commit behavior introduces a dependency among files in the commit rule, expanding opportunities to - leverage temporal parallelism for I/O operations across different workflow steps.The following Gant diagram visually + leverage temporal parallelism for I/O operations across different workflow steps. Multiple dependencies use AND + semantics and are evaluated transitively; unresolved dependency cycles remain uncommitted. Runtime dependencies are + concrete literal paths, and dependency glob patterns do not resolve. The following Gant diagram visually explains the CoF semantics: ![The Commit on File rule](media/cof.png){ width=60% } diff --git a/src/Engine.cpp b/src/Engine.cpp index 321441d..dcce585 100644 --- a/src/Engine.cpp +++ b/src/Engine.cpp @@ -2,6 +2,7 @@ #include #include #include +#include #include "calf/StdOutLogger.h" #include "calf/StlLogger.h" @@ -32,6 +33,13 @@ template class shared_lock_guard { SharedMutex &mutex_; }; +namespace { +std::filesystem::path normalize_runtime_path(const std::filesystem::path &path) { + return std::filesystem::absolute(path).lexically_normal(); +} + +} // namespace + void capiocl::engine::Engine::print() const { START_LOG(calf_current_tid(), "call()"); UPDATE_CALF_WORKFLOW_NAME(workflow_name); @@ -188,7 +196,10 @@ void capiocl::engine::Engine::_newFile(const std::filesystem::path &path) const std::string matchKey; size_t matchSize = 0; for (const auto &[filename, data] : _capio_cl_entries) { - if (const bool match = fnmatch(filename.c_str(), path.c_str(), FNM_NOESCAPE) == 0; + const auto normalized_pattern = normalize_runtime_path(filename).string(); + if (const bool match = + fnmatch(filename.c_str(), path.c_str(), FNM_NOESCAPE) == 0 || + fnmatch(normalized_pattern.c_str(), path.c_str(), FNM_NOESCAPE) == 0; match && filename.length() > matchSize) { matchSize = filename.length(); matchKey = filename; @@ -222,6 +233,13 @@ void capiocl::engine::Engine::_newFile(const std::filesystem::path &path) const } } +capiocl::engine::CapioCLEntry +capiocl::engine::Engine::materializeRuntimeEntry(const std::filesystem::path &path) const { + std::lock_guard lock(_shared_mutex); + _newFile(path); + return _capio_cl_entries.at(path); +} + void capiocl::engine::Engine::compute_directory_entry_count( const std::filesystem::path &path) const { START_LOG(calf_current_tid(), "call()"); @@ -284,6 +302,9 @@ void capiocl::engine::Engine::add(std::filesystem::path &path, std::vector resolved; + std::unordered_set visiting; + + const auto evaluate = [&](auto &&self, const std::filesystem::path &candidate) -> bool { + if (candidate.empty() || candidate.string().find_first_of("*?[") != std::string::npos) { + return false; + } + const auto normalized = normalize_runtime_path(candidate); + const auto key = normalized.string(); + if (monitor.isCommitted(normalized)) { + resolved[key] = true; + return true; + } + + if (const auto result = resolved.find(key); result != resolved.end()) { + return result->second; + } + if (visiting.find(key) != visiting.end()) { + return false; + } + visiting.insert(key); + + const auto entry = materializeRuntimeEntry(normalized); + + bool committed = false; + if (entry.commit_rule == commitRules::ON_FILE && !entry.file_dependencies.empty()) { + committed = std::all_of(entry.file_dependencies.begin(), entry.file_dependencies.end(), + [&](const auto &dependency) { return self(self, dependency); }); + if (committed) { + setCommitted(normalized); + } + } + visiting.erase(key); + resolved[key] = committed; + return committed; + }; + + return evaluate(evaluate, path); +} + +bool capiocl::engine::Engine::increaseCloseCount(const std::filesystem::path &path) const { + START_LOG(calf_current_tid(), "call()"); + if (path.empty()) { + return false; + } + const auto normalized = normalize_runtime_path(path); + if (monitor.isCommitted(normalized)) { + return true; + } + + const auto entry = materializeRuntimeEntry(normalized); + const auto &rule = entry.commit_rule; + const long threshold = entry.commit_on_close_count; + + if (rule != commitRules::ON_CLOSE) { + return isCommitted(normalized); + } + if (threshold <= 1) { + setCommitted(normalized); + return monitor.isCommitted(normalized); + } + return monitor.increaseCloseCount(normalized, threshold) || isCommitted(normalized); } void capiocl::engine::Engine::setCommitted(const std::filesystem::path &path) const { @@ -661,6 +743,9 @@ bool capiocl::engine::Engine::isDirectory(const std::filesystem::path &path) con void capiocl::engine::Engine::setCommitedCloseNumber(const std::filesystem::path &path, const long num) { START_LOG(calf_current_tid(), "call()"); + if (num < 0) { + throw std::invalid_argument("ON_CLOSE threshold cannot be negative"); + } if (path.empty()) { return; } diff --git a/src/Monitor.cpp b/src/Monitor.cpp index c742f3a..dddbce0 100644 --- a/src/Monitor.cpp +++ b/src/Monitor.cpp @@ -10,17 +10,45 @@ capiocl::monitor::MonitorException::MonitorException(const std::string &msg) : m bool capiocl::monitor::Monitor::isCommitted(const std::filesystem::path &path) const { START_LOG(calf_current_tid(), "call()"); - return std::any_of(interfaces.begin(), interfaces.end(), - [&path](const auto &interface) { return interface->isCommitted(path); }); + const auto normalized = std::filesystem::absolute(path).lexically_normal(); + return std::any_of(interfaces.begin(), interfaces.end(), [&normalized](const auto &interface) { + return interface->isCommitted(normalized); + }); } void capiocl::monitor::Monitor::setCommitted(std::filesystem::path path) const { START_LOG(calf_current_tid(), "call()"); + path = std::filesystem::absolute(path).lexically_normal(); LOG("setting committed path=%s backends=%zu", path.string().c_str(), interfaces.size()); std::for_each(interfaces.begin(), interfaces.end(), [&path](const auto &interface) { interface->setCommitted(path); }); } +bool capiocl::monitor::Monitor::increaseCloseCount(const std::filesystem::path &path, + const long threshold) const { + START_LOG(calf_current_tid(), "call()"); + if (threshold <= 1) { + throw std::invalid_argument("Persistent ON_CLOSE threshold must be greater than one"); + } + const auto normalized = std::filesystem::absolute(path).lexically_normal(); + bool supported = false; + bool reached = false; + for (const auto &interface : interfaces) { + if (const auto committed = interface->increaseCloseCount(normalized, threshold); + committed.has_value()) { + supported = true; + reached |= *committed; + } + } + if (!supported) { + throw MonitorException("ON_CLOSE count greater than one requires a counting monitor"); + } + if (reached) { + setCommitted(normalized); + } + return reached; +} + void capiocl::monitor::Monitor::registerMonitorBackend(const MonitorInterface *interface) { START_LOG(calf_current_tid(), "call()"); interfaces.emplace_back(interface); diff --git a/src/monitors/FileSystem.cpp b/src/monitors/FileSystem.cpp index c58992f..f5c03cb 100644 --- a/src/monitors/FileSystem.cpp +++ b/src/monitors/FileSystem.cpp @@ -1,10 +1,166 @@ +#include +#include +#include #include +#include +#include +#include #include +#include #include "calf/StlLogger.h" #include "capiocl.hpp" #include "capiocl/monitor.h" +[[noreturn]] void monitor_error(const std::string &operation, const std::filesystem::path &path, + const int error = errno) { + throw capiocl::monitor::MonitorException(operation + " " + path.string() + ": " + + strerror(error)); +} + +struct CloseMetadataPaths final { + std::filesystem::path counter; + std::filesystem::path lock; +}; + +CloseMetadataPaths close_metadata_paths(const std::filesystem::path &path) { + const char *configured = std::getenv("CAPIO_METADATA_DIR"); + if (configured == nullptr || configured[0] == '\0') { + throw capiocl::monitor::MonitorException( + "Counted ON_CLOSE requires CAPIO_METADATA_DIR to name a trusted, unique workflow " + "metadata directory"); + } + const auto root = std::filesystem::absolute(configured).lexically_normal() / "capiocl"; + const auto normalized = std::filesystem::absolute(path).lexically_normal().generic_string(); + static constexpr char digits[] = "0123456789abcdef"; + std::string encoded; + encoded.reserve(normalized.size() * 2); + for (const unsigned char character : normalized) { + encoded.push_back(digits[character >> 4]); + encoded.push_back(digits[character & 0x0f]); + } + + auto make_directory = [&](const char *subtree) { + auto directory = root / subtree; + for (size_t offset = 0; offset < encoded.size(); offset += 64) { + directory /= encoded.substr(offset, 64); + } + std::error_code error; + std::filesystem::create_directories(directory, error); + const auto status = std::filesystem::symlink_status(directory, error); + if (error || std::filesystem::is_symlink(status) || + !std::filesystem::is_directory(status)) { + throw capiocl::monitor::MonitorException("Unsafe CAPIO-CL metadata directory " + + directory.string()); + } + return directory; + }; + return {make_directory("close-counts") / "count", make_directory("close-locks") / "lock"}; +} + +long read_counter(const std::filesystem::path &path) { + std::error_code error; + const auto status = std::filesystem::symlink_status(path, error); + if (error == std::errc::no_such_file_or_directory) { + return 0; + } + if (error || !std::filesystem::exists(status) || std::filesystem::is_symlink(status) || + !std::filesystem::is_regular_file(status)) { + throw capiocl::monitor::MonitorException("Unsafe close counter " + path.string()); + } + std::ifstream file(path); + long value = 0; + if (!(file >> value) || value < 0) { + throw capiocl::monitor::MonitorException("Malformed close counter " + path.string()); + } + file >> std::ws; + if (!file.eof()) { + throw capiocl::monitor::MonitorException("Malformed close counter " + path.string()); + } + return value; +} + +// ponytail: atomic rename avoids partial counters; a machine crash can still lose the latest close. +void persist_counter(const std::filesystem::path &counter, const long value) { + auto temporary = counter; + temporary += ".tmp"; + std::error_code error; + if (std::filesystem::is_symlink(std::filesystem::symlink_status(temporary, error))) { + throw capiocl::monitor::MonitorException("Unsafe temporary close counter " + + temporary.string()); + } + try { + { + std::ofstream file(temporary, std::ios::trunc); + file << value << '\n'; + file.close(); + if (!file.good()) { + throw capiocl::monitor::MonitorException("Unable to write close counter " + + temporary.string()); + } + } + const auto status = std::filesystem::symlink_status(counter, error); + if (!error && std::filesystem::is_symlink(status)) { + throw capiocl::monitor::MonitorException("Unsafe close counter " + counter.string()); + } + std::filesystem::rename(temporary, counter, error); + if (error) { + throw capiocl::monitor::MonitorException("Unable to replace close counter " + + counter.string() + ": " + error.message()); + } + } catch (...) { + std::filesystem::remove(temporary, error); + throw; + } +} + +class FileLock final { + std::filesystem::path path; + int fd = -1; + + public: + explicit FileLock(std::filesystem::path lock_path) : path(std::move(lock_path)) { + while (true) { + fd = open(path.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC | O_NOFOLLOW, 0600); + if (fd != -1) { + return; + } + const int error = errno; + if (error != EEXIST) { + monitor_error("Unable to acquire close counter lock", path, error); + } + struct stat status{}; + if (lstat(path.c_str(), &status) == 0) { + if (!S_ISREG(status.st_mode)) { + throw capiocl::monitor::MonitorException("Unsafe close counter lock " + + path.string()); + } + } else if (errno != ENOENT) { + monitor_error("Unable to inspect close counter lock", path); + } + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + } + + ~FileLock() { + if (fd == -1) { + return; + } + struct stat owned{}, current{}; + const bool still_owned = fstat(fd, &owned) == 0 && lstat(path.c_str(), ¤t) == 0 && + owned.st_dev == current.st_dev && owned.st_ino == current.st_ino; + close(fd); + if (still_owned) { + unlink(path.c_str()); + } + } + + FileLock(const FileLock &) = delete; + FileLock &operator=(const FileLock &) = delete; + FileLock(FileLock &&) = delete; + FileLock &operator=(FileLock &&) = delete; +}; + std::filesystem::path capiocl::monitor::FileSystemMonitor::compute_capiocl_token_name(const std::filesystem::path &path, CAPIO_CL_COMMIT_TOKEN_TYPES type) { @@ -13,11 +169,11 @@ capiocl::monitor::FileSystemMonitor::compute_capiocl_token_name(const std::files if (type == COMMIT) { token_type = ".commit"; - } else { + } else if (type == HOME_NODE) { token_type = ".home_node"; } - const auto abs = std::filesystem::absolute(path); + const auto abs = std::filesystem::absolute(path).lexically_normal(); const auto new_filename = "." + abs.filename().string() + token_type; return abs.parent_path() / new_filename; } @@ -43,17 +199,19 @@ void capiocl::monitor::FileSystemMonitor::generate_home_node_token( void capiocl::monitor::FileSystemMonitor::generate_commit_token(const std::filesystem::path &path) { START_LOG(calf_current_tid(), "call()"); - if (const auto token_name = compute_capiocl_token_name(path, COMMIT); - !std::filesystem::exists(token_name)) { - std::filesystem::create_directories(token_name.parent_path()); - std::ofstream file(token_name); - if (!file.good()) { - LOG("failed to create commit token=%s", token_name.string().c_str()); - } else { - LOG("created commit token=%s", token_name.string().c_str()); - } - file.close(); + const auto token_name = compute_capiocl_token_name(path, COMMIT); + std::filesystem::create_directories(token_name.parent_path()); + std::error_code error; + const auto status = std::filesystem::symlink_status(token_name, error); + if (!error && + (std::filesystem::is_symlink(status) || !std::filesystem::is_regular_file(status))) { + throw MonitorException("Unsafe commit token " + token_name.string()); + } + std::ofstream file(token_name, std::ios::app); + if (!file.good()) { + throw MonitorException("Unable to create commit token " + token_name.string()); } + LOG("created commit token=%s", token_name.string().c_str()); } capiocl::monitor::FileSystemMonitor::FileSystemMonitor() { @@ -73,6 +231,28 @@ bool capiocl::monitor::FileSystemMonitor::isCommitted(const std::filesystem::pat return std::filesystem::exists(compute_capiocl_token_name(path)); } +std::optional +capiocl::monitor::FileSystemMonitor::increaseCloseCount(const std::filesystem::path &path, + const long threshold) const { + START_LOG(calf_current_tid(), "call()"); + if (threshold <= 1) { + throw std::invalid_argument("Persistent ON_CLOSE threshold must be greater than one"); + } + const auto metadata = close_metadata_paths(path); + FileLock lock(metadata.lock); + long value = read_counter(metadata.counter); + + if (value == std::numeric_limits::max()) { + throw MonitorException("Close counter overflow for " + metadata.counter.string()); + } + ++value; + if (value >= threshold) { + generate_commit_token(path); + } + persist_counter(metadata.counter, value); + return value >= threshold; +} + void capiocl::monitor::FileSystemMonitor::setHomeNode(const std::filesystem::path &path) const { START_LOG(calf_current_tid(), "call()"); generate_home_node_token(path, _hostname); diff --git a/src/monitors/Interface.cpp b/src/monitors/Interface.cpp index d30fdeb..40ed04e 100644 --- a/src/monitors/Interface.cpp +++ b/src/monitors/Interface.cpp @@ -15,6 +15,16 @@ void capiocl::monitor::MonitorInterface::setCommitted(const std::filesystem::pat msg += path.string(); throw MonitorException(msg); } + +std::optional +capiocl::monitor::MonitorInterface::increaseCloseCount(const std::filesystem::path &, + const long threshold) const { + if (threshold <= 1) { + throw std::invalid_argument("Persistent ON_CLOSE threshold must be greater than one"); + } + return std::nullopt; +} + void capiocl::monitor::MonitorInterface::setHomeNode(const std::filesystem::path &path) const { START_LOG(calf_current_tid(), "call()"); std::string msg = "Attempted to use MonitorInterface as Monitor backend to set commit for: "; diff --git a/src/monitors/Multicast.cpp b/src/monitors/Multicast.cpp index e59f0cf..e4bd134 100644 --- a/src/monitors/Multicast.cpp +++ b/src/monitors/Multicast.cpp @@ -1,4 +1,8 @@ +#include #include +#include +#include +#include #include #include #include @@ -8,6 +12,9 @@ #include "capiocl.hpp" #include "capiocl/monitor.h" + +std::atomic close_count_origin_sequence{0}; + static std::tuple outgoing_socket_multicast(const std::string &address, const int port) { START_LOG(calf_current_tid(), "call()"); @@ -99,11 +106,11 @@ static int incoming_socket_multicast(const std::string &address_ip, const int po return _socket; } -void capiocl::monitor::MulticastMonitor::commit_listener(std::vector &committed_files, - std::mutex &lock, - const std::string &ip_addr, - const int ip_port, - const std::atomic *terminate) { +void capiocl::monitor::MulticastMonitor::commit_listener( + std::vector &committed_files, std::mutex &lock, + std::unordered_map> &close_counts, + std::mutex &close_count_lock, const std::string &ip_addr, const int ip_port, + const std::atomic *terminate) { START_LOG(calf_current_tid(), "call()"); sockaddr_in addr_in = {}; socklen_t addr_len = {}; @@ -113,8 +120,8 @@ void capiocl::monitor::MulticastMonitor::commit_listener(std::vector(&addr_in); - char incoming_message[MESSAGE_SIZE] = {0}; + const auto addr = reinterpret_cast(&addr_in); + char incoming_message[MESSAGE_SIZE + 1] = {0}; // Polling for non blocking pollfd pfd = {}; @@ -122,30 +129,32 @@ void capiocl::monitor::MulticastMonitor::commit_listener(std::vector MESSAGE_SIZE) { + LOG("multicast message discarded reason=oversize size=%zd limit=%d", incoming_size, + MESSAGE_SIZE); + continue; + } const std::string msg(incoming_message, static_cast(incoming_size)); - if (msg.size() < 2) { + if (msg.size() < 3 || msg[1] != ' ') { continue; } const auto path = msg.substr(2); @@ -158,13 +167,51 @@ void capiocl::monitor::MulticastMonitor::commit_listener(std::vector= path.size()) { + continue; + } + const auto origin = path.substr(0, origin_end); + if (origin.find_first_of(" \t\r\n") != std::string::npos) { + continue; + } + std::uint64_t count = 0; + const auto count_text = path.substr(origin_end + 1, count_end - origin_end - 1); + const auto parsed = + std::from_chars(count_text.data(), count_text.data() + count_text.size(), count); + if (count == 0 || parsed.ec != std::errc{} || + parsed.ptr != count_text.data() + count_text.size()) { + continue; + } + const auto count_path = + std::filesystem::absolute(path.substr(count_end + 1)).lexically_normal().string(); + std::lock_guard count_guard(close_count_lock); + auto &known = close_counts[count_path][origin]; + known = std::max(known, count); + } else if (command == COUNT_GET && !path.empty()) { + const auto count_path = std::filesystem::absolute(path).lexically_normal().string(); + std::vector> snapshots; + { + std::lock_guard count_guard(close_count_lock); + if (const auto known = close_counts.find(count_path); known != close_counts.end()) { + snapshots.assign(known->second.begin(), known->second.end()); + } + } + for (const auto &[origin, count] : snapshots) { + _send_message(ip_addr, ip_port, + origin + " " + std::to_string(count) + " " + count_path, COUNT_SET); + } } } while (true); } @@ -186,10 +233,14 @@ void capiocl::monitor::MulticastMonitor::home_node_listener( return; } - const auto addr = reinterpret_cast(&addr_in); - char incoming_message[MESSAGE_SIZE] = {0}; + const auto addr = reinterpret_cast(&addr_in); + char incoming_message[MESSAGE_SIZE + 1] = {0}; do { + if (*terminate) { + close(socket); + return; + } bzero(incoming_message, sizeof(incoming_message)); // Polling for non blocking @@ -199,25 +250,26 @@ void capiocl::monitor::MulticastMonitor::home_node_listener( // TODO: migrate to epoll for linux and kqueue on MacOS if (poll(&pfd, 1, MULTICAST_THREAD_POLL_INTERVAL) == 0) { - // No data from incoming socket. Continue, awaking thread ensuring pthread_cancel points - // can be reached - if (*terminate) { - close(socket); - return; - } - continue; } // LCOV_EXCL_START - const auto incoming_size = - recvfrom(socket, incoming_message, MESSAGE_SIZE, MSG_DONTWAIT, addr, &addr_len); + const auto incoming_size = recvfrom(socket, incoming_message, sizeof(incoming_message), + MSG_DONTWAIT, addr, &addr_len); if (incoming_size < 0) { continue; } // LCOV_EXCL_STOP + if (incoming_size > MESSAGE_SIZE) { + LOG("multicast home-node message discarded reason=oversize size=%zd limit=%d", + incoming_size, MESSAGE_SIZE); + continue; + } std::string incoming_message_str(incoming_message, incoming_size); + if (incoming_message_str.size() < 3 || incoming_message_str[1] != ' ') { + continue; + } std::vector tokens; size_t start = 0, end = incoming_message_str.find(' '); @@ -248,7 +300,7 @@ void capiocl::monitor::MulticastMonitor::home_node_listener( std::lock_guard lg(lock); home_nodes[path] = home_node; LOG("received home node path=%s node=%s", path.c_str(), home_node.c_str()); - } else { + } else if (command == GET) { // Received a query for a home node, Message begins with capiocl::Monitor::REQUEST if (tokens.size() < 2) { // need "? " -> malformed message, skip @@ -267,19 +319,24 @@ void capiocl::monitor::MulticastMonitor::home_node_listener( } void capiocl::monitor::MulticastMonitor::_send_message(const std::string &ip_addr, - const int ip_port, const std::string &path, + const int ip_port, + const std::string &payload, const MESSAGE_COMMANDS action) { START_LOG(calf_current_tid(), "call()"); - char message[MESSAGE_SIZE] = {0}; - snprintf(message, sizeof(message), "%c %s", action, path.c_str()); + const std::string message = static_cast(action) + std::string(" ") + payload; + if (message.size() > MESSAGE_SIZE) { + LOG("multicast message rejected reason=oversize size=%zu limit=%d", message.size(), + MESSAGE_SIZE); + return; + } auto [out_s, addr] = outgoing_socket_multicast(ip_addr, ip_port); - if (sendto(out_s, message, strlen(message), 0, reinterpret_cast(&addr), + if (sendto(out_s, message.data(), message.size(), 0, reinterpret_cast(&addr), sizeof(addr)) < 0) { LOG("multicast send failed address=%s port=%d action=%c path=%s errno=%d", ip_addr.c_str(), - ip_port, action, path.c_str(), errno); + ip_port, action, payload.c_str(), errno); } else { LOG("multicast message sent address=%s port=%d action=%c path=%s", ip_addr.c_str(), ip_port, - action, path.c_str()); + action, payload.c_str()); } close(out_s); } @@ -292,20 +349,28 @@ capiocl::monitor::MulticastMonitor::MulticastMonitor( config.getParameter("monitor.mcast.homenode.ip", &MULTICAST_HOME_NODE_ADDR); config.getParameter("monitor.mcast.homenode.port", &MULTICAST_HOME_NODE_PORT); config.getParameter("monitor.mcast.delay_ms", &MULTICAST_DELAY_MILLIS); + gethostname(_hostname, sizeof(_hostname)); + _hostname[sizeof(_hostname) - 1] = '\0'; + std::string origin_host(_hostname); + std::replace_if( + origin_host.begin(), origin_host.end(), + [](const unsigned char character) { return std::isspace(character); }, '_'); + close_count_origin = + origin_host + ":" + std::to_string(getpid()) + ":" + + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()) + ":" + + std::to_string(close_count_origin_sequence++); LOG("multicast monitor configured commit=%s:%d home_node=%s:%d delay_ms=%d", MULTICAST_COMMIT_ADDR.c_str(), MULTICAST_COMMIT_PORT, MULTICAST_HOME_NODE_ADDR.c_str(), MULTICAST_HOME_NODE_PORT, MULTICAST_DELAY_MILLIS); commit_thread = std::thread(&commit_listener, std::ref(_committed_files), std::ref(committed_lock), - MULTICAST_COMMIT_ADDR, MULTICAST_COMMIT_PORT, &this->terminate); + std::ref(close_counts), std::ref(close_count_lock), MULTICAST_COMMIT_ADDR, + MULTICAST_COMMIT_PORT, &this->terminate); home_node_thread = std::thread(&home_node_listener, std::ref(_home_nodes), std::ref(home_node_lock), MULTICAST_HOME_NODE_ADDR, MULTICAST_HOME_NODE_PORT, &this->terminate); - - gethostname(_hostname, sizeof(_hostname)); - _hostname[sizeof(_hostname) - 1] = '\0'; } capiocl::monitor::MulticastMonitor::~MulticastMonitor() { @@ -334,6 +399,51 @@ bool capiocl::monitor::MulticastMonitor::isCommitted(const std::filesystem::path } } +std::optional +capiocl::monitor::MulticastMonitor::increaseCloseCount(const std::filesystem::path &path, + const long threshold) const { + START_LOG(calf_current_tid(), "call()"); + if (threshold <= 1) { + throw std::invalid_argument("Persistent ON_CLOSE threshold must be greater than one"); + } + const auto normalized = std::filesystem::absolute(path).lexically_normal().string(); + const auto reached = [&] { + std::uint64_t remaining = static_cast(threshold); + if (const auto known = close_counts.find(normalized); known != close_counts.end()) { + for (const auto &[_, count] : known->second) { + if (count >= remaining) { + return true; + } + remaining -= count; + } + } + return false; + }; + + std::uint64_t snapshot; + bool threshold_reached; + { + std::lock_guard guard(close_count_lock); + auto &own_count = close_counts[normalized][close_count_origin]; + if (own_count == std::numeric_limits::max()) { + throw MonitorException("Multicast close counter overflow for " + normalized); + } + snapshot = ++own_count; + threshold_reached = reached(); + } + _send_message(MULTICAST_COMMIT_ADDR, MULTICAST_COMMIT_PORT, + close_count_origin + " " + std::to_string(snapshot) + " " + normalized, + COUNT_SET); + if (threshold_reached) { + return true; + } + + _send_message(MULTICAST_COMMIT_ADDR, MULTICAST_COMMIT_PORT, normalized, COUNT_GET); + std::this_thread::sleep_for(std::chrono::milliseconds(MULTICAST_DELAY_MILLIS)); + std::lock_guard guard(close_count_lock); + return reached(); +} + void capiocl::monitor::MulticastMonitor::setCommitted(const std::filesystem::path &path) const { START_LOG(calf_current_tid(), "call()"); _send_message(MULTICAST_COMMIT_ADDR, MULTICAST_COMMIT_PORT, std::filesystem::path(path), SET); diff --git a/tests/cpp/main.cpp b/tests/cpp/main.cpp index 1163cb9..adb80a5 100644 --- a/tests/cpp/main.cpp +++ b/tests/cpp/main.cpp @@ -25,4 +25,5 @@ template std::string demangled_name(const T &obj) { #include "test_engine.hpp" #include "test_exceptions.hpp" #include "test_monitor.hpp" -#include "test_serialize_deserialize.hpp" \ No newline at end of file +#include "test_runtime_commit.hpp" +#include "test_serialize_deserialize.hpp" diff --git a/tests/cpp/test_runtime_commit.hpp b/tests/cpp/test_runtime_commit.hpp new file mode 100644 index 0000000..4e2b882 --- /dev/null +++ b/tests/cpp/test_runtime_commit.hpp @@ -0,0 +1,533 @@ +#ifndef CAPIO_CL_RUNTIME_COMMIT_HPP +#define CAPIO_CL_RUNTIME_COMMIT_HPP + +#include +#include +#include +#include +#include +#include +#include +#include + +#define RUNTIME_COMMIT_SUITE_NAME testRuntimeCommit + +namespace { +class RuntimeTestScope final { + static std::atomic sequence; + bool had_metadata_root = false; + std::string previous_metadata_root; + + static std::string hexEncode(const std::string &value) { + static constexpr char digits[] = "0123456789abcdef"; + std::string encoded; + for (const unsigned char character : value) { + encoded.push_back(digits[character >> 4]); + encoded.push_back(digits[character & 0x0f]); + } + return encoded; + } + + public: + std::filesystem::path root; + std::filesystem::path data; + std::filesystem::path metadata; + + RuntimeTestScope() + : root(std::filesystem::temp_directory_path() / + ("capiocl-runtime-" + std::to_string(getpid()) + "-" + + std::to_string(sequence++))), + data(root / "data"), metadata(root / "metadata") { + if (const char *current = std::getenv("CAPIO_METADATA_DIR")) { + had_metadata_root = true; + previous_metadata_root = current; + } + std::filesystem::remove_all(root); + std::filesystem::create_directories(data); + setenv("CAPIO_METADATA_DIR", metadata.c_str(), 1); + } + + ~RuntimeTestScope() { + if (had_metadata_root) { + setenv("CAPIO_METADATA_DIR", previous_metadata_root.c_str(), 1); + } else { + unsetenv("CAPIO_METADATA_DIR"); + } + std::filesystem::remove_all(root); + } + + [[nodiscard]] std::filesystem::path file(const std::string &name) const { + return data / name; + } + + [[nodiscard]] std::filesystem::path closeMetadata(const std::filesystem::path &path, + const std::string &name) const { + const auto encoded = hexEncode( + std::filesystem::absolute(path).lexically_normal().generic_string()); + auto directory = metadata / "capiocl" / "close-counts"; + for (size_t offset = 0; offset < encoded.size(); offset += 64) { + directory /= encoded.substr(offset, 64); + } + return directory / name; + } + + [[nodiscard]] std::filesystem::path closeLock(const std::filesystem::path &path) const { + const auto encoded = hexEncode( + std::filesystem::absolute(path).lexically_normal().generic_string()); + auto directory = metadata / "capiocl" / "close-locks"; + for (size_t offset = 0; offset < encoded.size(); offset += 64) { + directory /= encoded.substr(offset, 64); + } + return directory / "lock"; + } +}; + +std::atomic RuntimeTestScope::sequence{0}; + +void useFilesystemMonitor(capiocl::engine::Engine &engine) { + engine.loadConfiguration("/tmp/capio_cl_tomls/runtime_fs_only.toml"); +} + +void useMulticastMonitor(capiocl::engine::Engine &engine) { + engine.loadConfiguration("/tmp/capio_cl_tomls/runtime_multicast_only.toml"); +} + +void sendMulticastCount(const std::string &message) { + const int socket_fd = socket(AF_INET, SOCK_DGRAM, 0); + ASSERT_NE(socket_fd, -1); + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = inet_addr("224.224.226.10"); + address.sin_port = htons(26101); + ASSERT_EQ(sendto(socket_fd, message.data(), message.size(), 0, + reinterpret_cast(&address), sizeof(address)), + static_cast(message.size())); + close(socket_fd); +} + +void configureClose(capiocl::engine::Engine &engine, const std::filesystem::path &path, + const long threshold) { + engine.setCommitRule(path, capiocl::commitRules::ON_CLOSE); + engine.setCommitedCloseNumber(path, threshold); +} + +void configureFile(capiocl::engine::Engine &engine, const std::filesystem::path &path, + const std::vector &dependencies) { + engine.setCommitRule(path, capiocl::commitRules::ON_FILE); + if (!dependencies.empty()) { + engine.setFileDeps(path, dependencies); + } +} +} // namespace + +TEST(RUNTIME_COMMIT_SUITE_NAME, onClosePlainAndOneCommitOnFirstClose) { + RuntimeTestScope scope; + for (const long threshold : {0L, 1L}) { + const auto path = scope.file("plain-" + std::to_string(threshold)); + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureClose(engine, path, threshold); + + EXPECT_TRUE(engine.increaseCloseCount(path)); + EXPECT_TRUE(engine.isCommitted(path)); + EXPECT_FALSE(std::filesystem::exists(scope.closeMetadata(path, "count"))); + } +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, countedCloseRequiresConfiguredMetadataRoot) { + RuntimeTestScope scope; + unsetenv("CAPIO_METADATA_DIR"); + const auto counted = scope.file("missing-metadata"); + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureClose(engine, counted, 2); + + try { + engine.increaseCloseCount(counted); + FAIL() << "counted ON_CLOSE accepted a missing CAPIO_METADATA_DIR"; + } catch (const capiocl::monitor::MonitorException &error) { + EXPECT_NE(std::string(error.what()).find("CAPIO_METADATA_DIR"), std::string::npos); + } + EXPECT_FALSE(engine.isCommitted(counted)); + EXPECT_FALSE(std::filesystem::exists(counted.parent_path() / + ("." + counted.filename().string() + ".commit"))); + + const auto plain = scope.file("plain-without-metadata"); + configureClose(engine, plain, 1); + EXPECT_TRUE(engine.increaseCloseCount(plain)); + EXPECT_TRUE(engine.isCommitted(plain)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, onCloseThresholdTwoAndNonClose) { + RuntimeTestScope scope; + const auto path = scope.file("threshold-two"); + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureClose(engine, path, 2); + + EXPECT_FALSE(engine.increaseCloseCount(path)); + EXPECT_TRUE(std::filesystem::exists(scope.closeMetadata(path, "count"))); + EXPECT_NE(scope.closeMetadata(path, "count").parent_path(), + scope.closeLock(path).parent_path()); + EXPECT_FALSE(std::filesystem::exists(scope.closeLock(path))); + EXPECT_TRUE(engine.increaseCloseCount(path)); + EXPECT_FALSE(std::filesystem::exists(scope.closeLock(path))); + EXPECT_TRUE(engine.increaseCloseCount(path)); + + const auto other = scope.file("not-on-close"); + EXPECT_FALSE(engine.increaseCloseCount(other)); + EXPECT_FALSE(std::filesystem::exists(scope.closeMetadata(other, "count"))); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, onCloseCountPersistsAcrossEngines) { + RuntimeTestScope scope; + const auto path = scope.file("persistent"); + { + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureClose(engine, path, 3); + EXPECT_FALSE(engine.increaseCloseCount(path)); + EXPECT_FALSE(engine.increaseCloseCount(path)); + } + { + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureClose(engine, path, 3); + EXPECT_TRUE(engine.increaseCloseCount(path)); + } +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, onCloseIsThreadSafe) { + RuntimeTestScope scope; + const auto path = scope.file("threads"); + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureClose(engine, path, 16); + std::vector threads; + for (int i = 0; i < 16; ++i) { + threads.emplace_back([&] { engine.increaseCloseCount(path); }); + } + for (auto &thread : threads) { + thread.join(); + } + EXPECT_TRUE(engine.isCommitted(path)); + EXPECT_FALSE(std::filesystem::exists(scope.closeLock(path))); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, multicastCloseCountAcrossOriginsAndSpaces) { + RuntimeTestScope scope; + const auto path = scope.file("multicast path with spaces"); + capiocl::engine::Engine first(false), second(false); + useMulticastMonitor(first); + useMulticastMonitor(second); + configureClose(first, path, 2); + configureClose(second, path, 2); + + EXPECT_FALSE(first.increaseCloseCount(path)); + EXPECT_TRUE(second.increaseCloseCount(path)); + EXPECT_TRUE(first.isCommitted(path)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, multicastCloseCountReconcilesLateObserver) { + RuntimeTestScope scope; + const auto path = scope.file("late-observer"); + capiocl::engine::Engine first(false); + useMulticastMonitor(first); + configureClose(first, path, 2); + EXPECT_FALSE(first.increaseCloseCount(path)); + + capiocl::engine::Engine second(false); + useMulticastMonitor(second); + configureClose(second, path, 2); + EXPECT_TRUE(second.increaseCloseCount(path)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, multicastSnapshotsMergeByMaximum) { + RuntimeTestScope scope; + const auto path = scope.file("snapshot-merge"); + capiocl::engine::Engine engine(false); + useMulticastMonitor(engine); + configureClose(engine, path, 6); + usleep(50000); + const auto normalized = std::filesystem::absolute(path).lexically_normal().string(); + sendMulticastCount("C injected 5 " + normalized); + sendMulticastCount("C injected 2 " + normalized); + + EXPECT_TRUE(engine.increaseCloseCount(path)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, multicastRejectsMalformedAndOversizedMessages) { + RuntimeTestScope scope; + const auto path = scope.file("malformed-multicast"); + const auto normalized = std::filesystem::absolute(path).lexically_normal().string(); + capiocl::engine::Engine engine(false); + useMulticastMonitor(engine); + configureClose(engine, path, 2); + usleep(50000); + + sendMulticastCount("!" + normalized); + sendMulticastCount("C injected 18446744073709551616 " + normalized); + sendMulticastCount("C injected 1junk " + normalized); + EXPECT_FALSE(engine.increaseCloseCount(path)); + + const std::string oversized_path = "/" + std::string(8191, 'x'); + sendMulticastCount("! " + oversized_path + "x"); + EXPECT_FALSE(engine.isCommitted(oversized_path)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, multicastConcurrentClosesReconcile) { + RuntimeTestScope scope; + const auto path = scope.file("concurrent-multicast"); + capiocl::engine::Engine first(false), second(false); + useMulticastMonitor(first); + useMulticastMonitor(second); + configureClose(first, path, 2); + configureClose(second, path, 2); + + std::atomic ready{0}; + std::atomic start{false}; + bool first_result = false, second_result = false; + std::thread first_close([&] { + ++ready; + while (!start) { + std::this_thread::yield(); + } + first_result = first.increaseCloseCount(path); + }); + std::thread second_close([&] { + ++ready; + while (!start) { + std::this_thread::yield(); + } + second_result = second.increaseCloseCount(path); + }); + while (ready != 2) { + std::this_thread::yield(); + } + start = true; + first_close.join(); + second_close.join(); + + EXPECT_TRUE(first_result || second_result); + EXPECT_TRUE(first.isCommitted(path)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, mixedBackendsDoNotDoubleCount) { + RuntimeTestScope scope; + const auto path = scope.file("mixed-backends"); + capiocl::engine::Engine engine(false); + engine.loadConfiguration("/tmp/capio_cl_tomls/runtime_mixed_monitors.toml"); + configureClose(engine, path, 2); + + EXPECT_FALSE(engine.increaseCloseCount(path)); + EXPECT_TRUE(engine.increaseCloseCount(path)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, simultaneousFirstCloseIsProcessSafe) { + RuntimeTestScope scope; + const auto path = scope.file("processes"); + int start[2]; + ASSERT_EQ(pipe(start), 0); + std::vector children; + for (int i = 0; i < 2; ++i) { + const pid_t child = fork(); + ASSERT_NE(child, -1); + if (child == 0) { + close(start[1]); + char signal; + if (read(start[0], &signal, 1) != 1) { + _exit(2); + } + try { + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureClose(engine, path, 2); + engine.increaseCloseCount(path); + _exit(0); + } catch (...) { + _exit(3); + } + } + children.push_back(child); + } + close(start[0]); + ASSERT_EQ(write(start[1], "xx", 2), 2); + close(start[1]); + for (const pid_t child : children) { + int status; + ASSERT_EQ(waitpid(child, &status, 0), child); + EXPECT_TRUE(WIFEXITED(status)); + EXPECT_EQ(WEXITSTATUS(status), 0); + } + + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureClose(engine, path, 2); + EXPECT_TRUE(engine.isCommitted(path)); + EXPECT_FALSE(std::filesystem::exists(scope.closeLock(path))); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, concurrentExplicitCommitWinsBelowThreshold) { + RuntimeTestScope scope; + const auto path = scope.file("explicit-race"); + const auto lock_path = scope.closeLock(path); + std::filesystem::create_directories(lock_path.parent_path()); + int ready[2], release[2]; + ASSERT_EQ(pipe(ready), 0); + ASSERT_EQ(pipe(release), 0); + const pid_t locker = fork(); + ASSERT_NE(locker, -1); + if (locker == 0) { + close(ready[0]); + close(release[1]); + const int lock_fd = + open(lock_path.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC | O_NOFOLLOW, 0600); + if (lock_fd == -1 || write(ready[1], "r", 1) != 1) { + _exit(2); + } + char signal; + if (read(release[0], &signal, 1) != 1) { + _exit(3); + } + close(lock_fd); + unlink(lock_path.c_str()); + _exit(0); + } + close(ready[1]); + close(release[0]); + char signal; + ASSERT_EQ(read(ready[0], &signal, 1), 1); + close(ready[0]); + + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureClose(engine, path, 10); + std::atomic started{false}; + bool result = false; + std::thread closer([&] { + started = true; + result = engine.increaseCloseCount(path); + }); + while (!started) { + std::this_thread::yield(); + } + usleep(50000); + engine.setCommitted(path); + ASSERT_EQ(write(release[1], "x", 1), 1); + close(release[1]); + closer.join(); + int status; + ASSERT_EQ(waitpid(locker, &status, 0), locker); + ASSERT_TRUE(WIFEXITED(status)); + ASSERT_EQ(WEXITSTATUS(status), 0); + EXPECT_TRUE(result); + EXPECT_FALSE(std::filesystem::exists(lock_path)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, relativeAndAbsolutePathsShareRuntimeIdentity) { + RuntimeTestScope scope; + const auto absolute = scope.file("alias"); + const auto relative = std::filesystem::relative(absolute, std::filesystem::current_path()); + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureClose(engine, relative, 2); + + EXPECT_FALSE(engine.increaseCloseCount(absolute)); + EXPECT_TRUE(engine.increaseCloseCount(relative)); + EXPECT_TRUE(engine.isCommitted(absolute)); + EXPECT_TRUE(engine.isCommitted(relative)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, onCloseRejectsMalformedAndUnsupportedCounters) { + RuntimeTestScope scope; + const auto malformed = scope.file("malformed"); + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureClose(engine, malformed, 2); + const auto counter = scope.closeMetadata(malformed, "count"); + std::filesystem::create_directories(counter.parent_path()); + std::ofstream(counter) << "not-a-number\n"; + EXPECT_THROW(engine.increaseCloseCount(malformed), capiocl::monitor::MonitorException); + EXPECT_FALSE(std::filesystem::exists(scope.closeLock(malformed))); + EXPECT_THROW(engine.setCommitedCloseNumber(malformed, -1), std::invalid_argument); + capiocl::engine::CapioCLEntry invalid; + invalid.commit_on_close_count = -1; + EXPECT_THROW(engine.add(malformed, invalid), std::invalid_argument); + + const auto unsupported = scope.file("unsupported"); + capiocl::engine::Engine no_backend(false); + configureClose(no_backend, unsupported, 2); + EXPECT_THROW(no_backend.increaseCloseCount(unsupported), capiocl::monitor::MonitorException); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, lockAcquisitionErrorsDoNotSpin) { + if (geteuid() == 0) { + GTEST_SKIP() << "permission failure cannot be induced as root"; + } + RuntimeTestScope scope; + const auto path = scope.file("lock-error"); + const auto lock = scope.closeLock(path); + std::filesystem::create_directories(lock.parent_path()); + ASSERT_EQ(chmod(lock.parent_path().c_str(), 0500), 0); + + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureClose(engine, path, 2); + EXPECT_THROW(engine.increaseCloseCount(path), capiocl::monitor::MonitorException); + + ASSERT_EQ(chmod(lock.parent_path().c_str(), 0700), 0); + EXPECT_FALSE(std::filesystem::exists(lock)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, onFileChainAndFanIn) { + RuntimeTestScope scope; + const auto a = scope.file("a"); + const auto b = scope.file("b"); + const auto c = scope.file("c"); + const auto d = scope.file("d"); + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureFile(engine, a, {b, c}); + configureFile(engine, b, {d}); + + engine.setCommitted(d); + EXPECT_FALSE(engine.isCommitted(a)); + engine.setCommitted(c); + EXPECT_TRUE(engine.isCommitted(a)); + EXPECT_TRUE(engine.isCommitted(b)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, onFileEmptyAndCyclesRemainUncommitted) { + RuntimeTestScope scope; + const auto empty = scope.file("empty"); + const auto self = scope.file("self"); + const auto a = scope.file("a"); + const auto b = scope.file("b"); + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureFile(engine, empty, {}); + configureFile(engine, self, {self}); + configureFile(engine, a, {b}); + configureFile(engine, b, {a}); + + EXPECT_FALSE(engine.isCommitted(empty)); + EXPECT_FALSE(engine.isCommitted(self)); + EXPECT_FALSE(engine.isCommitted(a)); + EXPECT_FALSE(engine.isCommitted(b)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, onFileCycleResolvesFromRawSeed) { + RuntimeTestScope scope; + const auto a = scope.file("a"); + const auto b = scope.file("b"); + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + configureFile(engine, a, {b}); + configureFile(engine, b, {a}); + + engine.setCommitted(a); + EXPECT_TRUE(engine.isCommitted(b)); + EXPECT_TRUE(engine.isCommitted(a)); +} + +#endif // CAPIO_CL_RUNTIME_COMMIT_HPP diff --git a/tests/tomls/runtime_fs_only.toml b/tests/tomls/runtime_fs_only.toml new file mode 100644 index 0000000..e9be950 --- /dev/null +++ b/tests/tomls/runtime_fs_only.toml @@ -0,0 +1,5 @@ +[monitor.filesystem] +enabled = true + +[monitor.mcast] +enabled = false diff --git a/tests/tomls/runtime_mixed_monitors.toml b/tests/tomls/runtime_mixed_monitors.toml new file mode 100644 index 0000000..9580427 --- /dev/null +++ b/tests/tomls/runtime_mixed_monitors.toml @@ -0,0 +1,14 @@ +[monitor.filesystem] +enabled = true + +[monitor.mcast] +delay_ms = 50 +enabled = true + +[monitor.mcast.commit] +ip = "224.224.226.12" +port = 26103 + +[monitor.mcast.homenode] +ip = "224.224.226.13" +port = 26104 diff --git a/tests/tomls/runtime_multicast_only.toml b/tests/tomls/runtime_multicast_only.toml new file mode 100644 index 0000000..589d9ee --- /dev/null +++ b/tests/tomls/runtime_multicast_only.toml @@ -0,0 +1,14 @@ +[monitor.filesystem] +enabled = false + +[monitor.mcast] +delay_ms = 50 +enabled = true + +[monitor.mcast.commit] +ip = "224.224.226.10" +port = 26101 + +[monitor.mcast.homenode] +ip = "224.224.226.11" +port = 26102 From d58bb890b725dd3b6368f50fe20cc7fb3aeba4d4 Mon Sep 17 00:00:00 2001 From: Marco Edoardo Santimaria Date: Wed, 23 Sep 2026 13:56:23 +0200 Subject: [PATCH 2/5] Fixed github CI/CD --- .github/workflows/ci-test.yml | 27 ++++++++++++++++----------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/.github/workflows/ci-test.yml b/.github/workflows/ci-test.yml index 1f58684..433b289 100644 --- a/.github/workflows/ci-test.yml +++ b/.github/workflows/ci-test.yml @@ -22,17 +22,22 @@ jobs: - Debug - Release compiler: - - gcc:9 - - gcc:10 - - gcc:11 - - gcc:12 - - gcc:13 - - silkeh/clang:13 - - silkeh/clang:14 - - silkeh/clang:15 - - silkeh/clang:16 - - silkeh/clang:17 - - silkeh/clang:18 + - gcc:12 + - gcc:13 + - gcc:14 + - gcc:15 + - gcc:16 + - zhongruoyu/llvm-ports:13-trixie + - zhongruoyu/llvm-ports:14-trixie + - zhongruoyu/llvm-ports:15-trixie + - zhongruoyu/llvm-ports:16-trixie + - zhongruoyu/llvm-ports:17-trixie + - zhongruoyu/llvm-ports:18-trixie + - zhongruoyu/llvm-ports:19-trixie + - zhongruoyu/llvm-ports:20-trixie + - zhongruoyu/llvm-ports:21-trixie + - zhongruoyu/llvm-ports:22-trixie + - zhongruoyu/llvm-ports:23-trixie container: image: ${{ matrix.compiler }} From fd44fa6f880c051b9cced90726dc28c3a5ff87c1 Mon Sep 17 00:00:00 2001 From: Marco Edoardo Santimaria Date: Wed, 23 Sep 2026 14:08:48 +0200 Subject: [PATCH 3/5] Dropped mac os intel runners --- .github/workflows/python-bindings.yml | 2 +- README.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/python-bindings.yml b/.github/workflows/python-bindings.yml index 10e701f..2d8c095 100644 --- a/.github/workflows/python-bindings.yml +++ b/.github/workflows/python-bindings.yml @@ -14,7 +14,7 @@ jobs: name: "Test Python bindings" strategy: matrix: - on: [ 'ubuntu-24.04', 'macos-15-intel', 'macos-15' ] + on: [ 'ubuntu-24.04', 'macos-15' ] python: [ '3.10', '3.11', '3.12', '3.13', '3.14' ] runs-on: ${{ matrix.on }} diff --git a/README.md b/README.md index 218def1..a8e6e61 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ | OS / Arch | ![x86_64](https://img.shields.io/badge/x86__64-121212?logo=intel&logoColor=blue) | ![ARM](https://img.shields.io/badge/ARM-121212?logo=arm&logoColor=0091BD) | ![RISC-V](https://img.shields.io/badge/RISC--V-121212?logo=riscv&logoColor=F9A825) | |------------------------------------------------------------------------------------|----------------------------------------------------------------------------------|---------------------------------------------------------------------------|------------------------------------------------------------------------------------| | ![Ubuntu](https://img.shields.io/badge/Ubuntu-121212?logo=ubuntu&logoColor=E95420) | YES | YES | YES | -| ![macOS](https://img.shields.io/badge/macOS-121212?logo=apple&logoColor=white) | YES | YES | N.A. | +| ![macOS](https://img.shields.io/badge/macOS-121212?logo=apple&logoColor=white) | No CI/CD support | YES | N.A. | #### Documentation From 62d6742c3d4b9bfef9d86d95fdd3bc565bc61cd6 Mon Sep 17 00:00:00 2001 From: Marco Edoardo Santimaria Date: Wed, 23 Sep 2026 14:28:26 +0200 Subject: [PATCH 4/5] improved coverage computation --- .github/workflows/ci-test.yml | 142 ++++++++++++------------------ capiocl/monitor.h | 15 ++-- src/monitors/Multicast.cpp | 1 - tests/cpp/test_monitor.hpp | 8 ++ tests/cpp/test_runtime_commit.hpp | 92 +++++++++++++++++++ 5 files changed, 165 insertions(+), 93 deletions(-) diff --git a/.github/workflows/ci-test.yml b/.github/workflows/ci-test.yml index 433b289..1c1cbdf 100644 --- a/.github/workflows/ci-test.yml +++ b/.github/workflows/ci-test.yml @@ -13,8 +13,6 @@ jobs: unit-tests: name: "Build ${{ matrix.build_type }} with ${{ matrix.compiler }}" runs-on: ubuntu-latest - env: - CALF_LOG_DIR: ${{ github.workspace }}/capiocl_logs strategy: fail-fast: false matrix: @@ -22,22 +20,22 @@ jobs: - Debug - Release compiler: - - gcc:12 - - gcc:13 - - gcc:14 - - gcc:15 - - gcc:16 - - zhongruoyu/llvm-ports:13-trixie - - zhongruoyu/llvm-ports:14-trixie - - zhongruoyu/llvm-ports:15-trixie - - zhongruoyu/llvm-ports:16-trixie - - zhongruoyu/llvm-ports:17-trixie - - zhongruoyu/llvm-ports:18-trixie - - zhongruoyu/llvm-ports:19-trixie - - zhongruoyu/llvm-ports:20-trixie - - zhongruoyu/llvm-ports:21-trixie - - zhongruoyu/llvm-ports:22-trixie - - zhongruoyu/llvm-ports:23-trixie + - gcc:12 + - gcc:13 + - gcc:14 + - gcc:15 + - gcc:16 + - zhongruoyu/llvm-ports:13-trixie + - zhongruoyu/llvm-ports:14-trixie + - zhongruoyu/llvm-ports:15-trixie + - zhongruoyu/llvm-ports:16-trixie + - zhongruoyu/llvm-ports:17-trixie + - zhongruoyu/llvm-ports:18-trixie + - zhongruoyu/llvm-ports:19-trixie + - zhongruoyu/llvm-ports:20-trixie + - zhongruoyu/llvm-ports:21-trixie + - zhongruoyu/llvm-ports:22-trixie + - zhongruoyu/llvm-ports:23-trixie container: image: ${{ matrix.compiler }} @@ -53,75 +51,57 @@ jobs: python3 python3-pip python3-venv \ libcurl4-gnutls-dev libprotobuf-dev protobuf-compiler + - name: "Set log directory" + shell: bash + run: echo "CALF_LOG_DIR=$GITHUB_WORKSPACE/capiocl_logs" >> "$GITHUB_ENV" + - name: "Run CMake" shell: bash run: | cmake -DCMAKE_BUILD_TYPE=${{ matrix.build_type }} \ - -DENABLE_COVERAGE=OFF \ + -DENABLE_COVERAGE=${{ matrix.build_type == 'Debug' && matrix.compiler == 'gcc:12' && 'ON' || 'OFF' }} \ -DBUILD_PYTHON_BINDINGS=OFF \ -DCAPIO_CL_BUILD_TESTS=ON \ -G Ninja \ - -B ${{ github.workspace }}/build \ - . - cmake --build ${{ github.workspace }}/build -j $(nproc) - sudo cmake --install ${{ github.workspace }}/build --prefix /usr/local + -B "$GITHUB_WORKSPACE/build" \ + -S "$GITHUB_WORKSPACE" + cmake --build "$GITHUB_WORKSPACE/build" -j $(nproc) + sudo cmake --install "$GITHUB_WORKSPACE/build" --prefix /usr/local - name: Run unit tests shell: bash run: CAPIO_CL_tests --gtest_break_on_failure --gtest_print_time=1 - - name: Zip logs - if: always() + - name: Generate coverage report + if: matrix.build_type == 'Debug' && matrix.compiler == 'gcc:12' + shell: bash run: | - if [ -d "capiocl_logs" ]; then - cmake -E tar cf capiocl_logs.zip --format=zip capiocl_logs - fi + python3 -m venv /tmp/gcovr + /tmp/gcovr/bin/pip install gcovr + /tmp/gcovr/bin/gcovr \ + --root "$GITHUB_WORKSPACE" \ + --exclude '^build/' \ + --exclude '^tests/' \ + --exclude-throw-branches \ + --gcov-ignore-parse-errors=negative_hits.warn \ + --xml coverage.xml \ + --txt-summary \ + "$GITHUB_WORKSPACE/build" + # Fail here instead of silently uploading an empty report to Codecov + grep -q ' &committed_files, std::mutex &lock, - std::unordered_map> - &close_counts, - std::mutex &close_count_lock, - const std::string &ip_addr, int ip_port, - const std::atomic *terminate); + static void + commit_listener(std::vector &committed_files, std::mutex &lock, + std::unordered_map> + &close_counts, + std::mutex &close_count_lock, const std::string &ip_addr, int ip_port, + const std::atomic *terminate); /** * @brief Background thread function to listen for commit messages. diff --git a/src/monitors/Multicast.cpp b/src/monitors/Multicast.cpp index e4bd134..cf8a7fc 100644 --- a/src/monitors/Multicast.cpp +++ b/src/monitors/Multicast.cpp @@ -12,7 +12,6 @@ #include "capiocl.hpp" #include "capiocl/monitor.h" - std::atomic close_count_origin_sequence{0}; static std::tuple outgoing_socket_multicast(const std::string &address, diff --git a/tests/cpp/test_monitor.hpp b/tests/cpp/test_monitor.hpp index ba0f127..2077e9d 100644 --- a/tests/cpp/test_monitor.hpp +++ b/tests/cpp/test_monitor.hpp @@ -66,6 +66,14 @@ TEST(MONITOR_SUITE_NAME, testIssueExceptionOnMonitorInstance) { } } +TEST(MONITOR_SUITE_NAME, defaultCloseCountContract) { + const capiocl::monitor::MonitorInterface interface; + + EXPECT_EQ(interface.increaseCloseCount("/test", 2), std::nullopt); + EXPECT_THROW(interface.increaseCloseCount("/test", 1), std::invalid_argument); + EXPECT_THROW(interface.increaseCloseCount("/test", 0), std::invalid_argument); +} + TEST(MONITOR_SUITE_NAME, testHomeNodeAcrossDifferentThreads) { const auto e1 = new capiocl::engine::Engine(); const auto e2 = new capiocl::engine::Engine(); diff --git a/tests/cpp/test_runtime_commit.hpp b/tests/cpp/test_runtime_commit.hpp index 4e2b882..e96dd7f 100644 --- a/tests/cpp/test_runtime_commit.hpp +++ b/tests/cpp/test_runtime_commit.hpp @@ -460,6 +460,98 @@ TEST(RUNTIME_COMMIT_SUITE_NAME, onCloseRejectsMalformedAndUnsupportedCounters) { EXPECT_THROW(no_backend.increaseCloseCount(unsupported), capiocl::monitor::MonitorException); } +TEST(RUNTIME_COMMIT_SUITE_NAME, fileSystemCloseMetadataRejectsSymlinks) { + RuntimeTestScope scope; + const capiocl::monitor::FileSystemMonitor monitor; + const auto target = scope.root / "target"; + std::ofstream(target) << "unchanged\n"; + const auto expect_target_unchanged = [&] { + std::ifstream input(target); + std::string value; + std::getline(input, value); + EXPECT_EQ(value, "unchanged"); + }; + + const auto counter_path = scope.file("counter-symlink"); + const auto counter = scope.closeMetadata(counter_path, "count"); + std::filesystem::create_directories(counter.parent_path()); + std::filesystem::create_symlink(target, counter); + EXPECT_THROW(monitor.increaseCloseCount(counter_path, 2), + capiocl::monitor::MonitorException); + EXPECT_FALSE(std::filesystem::exists(scope.closeLock(counter_path))); + expect_target_unchanged(); + + const auto temporary_path = scope.file("temporary-symlink"); + const auto temporary = scope.closeMetadata(temporary_path, "count.tmp"); + std::filesystem::create_directories(temporary.parent_path()); + std::filesystem::create_symlink(target, temporary); + EXPECT_THROW(monitor.increaseCloseCount(temporary_path, 2), + capiocl::monitor::MonitorException); + EXPECT_FALSE(std::filesystem::exists(scope.closeLock(temporary_path))); + expect_target_unchanged(); + + const auto commit_path = scope.file("commit-symlink"); + const auto commit = commit_path.parent_path() / ".commit-symlink.commit"; + const auto commit_count = scope.closeMetadata(commit_path, "count"); + std::filesystem::create_directories(commit_count.parent_path()); + std::ofstream(commit_count) << "1\n"; + std::filesystem::create_symlink(target, commit); + EXPECT_THROW(monitor.increaseCloseCount(commit_path, 2), + capiocl::monitor::MonitorException); + EXPECT_FALSE(std::filesystem::exists(scope.closeLock(commit_path))); + expect_target_unchanged(); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, fileSystemCloseCounterRejectsTrailingDataAndOverflow) { + RuntimeTestScope scope; + const capiocl::monitor::FileSystemMonitor monitor; + + const auto trailing_path = scope.file("trailing"); + const auto trailing = scope.closeMetadata(trailing_path, "count"); + std::filesystem::create_directories(trailing.parent_path()); + std::ofstream(trailing) << "1 trailing\n"; + EXPECT_THROW(monitor.increaseCloseCount(trailing_path, 2), + capiocl::monitor::MonitorException); + EXPECT_FALSE(std::filesystem::exists(scope.closeLock(trailing_path))); + std::ifstream trailing_input(trailing); + std::string trailing_value; + std::getline(trailing_input, trailing_value); + EXPECT_EQ(trailing_value, "1 trailing"); + + const auto overflow_path = scope.file("overflow"); + const auto overflow = scope.closeMetadata(overflow_path, "count"); + std::filesystem::create_directories(overflow.parent_path()); + std::ofstream(overflow) << std::numeric_limits::max() << '\n'; + EXPECT_THROW(monitor.increaseCloseCount(overflow_path, 2), + capiocl::monitor::MonitorException); + EXPECT_FALSE(std::filesystem::exists(scope.closeLock(overflow_path))); + std::ifstream input(overflow); + long persisted = 0; + input >> persisted; + EXPECT_EQ(persisted, std::numeric_limits::max()); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, multicastOversizedOutboundCommitIsLocallyRecorded) { + RuntimeTestScope scope; + capiocl::engine::Engine engine(false); + useMulticastMonitor(engine); + const std::filesystem::path oversized = "/" + std::string(8192, 'x'); + + EXPECT_NO_THROW(engine.setCommitted(oversized)); + EXPECT_TRUE(engine.isCommitted(oversized)); +} + +TEST(RUNTIME_COMMIT_SUITE_NAME, earlyEmptyCloseAndWildcardDependencyRemainUncommitted) { + RuntimeTestScope scope; + capiocl::engine::Engine engine(false); + useFilesystemMonitor(engine); + + EXPECT_FALSE(engine.increaseCloseCount({})); + const auto path = scope.file("wildcard-dependency"); + configureFile(engine, path, {scope.file("*")}); + EXPECT_FALSE(engine.isCommitted(path)); +} + TEST(RUNTIME_COMMIT_SUITE_NAME, lockAcquisitionErrorsDoNotSpin) { if (geteuid() == 0) { GTEST_SKIP() << "permission failure cannot be induced as root"; From e2ade719a6be2f62aac0c540471eb1c368218c56 Mon Sep 17 00:00:00 2001 From: Marco Edoardo Santimaria Date: Wed, 23 Sep 2026 15:46:42 +0200 Subject: [PATCH 5/5] cleanup --- .github/workflows/ci-test.yml | 4 ++-- capiocl/engine.h | 3 --- src/Engine.cpp | 21 ++++++++++++--------- 3 files changed, 14 insertions(+), 14 deletions(-) diff --git a/.github/workflows/ci-test.yml b/.github/workflows/ci-test.yml index 1c1cbdf..90ee736 100644 --- a/.github/workflows/ci-test.yml +++ b/.github/workflows/ci-test.yml @@ -59,7 +59,7 @@ jobs: shell: bash run: | cmake -DCMAKE_BUILD_TYPE=${{ matrix.build_type }} \ - -DENABLE_COVERAGE=${{ matrix.build_type == 'Debug' && matrix.compiler == 'gcc:12' && 'ON' || 'OFF' }} \ + -DENABLE_COVERAGE=${{ matrix.build_type == 'Debug' && matrix.compiler == 'gcc:16' && 'ON' || 'OFF' }} \ -DBUILD_PYTHON_BINDINGS=OFF \ -DCAPIO_CL_BUILD_TESTS=ON \ -G Ninja \ @@ -73,7 +73,7 @@ jobs: run: CAPIO_CL_tests --gtest_break_on_failure --gtest_print_time=1 - name: Generate coverage report - if: matrix.build_type == 'Debug' && matrix.compiler == 'gcc:12' + if: matrix.build_type == 'Debug' && matrix.compiler == 'gcc:16' shell: bash run: | python3 -m venv /tmp/gcovr diff --git a/capiocl/engine.h b/capiocl/engine.h index 1b6437e..f3f8e60 100644 --- a/capiocl/engine.h +++ b/capiocl/engine.h @@ -130,9 +130,6 @@ class Engine final { */ void _newFile(const std::filesystem::path &path) const; - /// Materialize and snapshot a normalized concrete runtime entry under one lock. - CapioCLEntry materializeRuntimeEntry(const std::filesystem::path &path) const; - /** * @brief Updates the number of entries in the parent directory of the given path. * diff --git a/src/Engine.cpp b/src/Engine.cpp index dcce585..da59bd8 100644 --- a/src/Engine.cpp +++ b/src/Engine.cpp @@ -233,13 +233,6 @@ void capiocl::engine::Engine::_newFile(const std::filesystem::path &path) const } } -capiocl::engine::CapioCLEntry -capiocl::engine::Engine::materializeRuntimeEntry(const std::filesystem::path &path) const { - std::lock_guard lock(_shared_mutex); - _newFile(path); - return _capio_cl_entries.at(path); -} - void capiocl::engine::Engine::compute_directory_entry_count( const std::filesystem::path &path) const { START_LOG(calf_current_tid(), "call()"); @@ -593,7 +586,12 @@ bool capiocl::engine::Engine::isCommitted(const std::filesystem::path &path) con } visiting.insert(key); - const auto entry = materializeRuntimeEntry(normalized); + CapioCLEntry entry; + { + std::lock_guard lock(_shared_mutex); + _newFile(normalized); + entry = _capio_cl_entries.at(normalized); + } bool committed = false; if (entry.commit_rule == commitRules::ON_FILE && !entry.file_dependencies.empty()) { @@ -621,7 +619,12 @@ bool capiocl::engine::Engine::increaseCloseCount(const std::filesystem::path &pa return true; } - const auto entry = materializeRuntimeEntry(normalized); + CapioCLEntry entry; + { + std::lock_guard lock(_shared_mutex); + _newFile(normalized); + entry = _capio_cl_entries.at(normalized); + } const auto &rule = entry.commit_rule; const long threshold = entry.commit_on_close_count;