Streaming Support

[franzpoeschel]

Add support for streaming IO to the openPMD API, including frontend semantics and a backend implementation for the ADIOS2 IO backend.

The basic idea is to split the former concept of flushing a Series into two separate concepts:

• flush: frontend guarantees, application-side buffers may be used after this call returns
• advance: backend guarantees, data has moved to/from the backend storage after this call returns. In a streaming context, this means that data packets are sent/received.

This concept is now purely internal, with a much simpler API exposed to users. The API is documented in streaming.rst. It is more restrictive than the traditional random-access based API and hence more generic, being usable for file-based as well as for streaming-based workflows.

Review entry points:

• Streaming user-api: Series.hpp classes ReadIterations and WriteIterations
• new IO tasks
• Iteration::beginStep()/endStep()/close()

Major additions include:

• Additional functionality to deal with partial datasets in Dataset.hpp and RecordComponent.hpp Since the high-level API now correlates streaming steps with iterations, single datasets are expected to be fully available or not available at all. Finer-grained accessing methods have hence been removed again.
• Implementation of streaming mode for the ADIOS2 backend. This does currently not include a Read-Write mode as provided by ADIOS2.
• Four new IO tasks: Advance, inquiring available chunks, closing files (see below for a reason why this is not a separate PR) and closing paths. The latter command indicates that this group will not be written to any longer, allowing for optimizations in the backend.
  Update: Closing files is already merged separately, inquiring available chunks extracted to PR Add ::availableChunks call to Record Component types #802.~~

• The Series::advance method as described above. Just like flush, this returns a std::future, but for the first time in the openPMD API, those futures are actually used. If discarding them, the destructor of std::future will immediately run, blocking until the future returns. If storing them, the result can be awaited at a later time, continuing with computation asynchronously. This currently makes sense mostly on the reader end, where the future will block until a next data packet has been received.
  CAUTION: As described in Design limitations in the frontend resulting in unexpected behaviours (segfaults, premature destructor calls) #534, the Series object should currently not be copied due to unclear semantics, the the returned future wraps a reference to the Series. Deleting the Series before the returned future will result in use after free.
  Since the class std::future in the standard library is rather limited at the moment, this includes an implementation of an extended future class that allows to chain futures.
  TODO: "Advance" tasks are currently only passed through to the backend if choosing group-based layouts. In file-based layouts, this currently falls back to flushing the Series.

Update: Now internal, not part of the public API.

• Additionally to the random-access based chunk loading procedures and their aforementioned extensions, functionality to automatically distribute incoming chunks in Streaming.hpp. This obviously requires that reading applications can deal with having chunks arranged by the IO library. The strategies are flexibly interchangeable. They receive as input meta information about the MPI ranks of writers and readers, e.g. the hostnames. MPI.hpp contains functionality to collect this meta information, the writer can then set its meta information as a top-level attribute in Series.hpp.

Update: Not part of this PR, will either come as a later PR or as part of external tooling.

• A hopefully easy-to-use class to dump times to a file. Can be deactivated at compile time via a template parameter.

Update: Removed.

• Adaptations for re-parsing a Series after receiving a new packet.
• Update: A high-level iterator-based Streaming API for reading (ReadIterations class) and a high-level container-based Streaming-API for writing (WriteIterations class)
• Update: Python bindings

TODO:

• Document all of this
• Add new examples for streaming
• I mostly did ad-hoc testing so far, integrate those tests with the real ones.
• Formatting and cleanup
• Figure out why Windows tests hang – MSVC used the wrong method overload in one place, leading to an infinite recursion. Explicit typecasting solved the issue.

(I will be on holidays in the second half of october and likely without Internet during that time, just leave comments, but I might take a while to react :) )

[ax3l]

I updated the CI with further ADIOS2 tests in #568.
Please rebase against the current dev :)

[ax3l]

Hi @franzpoeschel, happy new year! 🌟
Are you continuing on this PR this week? :)

[franzpoeschel]

Hi, @ax3l, I will continue on thursday/friday, yes.
Happy new year to you, too!

[franzpoeschel]

I have temporarily moved this to topic-streaming-bak to perform cleanup work on this branch.

[ax3l]

Can we implement that a Series::advance() for regular file operations closes the opened file(s), e.g. via H5Fclose()? Currently files are kept open "too long" during simulations, which especially for HDF5 files increases the time frame on which corruption of the written data can occur. (And it's problematic for semi-live post-processing, too.)
Related to: BLAST-WarpX/warpx#645

[franzpoeschel]

I am currently planning to move Series::advance() to Iteration::advance() since that makes more sense for file-based layout (and maybe keep Series::advance() as a short-cut operation that advances all iterations). This being said, using Series::advance() to close a file would give different semantics to this call depending on the backend, making it potentially impossible to quickly switch backends. I have used Iteration::finalize() to pass a CLOSE_FILE command through to the backends and implemented that in the ADIOS2 backend already.
Hence, in order to make use of HF5Close(), you should be fine with implementing AbstractIOHandlerImpl::closeFile() which is triggered by Iteration::finalize() independently of the backend.

[ax3l]

Iteration::finalize() sounds legit, too.

[franzpoeschel]

I've implemented the streaming API in openPMD-ls. Given that draining a stream is definitely a a non-const thing to do, this implied removing some consts.

[ax3l]

This is fantastic work, thank you for your dedication and the high quality. Congratulations and happy holidays! 🎄

[ax3l]

Suggested change

	#endif
	#endif