From 30849dfb36c00e2775c413f11ca98962844d4156 Mon Sep 17 00:00:00 2001 From: Marco Edoardo Santimaria Date: Thu, 11 Dec 2025 13:29:12 +0100 Subject: [PATCH 1/4] Added serialization compression flag --- bindings/python_bindings.cpp | 5 +- capiocl/serializer.h | 9 ++-- src/Serializer.cpp | 6 +-- src/serializers/v1.1.cpp | 28 ++++++++++- src/serializers/v1.cpp | 62 ++++++++++++++++++++++-- tests/cpp/test_exceptions.hpp | 5 +- tests/cpp/test_serialize_deserialize.hpp | 8 +-- 7 files changed, 103 insertions(+), 20 deletions(-) diff --git a/bindings/python_bindings.cpp b/bindings/python_bindings.cpp index 2edab01..b3c4512 100644 --- a/bindings/python_bindings.cpp +++ b/bindings/python_bindings.cpp @@ -129,7 +129,8 @@ PYBIND11_MODULE(_py_capio_cl, m) { }); m.def("serialize", &capiocl::serializer::Serializer::dump, py::arg("engine"), - py::arg("filename"), py::arg("version") = capiocl::CAPIO_CL_VERSION::V1); + py::arg("filename"), py::arg("compress") = false, + py::arg("version") = capiocl::CAPIO_CL_VERSION::V1); py::class_(m, "CapioCLEntry") .def(py::init<>()) @@ -150,4 +151,4 @@ PYBIND11_MODULE(_py_capio_cl, m) { .def_readwrite("is_file", &capiocl::engine::CapioCLEntry::is_file) .def_static("from_json", &capiocl::engine::CapioCLEntry::fromJson, py::arg("in")) .def("to_json", &capiocl::engine::CapioCLEntry::toJson); -} \ No newline at end of file +} diff --git a/capiocl/serializer.h b/capiocl/serializer.h index 1c20ef7..d9b48ec 100644 --- a/capiocl/serializer.h +++ b/capiocl/serializer.h @@ -37,10 +37,11 @@ class Serializer final { * * @param engine instance of Engine to dump * @param filename path of output file + * @param compress Compress the serialized output * @throws SerializerException */ static void serialize_v1(const engine::Engine &engine, - const std::filesystem::path &filename); + const std::filesystem::path &filename, bool compress = false); /** * @brief Dump the current configuration loaded into an instance of Engine to a CAPIO-CL @@ -48,10 +49,11 @@ class Serializer final { * * @param engine instance of Engine to dump * @param filename path of output file + * @param compress Compress the serialized output * @throws SerializerException */ static void serialize_v1_1(const engine::Engine &engine, - const std::filesystem::path &filename); + const std::filesystem::path &filename, bool compress = false); }; public: @@ -61,10 +63,11 @@ class Serializer final { * * @param engine instance of Engine to dump * @param filename path of output file + * @param compress Compress directories entries when possible * @param version Version of CAPIO-CL used to generate configuration files. */ static void dump(const engine::Engine &engine, const std::filesystem::path &filename, - const std::string &version = CAPIO_CL_VERSION::V1); + bool compress = false, const std::string &version = CAPIO_CL_VERSION::V1); }; } // namespace capiocl::serializer #endif // CAPIO_CL_SERIALIZER_H \ No newline at end of file diff --git a/src/Serializer.cpp b/src/Serializer.cpp index 27698bb..316948f 100644 --- a/src/Serializer.cpp +++ b/src/Serializer.cpp @@ -10,15 +10,15 @@ void capiocl::serializer::Serializer::dump(const engine::Engine &engine, const std::filesystem::path &filename, - const std::string &version) { + const bool compress, const std::string &version) { START_LOG(calf_current_tid(), "call()"); UPDATE_CALF_WORKFLOW_NAME(engine.getWorkflowName()); if (version == CAPIO_CL_VERSION::V1) { CALF_PRINT_COLOR(CALF_CLI_LEVEL_INFO, "Serializing engine with V1 specification"); - available_serializers::serialize_v1(engine, filename); + available_serializers::serialize_v1(engine, filename, compress); } else if (version == CAPIO_CL_VERSION::V1_1) { CALF_PRINT_COLOR(CALF_CLI_LEVEL_INFO, "Serializing engine with V1.1 specification"); - available_serializers::serialize_v1_1(engine, filename); + available_serializers::serialize_v1_1(engine, filename, compress); } else { LOG("serializer unavailable version=%s workflow=%s output=%s", version.c_str(), engine.getWorkflowName().c_str(), filename.string().c_str()); diff --git a/src/serializers/v1.1.cpp b/src/serializers/v1.1.cpp index 677780c..0e709d1 100644 --- a/src/serializers/v1.1.cpp +++ b/src/serializers/v1.1.cpp @@ -7,7 +7,7 @@ #include "capiocl/serializer.h" void capiocl::serializer::Serializer::available_serializers::serialize_v1_1( - const engine::Engine &engine, const std::filesystem::path &filename) { + const engine::Engine &engine, const std::filesystem::path &filename, const bool compress) { START_LOG(calf_current_tid(), "call()"); UPDATE_CALF_WORKFLOW_NAME(engine.getWorkflowName()); jsoncons::json doc; @@ -16,6 +16,18 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1_1( const auto files = engine._capio_cl_entries; + std::vector keys; + keys.reserve(files.size()); + for (const auto &[k, v] : files) { + keys.push_back(k); + } + std::sort(keys.begin(), keys.end(), [](const std::string &a, const std::string &b) { + if (a.length() != b.length()) { + return a.length() < b.length(); + } + return a < b; + }); + std::unordered_map> app_inputs; std::unordered_map> app_outputs; @@ -27,7 +39,19 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1_1( jsoncons::json storage = jsoncons::json::object(); jsoncons::json io_graph = jsoncons::json::array(); - for (const auto &[path, entry] : files) { + for (const auto &path : keys) { + const auto entry = files.at(path); + + if (compress) { + if (const std::filesystem::path p(path); files.find(p.parent_path()) != files.end()) { + if (const auto &parent = files.at(p.parent_path()); + parent.fire_rule == entry.fire_rule && + parent.commit_rule == entry.commit_rule && entry.is_file) { + continue; + } + } + } + if (entry.permanent) { permanent.push_back(path); } diff --git a/src/serializers/v1.cpp b/src/serializers/v1.cpp index 5151843..f16b3b1 100644 --- a/src/serializers/v1.cpp +++ b/src/serializers/v1.cpp @@ -7,14 +7,31 @@ #include "capiocl/serializer.h" void capiocl::serializer::Serializer::available_serializers::serialize_v1( - const engine::Engine &engine, const std::filesystem::path &filename) { + const engine::Engine &engine, const std::filesystem::path &filename, const bool compress) { START_LOG(calf_current_tid(), "call()"); UPDATE_CALF_WORKFLOW_NAME(engine.getWorkflowName()); + + if (compress) { + CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, + "Using configuration compression to directories!"); + } jsoncons::json doc; doc["name"] = engine.getWorkflowName(); const auto files = engine._capio_cl_entries; + std::vector keys; + keys.reserve(files.size()); + for (const auto &[k, v] : files) { + keys.push_back(k); + } + std::sort(keys.begin(), keys.end(), [](const std::string &a, const std::string &b) { + if (a.length() != b.length()) { + return a.length() > b.length(); + } + return a > b; + }); + std::unordered_map> app_inputs; std::unordered_map> app_outputs; @@ -26,7 +43,20 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1( jsoncons::json storage = jsoncons::json::object(); jsoncons::json io_graph = jsoncons::json::array(); - for (const auto &[path, entry] : files) { + for (const auto &path : keys) { + const auto entry = files.at(path); + + if (compress) { + if (const std::filesystem::path p(path); files.find(p.parent_path()) != files.end()) { + if (const auto &parent = files.at(p.parent_path()); + parent.fire_rule == entry.fire_rule && + parent.commit_rule == entry.commit_rule && entry.is_file) { + CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, "Compressing entry %s", path.c_str()); + continue; + } + } + } + if (entry.permanent) { permanent.push_back(path); } @@ -43,13 +73,37 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1( } } - for (const auto &[app_name, outputs] : app_outputs) { + for (auto &[app_name, outputs] : app_outputs) { jsoncons::json app = jsoncons::json::object(); jsoncons::json streaming = jsoncons::json::array(); + std::sort(outputs.begin(), outputs.end(), [](const std::string &a, const std::string &b) { + if (a.length() != b.length()) { + return a.length() > b.length(); + } + return a > b; + }); + + std::vector filtered_outputs; + for (const auto &path : outputs) { const auto &entry = files.at(path); + if (compress) { + if (const std::filesystem::path p(path); + files.find(p.parent_path()) != files.end()) { + if (const auto &parent = files.at(p.parent_path()); + parent.fire_rule == entry.fire_rule && + parent.commit_rule == entry.commit_rule && entry.is_file) { + CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, "Compressing entry %s", + path.c_str()); + continue; + } + } + } + + filtered_outputs.push_back(path); + jsoncons::json streaming_item = jsoncons::json::object(); std::string committed = entry.commit_rule; const char *name_kind = entry.is_file ? "name" : "dirname"; @@ -89,7 +143,7 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1( app["name"] = app_name; app["input_stream"] = app_inputs[app_name]; - app["output_stream"] = outputs; + app["output_stream"] = filtered_outputs; app["streaming"] = streaming; io_graph.push_back(app); diff --git a/tests/cpp/test_exceptions.hpp b/tests/cpp/test_exceptions.hpp index c4bcd63..db1ea16 100644 --- a/tests/cpp/test_exceptions.hpp +++ b/tests/cpp/test_exceptions.hpp @@ -36,8 +36,9 @@ TEST(EXCEPTION_SUITE_NAME, testFailedserializeVersion) { const std::filesystem::path source = "/tmp/capio_cl_jsons/V" + version + "/test24.json"; auto engine = capiocl::parser::Parser::parse(source, "/tmp"); - EXPECT_THROW(capiocl::serializer::Serializer::dump(*engine, "test.json", "1234.5678"), - capiocl::serializer::SerializerException); + EXPECT_THROW( + capiocl::serializer::Serializer::dump(*engine, "test.json", false, "1234.5678"), + capiocl::serializer::SerializerException); } } diff --git a/tests/cpp/test_serialize_deserialize.hpp b/tests/cpp/test_serialize_deserialize.hpp index 96eadb2..4c0b0e5 100644 --- a/tests/cpp/test_serialize_deserialize.hpp +++ b/tests/cpp/test_serialize_deserialize.hpp @@ -42,7 +42,7 @@ TEST(SERIALIZE_DESERIALIZE_SUITE_NAME, testSerializeParseCAPIOCLV1) { engine.print(); - capiocl::serializer::Serializer::dump(engine, path, _cl_version); + capiocl::serializer::Serializer::dump(engine, path, false, _cl_version); std::filesystem::path resolve = ""; auto new_engine = capiocl::parser::Parser::parse(path, resolve); @@ -73,7 +73,7 @@ TEST(SERIALIZE_DESERIALIZE_SUITE_NAME, testSerializeParseCAPIOCLV1NcloseNfiles) engine.addProducer(file_1_name, producer_name); engine.addConsumer(file_1_name, consumer_name); - capiocl::serializer::Serializer::dump(engine, path, _cl_version); + capiocl::serializer::Serializer::dump(engine, path, false, _cl_version); std::filesystem::path resolve = ""; auto new_engine = capiocl::parser::Parser::parse(path, resolve); @@ -108,7 +108,7 @@ TEST(SERIALIZE_DESERIALIZE_SUITE_NAME, testSerializeParseCAPIOCLV1FileDeps) { engine.setFileDeps(file_3_name, {file_1_name, file_2_name}); engine.print(); - capiocl::serializer::Serializer::dump(engine, path, _cl_version); + capiocl::serializer::Serializer::dump(engine, path, false, _cl_version); std::filesystem::path resolve = ""; auto new_engine = capiocl::parser::Parser::parse(path, resolve); @@ -136,7 +136,7 @@ TEST(SERIALIZE_DESERIALIZE_SUITE_NAME, testSerializeCommitOnCloseCountNoCommitRu engine.setCommitedCloseNumber(file_1_name, 10); engine.print(); - capiocl::serializer::Serializer::dump(engine, path, _cl_version); + capiocl::serializer::Serializer::dump(engine, path, false, _cl_version); std::filesystem::path resolve = ""; auto new_engine = capiocl::parser::Parser::parse(path, resolve); From c94f0dd8f30288965b7200c24314de4c0bc82209 Mon Sep 17 00:00:00 2001 From: Marco Edoardo Santimaria Date: Thu, 11 Dec 2025 17:21:37 +0100 Subject: [PATCH 2/4] Code refactor --- capiocl/serializer.h | 16 ++++++++++++ src/Serializer.cpp | 31 ++++++++++++++++++++++ src/serializers/v1.1.cpp | 50 ++++++++++++++++++++---------------- src/serializers/v1.cpp | 55 +++++++++++----------------------------- 4 files changed, 90 insertions(+), 62 deletions(-) diff --git a/capiocl/serializer.h b/capiocl/serializer.h index d9b48ec..63f3a18 100644 --- a/capiocl/serializer.h +++ b/capiocl/serializer.h @@ -28,6 +28,22 @@ class SerializerException final : public std::exception { /// @brief Dump the current loaded CAPIO-CL configuration from class Engine to a CAPIO-CL /// configuration file. class Serializer final { + /** + * Check whether a CAPIO-CL entry has a parent entry for which the same rules applies, and tell + * whether this entry can be omitted by using rule inheritance. + * @param compress + * @param path + * @param engine + * @return + */ + static bool entryCanBeCompressed(bool compress, const std::filesystem::path &path, + const engine::Engine &engine); + + /** + * Sort path entries from longest to shortest + * @param paths + */ + static void sortPathsByDecreasingLength(std::vector &paths); /// @brief Available serializers for CAPIO-CL struct available_serializers { diff --git a/src/Serializer.cpp b/src/Serializer.cpp index 316948f..fdb972a 100644 --- a/src/Serializer.cpp +++ b/src/Serializer.cpp @@ -33,3 +33,34 @@ capiocl::serializer::SerializerException::SerializerException(const std::string UPDATE_CALF_WORKFLOW_NAME(""); CALF_PRINT_COLOR(CALF_CLI_LEVEL_ERROR, "%s", msg.c_str()); } + +bool capiocl::serializer::Serializer::entryCanBeCompressed(const bool compress, + const std::filesystem::path &path, + const engine::Engine &engine) { + + if (!compress) { + return false; + } + + if (engine.isDirectory(path)) { + return false; + } + + const auto parent_path = path.parent_path(); + + if (!engine.contains(parent_path)) { + return false; + } + + return engine.getCommitRule(path) == engine.getCommitRule(parent_path) && + engine.getFireRule(path) == engine.getFireRule(parent_path); +} + +void capiocl::serializer::Serializer::sortPathsByDecreasingLength(std::vector &paths) { + std::sort(paths.begin(), paths.end(), [](const std::string &a, const std::string &b) { + if (a.length() != b.length()) { + return a.length() > b.length(); + } + return a > b; + }); +} diff --git a/src/serializers/v1.1.cpp b/src/serializers/v1.1.cpp index 0e709d1..297029f 100644 --- a/src/serializers/v1.1.cpp +++ b/src/serializers/v1.1.cpp @@ -10,24 +10,18 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1_1( const engine::Engine &engine, const std::filesystem::path &filename, const bool compress) { START_LOG(calf_current_tid(), "call()"); UPDATE_CALF_WORKFLOW_NAME(engine.getWorkflowName()); + + if (compress) { + CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, + "Using configuration compression to directories!"); + } + jsoncons::json doc; doc["version"] = 1.1; doc["name"] = engine.getWorkflowName(); const auto files = engine._capio_cl_entries; - std::vector keys; - keys.reserve(files.size()); - for (const auto &[k, v] : files) { - keys.push_back(k); - } - std::sort(keys.begin(), keys.end(), [](const std::string &a, const std::string &b) { - if (a.length() != b.length()) { - return a.length() < b.length(); - } - return a < b; - }); - std::unordered_map> app_inputs; std::unordered_map> app_outputs; @@ -39,17 +33,20 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1_1( jsoncons::json storage = jsoncons::json::object(); jsoncons::json io_graph = jsoncons::json::array(); + std::vector keys; + keys.reserve(files.size()); + for (const auto &[k, v] : files) { + keys.push_back(k); + } + + sortPathsByDecreasingLength(keys); + for (const auto &path : keys) { const auto entry = files.at(path); - if (compress) { - if (const std::filesystem::path p(path); files.find(p.parent_path()) != files.end()) { - if (const auto &parent = files.at(p.parent_path()); - parent.fire_rule == entry.fire_rule && - parent.commit_rule == entry.commit_rule && entry.is_file) { - continue; - } - } + if (entryCanBeCompressed(compress, path, engine)) { + CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, "Compressing entry %s", path.c_str()); + continue; } if (entry.permanent) { @@ -68,13 +65,22 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1_1( } } - for (const auto &[app_name, outputs] : app_outputs) { + for (auto &[app_name, outputs] : app_outputs) { jsoncons::json app = jsoncons::json::object(); jsoncons::json streaming = jsoncons::json::array(); + std::vector filtered_outputs; + + sortPathsByDecreasingLength(outputs); for (const auto &path : outputs) { const auto &entry = files.at(path); + if (entryCanBeCompressed(compress, path, engine)) { + continue; + } + + filtered_outputs.push_back(path); + jsoncons::json streaming_item = jsoncons::json::object(); std::string committed = entry.commit_rule; const char *name_kind = entry.is_file ? "name" : "dirname"; @@ -114,7 +120,7 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1_1( app["name"] = app_name; app["input_stream"] = app_inputs[app_name]; - app["output_stream"] = outputs; + app["output_stream"] = filtered_outputs; app["streaming"] = streaming; io_graph.push_back(app); diff --git a/src/serializers/v1.cpp b/src/serializers/v1.cpp index f16b3b1..18f318e 100644 --- a/src/serializers/v1.cpp +++ b/src/serializers/v1.cpp @@ -20,18 +20,6 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1( const auto files = engine._capio_cl_entries; - std::vector keys; - keys.reserve(files.size()); - for (const auto &[k, v] : files) { - keys.push_back(k); - } - std::sort(keys.begin(), keys.end(), [](const std::string &a, const std::string &b) { - if (a.length() != b.length()) { - return a.length() > b.length(); - } - return a > b; - }); - std::unordered_map> app_inputs; std::unordered_map> app_outputs; @@ -43,18 +31,20 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1( jsoncons::json storage = jsoncons::json::object(); jsoncons::json io_graph = jsoncons::json::array(); + std::vector keys; + keys.reserve(files.size()); + for (const auto &[k, v] : files) { + keys.push_back(k); + } + + sortPathsByDecreasingLength(keys); + for (const auto &path : keys) { const auto entry = files.at(path); - if (compress) { - if (const std::filesystem::path p(path); files.find(p.parent_path()) != files.end()) { - if (const auto &parent = files.at(p.parent_path()); - parent.fire_rule == entry.fire_rule && - parent.commit_rule == entry.commit_rule && entry.is_file) { - CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, "Compressing entry %s", path.c_str()); - continue; - } - } + if (entryCanBeCompressed(compress, path, engine)) { + CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, "Compressing entry %s", path.c_str()); + continue; } if (entry.permanent) { @@ -76,30 +66,15 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1( for (auto &[app_name, outputs] : app_outputs) { jsoncons::json app = jsoncons::json::object(); jsoncons::json streaming = jsoncons::json::array(); - - std::sort(outputs.begin(), outputs.end(), [](const std::string &a, const std::string &b) { - if (a.length() != b.length()) { - return a.length() > b.length(); - } - return a > b; - }); - std::vector filtered_outputs; + sortPathsByDecreasingLength(outputs); + for (const auto &path : outputs) { const auto &entry = files.at(path); - if (compress) { - if (const std::filesystem::path p(path); - files.find(p.parent_path()) != files.end()) { - if (const auto &parent = files.at(p.parent_path()); - parent.fire_rule == entry.fire_rule && - parent.commit_rule == entry.commit_rule && entry.is_file) { - CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, "Compressing entry %s", - path.c_str()); - continue; - } - } + if (entryCanBeCompressed(compress, path, engine)) { + continue; } filtered_outputs.push_back(path); From 153f9b0b0e096cc1e7cfb26644b5e11a2ae7bcaa Mon Sep 17 00:00:00 2001 From: marcoSanti Date: Mon, 3 Aug 2026 14:39:52 +0200 Subject: [PATCH 3/4] BEta implementation --- capiocl/serializer.h | 18 ++--- src/Serializer.cpp | 96 ++++++++++++++++++------ src/serializers/v1.1.cpp | 21 +++--- src/serializers/v1.cpp | 21 +++--- tests/cpp/test_serialize_deserialize.hpp | 54 ++++++++++++- 5 files changed, 153 insertions(+), 57 deletions(-) diff --git a/capiocl/serializer.h b/capiocl/serializer.h index 63f3a18..5502a24 100644 --- a/capiocl/serializer.h +++ b/capiocl/serializer.h @@ -1,6 +1,10 @@ #ifndef CAPIO_CL_SERIALIZER_H #define CAPIO_CL_SERIALIZER_H +#include +#include +#include + #include "capiocl.hpp" /// @brief Namespace containing the CAPIO-CL Serializer component @@ -28,16 +32,8 @@ class SerializerException final : public std::exception { /// @brief Dump the current loaded CAPIO-CL configuration from class Engine to a CAPIO-CL /// configuration file. class Serializer final { - /** - * Check whether a CAPIO-CL entry has a parent entry for which the same rules applies, and tell - * whether this entry can be omitted by using rule inheritance. - * @param compress - * @param path - * @param engine - * @return - */ - static bool entryCanBeCompressed(bool compress, const std::filesystem::path &path, - const engine::Engine &engine); + static std::vector> + compressedPaths(const engine::Engine &engine); /** * Sort path entries from longest to shortest @@ -86,4 +82,4 @@ class Serializer final { bool compress = false, const std::string &version = CAPIO_CL_VERSION::V1); }; } // namespace capiocl::serializer -#endif // CAPIO_CL_SERIALIZER_H \ No newline at end of file +#endif // CAPIO_CL_SERIALIZER_H diff --git a/src/Serializer.cpp b/src/Serializer.cpp index fdb972a..1f7704d 100644 --- a/src/Serializer.cpp +++ b/src/Serializer.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -34,33 +35,86 @@ capiocl::serializer::SerializerException::SerializerException(const std::string CALF_PRINT_COLOR(CALF_CLI_LEVEL_ERROR, "%s", msg.c_str()); } -bool capiocl::serializer::Serializer::entryCanBeCompressed(const bool compress, - const std::filesystem::path &path, - const engine::Engine &engine) { +void capiocl::serializer::Serializer::sortPathsByDecreasingLength(std::vector &paths) { + std::sort(paths.begin(), paths.end(), [](const std::string &a, const std::string &b) { + if (a.length() != b.length()) { + return a.length() > b.length(); + } + return a > b; + }); +} - if (!compress) { - return false; - } +std::vector> +capiocl::serializer::Serializer::compressedPaths(const engine::Engine &engine) { + std::unordered_map paths; + std::vector directories; + + for (const auto &[path, entry] : engine._capio_cl_entries) { + paths.emplace(path, path); + if (!entry.is_file || path.find_first_of("*?[") != std::string::npos) { + continue; + } - if (engine.isDirectory(path)) { - return false; + for (auto parent = std::filesystem::path(path).parent_path();;) { + if (const auto value = parent.string(); + std::find(directories.begin(), directories.end(), value) == directories.end()) { + directories.push_back(value); + } + if (parent.empty() || parent == parent.root_path()) { + break; + } + parent = parent.parent_path(); + } } - const auto parent_path = path.parent_path(); + sortPathsByDecreasingLength(directories); - if (!engine.contains(parent_path)) { - return false; - } + for (const auto &directory : directories) { + const auto wildcard = (std::filesystem::path(directory) / "*").string(); + if (paths.find(wildcard) != paths.end()) { + continue; + } - return engine.getCommitRule(path) == engine.getCommitRule(parent_path) && - engine.getFireRule(path) == engine.getFireRule(parent_path); -} + std::vector> groups; + for (const auto &[output, source] : paths) { + if (!engine._capio_cl_entries.at(source).is_file || + (output == source && output.find_first_of("*?[") != std::string::npos)) { + continue; + } + const std::filesystem::path output_path(output); + const auto relative = output_path.lexically_relative(directory); + if ((!directory.empty() && + (relative.empty() || *relative.begin() == std::filesystem::path(".."))) || + (directory.empty() && output_path.is_absolute())) { + continue; + } -void capiocl::serializer::Serializer::sortPathsByDecreasingLength(std::vector &paths) { - std::sort(paths.begin(), paths.end(), [](const std::string &a, const std::string &b) { - if (a.length() != b.length()) { - return a.length() > b.length(); + auto group = std::find_if(groups.begin(), groups.end(), [&](const auto &candidate) { + return engine._capio_cl_entries.at(paths.at(candidate.front())) == + engine._capio_cl_entries.at(source); + }); + (group == groups.end() ? groups.emplace_back() : *group).push_back(output); } - return a > b; - }); + for (auto &group : groups) { + std::sort(group.begin(), group.end()); + } + + const auto largest = + std::max_element(groups.begin(), groups.end(), [](const auto &left, const auto &right) { + return left.size() != right.size() ? left.size() < right.size() + : left.front() > right.front(); + }); + if (largest == groups.end() || largest->size() < 2) { + continue; + } + + const auto source = paths.at(largest->front()); + for (const auto &path : *largest) { + paths.erase(path); + } + paths.emplace(wildcard, source); + CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, "Compressing entries to %s", wildcard.c_str()); + } + + return {paths.begin(), paths.end()}; } diff --git a/src/serializers/v1.1.cpp b/src/serializers/v1.1.cpp index 297029f..35a6b1e 100644 --- a/src/serializers/v1.1.cpp +++ b/src/serializers/v1.1.cpp @@ -12,15 +12,21 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1_1( UPDATE_CALF_WORKFLOW_NAME(engine.getWorkflowName()); if (compress) { - CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, - "Using configuration compression to directories!"); + CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, "Using configuration compression to directories!"); } jsoncons::json doc; doc["version"] = 1.1; doc["name"] = engine.getWorkflowName(); - const auto files = engine._capio_cl_entries; + auto files = engine._capio_cl_entries; + if (compress) { + decltype(files) compressed; + for (const auto &[path, source] : compressedPaths(engine)) { + compressed.emplace(path, files.at(source)); + } + files = std::move(compressed); + } std::unordered_map> app_inputs; std::unordered_map> app_outputs; @@ -44,11 +50,6 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1_1( for (const auto &path : keys) { const auto entry = files.at(path); - if (entryCanBeCompressed(compress, path, engine)) { - CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, "Compressing entry %s", path.c_str()); - continue; - } - if (entry.permanent) { permanent.push_back(path); } @@ -75,10 +76,6 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1_1( for (const auto &path : outputs) { const auto &entry = files.at(path); - if (entryCanBeCompressed(compress, path, engine)) { - continue; - } - filtered_outputs.push_back(path); jsoncons::json streaming_item = jsoncons::json::object(); diff --git a/src/serializers/v1.cpp b/src/serializers/v1.cpp index 18f318e..761ebf9 100644 --- a/src/serializers/v1.cpp +++ b/src/serializers/v1.cpp @@ -12,13 +12,19 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1( UPDATE_CALF_WORKFLOW_NAME(engine.getWorkflowName()); if (compress) { - CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, - "Using configuration compression to directories!"); + CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, "Using configuration compression to directories!"); } jsoncons::json doc; doc["name"] = engine.getWorkflowName(); - const auto files = engine._capio_cl_entries; + auto files = engine._capio_cl_entries; + if (compress) { + decltype(files) compressed; + for (const auto &[path, source] : compressedPaths(engine)) { + compressed.emplace(path, files.at(source)); + } + files = std::move(compressed); + } std::unordered_map> app_inputs; std::unordered_map> app_outputs; @@ -42,11 +48,6 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1( for (const auto &path : keys) { const auto entry = files.at(path); - if (entryCanBeCompressed(compress, path, engine)) { - CALF_PRINT_COLOR(CALF_CLI_LEVEL_WARNING, "Compressing entry %s", path.c_str()); - continue; - } - if (entry.permanent) { permanent.push_back(path); } @@ -73,10 +74,6 @@ void capiocl::serializer::Serializer::available_serializers::serialize_v1( for (const auto &path : outputs) { const auto &entry = files.at(path); - if (entryCanBeCompressed(compress, path, engine)) { - continue; - } - filtered_outputs.push_back(path); jsoncons::json streaming_item = jsoncons::json::object(); diff --git a/tests/cpp/test_serialize_deserialize.hpp b/tests/cpp/test_serialize_deserialize.hpp index 4c0b0e5..9feb6c7 100644 --- a/tests/cpp/test_serialize_deserialize.hpp +++ b/tests/cpp/test_serialize_deserialize.hpp @@ -150,6 +150,58 @@ TEST(SERIALIZE_DESERIALIZE_SUITE_NAME, testSerializeCommitOnCloseCountNoCommitRu } } +TEST(SERIALIZE_DESERIALIZE_SUITE_NAME, testCompressedSerializationUsesLongestPrefix) { + for (const auto &_cl_version : CAPIO_CL_AVAIL_VERSIONS) { + const std::filesystem::path config_path("./compressed-config.json"); + const std::vector regular = {"/data/one", "/data/two"}; + const std::vector special = {"/data/special/one", "/data/special/two"}; + std::string producer = "producer", consumer = "consumer"; + + capiocl::engine::Engine engine; + for (const auto &path : regular) { + engine.addProducer(path, producer); + engine.addConsumer(path, consumer); + engine.setCommitRule(path, capiocl::commitRules::ON_CLOSE); + engine.setCommitedCloseNumber(path, 2); + engine.setFireRule(path, capiocl::fireRules::NO_UPDATE); + engine.setStoreFileInMemory(path); + engine.setPermanent(path, true); + } + for (const auto &path : special) { + engine.addProducer(path, producer); + engine.addConsumer(path, consumer); + engine.setCommitRule(path, capiocl::commitRules::ON_TERMINATION); + engine.setFireRule(path, capiocl::fireRules::UPDATE); + } + + capiocl::serializer::Serializer::dump(engine, config_path, true, _cl_version); + auto compressed = capiocl::parser::Parser::parse(config_path, ""); + const auto paths = compressed->getPaths(); + + EXPECT_EQ(paths.size(), 2); + EXPECT_NE(std::find(paths.begin(), paths.end(), "/data/*"), paths.end()); + EXPECT_NE(std::find(paths.begin(), paths.end(), "/data/special/*"), paths.end()); + + EXPECT_EQ(compressed->getCommitRule(regular.front()), capiocl::commitRules::ON_CLOSE); + EXPECT_EQ(compressed->getCommitCloseCount(regular.front()), 2); + EXPECT_EQ(compressed->getFireRule(regular.front()), capiocl::fireRules::NO_UPDATE); + EXPECT_TRUE(compressed->isStoredInMemory(regular.front())); + EXPECT_TRUE(compressed->isPermanent(regular.front())); + EXPECT_TRUE(compressed->isProducer(regular.front(), "producer")); + EXPECT_TRUE(compressed->isConsumer(regular.front(), "consumer")); + + EXPECT_EQ(compressed->getCommitRule(special.front()), + capiocl::commitRules::ON_TERMINATION); + EXPECT_EQ(compressed->getFireRule(special.front()), capiocl::fireRules::UPDATE); + EXPECT_FALSE(compressed->isStoredInMemory(special.front())); + EXPECT_FALSE(compressed->isPermanent(special.front())); + EXPECT_TRUE(compressed->isProducer(special.front(), "producer")); + EXPECT_TRUE(compressed->isConsumer(special.front(), "consumer")); + + std::filesystem::remove(config_path); + } +} + TEST(SERIALIZE_DESERIALIZE_SUITE_NAME, testParserResolveAbsolute) { for (const auto &_cl_version : CAPIO_CL_AVAIL_VERSIONS) { const std::filesystem::path json_path("/tmp/capio_cl_jsons/V" + _cl_version + @@ -174,4 +226,4 @@ TEST(SERIALIZE_DESERIALIZE_SUITE_NAME, testNoStorageSection) { EXPECT_TRUE(engine->contains("/tmp/file1")); } } -#endif // CAPIO_CL_TEST_SERIALIZE_DESERIALIZE_HPP \ No newline at end of file +#endif // CAPIO_CL_TEST_SERIALIZE_DESERIALIZE_HPP From 77f62de29b6079fafe38e7e5551610f75351aa44 Mon Sep 17 00:00:00 2001 From: marcoSanti Date: Mon, 3 Aug 2026 14:57:22 +0200 Subject: [PATCH 4/4] More tests --- capiocl/serializer.h | 5 ++++ tests/cpp/test_serialize_deserialize.hpp | 32 ++++++++++++++++++++++++ 2 files changed, 37 insertions(+) diff --git a/capiocl/serializer.h b/capiocl/serializer.h index 5502a24..6e434d6 100644 --- a/capiocl/serializer.h +++ b/capiocl/serializer.h @@ -32,6 +32,11 @@ class SerializerException final : public std::exception { /// @brief Dump the current loaded CAPIO-CL configuration from class Engine to a CAPIO-CL /// configuration file. class Serializer final { + /** + * Compress entries from a CAPIO-CL engine into entries using wildcards. + * @param engine + * @return + */ static std::vector> compressedPaths(const engine::Engine &engine); diff --git a/tests/cpp/test_serialize_deserialize.hpp b/tests/cpp/test_serialize_deserialize.hpp index 9feb6c7..aec709b 100644 --- a/tests/cpp/test_serialize_deserialize.hpp +++ b/tests/cpp/test_serialize_deserialize.hpp @@ -1,6 +1,8 @@ #ifndef CAPIO_CL_TEST_SERIALIZE_DESERIALIZE_HPP #define CAPIO_CL_TEST_SERIALIZE_DESERIALIZE_HPP +#include + #define SERIALIZE_DESERIALIZE_SUITE_NAME TestSerializeAndDeserialize TEST(SERIALIZE_DESERIALIZE_SUITE_NAME, testSerializeParseCAPIOCLV1) { @@ -202,6 +204,36 @@ TEST(SERIALIZE_DESERIALIZE_SUITE_NAME, testCompressedSerializationUsesLongestPre } } +TEST(SERIALIZE_DESERIALIZE_SUITE_NAME, testCompressedSerializationGroupsThousandsOfFiles) { + for (const auto &_cl_version : CAPIO_CL_AVAIL_VERSIONS) { + const std::filesystem::path config_path("./large-compressed-config.json"); + std::string producer = "producer"; + capiocl::engine::Engine engine; + + for (int i = 0; i < 5000; ++i) { + const auto path = "/data/regular/file-" + std::to_string(i); + engine.addProducer(path, producer); + engine.setCommitRule(path, capiocl::commitRules::ON_CLOSE); + } + for (int i = 0; i < 1000; ++i) { + const auto path = "/data/special/file-" + std::to_string(i); + engine.addProducer(path, producer); + engine.setCommitRule(path, capiocl::commitRules::ON_TERMINATION); + } + + capiocl::serializer::Serializer::dump(engine, config_path, true, _cl_version); + std::ifstream config(config_path); + const auto doc = jsoncons::json::parse(config); + const auto paths = doc["IO_Graph"][0]["output_stream"].as>(); + + EXPECT_EQ(paths.size(), 2); + EXPECT_NE(std::find(paths.begin(), paths.end(), "/data/regular/*"), paths.end()); + EXPECT_NE(std::find(paths.begin(), paths.end(), "/data/special/*"), paths.end()); + + std::filesystem::remove(config_path); + } +} + TEST(SERIALIZE_DESERIALIZE_SUITE_NAME, testParserResolveAbsolute) { for (const auto &_cl_version : CAPIO_CL_AVAIL_VERSIONS) { const std::filesystem::path json_path("/tmp/capio_cl_jsons/V" + _cl_version +