Skip to content

Immediate flush mode - #1846

Open
franzpoeschel wants to merge 86 commits into
openPMD:devfrom
franzpoeschel:synchronous-flush
Open

franzpoeschel wants to merge 86 commits into
openPMD:devfrom
franzpoeschel:synchronous-flush

Conversation

@franzpoeschel

@franzpoeschel franzpoeschel commented Feb 2, 2026 •

Copy link
Copy Markdown
Contributor

Support immediate (synchronous) flushing via flush_immediately

Base branch: adios2-memory-selection · Feature branch: synchronous-flush. Use this for the Diff: Diff until then: https://github.com/franzpoeschel/openPMD-api/compare/adios2-memory-selection...franzpoeschel:openPMD-api:synchronous-flush?expand=1

Motivation

The openPMD-api is a deferred data API: load/store operations are enqueued and only executed at explicit flush points. This is efficient but easy to get wrong (missed flushes, ordering surprises). This PR adds an opt-in immediate flush mode where every load/store call acts as an implicit flush point.

What changes

  • New option flush_immediately (JSON/TOML key or OPENPMD_FLUSH_IMMEDIATELY=1). Default: false in C++, true in Python (merged in via json::merge()).
  • New FlushLevel::ImmediateFlush with updated flush_level::* helpers, so an immediate flush can write datasets/attributes/hierarchy like a UserFlush without being a user-visible global flush point; determineUnsetDirty() treats it as a no-op.
  • Immediate path in push_chunk(): performs a frontend hierarchy flush at ImmediateFlush, enqueues the chunk directly, and flushes the backend queue
    all within the store/load call. RecordComponent::flush() fixed so the read path only drains m_chunks at a true global flush point.
  • LS_API { legacy, chaining }: only the legacy API (and PatchRecordComponent) respects immediate flushing; the chaining API manages its own deferred flush and stays deferred; storeChunkSpan forces flush_immediately=false to keep span buffers valid.
  • IOHandler refactor: shared state extracted into internal::GlobalParameters (from which AbstractIOHandler derives), plus internal::AbstractIOHandlerInitFrom replacing the (path, access, initialize_from) packs in all handler constructors / createIOHandler(). Deferred init now carries options into the real handler via GlobalParameters. Adds Series::flushImmediately().
  • Tests & CI: full test suite now runs in both deferred and immediate modes on Linux/macOS; tests updated to branch on the active mode (chunk-queue sizes, buffer use-counts, adios2_bp5_flush size, immediate throw on closed-iteration, skips for tests needing non-collective deferred store).
  • Docs: backendconfig.rst and workflow.rst document the option and that immediate flushing introduces implicit flush points.

Compatibility

Deferred default for C++ is fully preserved; purely additive.

Testing

  • Full build clean; Core/Auxiliary/JSON/Serial and Python unit suites pass (Python exercises the immediate default).
  • Full serial CTest passes with both OPENPMD_FLUSH_IMMEDIATELY=0 and =1 (36/36 each).
  • Parallel (MPI) suite passes in both modes, except the pre-existing adios2_streaming (SST) abort that also reproduces on the base branch (environment-related, not introduced here).

TODO:

@franzpoeschel franzpoeschel added the api: new additions to the API label Feb 2, 2026
Comment thread src/Mesh.cpp Fixed
Comment thread src/Series.cpp
auto init_directly = [this, &comm..., &filepath](
std::unique_ptr<ParsedInput> parsed_input,
json::TracingJSON tracing_json) {
json::TracingJSON tracing_json,

Check notice

Code scanning / CodeQL

Large object passed by value

This parameter of type [TracingJSON](1) is 144 bytes - consider passing a const pointer/reference instead.
Comment thread src/backend/ScientificDefaults.cpp Fixed
Comment thread test/CoreTest.cpp Fixed
Comment thread test/ParallelIOTest.cpp
else
{
auto read_again = E_x_read.loadChunk<int>({0, 0}, {mpi_size, 4});
// REQUIRE_THROWS(read.flush());

Check notice

Code scanning / CodeQL

Commented-out code

This comment appears to contain commented-out code.
std::move(ls_cfg),
/*flush_immediately=*/false);
}
// storeChunk(std::move(data), std::move(o), std::move(e));

Check notice

Code scanning / CodeQL

Commented-out code

This comment appears to contain commented-out code.
Comment thread src/Iteration.cpp Fixed
Comment thread src/RecordComponent.cpp
template <typename T>
void RecordComponent::loadChunk(std::shared_ptr<T> data, Offset o, Extent e)
void RecordComponent::loadChunk_impl(
std::shared_ptr<T> const &data, internal::LoadStoreConfigWithBuffer cfg)

Check notice

Code scanning / CodeQL

Large object passed by value

This parameter of type [LoadStoreConfigWithBuffer](1) is 104 bytes - consider passing a const pointer/reference instead.
Comment thread src/RecordComponent.cpp
std::static_pointer_cast<T>(std::move(ptr)),
std::move(offset),
std::move(extent));
// static_assert(!std::is_same_v<T_with_extent, std::string>, "EVIL");

Check notice

Code scanning / CodeQL

Commented-out code

This comment appears to contain commented-out code.
Comment thread test/AuxiliaryTest.cpp
#endif
}

TEST_CASE("future_test", "[auxiliary]")

Check notice

Code scanning / CodeQL

Unused static function

Static function CATCH2_INTERNAL_TEST_18 is unreachable ([autoRegistrar19](1) must be removed at the same time)
inline DynamicMemoryView<T> RecordComponent::storeChunkSpanCreateBuffer_impl(
internal::LoadStoreConfig cfg, F &&createBuffer)
{
[[maybe_unused]] auto [o, e, api] = std::move(cfg);

Check notice

Code scanning / CodeQL

Unused local variable

Variable api is not used.
Comment thread test/CoreTest.cpp
#endif
}

TEST_CASE("unsafe_no_automatic_flush_immediate_flush_test", "[core]")

Check notice

Code scanning / CodeQL

Unused static function

Static function CATCH2_INTERNAL_TEST_38 is unreachable ([autoRegistrar39](1) must be removed at the same time)
@franzpoeschel
franzpoeschel force-pushed the synchronous-flush branch 3 times, most recently from 2bca70c to f29b628 Compare September 17, 2026 09:31
franzpoeschel and others added 13 commits September 29, 2026 11:51
The "clean up joined dim logic" change centralized the joined-dimension
offset rule in computeOffset() (empty offset for joined dims, throw on a
non-empty one) but only updated the container/allocating overloads. The
buffer-based overloads were left inconsistent:

- loadChunk(shared_ptr) still forwarded a {0} offset (stale `dim <= 1u`
  condition), which computeOffset() then rejected for a joined dimension.
- loadChunk_impl() lacked the joined-dimension branch that verifyChunk()
  (the store path) has, so it rejected the empty offset that joined
  dimensions require with a dimensionality error. A joined dimension
  therefore could not be loaded through the buffer-based load overloads
  at all.
- The buffer-based storeChunk overloads and the span-based allocating
  storeChunk forwarded a {0} offset / {-1} extent unconditionally, so a
  joined-dimension store with the default {0} offset threw.

Make all buffer-based and span-based loadChunk/storeChunk overloads
detect the default {0} offset and {-1} extent (leaving joined-dimension
handling to computeOffset/computeExtent), and add the joined-dimension
branch to loadChunk_impl() mirroring verifyChunk().

Add a regression test (joined_dim_buffer_api) validating the frontend
offset/extent handling for a joined dimension through the buffer-based
overloads on the JSON backend; the argument handling is backend agnostic
(a joined dimension is only readable back on ADIOS2).
The handles returned by RecordComponent::prepareLoadStore().store()/load()
perform the data operation and, by default, an automatic Series flush. Since
Series::flush() is MPI-collective, document this behavior in the MPI overview,
the deferred data API contract and the API headers, including the fact that
per-component flush counters may skip the flush, and that handles are
[[nodiscard]] so the flush cannot happen unnoticed.
Split the joined-dimension buffer-API regression test out of
SerialIOTest.cpp into test/Files_SerialIO/joined_dim_buffer_api.cpp and
register it in CMakeLists.txt, matching the convention of the other
per-issue SerialIO test files.
A joined dimension's extent is only known once all writers have flushed and
the data is read back (close + reopen). During the write session the extent
is Dataset::JOINED_DIMENSION (max), so loading a chunk of a joined array is
not a meaningful operation: the total size is unknown and the streaming
engine cannot read the data back mid-write. After a reopen the extent is
concrete, joinedDimension() is nullopt, and the load goes through the normal
non-joined path (the case exercised by the joined_dim test).

The previous change made loadChunk_impl mirror the store path by accepting an
empty offset for joined arrays, which only validated the in-session case and
then queued a READ_DATASET for 2^64 elements. Reject joined arrays in
loadChunk_impl with a clear error instead; the store path (append) is
unchanged.

Update the joined_dim_buffer_api regression test: store with an empty {} and
a {0} offset still passes the frontend checks, while every buffer-based load
overload now throws error::WrongAPIUsage.
Rejecting a memory selection inside the backend's writeDataset() throws
while an IO task is being flushed. AbstractIOHandlerImpl::flush() reacts to
an exception in an IO task by clearing the whole IO queue and rethrowing
("Clearing IO queue and passing on the exception"), so every other pending
chunk and the file's root attributes are dropped. The subsequent flush()
and close() then report success, but the file can no longer be read back
(AttributeNotFound: openPMD) in HDF5 and JSON.

Whether a backend supports non-contiguous memory selections is known when
the chunk is enqueued, so check it in RecordComponent::storeChunk_impl()
and throw error::OperationUnsupportedInBackend there, before anything is
queued. Add AbstractIOHandler::supportsMemorySelection(), defaulting to
false and overridden to true only in ADIOS2IOHandler. The backend-side
throws stay as a safety net.

Add a regression test that stores a valid chunk, attempts a rejected
memory selection on another component, and verifies that the Series stays
usable and that the valid chunk and a root attribute survive the roundtrip
in HDF5 and JSON. Without the early check the test fails at flush time.
A memory selection could not be reset before ADIOS2 v2.10.1 (the reset
capability was added upstream in v2.11.0 and backported to 2.10.1). On
older versions a stale memory selection silently leaked into every
following store operation of the same variable, producing wrong data or
out-of-bounds reads. This affected later storeChunk() calls, later steps
and the unique_ptr path.

Instead of applying a workaround, report ADIOS2 as not supporting memory
selections in that case via AbstractIOHandler::supportsMemorySelection(),
so the frontend rejects them up front, and keep the backend-side check in
verifyDataset() as a safety net. Both reuse the existing
CanTheMemorySelectionBeReset trait. Remove the now-obsolete one-time
warning and its bookkeeping.

Adapt available_chunks_test and memory_selection_rejected_before_flush to
use an equivalent contiguous buffer / expectation where memory selections
are unsupported, and add a dedicated memory_selection_old_adios2 test.
Pin the ADIOS2 v2.10 CI job to 2.10.0 to exercise the rejection path.
franzpoeschel and others added 15 commits September 30, 2026 18:24
…tics

The chaining API's push_chunk() ignores OPENPMD_FLUSH_IMMEDIATELY for
safety. When unsafeNoAutomaticFlush() is used, that guarantee no longer
applies, so run legacy flushing semantics instead:
unsafeNoAutomaticFlush_impl() now sets the internal api to
LS_API::legacy, making push_chunk() honor the immediate-flush setting
again.

Adds a Core test (unsafe_no_automatic_flush_immediate_flush_test)
verifying that chunks are flushed immediately when
m_flush_immediately is set and buffered otherwise.
Explain how the synchronous flush option (flush_immediately/
OPENPMD_FLUSH_IMMEDIATELY) interacts with the two load/store APIs:

- Legacy API (storeChunk()/loadChunk() and friends) only enqueues
  operations performed at flush points, so the sync-flush option makes
  each such call its own flush point.
- Chaining API (prepareLoadStore()) flushes automatically upon
  evaluation of its DeferredComputation object and is unaffected by the
  sync-flush option, unless unsafeNoAutomaticFlush() is used, which
  falls back to legacy flushing semantics.
- The span-based API always defers; Python defaults to immediate
  flushing.

Documented in Doxygen/comments in the source code, in the .rst docs
(workflow.rst and backendconfig.rst) and in the Python bindings
docstring.

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

api: new additions to the API

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants