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..6e434d6 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,6 +32,19 @@ 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); + + /** + * Sort path entries from longest to shortest + * @param paths + */ + static void sortPathsByDecreasingLength(std::vector &paths); /// @brief Available serializers for CAPIO-CL struct available_serializers { @@ -37,10 +54,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 +66,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 +80,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 +#endif // CAPIO_CL_SERIALIZER_H diff --git a/src/Serializer.cpp b/src/Serializer.cpp index 27698bb..1f7704d 100644 --- a/src/Serializer.cpp +++ b/src/Serializer.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -10,15 +11,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()); @@ -33,3 +34,87 @@ capiocl::serializer::SerializerException::SerializerException(const std::string UPDATE_CALF_WORKFLOW_NAME(""); CALF_PRINT_COLOR(CALF_CLI_LEVEL_ERROR, "%s", msg.c_str()); } + +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; + }); +} + +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; + } + + 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(); + } + } + + sortPathsByDecreasingLength(directories); + + for (const auto &directory : directories) { + const auto wildcard = (std::filesystem::path(directory) / "*").string(); + if (paths.find(wildcard) != paths.end()) { + continue; + } + + 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; + } + + 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); + } + 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 677780c..35a6b1e 100644 --- a/src/serializers/v1.1.cpp +++ b/src/serializers/v1.1.cpp @@ -7,14 +7,26 @@ #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()); + + 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; + 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; @@ -27,7 +39,17 @@ 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) { + 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 (entry.permanent) { permanent.push_back(path); } @@ -44,13 +66,18 @@ 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); + 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"; @@ -90,7 +117,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 5151843..761ebf9 100644 --- a/src/serializers/v1.cpp +++ b/src/serializers/v1.cpp @@ -7,13 +7,24 @@ #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; + 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; @@ -26,7 +37,17 @@ 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) { + 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 (entry.permanent) { permanent.push_back(path); } @@ -43,13 +64,18 @@ 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::vector filtered_outputs; + + sortPathsByDecreasingLength(outputs); for (const auto &path : outputs) { const auto &entry = files.at(path); + 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 +115,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..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) { @@ -42,7 +44,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 +75,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 +110,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 +138,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); @@ -150,6 +152,88 @@ 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, 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 + @@ -174,4 +258,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