From 4a7b48bebd77ddebcf96b3b282ac598f280d3cfb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Franz=20P=C3=B6schel?= Date: Fri, 21 Jun 2024 10:59:24 +0200 Subject: [PATCH 1/5] Main implementation --- src/IO/ADIOS/ADIOS2IOHandler.cpp | 38 ++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/src/IO/ADIOS/ADIOS2IOHandler.cpp b/src/IO/ADIOS/ADIOS2IOHandler.cpp index d9d1e3eeb3..8f50b086b9 100644 --- a/src/IO/ADIOS/ADIOS2IOHandler.cpp +++ b/src/IO/ADIOS/ADIOS2IOHandler.cpp @@ -30,6 +30,7 @@ #include "openPMD/IterationEncoding.hpp" #include "openPMD/auxiliary/Environment.hpp" #include "openPMD/auxiliary/Filesystem.hpp" +#include "openPMD/auxiliary/JSON_internal.hpp" #include "openPMD/auxiliary/Mpi.hpp" #include "openPMD/auxiliary/StringManip.hpp" #include "openPMD/auxiliary/TypeTraits.hpp" @@ -40,6 +41,7 @@ #include #include #include +#include #include #include #include @@ -1496,6 +1498,42 @@ void ADIOS2IOHandlerImpl::touch( adios2::Mode ADIOS2IOHandlerImpl::adios2AccessMode(std::string const &fullPath) { + if (m_config.json().contains("engine") && + m_config["engine"].json().contains("access_mode")) + { + auto const &access_mode_json = m_config["engine"]["access_mode"].json(); + auto maybe_access_mode_string = + json::asLowerCaseStringDynamic(access_mode_json); + if (!maybe_access_mode_string.has_value()) + { + throw error::BackendConfigSchema( + {"adios2", "engine", "access_mode"}, "Must be of string type."); + } + auto access_mode_string = *maybe_access_mode_string; + using pair_t = std::pair; + constexpr std::array modeNames{ + pair_t{"write", adios2::Mode::Write}, + pair_t{"read", adios2::Mode::Read}, + pair_t{"append", adios2::Mode::Append}, + pair_t{"readrandomaccess", adios2::Mode::ReadRandomAccess}}; + for (auto const &[name, mode] : modeNames) + { + if (name == access_mode_string) + { + return mode; + } + } + std::stringstream error; + error << "Unsupported value '" << access_mode_string + << "'. Must be one of:"; + for (auto const &pair : modeNames) + { + error << " '" << pair.first << "'"; + } + error << '.'; + throw error::BackendConfigSchema( + {"adios2", "engine", "access_mode"}, error.str()); + } switch (m_handler->m_backendAccess) { case Access::CREATE: From 22920de7b91902151ee8fac8b156a9274c732cdc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Franz=20P=C3=B6schel?= Date: Fri, 21 Jun 2024 11:35:26 +0200 Subject: [PATCH 2/5] Add test --- test/SerialIOTest.cpp | 52 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/test/SerialIOTest.cpp b/test/SerialIOTest.cpp index feaab30788..92675110aa 100644 --- a/test/SerialIOTest.cpp +++ b/test/SerialIOTest.cpp @@ -4475,6 +4475,58 @@ TEST_CASE("adios2_flush_via_step") } } #endif + + /* + * Now emulate restarting from a checkpoint after a crash and continuing to + * write to the output Series. The semantics of openPMD::Access::APPEND + * don't fully fit here since that mode is for adding new Iterations to an + * existing Series. What we truly want to do is to continue writing to an + * Iteration without replacing it with a new one. So we must use the option + * adios2.engine.access_mode = "append" to tell the ADIOS2 backend that new + * steps should be added to an existing Iteration file. + */ + + write = Series( + "../samples/adios2_flush_via_step/simData_%T.bp5", + Access::APPEND, + R"( + [adios2.engine] + access_mode = "append" + parameters.FlattenSteps = "on" + )"); + for (Iteration::IterationIndex_t i = 0; i < 5; ++i) + { + Iteration it = write.writeIterations()[i]; + auto E_x = it.meshes["E"]["y"]; + E_x.resetDataset({Datatype::FLOAT, {10, 10}}); + for (Extent::value_type j = 0; j < 10; ++j) + { + std::iota(data.begin(), data.end(), i * 100 + j * 10); + E_x.storeChunk(data, {j, 0}, {1, 10}); + write.flush(R"(adios2.engine.preferred_flush_target = "new_step")"); + } + it.close(); + } + +#if openPMD_HAS_ADIOS_2_10_1 + for (auto access : {Access::READ_RANDOM_ACCESS, Access::READ_LINEAR}) + { + Series read("../samples/adios2_flush_via_step/simData_%T.%E", access); + std::vector load_data(100); + data.resize(100); + for (auto iteration : read.readIterations()) + { + std::iota(data.begin(), data.end(), iteration.iterationIndex * 100); + iteration.meshes["E"]["x"].loadChunkRaw( + load_data.data(), {0, 0}, {10, 10}); + iteration.meshes["E"]["y"].loadChunkRaw( + load_data.data(), {0, 0}, {10, 10}); + iteration.close(); + REQUIRE(load_data == data); + REQUIRE(load_data == data); + } + } +#endif } #endif From cfa96422e6fb2a4209f69fab2da27e66025b33b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Franz=20P=C3=B6schel?= Date: Fri, 21 Jun 2024 11:47:48 +0200 Subject: [PATCH 3/5] Documentation --- docs/source/details/backendconfig.rst | 3 +++ docs/source/usage/workflow.rst | 2 ++ 2 files changed, 5 insertions(+) diff --git a/docs/source/details/backendconfig.rst b/docs/source/details/backendconfig.rst index fae114a01c..49cccebe69 100644 --- a/docs/source/details/backendconfig.rst +++ b/docs/source/details/backendconfig.rst @@ -122,6 +122,9 @@ Explanation of the single keys: * ``adios2.engine.type``: A string that is passed directly to ``adios2::IO:::SetEngine`` for choosing the ADIOS2 engine to be used. Please refer to the `official ADIOS2 documentation `_ for a list of available engines. +* ``adios2.engine.access_mode``: One of ``"Write", "Read", "Append", "ReadRandomAccess"``. + Only needed in specific use cases, the access mode is usually determined from the specified ``openPMD::Access``. + Useful for finetuning the backend-specific behavior of ADIOS2 when overwriting existing Iterations in file-based Append mode. * ``adios2.engine.parameters``: An associative array of string-formatted engine parameters, passed directly through to ``adios2::IO::SetParameters``. Please refer to the `official ADIOS2 documentation `_ for the available engine parameters. The openPMD-api does not interpret these values and instead simply forwards them to ADIOS2. diff --git a/docs/source/usage/workflow.rst b/docs/source/usage/workflow.rst index ec44e2e70f..03fa4ce9d4 100644 --- a/docs/source/usage/workflow.rst +++ b/docs/source/usage/workflow.rst @@ -102,6 +102,8 @@ The openPMD-api distinguishes between a number of different access modes: We suggest to fully define iterations when using Append mode (i.e. as if using Create mode) to avoid implementation-specific behavior. Appending to an openPMD Series is only supported on a per-iteration level. + **Tip:** Use the ``adios2.engine.access_mode`` :ref:`backend key ` of the :ref:`ADIOS2 backend ` to finetune the backend-specific behavior of Append mode for niche use cases. + **Warning:** There is no reading involved in using Append mode. It is a user's responsibility to ensure that the appended dataset and the appended-to dataset are compatible with each other. The results of using incompatible backend configurations are undefined. From ef769753c44eaadf709a70f094af8d6f0a07da66 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Franz=20P=C3=B6schel?= Date: Fri, 21 Jun 2024 14:00:16 +0200 Subject: [PATCH 4/5] Guard against ADIOS2 v2.7 --- src/IO/ADIOS/ADIOS2IOHandler.cpp | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/IO/ADIOS/ADIOS2IOHandler.cpp b/src/IO/ADIOS/ADIOS2IOHandler.cpp index 8f50b086b9..d97769d19b 100644 --- a/src/IO/ADIOS/ADIOS2IOHandler.cpp +++ b/src/IO/ADIOS/ADIOS2IOHandler.cpp @@ -1514,8 +1514,12 @@ adios2::Mode ADIOS2IOHandlerImpl::adios2AccessMode(std::string const &fullPath) constexpr std::array modeNames{ pair_t{"write", adios2::Mode::Write}, pair_t{"read", adios2::Mode::Read}, - pair_t{"append", adios2::Mode::Append}, - pair_t{"readrandomaccess", adios2::Mode::ReadRandomAccess}}; + pair_t{"append", adios2::Mode::Append} +#if openPMD_HAS_ADIOS_2_8 + , + pair_t{"readrandomaccess", adios2::Mode::ReadRandomAccess} +#endif + }; for (auto const &[name, mode] : modeNames) { if (name == access_mode_string) From 5132b6d2a99925100dd6df2ff45069bfae8bb5c9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Franz=20P=C3=B6schel?= Date: Wed, 17 Jul 2024 11:53:35 +0200 Subject: [PATCH 5/5] Extend the test to parallel --- test/ParallelIOTest.cpp | 115 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 115 insertions(+) diff --git a/test/ParallelIOTest.cpp b/test/ParallelIOTest.cpp index e803e1a0f7..750c3fe50e 100644 --- a/test/ParallelIOTest.cpp +++ b/test/ParallelIOTest.cpp @@ -2082,4 +2082,119 @@ TEST_CASE("joined_dim", "[parallel]") } } +#if openPMD_HAVE_ADIOS2_BP5 +// Parallel version of the same test from SerialIOTest.cpp +TEST_CASE("adios2_flush_via_step") +{ + int size_i(0), rank_i(0); + MPI_Comm_rank(MPI_COMM_WORLD, &rank_i); + MPI_Comm_size(MPI_COMM_WORLD, &size_i); + Extent::value_type const size(size_i), rank(rank_i); + + Series write( + "../samples/adios2_flush_via_step_parallel/simData_%T.bp5", + Access::CREATE, + MPI_COMM_WORLD, + R"(adios2.engine.parameters.FlattenSteps = "on")"); + std::vector data(10); + for (Iteration::IterationIndex_t i = 0; i < 5; ++i) + { + Iteration it = write.writeIterations()[i]; + auto E_x = it.meshes["E"]["x"]; + E_x.resetDataset({Datatype::FLOAT, {size, 10, 10}}); + for (Extent::value_type j = 0; j < 10; ++j) + { + std::iota( + data.begin(), data.end(), i * 100 * size + rank * 100 + j * 10); + E_x.storeChunk(data, {rank, j, 0}, {1, 1, 10}); + write.flush(R"(adios2.engine.preferred_flush_target = "new_step")"); + } + it.close(); + } + +#if openPMD_HAS_ADIOS_2_10_1 + for (auto access : {Access::READ_RANDOM_ACCESS, Access::READ_LINEAR}) + { + Series read( + "../samples/adios2_flush_via_step_parallel/simData_%T.%E", + access, + MPI_COMM_WORLD); + std::vector load_data(100 * size); + data.resize(100 * size); + for (auto iteration : read.readIterations()) + { + std::iota( + data.begin(), + data.end(), + iteration.iterationIndex * size * 100); + iteration.meshes["E"]["x"].loadChunkRaw( + load_data.data(), {0, 0, 0}, {size, 10, 10}); + iteration.close(); + REQUIRE(load_data == data); + } + } +#endif + + /* + * Now emulate restarting from a checkpoint after a crash and continuing to + * write to the output Series. The semantics of openPMD::Access::APPEND + * don't fully fit here since that mode is for adding new Iterations to an + * existing Series. What we truly want to do is to continue writing to an + * Iteration without replacing it with a new one. So we must use the option + * adios2.engine.access_mode = "append" to tell the ADIOS2 backend that new + * steps should be added to an existing Iteration file. + */ + + write = Series( + "../samples/adios2_flush_via_step_parallel/simData_%T.bp5", + Access::APPEND, + MPI_COMM_WORLD, + R"( + [adios2.engine] + access_mode = "append" + parameters.FlattenSteps = "on" + )"); + for (Iteration::IterationIndex_t i = 0; i < 5; ++i) + { + Iteration it = write.writeIterations()[i]; + auto E_x = it.meshes["E"]["y"]; + E_x.resetDataset({Datatype::FLOAT, {size, 10, 10}}); + for (Extent::value_type j = 0; j < 10; ++j) + { + std::iota( + data.begin(), data.end(), i * 100 * size + rank * 100 + j * 10); + E_x.storeChunk(data, {rank, j, 0}, {1, 1, 10}); + write.flush(R"(adios2.engine.preferred_flush_target = "new_step")"); + } + it.close(); + } + +#if openPMD_HAS_ADIOS_2_10_1 + for (auto access : {Access::READ_RANDOM_ACCESS, Access::READ_LINEAR}) + { + Series read( + "../samples/adios2_flush_via_step_parallel/simData_%T.%E", + access, + MPI_COMM_WORLD); + std::vector load_data(100 * size); + data.resize(100 * size); + for (auto iteration : read.readIterations()) + { + std::iota( + data.begin(), + data.end(), + iteration.iterationIndex * size * 100); + iteration.meshes["E"]["x"].loadChunkRaw( + load_data.data(), {0, 0, 0}, {size, 10, 10}); + iteration.meshes["E"]["y"].loadChunkRaw( + load_data.data(), {0, 0, 0}, {size, 10, 10}); + iteration.close(); + REQUIRE(load_data == data); + REQUIRE(load_data == data); + } + } +#endif +} +#endif + #endif // openPMD_HAVE_ADIOS2 && openPMD_HAVE_MPI