From 40a0b070e0d23f3ac523b0f06c76f302ef175f17 Mon Sep 17 00:00:00 2001 From: desp0042 Date: Thu, 30 Jul 2026 07:16:56 +0200 Subject: [PATCH 1/3] refactor(output): require observer-owned HDF5 communication (ADC-683) --- .../pops/runtime/output/hdf5_collective.hpp | 10 - .../bindings/core/init/init_parallel_hdf5.cpp | 22 +- python/pops/output/_writers/hdf5.py | 2 +- src/runtime/output/hdf5_collective.cpp | 195 ++++++++---------- 4 files changed, 90 insertions(+), 139 deletions(-) diff --git a/include/pops/runtime/output/hdf5_collective.hpp b/include/pops/runtime/output/hdf5_collective.hpp index 9a5e4bf51..c89db165b 100644 --- a/include/pops/runtime/output/hdf5_collective.hpp +++ b/include/pops/runtime/output/hdf5_collective.hpp @@ -6,10 +6,6 @@ #include -namespace pops { -class WorldCommunicator; -} - namespace pops::runtime::output { /// Non-owning, contiguous NumPy-compatible array view used by the native HDF5 adapter. @@ -56,8 +52,6 @@ struct ParallelHdf5Capability { /// rank is allowed to enter HDF5. An empty string means that local validation succeeded. void collective_hdf5_input_consensus(const CommunicatorView& communicator, const std::string& local_error); -void collective_hdf5_input_consensus(const WorldCommunicator& world, - const std::string& local_error); /// Write one exact scientific-output artifact collectively on an explicit native communicator. /// @@ -71,9 +65,5 @@ void write_collective_hdf5(const CommunicatorView& communicator, const std::stri const std::string& manifest_json, const std::vector& root_arrays, const std::vector& fields); -void write_collective_hdf5(const WorldCommunicator& world, const std::string& path, - const std::string& manifest_json, - const std::vector& root_arrays, - const std::vector& fields); } // namespace pops::runtime::output diff --git a/python/bindings/core/init/init_parallel_hdf5.cpp b/python/bindings/core/init/init_parallel_hdf5.cpp index d2427257c..5e8cf2bde 100644 --- a/python/bindings/core/init/init_parallel_hdf5.cpp +++ b/python/bindings/core/init/init_parallel_hdf5.cpp @@ -1,7 +1,6 @@ #include "../bindings_detail.hpp" #include -#include #include #include @@ -97,21 +96,12 @@ void init_parallel_hdf5(py::module_& m) { [](const py::object& communicator_value, const py::object& path_value, const py::object& manifest_value, const py::object& root_arrays_value, const py::object& field_rows_value) { - pops::CommunicatorView communicator; - if (py::isinstance(communicator_value)) { - auto& world = communicator_value.cast(); - if (&world != &pops::WorldCommunicator::world()) - throw py::value_error("native HDF5 requires the exact process-world authority"); - communicator = world.communicator(); - } else if (py::isinstance(communicator_value)) { - auto& lane = communicator_value.cast(); - if (!lane.active()) - throw py::value_error("native HDF5 observer lane is closed"); - communicator = lane.communicator(); - } else { - throw py::type_error( - "native HDF5 requires a PoPS world communicator or observer MPI lane"); - } + if (!py::isinstance(communicator_value)) + throw py::type_error("native HDF5 requires an exact duplicated observer MPI lane"); + auto& lane = communicator_value.cast(); + if (!lane.active()) + throw py::value_error("native HDF5 observer lane is closed"); + const pops::CommunicatorView communicator = lane.communicator(); std::vector owners; std::vector arrays; std::vector fields; diff --git a/python/pops/output/_writers/hdf5.py b/python/pops/output/_writers/hdf5.py index 0e5c88609..6356eed6b 100644 --- a/python/pops/output/_writers/hdf5.py +++ b/python/pops/output/_writers/hdf5.py @@ -177,7 +177,7 @@ def _parallel_snapshot_data( if request.parallel_mode is not ParallelMode.COLLECTIVE: raise ValueError( "a resolved communicator is valid only for HDF5 COLLECTIVE output") - require_communicator(communicator) + require_communicator(communicator, allow_world=False) if request.rank != rank(communicator): raise ValueError("collective HDF5 request rank differs from its native communicator") native, capability = _require_native_parallel_hdf5() diff --git a/src/runtime/output/hdf5_collective.cpp b/src/runtime/output/hdf5_collective.cpp index 32781070d..aaa84e31c 100644 --- a/src/runtime/output/hdf5_collective.cpp +++ b/src/runtime/output/hdf5_collective.cpp @@ -1,5 +1,4 @@ #include -#include #include #include @@ -160,9 +159,9 @@ template template [[nodiscard]] AgreedFailure collective_phase(int rank, MPI_Comm communicator, - Operation&& operation) { - return agree_failure( - rank, capture_local_failure(std::forward(operation)), communicator); + Operation&& operation) { + return agree_failure(rank, capture_local_failure(std::forward(operation)), + communicator); } [[noreturn]] void throw_collective_failure(std::string_view phase, std::string_view subject, @@ -186,8 +185,8 @@ void require_collective_success(std::string_view phase, std::string_view subject [[nodiscard]] AgreedFailure require_identical_text(int rank, std::string_view local, MPI_Comm communicator) { - int overflow = local.size() > static_cast( - std::numeric_limits::max()); + int overflow = + local.size() > static_cast(std::numeric_limits::max()); require_mpi(MPI_Allreduce(MPI_IN_PLACE, &overflow, 1, MPI_INT, MPI_MAX, communicator), "MPI_Allreduce(schema length overflow)"); if (overflow != 0) { @@ -217,8 +216,7 @@ void require_collective_success(std::string_view phase, std::string_view subject length - offset, static_cast(std::numeric_limits::max()))); char* buffer = rank == 0 ? const_cast(local.data()) + static_cast(offset) : reference.data() + static_cast(offset); - require_mpi(MPI_Bcast(buffer, count, MPI_CHAR, 0, communicator), - "MPI_Bcast(schema bytes)"); + require_mpi(MPI_Bcast(buffer, count, MPI_CHAR, 0, communicator), "MPI_Bcast(schema bytes)"); offset += static_cast(count); } @@ -231,8 +229,7 @@ void require_collective_success(std::string_view phase, std::string_view subject using PieceDescriptor = std::array; -[[nodiscard]] std::vector piece_descriptors( - const std::vector& fields) { +[[nodiscard]] std::vector piece_descriptors(const std::vector& fields) { std::size_t count = 0; for (const auto& field : fields) count = checked_add(count, field.pieces.size(), "native HDF5 piece descriptor count"); @@ -251,11 +248,10 @@ using PieceDescriptor = std::array; } } for (const auto& piece : field.pieces) { - result.push_back({static_cast(field_index), - static_cast(piece.jlo), - static_cast(piece.ilo), - static_cast(piece.jhi), - static_cast(piece.ihi)}); + result.push_back( + {static_cast(field_index), static_cast(piece.jlo), + static_cast(piece.ilo), static_cast(piece.jhi), + static_cast(piece.ihi)}); } } return result; @@ -263,13 +259,14 @@ using PieceDescriptor = std::array; [[nodiscard]] bool pieces_overlap(const PieceDescriptor& left, const PieceDescriptor& right) noexcept { - return left[0] == right[0] && left[1] < right[3] && right[1] < left[3] && - left[2] < right[4] && right[2] < left[4]; + return left[0] == right[0] && left[1] < right[3] && right[1] < left[3] && left[2] < right[4] && + right[2] < left[4]; } -[[nodiscard]] AgreedFailure require_disjoint_rank_pieces( - int rank, int ranks, const std::vector& local, - const std::vector& fields, MPI_Comm communicator) { +[[nodiscard]] AgreedFailure require_disjoint_rank_pieces(int rank, int ranks, + const std::vector& local, + const std::vector& fields, + MPI_Comm communicator) { static_assert(sizeof(PieceDescriptor) == 5 * sizeof(unsigned long long)); int length_overflow = 0; if constexpr (sizeof(std::size_t) > sizeof(unsigned long long)) { @@ -301,8 +298,7 @@ using PieceDescriptor = std::array; return finish(type_failure); for (int owner = 0; owner < ranks; ++owner) { - unsigned long long count = - rank == owner ? static_cast(local.size()) : 0ULL; + unsigned long long count = rank == owner ? static_cast(local.size()) : 0ULL; require_mpi(MPI_Bcast(&count, 1, MPI_UNSIGNED_LONG_LONG, owner, communicator), "MPI_Bcast(piece descriptor count)"); @@ -321,8 +317,8 @@ using PieceDescriptor = std::array; while (offset < count) { const int chunk = static_cast(std::min( count - offset, static_cast(std::numeric_limits::max()))); - require_mpi(MPI_Bcast(buffer + static_cast(offset), chunk, descriptor_type, owner, - communicator), + require_mpi(MPI_Bcast(buffer + static_cast(offset), chunk, descriptor_type, + owner, communicator), "MPI_Bcast(piece descriptors)"); offset += static_cast(chunk); } @@ -335,12 +331,12 @@ using PieceDescriptor = std::array; if (!pieces_overlap(mine, theirs)) continue; const auto field_index = static_cast(mine[0]); - const std::string_view dataset = - field_index < fields.size() ? std::string_view{fields[field_index].dataset} - : std::string_view{""}; - throw std::invalid_argument( - "field pieces overlap across MPI ranks " + std::to_string(owner) + " and " + - std::to_string(rank) + " for dataset " + std::string(dataset)); + const std::string_view dataset = field_index < fields.size() + ? std::string_view{fields[field_index].dataset} + : std::string_view{""}; + throw std::invalid_argument("field pieces overlap across MPI ranks " + + std::to_string(owner) + " and " + std::to_string(rank) + + " for dataset " + std::string(dataset)); } } }); @@ -566,8 +562,7 @@ void validate_inputs(const std::string& path, const std::string& manifest, return result; } -[[nodiscard]] std::vector group_paths( - const std::vector& datasets) { +[[nodiscard]] std::vector group_paths(const std::vector& datasets) { std::vector groups; for (const auto& dataset : datasets) { std::size_t cursor = 0; @@ -590,8 +585,8 @@ struct DatasetCreatePlan { std::vector zero; }; -[[nodiscard]] DatasetCreatePlan prepare_dataset_creation( - const std::vector& shape, const std::string& dtype) { +[[nodiscard]] DatasetCreatePlan prepare_dataset_creation(const std::vector& shape, + const std::string& dtype) { const auto dimensions = hdf5_shape(shape); DatasetCreatePlan plan; plan.space = H5Handle( @@ -766,12 +761,9 @@ struct ManifestAttributePlan { ParallelHdf5Capability parallel_hdf5_capability() { #if defined(POPS_HAS_PARALLEL_HDF5) - std::lock_guard guard{parallel_hdf5_mutex()}; - int initialized = 0; - require_mpi(MPI_Initialized(&initialized), "MPI_Initialized"); const std::string version = std::to_string(H5_VERS_MAJOR) + "." + std::to_string(H5_VERS_MINOR) + "." + std::to_string(H5_VERS_RELEASE); - return {true, version, initialized ? "" : "MPI is compiled but not initialized"}; + return {true, version, ""}; #else return {false, "", "module was not built with MPI and a parallel HDF5 C library"}; #endif @@ -794,19 +786,10 @@ void collective_hdf5_input_consensus(const CommunicatorView& communicator, LocalFailure local; if (!local_error.empty()) set_failure(local, local_error); - require_collective_success( - "binding input validation", "", agree_failure(rank, local, native)); + require_collective_success("binding input validation", "", agree_failure(rank, local, native)); #endif } -void collective_hdf5_input_consensus(const WorldCommunicator& world, - const std::string& local_error) { - if (&world != &WorldCommunicator::world()) - throw std::invalid_argument( - "collective HDF5 requires the exact native process-world authority"); - collective_hdf5_input_consensus(world.communicator(), local_error); -} - void write_collective_hdf5(const CommunicatorView& communicator, const std::string& path, const std::string& manifest_json, const std::vector& root_arrays, @@ -831,24 +814,22 @@ void write_collective_hdf5(const CommunicatorView& communicator, const std::stri require_mpi(MPI_Comm_size(native, &ranks), "MPI_Comm_size"); require_collective_success("input validation", "", collective_phase(rank, native, [&] { - validate_inputs(path, manifest_json, root_arrays, fields); - })); + validate_inputs(path, manifest_json, root_arrays, fields); + })); std::string schema; require_collective_success("schema preparation", "", collective_phase(rank, native, [&] { - schema = schema_text(path, manifest_json, root_arrays, fields); - })); - require_collective_success( - "schema consensus", "", require_identical_text(rank, schema, native)); + schema = schema_text(path, manifest_json, root_arrays, fields); + })); + require_collective_success("schema consensus", "", require_identical_text(rank, schema, native)); std::vector descriptors; require_collective_success( - "piece descriptor preparation", "", collective_phase(rank, native, [&] { - descriptors = piece_descriptors(fields); - })); - require_collective_success("piece descriptor consensus", "", - require_disjoint_rank_pieces( - rank, ranks, descriptors, fields, native)); + "piece descriptor preparation", "", + collective_phase(rank, native, [&] { descriptors = piece_descriptors(fields); })); + require_collective_success( + "piece descriptor consensus", "", + require_disjoint_rank_pieces(rank, ranks, descriptors, fields, native)); std::vector dataset_names; std::vector groups; @@ -861,48 +842,49 @@ void write_collective_hdf5(const CommunicatorView& communicator, const std::stri H5Handle transfer; require_collective_success( "local HDF5 preparation", "", collective_phase(rank, native, [&] { - dataset_names.reserve(root_arrays.size() + fields.size()); - for (const auto& array : root_arrays) - dataset_names.push_back(array.dataset); - for (const auto& field : fields) - dataset_names.push_back(field.dataset); - groups = group_paths(dataset_names); - - root_creation_plans.reserve(root_arrays.size()); - for (const auto& array : root_arrays) - root_creation_plans.push_back( - prepare_dataset_creation(array.values.shape, array.values.dtype)); - field_creation_plans.reserve(fields.size()); - for (const auto& field : fields) - field_creation_plans.push_back(prepare_dataset_creation(field.shape, field.dtype)); - manifest_plan = prepare_manifest_attribute(); - group_creation = H5Handle(H5Pcreate(H5P_GROUP_CREATE), H5Pclose); - if (!group_creation || H5Pset_obj_track_times(group_creation.get(), false) < 0) - throw std::runtime_error("HDF5 deterministic group creation-property preparation failed"); - file_creation = H5Handle(H5Pcreate(H5P_FILE_CREATE), H5Pclose); - if (!file_creation || H5Pset_obj_track_times(file_creation.get(), false) < 0) - throw std::runtime_error("HDF5 deterministic file creation-property preparation failed"); - - access = H5Handle(H5Pcreate(H5P_FILE_ACCESS), H5Pclose); - if (!access || H5Pset_fapl_mpio(access.get(), native, MPI_INFO_NULL) < 0) - throw std::runtime_error("H5Pset_fapl_mpio(explicit communicator) failed"); + dataset_names.reserve(root_arrays.size() + fields.size()); + for (const auto& array : root_arrays) + dataset_names.push_back(array.dataset); + for (const auto& field : fields) + dataset_names.push_back(field.dataset); + groups = group_paths(dataset_names); + + root_creation_plans.reserve(root_arrays.size()); + for (const auto& array : root_arrays) + root_creation_plans.push_back( + prepare_dataset_creation(array.values.shape, array.values.dtype)); + field_creation_plans.reserve(fields.size()); + for (const auto& field : fields) + field_creation_plans.push_back(prepare_dataset_creation(field.shape, field.dtype)); + manifest_plan = prepare_manifest_attribute(); + group_creation = H5Handle(H5Pcreate(H5P_GROUP_CREATE), H5Pclose); + if (!group_creation || H5Pset_obj_track_times(group_creation.get(), false) < 0) + throw std::runtime_error("HDF5 deterministic group creation-property preparation failed"); + file_creation = H5Handle(H5Pcreate(H5P_FILE_CREATE), H5Pclose); + if (!file_creation || H5Pset_obj_track_times(file_creation.get(), false) < 0) + throw std::runtime_error("HDF5 deterministic file creation-property preparation failed"); + + access = H5Handle(H5Pcreate(H5P_FILE_ACCESS), H5Pclose); + if (!access || H5Pset_fapl_mpio(access.get(), native, MPI_INFO_NULL) < 0) + throw std::runtime_error("H5Pset_fapl_mpio(explicit communicator) failed"); #if H5_VERSION_GE(1, 10, 0) - if (H5Pset_all_coll_metadata_ops(access.get(), 1) < 0 || - H5Pset_coll_metadata_write(access.get(), 1) < 0) - throw std::runtime_error("parallel HDF5 collective metadata configuration failed"); + if (H5Pset_all_coll_metadata_ops(access.get(), 1) < 0 || + H5Pset_coll_metadata_write(access.get(), 1) < 0) + throw std::runtime_error("parallel HDF5 collective metadata configuration failed"); #endif - transfer = H5Handle(H5Pcreate(H5P_DATASET_XFER), H5Pclose); - if (!transfer || H5Pset_dxpl_mpio(transfer.get(), H5FD_MPIO_COLLECTIVE) < 0) - throw std::runtime_error("H5Pset_dxpl_mpio(H5FD_MPIO_COLLECTIVE) failed"); - })); + transfer = H5Handle(H5Pcreate(H5P_DATASET_XFER), H5Pclose); + if (!transfer || H5Pset_dxpl_mpio(transfer.get(), H5FD_MPIO_COLLECTIVE) < 0) + throw std::runtime_error("H5Pset_dxpl_mpio(H5FD_MPIO_COLLECTIVE) failed"); + })); H5Handle file; - require_collective_success("file creation", path, collective_phase(rank, native, [&] { - file = H5Handle(H5Fcreate(path.c_str(), H5F_ACC_TRUNC, file_creation.get(), access.get()), - H5Fclose); - if (!file) - throw std::runtime_error("H5Fcreate returned an invalid handle"); - })); + require_collective_success( + "file creation", path, collective_phase(rank, native, [&] { + file = H5Handle(H5Fcreate(path.c_str(), H5F_ACC_TRUNC, file_creation.get(), access.get()), + H5Fclose); + if (!file) + throw std::runtime_error("H5Fcreate returned an invalid handle"); + })); AgreedFailure transaction_failure; auto remember_failure = [&](const AgreedFailure& failure) noexcept { @@ -1028,24 +1010,13 @@ void write_collective_hdf5(const CommunicatorView& communicator, const std::stri } const auto close_failure = collective_phase(rank, native, [&] { - const hid_t handle = file.release(); - if (H5Fclose(handle) < 0) - throw std::runtime_error("H5Fclose failed"); + const hid_t handle = file.release(); + if (H5Fclose(handle) < 0) + throw std::runtime_error("H5Fclose failed"); }); remember_failure(close_failure); require_collective_success("transaction", "", transaction_failure); #endif } -void write_collective_hdf5(const WorldCommunicator& world, const std::string& path, - const std::string& manifest_json, - const std::vector& root_arrays, - const std::vector& fields) { - if (&world != &WorldCommunicator::world()) - throw std::invalid_argument( - "collective HDF5 requires the exact native process-world authority"); - write_collective_hdf5( - world.communicator(), path, manifest_json, root_arrays, fields); -} - } // namespace pops::runtime::output From c32d76ec02324192f95639cb287c696484b32c96 Mon Sep 17 00:00:00 2001 From: desp0042 Date: Thu, 30 Jul 2026 07:17:26 +0200 Subject: [PATCH 2/3] test(output): fence observer-owned HDF5 lanes (ADC-683) --- .../mpi/test_mpi_hdf5_collective.cpp | 14 ++++-- .../test_hdf5_observer_lane_fence.py | 33 ++++++++++++ .../integration/io/test_hdf5_parallel.py | 50 ++++++++++++------- .../mpi/test_scientific_output_mpi.py | 35 +++++++------ 4 files changed, 93 insertions(+), 39 deletions(-) create mode 100644 tests/python/architecture/test_hdf5_observer_lane_fence.py diff --git a/tests/cpp/integration/mpi/test_mpi_hdf5_collective.cpp b/tests/cpp/integration/mpi/test_mpi_hdf5_collective.cpp index 615d33c52..8f8a43df2 100644 --- a/tests/cpp/integration/mpi/test_mpi_hdf5_collective.cpp +++ b/tests/cpp/integration/mpi/test_mpi_hdf5_collective.cpp @@ -139,6 +139,7 @@ TEST(MpiHdf5Collective, WritesDisjointHyperslabsAndReopensNatively) { FAIL() << "this target must never be registered without native parallel HDF5"; #else auto& world = pops::WorldCommunicator::world(); + const auto communicator = world.communicator(); const int rank = world.rank(); const int ranks = world.size(); ASSERT_GE(rank, 0); @@ -180,7 +181,7 @@ TEST(MpiHdf5Collective, WritesDisjointHyperslabsAndReopensNatively) { local_values.size() * sizeof(double)}}}, }}; const std::string manifest = R"({"format":"native-test","version":1})"; - pops::runtime::output::write_collective_hdf5(world, path_text, manifest, arrays, fields); + pops::runtime::output::write_collective_hdf5(communicator, path_text, manifest, arrays, fields); std::string validation_error; if (rank == 0) { @@ -203,6 +204,7 @@ TEST(MpiHdf5Collective, RejectsOneRankInvalidDescriptorBeforeCreatingFile) { FAIL() << "this target must never be registered without native parallel HDF5"; #else auto& world = pops::WorldCommunicator::world(); + const auto communicator = world.communicator(); const int rank = world.rank(); const int ranks = world.size(); if (ranks < 2) @@ -240,7 +242,7 @@ TEST(MpiHdf5Collective, RejectsOneRankInvalidDescriptorBeforeCreatingFile) { std::string error; try { pops::runtime::output::write_collective_hdf5( - world, path, R"({"format":"native-invalid-test","version":1})", arrays, fields); + communicator, path, R"({"format":"native-invalid-test","version":1})", arrays, fields); } catch (const std::exception& failure) { error = failure.what(); } @@ -266,6 +268,7 @@ TEST(MpiHdf5Collective, RejectsCrossRankOverlappingHyperslabsBeforeCreatingFile) FAIL() << "this target must never be registered without native parallel HDF5"; #else auto& world = pops::WorldCommunicator::world(); + const auto communicator = world.communicator(); const int rank = world.rank(); const int ranks = world.size(); if (ranks < 2) @@ -303,7 +306,7 @@ TEST(MpiHdf5Collective, RejectsCrossRankOverlappingHyperslabsBeforeCreatingFile) std::string error; try { pops::runtime::output::write_collective_hdf5( - world, path, R"({"format":"native-overlap-test","version":1})", arrays, fields); + communicator, path, R"({"format":"native-overlap-test","version":1})", arrays, fields); } catch (const std::exception& failure) { error = failure.what(); } @@ -329,6 +332,7 @@ TEST(MpiHdf5Collective, RepeatedIdenticalWritesAreByteIdenticalAcrossTime) { FAIL() << "this target must never be registered without native parallel HDF5"; #else auto& world = pops::WorldCommunicator::world(); + const auto communicator = world.communicator(); const int rank = world.rank(); const int ranks = world.size(); const std::string first_path = shared_temporary_path(world, "native-parallel-hdf5-exact-a"); @@ -369,9 +373,9 @@ TEST(MpiHdf5Collective, RepeatedIdenticalWritesAreByteIdenticalAcrossTime) { }}; const std::string manifest = R"({"format":"native-exact-test","version":1})"; - pops::runtime::output::write_collective_hdf5(world, first_path, manifest, arrays, fields); + pops::runtime::output::write_collective_hdf5(communicator, first_path, manifest, arrays, fields); std::this_thread::sleep_for(std::chrono::milliseconds(1200)); - pops::runtime::output::write_collective_hdf5(world, second_path, manifest, arrays, fields); + pops::runtime::output::write_collective_hdf5(communicator, second_path, manifest, arrays, fields); std::string validation_error; if (rank == 0) { diff --git a/tests/python/architecture/test_hdf5_observer_lane_fence.py b/tests/python/architecture/test_hdf5_observer_lane_fence.py new file mode 100644 index 000000000..ee07cd388 --- /dev/null +++ b/tests/python/architecture/test_hdf5_observer_lane_fence.py @@ -0,0 +1,33 @@ +"""ADC-683 fences for observer-owned collective HDF5 communication.""" + +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[3] +HEADER = ROOT / "include/pops/runtime/output/hdf5_collective.hpp" +SOURCE = ROOT / "src/runtime/output/hdf5_collective.cpp" +BINDING = ROOT / "python/bindings/core/init/init_parallel_hdf5.cpp" +WRITER = ROOT / "python/pops/output/_writers/hdf5.py" + + +def test_native_hdf5_surface_has_no_process_world_overload_or_probe(): + header = HEADER.read_text(encoding="utf-8") + source = SOURCE.read_text(encoding="utf-8") + + assert "WorldCommunicator" not in header + assert "WorldCommunicator" not in source + assert "world_communicator.hpp" not in source + assert "MPI_COMM_WORLD" not in source + assert "MPI_Initialized" not in source + assert "const CommunicatorView& communicator" in header + + +def test_python_hdf5_route_requires_a_duplicated_observer_lane(): + binding = BINDING.read_text(encoding="utf-8") + writer = WRITER.read_text(encoding="utf-8") + + assert "WorldCommunicator" not in binding + assert "world_communicator.hpp" not in binding + assert "py::isinstance" in binding + assert "requires an exact duplicated observer MPI lane" in binding + assert "require_communicator(communicator, allow_world=False)" in writer diff --git a/tests/python/integration/io/test_hdf5_parallel.py b/tests/python/integration/io/test_hdf5_parallel.py index a5b7c1858..6643a7367 100644 --- a/tests/python/integration/io/test_hdf5_parallel.py +++ b/tests/python/integration/io/test_hdf5_parallel.py @@ -119,15 +119,15 @@ def snapshot(pieces): return snapshot((local_piece,)), snapshot(serial_pieces), key, global_values -def _parallel_hdf5_world(test_name: str): +def _parallel_hdf5_lane(test_name: str): try: import h5py # noqa: F401 -- serial native reopen verification except ImportError: _missing_mpi_requirement("collective HDF5 requires h5py") if getattr(_pops, "__has_parallel_hdf5__", False) is not True: _missing_mpi_requirement("collective HDF5 requires the compiled C++ parallel-HDF5 route") - communicator = _pops.mpi_world() - if world_size(communicator) == 1 and os.environ.get(_MPI_CHILD) != "1": + world = _pops.mpi_world() + if world_size(world) == 1 and os.environ.get(_MPI_CHILD) != "1": mpiexec = shutil.which("mpiexec") or shutil.which("mpirun") if mpiexec is None: _missing_mpi_requirement( @@ -154,12 +154,24 @@ def _parallel_hdf5_world(test_name: str): ) assert result.returncode == 0, result.stdout + result.stderr return None - assert world_size(communicator) >= 2, "MPI child did not start with two ranks" - return communicator + assert world_size(world) >= 2, "MPI child did not start with two ranks" + return world.duplicate_observer_lane("pytest-hdf5-" + test_name) -def test_collective_hdf5_roundtrip_matches_serial(tmp_path): - communicator = _parallel_hdf5_world(test_collective_hdf5_roundtrip_matches_serial.__name__) +@pytest.fixture +def parallel_hdf5_lane(request): + lane = _parallel_hdf5_lane(request.node.name) + if lane is None: + yield None + return + try: + yield lane + finally: + lane.close_collectively() + + +def test_collective_hdf5_roundtrip_matches_serial(tmp_path, parallel_hdf5_lane): + communicator = parallel_hdf5_lane if communicator is None: return rank = world_rank(communicator) @@ -237,10 +249,10 @@ def test_collective_hdf5_roundtrip_matches_serial(tmp_path): assert failure is None, failure -def test_collective_hdf5_refuses_rank_local_metadata_before_write(tmp_path): - communicator = _parallel_hdf5_world( - test_collective_hdf5_refuses_rank_local_metadata_before_write.__name__ - ) +def test_collective_hdf5_refuses_rank_local_metadata_before_write( + tmp_path, parallel_hdf5_lane, +): + communicator = parallel_hdf5_lane if communicator is None: return rank = world_rank(communicator) @@ -265,10 +277,10 @@ def test_collective_hdf5_refuses_rank_local_metadata_before_write(tmp_path): assert not tuple(shared_root.glob(".*must-not-exist*.tmp")) -def test_collective_hdf5_refuses_divergent_target_before_write(tmp_path): - communicator = _parallel_hdf5_world( - test_collective_hdf5_refuses_divergent_target_before_write.__name__ - ) +def test_collective_hdf5_refuses_divergent_target_before_write( + tmp_path, parallel_hdf5_lane, +): + communicator = parallel_hdf5_lane if communicator is None: return rank = world_rank(communicator) @@ -290,10 +302,10 @@ def test_collective_hdf5_refuses_divergent_target_before_write(tmp_path): assert not tuple(shared_root.glob("*must-not-exist.h5")) -def test_native_collective_hdf5_binding_failure_is_all_rank_consensus(tmp_path): - communicator = _parallel_hdf5_world( - test_native_collective_hdf5_binding_failure_is_all_rank_consensus.__name__ - ) +def test_native_collective_hdf5_binding_failure_is_all_rank_consensus( + tmp_path, parallel_hdf5_lane, +): + communicator = parallel_hdf5_lane if communicator is None: return rank = world_rank(communicator) diff --git a/tests/python/integration/mpi/test_scientific_output_mpi.py b/tests/python/integration/mpi/test_scientific_output_mpi.py index e01111287..189ce3610 100644 --- a/tests/python/integration/mpi/test_scientific_output_mpi.py +++ b/tests/python/integration/mpi/test_scientific_output_mpi.py @@ -128,6 +128,8 @@ def _shared_directory() -> Path: def _validate_native_binding_error_consensus(root: Path) -> None: """One malformed rank must fail before HDF5 while every peer receives the same cause.""" + lane = COMM.duplicate_observer_lane( + "scientific-output-mpi-hdf5-binding-validation") values = ( [[1.0, 2.0], [3.0, 4.0]] if RANK == 0 @@ -135,21 +137,24 @@ def _validate_native_binding_error_consensus(root: Path) -> None: ) error = None try: - _pops._write_parallel_hdf5( - COMM, - str(root / "binding-must-not-enter-hdf5.h5"), - "{}", - {"geometry/0000/coverage": np.zeros((2, 2), dtype=np.bool_)}, - ({ - "dataset": "fields/0000/values", - "dtype": np.dtype(np.float64).str, - "shape": (2, 2), - "pieces": ({"lower": (0, 0), "upper": (2, 2), "values": values},), - },), - ) - except RuntimeError as exc: - error = str(exc) - errors = allgather_value(COMM, error) + try: + _pops._write_parallel_hdf5( + lane, + str(root / "binding-must-not-enter-hdf5.h5"), + "{}", + {"geometry/0000/coverage": np.zeros((2, 2), dtype=np.bool_)}, + ({ + "dataset": "fields/0000/values", + "dtype": np.dtype(np.float64).str, + "shape": (2, 2), + "pieces": ({"lower": (0, 0), "upper": (2, 2), "values": values},), + },), + ) + except RuntimeError as exc: + error = str(exc) + errors = allgather_value(lane, error) + finally: + lane.close_collectively() if not all(item is not None and "binding input validation" in item for item in errors): raise AssertionError("rank-local binding fault did not reach all ranks: %r" % (errors,)) if len(set(errors)) != 1: From 836f2f2812e72a9bbb95de8c13c86d9a1bb02a44 Mon Sep 17 00:00:00 2001 From: desp0042 Date: Thu, 30 Jul 2026 07:17:31 +0200 Subject: [PATCH 3/3] docs(output): describe duplicated HDF5 observer lanes (ADC-683) --- .../SPECIFICATION_TECHNIQUE_FINALE_POPS_ARCHITECTURE.md | 5 +++-- docs/design/exact-output-consumers.md | 9 +++++---- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/docs/design/SPECIFICATION_TECHNIQUE_FINALE_POPS_ARCHITECTURE.md b/docs/design/SPECIFICATION_TECHNIQUE_FINALE_POPS_ARCHITECTURE.md index e8bb1baf1..af5e0e05d 100644 --- a/docs/design/SPECIFICATION_TECHNIQUE_FINALE_POPS_ARCHITECTURE.md +++ b/docs/design/SPECIFICATION_TECHNIQUE_FINALE_POPS_ARCHITECTURE.md @@ -1491,8 +1491,9 @@ scientifiques choisissent obligatoirement un `ParallelMode` typé : d'un unique writer rang 0, `COLLECTIVE` pour les hyperslabs HDF5 MPIO exacts, ou `PER_RANK` pour des artefacts locaux qualifiés par rang et un reçu agrégé. Le mode, le format, la sélection, la cible et l'identité de chaque pièce native (`global_box_index`, `owner_rank`, `replicated`) sont authentifiés -entre rangs avant toute écriture. La route `COLLECTIVE` appelle le backend C++ HDF5 parallèle sur -`MPI_COMM_WORLD`; `h5py` reste uniquement un lecteur/écrivain série optionnel et n'est jamais un +entre rangs avant toute écriture. La route `COLLECTIVE` appelle le backend C++ HDF5 parallèle avec +la lane MPI dupliquée possédée par la session observateur ; le writer ne redécouvre ni n'emprunte +`MPI_COMM_WORLD`. `h5py` reste uniquement un lecteur/écrivain série optionnel et n'est jamais un transport MPI. Une dépendance HDF5 parallèle native absente, un mode incompatible ou un backend Kokkos GPU/device handle non supporté est refusé avant le constructeur de `System`/`AmrSystem`; aucune route série implicite ne remplace une demande MPI. diff --git a/docs/design/exact-output-consumers.md b/docs/design/exact-output-consumers.md index 8b81f518b..e2ab048b5 100644 --- a/docs/design/exact-output-consumers.md +++ b/docs/design/exact-output-consumers.md @@ -85,10 +85,11 @@ count, target suffix, or writer availability: gather, but only rank 0 prepares, verifies and atomically publishes the single-file writer. Preparation failures and the final receipt are broadcast to every participant. - `COLLECTIVE` requires a distributed context, an authenticated collective resource plan and the - native C++ parallel-HDF5 provider. Each rank writes only its exact non-overlapping native - hyperslabs with exactly one MPIO collective transfer per dataset and rank (including a select-none - transfer for a rank with no patch). A replicated AMR coarse patch is assigned to rank 0 for this - mode so it cannot overlap. + native C++ parallel-HDF5 provider. The observer runtime owns a duplicated MPI lane for the complete + writer session; neither the Python writer nor the native HDF5 adapter borrows or rediscovers the + process world. Each rank writes only its exact non-overlapping native hyperslabs with exactly one + MPIO collective transfer per dataset and rank (including a select-none transfer for a rank with no + patch). A replicated AMR coarse patch is assigned to rank 0 for this mode so it cannot overlap. - `PER_RANK` requires a distributed context and preserves each rank's exact local pieces, including explicitly replicated coarse pieces. Targets are rank-qualified before any file is opened. The transaction succeeds only after it aggregates one deterministic receipt per contiguous rank.