diff --git a/.github/ci/sanitizer/clang/Leak.supp b/.github/ci/sanitizer/clang/Leak.supp index e33b7195d7..8691a8dc43 100644 --- a/.github/ci/sanitizer/clang/Leak.supp +++ b/.github/ci/sanitizer/clang/Leak.supp @@ -11,3 +11,6 @@ leak:/usr/bin/make* # ADIOS 1.13.1 leak:adios_read_bp_open_file leak:adios_inq_var +# ADIOS2 +leak:adios2::core::engine::SstReader::* +leak:adios2::core::engine::SstWriter::* diff --git a/CMakeLists.txt b/CMakeLists.txt index 2468a5a5d7..d12ba9fbd8 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -633,6 +633,8 @@ set(openPMD_EXAMPLE_NAMES 8_benchmark_parallel 8a_benchmark_write_parallel 8b_benchmark_read_parallel + 10_streaming_write + 10_streaming_read ) set(openPMD_PYTHON_EXAMPLE_NAMES 2_read_serial diff --git a/docs/source/backends/adios2.rst b/docs/source/backends/adios2.rst index 35c65332f9..ec4699a639 100644 --- a/docs/source/backends/adios2.rst +++ b/docs/source/backends/adios2.rst @@ -13,13 +13,31 @@ For further information, check out the :ref:`installation guide `, I/O Method ---------- -ADIOS2 has several engines for alternative file formats and other kinds of backends, yet natively writes to ``.bp`` files. At the moment, the openPMD API exclusively uses the BPFile engine. +ADIOS2 has several engines for alternative file formats and other kinds of backends, yet natively writes to ``.bp`` files. +The openPMD API uses the BP4 engine as the default file engine and the SST engine for streaming support. We currently leverage the default ADIOS2 transport parameters, i.e. ``POSIX`` on Unix systems and ``FStream`` on Windows. +Steps +----- + +ADIOS2 is optimized towards organizing the process of reading/writing data into IO steps. +In order to activate steps, it is imperative to use the :ref:`Streaming API ` (which can be used for either file-based or streaming-based workflows). +With ADIOS2 release 2.6.0 containing a bug (fixed in development versions, see `PR #2348 `_) that disallows random-accessing steps in file-based engines, step-based processing must currently be opted in to via use of the :ref:`JSON parameter` ``adios2.engine.usesteps = true`` when using a file-based engine such as BP3 or BP4. +With these ADIOS2 releases, files written in such a way may only be read using the streaming API. +Upon reading a file, the ADIOS2 backend will automatically recognize whether it has been written with or without steps, ignoring the JSON option mentioned above. +Steps are mandatory for streaming-based engines and trying to switch them off will result in a runtime error. + +.. note:: + + ADIOS2 will in general dump data to disk/transport only upon closing a file/engine or a step. + If not using steps, users are hence strongly encouraged to use file-based iteration layout (by creating a Series with a filename pattern such as ``simData_%06T.bp``) and enforce dumping to disk by ``Iteration::close()``-ing an iteration after writing to it. + Otherwise, out-of-memory errors are likely to occur. Backend-Specific Controls ------------------------- +The ADIOS2 SST engine for streaming can be picked by specifying the ending ``.sst`` instead of ``.bp``. + The following environment variables control ADIOS2 I/O behavior at runtime. Fine-tuning these is especially useful when running at large scale. diff --git a/docs/source/details/backendconfig.rst b/docs/source/details/backendconfig.rst index 0e089901a2..1bd099f1a8 100644 --- a/docs/source/details/backendconfig.rst +++ b/docs/source/details/backendconfig.rst @@ -46,8 +46,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.type``: An associative array of string-formatted engine parameters, passed directly through to ``adios2::IO::SetParameters``. +* ``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 allowable engine parameters. +* ``adios2.engine.usesteps``: Described more closely in the documentation for the :ref:`ADIOS2 backend`. * ``adios2.dataset.operators``: This key contains a list of ADIOS2 `operators `_, used to enable compression or dataset transformations. Each object in the list has two keys: diff --git a/docs/source/index.rst b/docs/source/index.rst index 70748ba053..c2bb0b1d07 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -73,6 +73,7 @@ Usage usage/firstread usage/serial usage/parallel + usage/streaming usage/benchmarks usage/examples diff --git a/docs/source/usage/10_streaming_read.cpp b/docs/source/usage/10_streaming_read.cpp new file mode 120000 index 0000000000..9192aef4e9 --- /dev/null +++ b/docs/source/usage/10_streaming_read.cpp @@ -0,0 +1 @@ +../../../examples/10_streaming_read.cpp \ No newline at end of file diff --git a/docs/source/usage/10_streaming_read.py b/docs/source/usage/10_streaming_read.py new file mode 120000 index 0000000000..c6201e54b1 --- /dev/null +++ b/docs/source/usage/10_streaming_read.py @@ -0,0 +1 @@ +../../../examples/10_streaming_read.py \ No newline at end of file diff --git a/docs/source/usage/10_streaming_write.cpp b/docs/source/usage/10_streaming_write.cpp new file mode 120000 index 0000000000..c819fb8e6e --- /dev/null +++ b/docs/source/usage/10_streaming_write.cpp @@ -0,0 +1 @@ +../../../examples/10_streaming_write.cpp \ No newline at end of file diff --git a/docs/source/usage/10_streaming_write.py b/docs/source/usage/10_streaming_write.py new file mode 120000 index 0000000000..99e766a6f9 --- /dev/null +++ b/docs/source/usage/10_streaming_write.py @@ -0,0 +1 @@ +../../../examples/10_streaming_write.py \ No newline at end of file diff --git a/docs/source/usage/streaming.rst b/docs/source/usage/streaming.rst new file mode 100644 index 0000000000..4be444a8e4 --- /dev/null +++ b/docs/source/usage/streaming.rst @@ -0,0 +1,93 @@ +.. _usage-streaming: + +Streaming +========= + +.. note:: + Data streaming is a novel backend and under active development. + At the moment, the internal data format is still changing rapidly and is likely not compatible between releases of the openPMD-api. + +The openPMD API includes a streaming-aware API as well as streaming-enabled backends (currently: ADIOS2). + +Unlike in file-based backends, the order in which data is put out becomes relevant in streaming-based backends. +Each iteration will be published as one atomical step by the streaming API (compare the concept of `steps in ADIOS2 `_). + +Reading +------- + +The reading end of the streaming API enforces further restrictions that become necessary through the nature of streaming. +It can be used to read any kind of openPMD-compatible dataset, stream-based and filesystem-based alike. + +C++ +^^^ + +The reading end of the streaming API is activated through use of ``Series::readIterations()`` instead of accessing the field ``Series::iterations`` directly. +The returned object of type ``ReadIterations`` can be used in a C++11 range-based for loop to iterate over objects of type ``IndexedIteration``. +This class extends the ``Iteration`` class with a field ``IndexedIteration::iterationIndex``, denoting this iteration's index. + +Users are encouraged to explicitly ``.close()`` the iteration after reading from it. +Closing the iteration will flush all pending operations on that iteration. +If an iteration is not closed until the beginning of the next iteration, it will be closed automatically. + +Note that a closed iteration cannot be reopened. +This pays tribute to the fact that in streaming mode, an iteration may be dropped by the data source once the data sink has finished reading from it. + +.. literalinclude:: 10_streaming_read.cpp + :language: cpp + +Python +^^^^^^ + +The reading end of the streaming API is activated through use of ``Series.read_iterations()`` instead of accessing the field ``Series.iterations`` directly. +The returned object of type ``ReadIterations`` can be used in a Python range-based for loop to iterate over objects of type ``IndexedIteration``. +This class extends the ``Iteration`` class with a field ``IndexedIteration.iteration_index``, denoting this iteration's index. + +Users are encouraged to explicitly ``.close()`` the iteration after reading from it. +Closing the iteration will flush all pending operations on that iteration. +If an iteration is not closed until the beginning of the next iteration, it will be closed automatically. + +Note that a closed iteration cannot be reopened. +This pays tribute to the fact that in streaming mode, an iteration may be dropped by the data source once the data sink has finished reading from it. + +.. literalinclude:: 10_streaming_read.py + :language: python3 + +Writing +------- + +The writing end of the streaming API enforces further restrictions that become necessary through the nature of streaming. +It can be used to write any kind of openPMD-compatible dataset, stream-based and filesystem-based alike. + +C++ +^^^ + +The writing end of the streaming API is activated through use of ``Series::writeIterations()`` instead of accessing the field ``Series::iterations`` directly. +The returned object of type ``WriteIterations`` wraps the field ``Series::iterations``, but exposes only a restricted subset of functionality. +Using ``WriteIterations::operator[]( uint64_t )`` will automatically open a streaming step for the corresponding iteration. + +Users are encouraged to explicitly ``.close()`` the iteration after writing to it. +Closing the iteration will flush all pending operations on that iteration. +If an iteration is not closed until the next iteration is accessed via ``WriteIterations::operator[]( uint64_t )``, it will be closed automatically. + +Note that a closed iteration cannot be reopened. +This pays tribute to the fact that in streaming mode, an iteration is sent to the sink upon closing it and the data source can no longer modify it. + +.. literalinclude:: 10_streaming_write.cpp + :language: cpp + +Python +^^^^^^ + +The writing end of the streaming API is activated through use of ``Series.write_iterations()`` instead of accessing the field ``Series.iterations`` directly. +The returned object of type ``WriteIterations`` wraps the field ``Series.iterations``, but exposes only a restricted subset of functionality. +Using ``WriteIterations.__getitem__(index)`` (i.e. the index operator ``series.writeIterations()[index]``) will automatically open a streaming step for the corresponding iteration. + +Users are encouraged to explicitly ``.close()`` the iteration after writing to it. +Closing the iteration will flush all pending operations on that iteration. +If an iteration is not closed until the next iteration is accessed via ``WriteIterations.__getitem__(index)``, it will be closed automatically. + +Note that a closed iteration cannot be reopened. +This pays tribute to the fact that in streaming mode, an iteration is sent to the sink upon closing it and the data source can no longer modify it. + +.. literalinclude:: 10_streaming_write.py + :language: python3 \ No newline at end of file diff --git a/examples/10_streaming_read.cpp b/examples/10_streaming_read.cpp new file mode 100644 index 0000000000..aa3479f039 --- /dev/null +++ b/examples/10_streaming_read.cpp @@ -0,0 +1,63 @@ +#define BUILD_STREAMING_EXAMPLE false +#if BUILD_STREAMING_EXAMPLE +#include + +#include +#include +#include + +using std::cout; +using namespace openPMD; + +int +main() +{ +#if openPMD_HAVE_ADIOS2 + using position_t = double; + Series series = Series( "electrons.sst", Access::READ_ONLY ); + + for( IndexedIteration iteration : series.readIterations() ) + { + std::cout << "Current iteration: " << iteration.iterationIndex + << std::endl; + Record electronPositions = iteration.particles[ "e" ][ "position" ]; + std::array< std::shared_ptr< position_t >, 3 > loadedChunks; + std::array< Extent, 3 > extents; + std::array< std::string, 3 > const dimensions{ { "x", "y", "z" } }; + + for( size_t i = 0; i < 3; ++i ) + { + std::string dim = dimensions[ i ]; + RecordComponent rc = electronPositions[ dim ]; + loadedChunks[ i ] = rc.loadChunk< position_t >( + Offset( rc.getDimensionality(), 0 ), rc.getExtent() ); + extents[ i ] = rc.getExtent(); + } + + iteration.close(); + + for( size_t i = 0; i < 3; ++i ) + { + std::string dim = dimensions[ i ]; + Extent const & extent = extents[ i ]; + std::cout << "\ndim: " << dim << "\n" << std::endl; + auto chunk = loadedChunks[ i ]; + for( size_t j = 0; j < extent[ 0 ]; ++j ) + { + std::cout << chunk.get()[ j ] << ", "; + } + std::cout << "\n----------\n" << std::endl; + } + } + + return 0; +#else + std::cout << "The streaming example requires that openPMD has been built " + "with ADIOS2." + << std::endl; + return 0; +#endif +} +#else +int main(){ return 0; } +#endif diff --git a/examples/10_streaming_read.py b/examples/10_streaming_read.py new file mode 100755 index 0000000000..eeae445b01 --- /dev/null +++ b/examples/10_streaming_read.py @@ -0,0 +1,35 @@ +#!/usr/bin/env python +import openpmd_api as io + +run_streaming_example = False + +if not run_streaming_example: + exit(0) + +if __name__ == "__main__": + if 'adios2' not in io.variants or not io.variants['adios2']: + print('This example requires ADIOS2') + exit(0) + + series = io.Series("stream.sst", io.Access_Type.read_only) + + for iteration in series.read_iterations(): + print("Current iteration {}".format(iteration.iteration_index)) + electronPositions = iteration.particles["e"]["position"] + loadedChunks = [] + shapes = [] + dimensions = ["x", "y", "z"] + + for i in range(3): + dim = dimensions[i] + rc = electronPositions[dim] + loadedChunks.append(rc.load_chunk([0], rc.shape)) + shapes.append(rc.shape) + iteration.close() + + for i in range(3): + dim = dimensions[i] + shape = shapes[i] + print("dim: {}".format(dim)) + chunk = loadedChunks[i] + print(chunk) diff --git a/examples/10_streaming_write.cpp b/examples/10_streaming_write.cpp new file mode 100644 index 0000000000..6aa1d4a185 --- /dev/null +++ b/examples/10_streaming_write.cpp @@ -0,0 +1,55 @@ +#define BUILD_STREAMING_EXAMPLE false +#if BUILD_STREAMING_EXAMPLE +#include + +#include +#include +#include // std::iota + +using std::cout; +using namespace openPMD; + +int +main() +{ +#if openPMD_HAVE_ADIOS2 + using position_t = double; + + // open file for writing + Series series = Series( "electrons.sst", Access::CREATE ); + + Datatype datatype = determineDatatype< position_t >(); + constexpr unsigned long length = 10ul; + Extent global_extent = { length }; + Dataset dataset = Dataset( datatype, global_extent ); + std::shared_ptr< position_t > local_data( + new position_t[ length ], + []( position_t const * ptr ) { delete[] ptr; } ); + + WriteIterations iterations = series.writeIterations(); + for( size_t i = 0; i < 100; ++i ) + { + Iteration iteration = iterations[ i ]; + Record electronPositions = iteration.particles[ "e" ][ "position" ]; + + std::iota( local_data.get(), local_data.get() + length, i * length ); + for( auto const & dim : { "x", "y", "z" } ) + { + RecordComponent pos = electronPositions[ dim ]; + pos.resetDataset( dataset ); + pos.storeChunk( local_data, Offset{ 0 }, global_extent ); + } + iteration.close(); + } + + return 0; +#else + std::cout << "The streaming example requires that openPMD has been built " + "with ADIOS2." + << std::endl; + return 0; +#endif +} +#else +int main(){ return 0; } +#endif diff --git a/examples/10_streaming_write.py b/examples/10_streaming_write.py new file mode 100755 index 0000000000..958a623df0 --- /dev/null +++ b/examples/10_streaming_write.py @@ -0,0 +1,31 @@ +#!/usr/bin/env python +import openpmd_api as io +import numpy as np + +run_streaming_example = False + +if not run_streaming_example: + exit(0) + +if __name__ == "__main__": + if 'adios2' not in io.variants or not io.variants['adios2']: + print('This example requires ADIOS2') + exit(0) + + series = io.Series("stream.sst", io.Access_Type.create) + datatype = np.dtype("double") + length = 10 + global_extent = [10] + dataset = io.Dataset(datatype, global_extent) + + iterations = series.write_iterations() + for i in range(100): + iteration = iterations[i] + electronPositions = iteration.particles["e"]["position"] + + local_data = np.arange(i * length, (i + 1) * length, dtype=datatype) + for dim in ["x", "y", "z"]: + pos = electronPositions[dim] + pos.reset_dataset(dataset) + pos[()] = local_data + iteration.close() diff --git a/include/openPMD/IO/ADIOS/ADIOS2Auxiliary.hpp b/include/openPMD/IO/ADIOS/ADIOS2Auxiliary.hpp index f39aa69fa4..c93b15ce28 100644 --- a/include/openPMD/IO/ADIOS/ADIOS2Auxiliary.hpp +++ b/include/openPMD/IO/ADIOS/ADIOS2Auxiliary.hpp @@ -133,9 +133,18 @@ namespace detail size_t operator( )( Params &&... ); }; + /** + * @brief Get openPMD datatype of attribute within given ADIOS IO. + * + * @param IO The IO within which to retrieve the attribute. + * @param attributeName The full ADIOS name of the attribute. + * @param verbose If true, print a warning if not finding the attribute. + * @return The openPMD datatype corresponding to the type of the attribute. + * UNDEFINED if attribute is not found. + */ Datatype attributeInfo( - adios2::IO &, + adios2::IO & IO, std::string const & attributeName, bool verbose ); } // namespace detail diff --git a/include/openPMD/IO/ADIOS/ADIOS2IOHandler.hpp b/include/openPMD/IO/ADIOS/ADIOS2IOHandler.hpp index 2b023b2c32..06276b6fb2 100644 --- a/include/openPMD/IO/ADIOS/ADIOS2IOHandler.hpp +++ b/include/openPMD/IO/ADIOS/ADIOS2IOHandler.hpp @@ -20,10 +20,11 @@ */ #pragma once -#include "ADIOS2FilePosition.hpp" #include "openPMD/IO/AbstractIOHandler.hpp" #include "openPMD/IO/AbstractIOHandlerImpl.hpp" #include "openPMD/IO/AbstractIOHandlerImplCommon.hpp" +#include "openPMD/IO/ADIOS/ADIOS2Auxiliary.hpp" +#include "openPMD/IO/ADIOS/ADIOS2FilePosition.hpp" #include "openPMD/IO/IOTask.hpp" #include "openPMD/IO/InvalidatableFile.hpp" #include "openPMD/auxiliary/JSON.hpp" @@ -33,14 +34,10 @@ #if openPMD_HAVE_ADIOS2 # include - -# include "openPMD/IO/ADIOS/ADIOS2Auxiliary.hpp" #endif - #if openPMD_HAVE_MPI # include #endif - #include #include @@ -48,11 +45,13 @@ #include #include #include // shared_ptr +#include #include #include #include // pair #include + namespace openPMD { #if openPMD_HAVE_ADIOS2 @@ -97,16 +96,18 @@ class ADIOS2IOHandlerImpl #if openPMD_HAVE_MPI - ADIOS2IOHandlerImpl( AbstractIOHandler *, MPI_Comm, nlohmann::json config ); - - MPI_Comm m_comm; + ADIOS2IOHandlerImpl( + AbstractIOHandler *, + MPI_Comm, + nlohmann::json config, + std::string engineType ); #endif // openPMD_HAVE_MPI explicit ADIOS2IOHandlerImpl( - AbstractIOHandler * - , nlohmann::json config - ); + AbstractIOHandler *, + nlohmann::json config, + std::string engineType ); ~ADIOS2IOHandlerImpl() override; @@ -136,6 +137,9 @@ class ADIOS2IOHandlerImpl void openPath( Writable *, Parameter< Operation::OPEN_PATH > const & ) override; + void closePath( Writable *, + Parameter< Operation::CLOSE_PATH > const & ) override; + void openDataset( Writable *, Parameter< Operation::OPEN_DATASET > & ) override; @@ -173,6 +177,9 @@ class ADIOS2IOHandlerImpl listAttributes( Writable *, Parameter< Operation::LIST_ATTS > & parameters ) override; + void + advance( Writable*, Parameter< Operation::ADVANCE > & ) override; + void availableChunks( Writable*, Parameter< Operation::AVAILABLE_CHUNKS > &) override; @@ -185,6 +192,10 @@ class ADIOS2IOHandlerImpl private: adios2::ADIOS m_ADIOS; + /** + * The ADIOS2 engine type, to be passed to adios2::IO::SetEngine + */ + std::string m_engineType; struct ParameterizedOperator { @@ -235,6 +246,9 @@ class ADIOS2IOHandlerImpl auxiliary::Option< std::vector< ParameterizedOperator > > getOperators(); + std::string + fileSuffix() const; + /* * We need to give names to IO objects. These names are irrelevant * within this application, since: @@ -326,7 +340,9 @@ namespace ADIOS2Defaults constexpr const_str str_engine = "engine"; constexpr const_str str_type = "type"; constexpr const_str str_params = "parameters"; -} + constexpr const_str str_usesteps = "usesteps"; + constexpr const_str str_usesstepsAttribute = "__openPMD_internal/useSteps"; +} // namespace ADIOS2Defaults namespace detail { @@ -423,7 +439,6 @@ namespace detail // Helper structs to help distinguish valid attribute/variable // datatypes from invalid ones - /* * This struct's purpose is to have specialisations * for vector and array types, as well as the boolean @@ -440,6 +455,26 @@ namespace detail static void readAttribute( adios2::IO & IO, std::string name, std::shared_ptr< Attribute::resource > resource ); + + /** + * @brief Is the attribute given by parameters name and val already + * defined exactly in that way within the given IO? + */ + static bool + attributeUnchanged( adios2::IO & IO, std::string name, BasicType val ) + { + auto attr = IO.InquireAttribute< BasicType >( name ); + if( !attr ) + { + return false; + } + std::vector< BasicType > data = attr.Data(); + if( data.size() != 1 ) + { + return false; + } + return data[ 0 ] == val; + } }; template< > struct AttributeTypes< std::complex< long double > > @@ -461,6 +496,14 @@ namespace detail throw std::runtime_error( "[ADIOS2] Internal error: no support for long double complex attribute types" ); } + + static bool + attributeUnchanged( + adios2::IO &, std::string, std::complex< long double > ) + { + throw std::runtime_error( + "[ADIOS2] Internal error: no support for long double complex attribute types" ); + } }; template< > struct AttributeTypes< std::vector< std::complex< long double > > > @@ -482,6 +525,16 @@ namespace detail throw std::runtime_error( "[ADIOS2] Internal error: no support for long double complex vector attribute types" ); } + + static bool + attributeUnchanged( + adios2::IO &, + std::string, + std::vector< std::complex< long double > > ) + { + throw std::runtime_error( + "[ADIOS2] Internal error: no support for long double complex vector attribute types" ); + } }; template < typename T > struct AttributeTypes< std::vector< T > > @@ -495,6 +548,32 @@ namespace detail static void readAttribute( adios2::IO & IO, std::string name, std::shared_ptr< Attribute::resource > resource ); + + static bool + attributeUnchanged( + adios2::IO & IO, + std::string name, + std::vector< T > val ) + { + auto attr = IO.InquireAttribute< BasicType >( name ); + if( !attr ) + { + return false; + } + std::vector< BasicType > data = attr.Data(); + if( data.size() != val.size() ) + { + return false; + } + for( size_t i = 0; i < val.size(); ++i ) + { + if( data[ i ] != val[ i ] ) + { + return false; + } + } + return true; + } }; template < typename T, size_t n > @@ -509,6 +588,32 @@ namespace detail static void readAttribute( adios2::IO & IO, std::string name, std::shared_ptr< Attribute::resource > resource ); + + static bool + attributeUnchanged( + adios2::IO & IO, + std::string name, + std::array< T, n > val ) + { + auto attr = IO.InquireAttribute< BasicType >( name ); + if( !attr ) + { + return false; + } + std::vector< BasicType > data = attr.Data(); + if( data.size() != n ) + { + return false; + } + for( size_t i = 0; i < n; ++i ) + { + if( data[ i ] != val[ i ] ) + { + return false; + } + } + return true; + } }; template <> struct AttributeTypes< bool > @@ -535,6 +640,22 @@ namespace detail { return r != 0; } + + static bool + attributeUnchanged( adios2::IO & IO, std::string name, bool val ) + { + auto attr = IO.InquireAttribute< BasicType >( name ); + if( !attr ) + { + return false; + } + std::vector< BasicType > data = attr.Data(); + if( data.size() != 1 ) + { + return false; + } + return data[ 0 ] == toRep( val ); + } }; @@ -698,20 +819,59 @@ namespace detail { BufferedActions( BufferedActions const & ) = delete; + /** + * The full path to the file created on disk, including the + * containing directory and the file extension, as determined + * by ADIOS2IOHandlerImpl::fileSuffix(). + * (Meaning, in case of the SST engine, no file suffix since the + * SST engine automatically adds its suffix unconditionally) + */ std::string m_file; - adios2::IO m_IO; - std::vector< std::unique_ptr< BufferedAction > > m_buffer; /** - * @brief std::optional would be more idiomatic, but it's not in - * the C++11 standard - * @todo replace with std::optional upon switching to C++17 + * ADIOS requires giving names to instances of adios2::IO. + * We make them different from the actual file name, because of the + * possible following workflow: + * + * 1. create file foo.bp + * -> would create IO object named foo.bp + * 2. delete that file + * (let's ignore that we don't support deletion yet and call it + * preplanning) + * 3. create file foo.bp a second time + * -> would create another IO object named foo.bp + * -> craash + * + * So, we just give out names based on a counter for IO objects. + * Hence, next to the actual file name, also store the name for the + * IO. */ - auxiliary::Option< adios2::Engine > m_engine; + std::string const m_IOName; + adios2::ADIOS & m_ADIOS; + adios2::IO m_IO; + std::vector< std::unique_ptr< BufferedAction > > m_buffer; adios2::Mode m_mode; - detail::WriteDataset m_writeDataset; - detail::DatasetReader m_readDataset; - detail::AttributeReader m_attributeReader; - ADIOS2IOHandlerImpl & m_impl; + detail::WriteDataset const m_writeDataset; + detail::DatasetReader const m_readDataset; + detail::AttributeReader const m_attributeReader; + + /* + * We call an attribute committed if the step during which it was + * written has been closed. + * A committed attribute cannot be modified. + */ + std::set< std::string > uncommittedAttributes; + + /* + * The openPMD API will generally create new attributes for each + * iteration. This results in a growing number of attributes over time. + * In streaming-based modes, these will be completely sent anew in each + * iteration. If the following boolean is true, old attributes will be + * removed upon CLOSE_GROUP. + * Should not be set to true in persistent backends. + * Will be automatically set by BufferedActions::configure_IO depending + * on chosen ADIOS2 engine and can not be explicitly overridden by user. + */ + bool optimizeAttributesStreaming = false; using AttributeMap_t = std::map< std::string, adios2::Params >; @@ -719,15 +879,53 @@ namespace detail ~BufferedActions( ); + /** + * Implementation of destructor, will only run once. + * + */ void - configure_IO( ADIOS2IOHandlerImpl & impl ); + finalize(); adios2::Engine & getEngine( ); + adios2::Engine & requireActiveStep( ); template < typename BA > void enqueue( BA && ba ); + template < typename BA > void enqueue( BA && ba, decltype( m_buffer ) & ); - void flush( ); + /** + * Flush deferred IO actions. + * + * @param performPutsGets A functor that takes as parameters (1) *this + * and (2) the ADIOS2 engine. + * Its task is to ensure that ADIOS2 performs Put/Get operations. + * Several options for this: + * * adios2::Engine::EndStep + * * adios2::Engine::Perform(Puts|Gets) + * * adios2::Engine::Close + * @param flushUnconditionally Whether to run the functor even if no + * deferred IO tasks had been queued. + */ + template< typename F > + void + flush( F && performPutsGets, bool flushUnconditionally ); + + /** + * Overload of flush() that uses adios2::Engine::Perform(Puts|Gets) + * and does not flush unconditionally. + * + */ + void + flush(); + + /** + * @brief Begin or end an ADIOS step. + * + * @param mode Whether to begin or end a step. + * @return AdvanceStatus + */ + AdvanceStatus + advance( AdvanceMode mode ); /* * Delete all buffered actions without running them. @@ -759,6 +957,90 @@ namespace detail invalidateVariablesMap(); private: + auxiliary::Option< adios2::Engine > m_engine; //! ADIOS engine + /** + * The ADIOS2 engine type, to be passed to adios2::IO::SetEngine + */ + std::string m_engineType; + /* + * streamStatus is NoStream for file-based ADIOS engines. + * This is relevant for the method BufferedActions::requireActiveStep, + * where a step is only opened if the status is OutsideOfStep, but not + * if NoStream. The rationale behind this is that parsing a Series + * works differently for file-based and for stream-based engines: + * * stream-based: Iterations are parsed as they arrive. For parsing an + * iteration, the iteration must be awaited. + * BufferedActions::requireActiveStep takes care of this. + * * file-based: The Series is parsed up front. If no step has been + * opened yet, ADIOS2 gives access to all variables and attributes + * from all steps. Upon opening a step, only the variables from that + * step are shown which hinders parsing. So, until a step is + * explicitly opened via ADIOS2IOHandlerImpl::advance, do not open + * one. + * (This is a workaround for the fact that attributes + * are not associated with steps in ADIOS -- seeing all attributes + * from all steps in file-based engines, but only the current + * variables breaks parsing otherwise.) + * + */ + enum class StreamStatus + { + /** + * A step is currently active. + */ + DuringStep, + /** + * A stream is active, but no step. + */ + OutsideOfStep, + /** + * Stream has ended. + */ + StreamOver, + /** + * File is not written is streaming fashion. + * Begin/EndStep will be replaced by simple flushes. + * Used for: + * 1) Writing BP4 files without steps despite using the Streaming + * API. This is due to the fact that ADIOS2.6.0 requires using + * steps to read BP4 files written with steps, so using steps + * is opt-in for now. + * 2) Reading with the Streaming API any file that has been written + * without steps. + */ + NoStream, + /** + * Necessary workaround under the following circumstances: + * 1) Using ADIOS2.6.0 + * 2) Using attribute-based layout + * 3) Reading from a file-based engine a Series written with steps + * Up until ADIOS2.6.0, attributes are not associated with ADIOS + * steps in file-based engines. As a consequence, parsing one + * ADIOS step will show only the variables of that step, but the + * attributes of all steps which breaks our parsing logic. + * Workaround: If parsing before opening any step, all variables + * and attributes in the file will be shown. + * Hence, streamStatus == Parsing means that the first step has yet + * to be opened. + */ + Parsing, + /** + * The stream status of a file-based engine will be decided upon + * opening the engine if in read mode. Up until then, this right + * here is the status. + */ + Undecided + }; + StreamStatus streamStatus = StreamStatus::OutsideOfStep; + adios2::StepStatus m_lastStepStatus = adios2::StepStatus::OK; + + /** + * See documentation for StreamStatus::Parsing. + * Will be set true under the circumstance described there in order to + * indicate that the first step should only be opened after parsing. + */ + bool delayOpeningTheFirstStep = false; + /* * ADIOS2 does not give direct access to its internal attribute and * variable maps, but will instead give access to copies of them. @@ -767,12 +1049,20 @@ namespace detail * the map whenever an attribute/variable is altered. In that case, we * fetch the map anew. * If empty, the buffered map has been invalidated and needs to be - * queried from ADIOS2 again. If true, the buffered map is equivalent to + * queried from ADIOS2 again. If full, the buffered map is equivalent to * the map that would be returned by a call to * IO::Available(Attributes|Variables). */ auxiliary::Option< AttributeMap_t > m_availableAttributes; auxiliary::Option< AttributeMap_t > m_availableVariables; + + /* + * finalize() will set this true to avoid running twice. + */ + bool finalized = false; + + void + configure_IO( ADIOS2IOHandlerImpl & impl ); }; @@ -817,12 +1107,16 @@ friend class ADIOS2IOHandlerImpl; std::string path, Access, MPI_Comm, - nlohmann::json options - ); + nlohmann::json options, + std::string engineType ); #endif - ADIOS2IOHandler(std::string path, Access, nlohmann::json options ); + ADIOS2IOHandler( + std::string path, + Access, + nlohmann::json options, + std::string engineType ); std::string backendName() const override { return "ADIOS2"; } diff --git a/include/openPMD/IO/AbstractIOHandlerImpl.hpp b/include/openPMD/IO/AbstractIOHandlerImpl.hpp index 773eb240f6..d204b8bc90 100644 --- a/include/openPMD/IO/AbstractIOHandlerImpl.hpp +++ b/include/openPMD/IO/AbstractIOHandlerImpl.hpp @@ -69,11 +69,14 @@ class AbstractIOHandlerImpl openFile(i.writable, deref_dynamic_cast< Parameter< O::OPEN_FILE > >(i.parameter.get())); break; case O::CLOSE_FILE: - closeFile(i.writable, *dynamic_cast< Parameter< O::CLOSE_FILE >* >(i.parameter.get())); + closeFile(i.writable, deref_dynamic_cast< Parameter< O::CLOSE_FILE > >(i.parameter.get())); break; case O::OPEN_PATH: openPath(i.writable, deref_dynamic_cast< Parameter< O::OPEN_PATH > >(i.parameter.get())); break; + case O::CLOSE_PATH: + closePath(i.writable, deref_dynamic_cast< Parameter< O::CLOSE_PATH > >(i.parameter.get())); + break; case O::OPEN_DATASET: openDataset(i.writable, deref_dynamic_cast< Parameter< O::OPEN_DATASET > >(i.parameter.get())); break; @@ -110,6 +113,9 @@ class AbstractIOHandlerImpl case O::LIST_ATTS: listAttributes(i.writable, deref_dynamic_cast< Parameter< O::LIST_ATTS > >(i.parameter.get())); break; + case O::ADVANCE: + advance(i.writable, deref_dynamic_cast< Parameter< O::ADVANCE > >(i.parameter.get())); + break; case O::AVAILABLE_CHUNKS: availableChunks(i.writable, deref_dynamic_cast< Parameter< O::AVAILABLE_CHUNKS > >(i.parameter.get())); break; @@ -130,6 +136,41 @@ class AbstractIOHandlerImpl */ virtual void closeFile( Writable *, Parameter< Operation::CLOSE_FILE > const & ) = 0; + + /** Advance the file/stream that this writable belongs to. + * + * If the backend is based around usage of IO steps (especially streaming + * backends), open or close an IO step. This is modeled closely after the + * step concept in ADIOS2. + * + * This task is used to implement streaming-aware semantics in the openPMD + * API by splitting data into packets that are written to and read from + * transport. + * + * IO actions up to the point of closing a step must be performed now. + * + * The advance mode is determined by parameters.mode. + * The return status code shall be stored as parameters.status. + */ + virtual void + advance( Writable *, Parameter< Operation::ADVANCE > & ) + {} + + /** Close an openPMD group. + * + * This is an optimization-enabling task and may be ignored by backends. + * Indicates that the group will not be accessed any further. + * Especially in step-based IO mode (e.g. streaming): + * Indicates that the group corresponding with the writable needs not be held + * in a parseable state for this and upcoming IO steps, allowing for deletion + * of metadata to be sent/stored (attributes, datasets, ..). + * Should fail if the writable is not written. + * Should fail if m_handler->accessType is AccessType::READ_ONLY. + * + */ + virtual void + closePath( Writable *, Parameter< Operation::CLOSE_PATH > const & ) + {} /** Report chunks that are available for loading from the dataset represented * by this writable. diff --git a/include/openPMD/IO/Format.hpp b/include/openPMD/IO/Format.hpp index 65c04290ea..66db72d37d 100644 --- a/include/openPMD/IO/Format.hpp +++ b/include/openPMD/IO/Format.hpp @@ -32,6 +32,7 @@ namespace openPMD HDF5, ADIOS1, ADIOS2, + ADIOS2_SST, JSON, DUMMY }; diff --git a/include/openPMD/IO/IOTask.hpp b/include/openPMD/IO/IOTask.hpp index 8b24ea4e3f..9b22a2d4fa 100644 --- a/include/openPMD/IO/IOTask.hpp +++ b/include/openPMD/IO/IOTask.hpp @@ -28,9 +28,11 @@ #include #include -#include #include #include +#include + +#include "openPMD/Streaming.hpp" namespace openPMD @@ -51,6 +53,7 @@ OPENPMDAPI_EXPORT_ENUM_CLASS(Operation) DELETE_FILE, CREATE_PATH, + CLOSE_PATH, OPEN_PATH, DELETE_PATH, LIST_PATHS, @@ -68,6 +71,7 @@ OPENPMDAPI_EXPORT_ENUM_CLASS(Operation) READ_ATT, LIST_ATTS, + ADVANCE, AVAILABLE_CHUNKS //!< Query chunks that can be loaded in a dataset }; // Operation @@ -177,6 +181,28 @@ struct OPENPMDAPI_EXPORT Parameter< Operation::CREATE_PATH > : public AbstractPa std::string path = ""; }; +template<> +struct OPENPMDAPI_EXPORT Parameter< Operation::CLOSE_PATH > : public AbstractParameter +{ + Parameter() = default; + Parameter( Parameter const & ) : AbstractParameter() + { + } + + Parameter & + operator=( Parameter const & ) + { + return *this; + } + + std::unique_ptr< AbstractParameter > + clone() const override + { + return std::unique_ptr< AbstractParameter >( + new Parameter< Operation::CLOSE_PATH >( *this ) ); + } +}; + template<> struct OPENPMDAPI_EXPORT Parameter< Operation::OPEN_PATH > : public AbstractParameter { @@ -463,6 +489,29 @@ struct OPENPMDAPI_EXPORT Parameter< Operation::LIST_ATTS > : public AbstractPara = std::make_shared< std::vector< std::string > >(); }; +template<> +struct OPENPMDAPI_EXPORT Parameter< Operation::ADVANCE > : public AbstractParameter +{ + Parameter() = default; + Parameter( Parameter const & p ) + : AbstractParameter(), mode( p.mode ), status( p.status ) + { + } + + std::unique_ptr< AbstractParameter > + clone() const override + { + return std::unique_ptr< AbstractParameter >( + new Parameter< Operation::ADVANCE >( *this ) ); + } + + //! input parameter + AdvanceMode mode; + //! output parameter + std::shared_ptr< AdvanceStatus > status = + std::make_shared< AdvanceStatus >( AdvanceStatus::OK ); +}; + template<> struct OPENPMDAPI_EXPORT Parameter< Operation::AVAILABLE_CHUNKS > : public AbstractParameter diff --git a/include/openPMD/Iteration.hpp b/include/openPMD/Iteration.hpp index fba4cdff24..ece4ba2182 100644 --- a/include/openPMD/Iteration.hpp +++ b/include/openPMD/Iteration.hpp @@ -20,10 +20,13 @@ */ #pragma once +#include "openPMD/auxiliary/Variant.hpp" #include "openPMD/backend/Attributable.hpp" #include "openPMD/backend/Container.hpp" +#include "openPMD/IterationEncoding.hpp" #include "openPMD/Mesh.hpp" #include "openPMD/ParticleSpecies.hpp" +#include "openPMD/Streaming.hpp" namespace openPMD @@ -41,6 +44,8 @@ class Iteration : public Attributable > friend class Container; friend class Series; + friend class WriteIterations; + friend class SeriesIterator; public: Iteration(Iteration const&); @@ -160,10 +165,63 @@ class Iteration : public Attributable * but not necessarily yet in the backend. * Will be propagated to the backend upon next flush. * Store the current status. + * Once an iteration has been closed, no further flushes shall be performed. + * If flushing a closed file, the old file may otherwise be overwritten. */ std::shared_ptr< CloseStatus > m_closed = std::make_shared< CloseStatus >( CloseStatus::Open ); + /** + * Whether a step is currently active for this iteration. + * Used for file-based iteration layout, see Series.hpp for + * group-based layout. + * Access via stepStatus() method to automatically select the correct + * one among both flags. + */ + std::shared_ptr< StepStatus > m_stepStatus = + std::make_shared< StepStatus >( StepStatus::NoStep ); + + /** + * @brief Begin an IO step on the IO file (or file-like object) + * containing this iteration. In case of group-based iteration + * layout, this will be the complete Series. + * + * @return AdvanceStatus + */ + AdvanceStatus + beginStep(); + + /** + * @brief End an IO step on the IO file (or file-like object) + * containing this iteration. In case of group-based iteration + * layout, this will be the complete Series. + * + * @return AdvanceStatus + */ + void + endStep(); + + /** + * @brief Is a step currently active for this iteration? + * + * In case of group-based iteration layout, this information is global + * (member of the Series object containing this iteration), + * in case of file-based iteration layout, it is local (member of this very + * object). + */ + StepStatus + getStepStatus(); + + /** + * @brief Set step activity status for this iteration. + * + * In case of group-based iteration layout, this information is set + * globally (member of the Series object containing this iteration), + * in case of file-based iteration layout, it is set locally (member of + * this very object). + */ + void setStepStatus( StepStatus ); + /* * @brief Check recursively whether this Iteration is dirty. * It is dirty if any attribute or dataset is read from or written to diff --git a/include/openPMD/RecordComponent.hpp b/include/openPMD/RecordComponent.hpp index 7b298cc6b4..19f62cbcf8 100644 --- a/include/openPMD/RecordComponent.hpp +++ b/include/openPMD/RecordComponent.hpp @@ -219,6 +219,12 @@ class RecordComponent : public BaseRecordComponent */ bool dirtyRecursive() const; + +protected: + /** + * Make sure to parse a RecordComponent only once. + */ + std::shared_ptr< bool > hasBeenRead = std::make_shared< bool >( false ); }; // RecordComponent diff --git a/include/openPMD/Series.hpp b/include/openPMD/Series.hpp index 80946528c3..3d9b043721 100644 --- a/include/openPMD/Series.hpp +++ b/include/openPMD/Series.hpp @@ -28,12 +28,19 @@ #include "openPMD/IO/Format.hpp" #include "openPMD/Iteration.hpp" #include "openPMD/IterationEncoding.hpp" +#include "openPMD/Streaming.hpp" +#include "openPMD/auxiliary/Option.hpp" +#include "openPMD/auxiliary/Variant.hpp" +#include "openPMD/backend/Attributable.hpp" +#include "openPMD/backend/Container.hpp" +#include "openPMD/config.hpp" #include "openPMD/version.hpp" #if openPMD_HAVE_MPI # include #endif +#include #include // expose private and protected members for invasive testing @@ -44,6 +51,9 @@ namespace openPMD { +class ReadIterations; +class WriteIterations; + /** @brief Root level of the openPMD hierarchy. * * Entry point and common link between all iterations of particle and mesh data. @@ -54,6 +64,7 @@ namespace openPMD class Series : public Attributable { friend class Iteration; + friend class SeriesIterator; public: #if openPMD_HAVE_MPI @@ -254,25 +265,104 @@ class Series : public Attributable */ void flush(); + /** + * @brief Entry point to the reading end of the streaming API. + * + * Creates and returns an instance of the ReadIterations class which can + * be used for iterating over the openPMD iterations in a C++11-style for + * loop. + * Look for the ReadIterations class for further documentation. + * + * @return ReadIterations + */ + ReadIterations + readIterations(); + + /** + * @brief Entry point to the writing end of the streaming API. + * + * Creates and returns an instance of the WriteIterations class which is a + * restricted container of iterations which takes care of + * streaming semantics. + * The created object is stored as member of the Series object, hence this + * method may be called as many times as a user wishes. + * Look for the WriteIterations class for further documentation. + * + * @return WriteIterations + */ + WriteIterations + writeIterations(); + Container< Iteration, uint64_t > iterations; OPENPMD_private: struct ParsedInput; + using iterations_iterator = decltype( iterations )::iterator; + std::unique_ptr< ParsedInput > parseInput(std::string); void init(std::shared_ptr< AbstractIOHandler >, std::unique_ptr< ParsedInput >); void initDefaults(); - template< typename IterationsContainer > - void flushFileBased( IterationsContainer && iterationsToFlush ); - template< typename IterationsContainer > - void flushGroupBased( IterationsContainer && iterationsToFlush ); + std::future< void > flush_impl( + iterations_iterator begin, iterations_iterator end ); + void flushFileBased( iterations_iterator begin, iterations_iterator end ); + void flushGroupBased( iterations_iterator begin, iterations_iterator end ); void flushMeshesPath(); void flushParticlesPath(); - void readFileBased(); - void readGroupBased(); - void readBase(); + void readFileBased( ); + /** + * Note on re-parsing of a Series: + * If init == false, the parsing process will seek for new + * Iterations/Records/Record Components etc. + * Re-parsing of objects that have already been parsed is not implemented + * as of yet. Such a facility will be required upon implementing things such + * as resizable datasets. + */ + void readGroupBased( bool init = true ); + void + readBase(); void read(); + std::string iterationFilename( uint64_t i ); void openIteration( uint64_t index, Iteration iteration ); + /** + * Find the given iteration in Series::iterations and return an iterator + * into Series::iterations at that place. + */ + iterations_iterator + indexOf( Iteration const & ); + + /** + * @brief In step-based IO mode, begin or end an IO step for the given + * iteration. + * + * Called internally by Iteration::beginStep and Iteration::endStep. + * + * @param mode Whether to begin or end a step. + * @param file The Attributable representing the iteration. In file-based + * iteration layout, this is an Iteration object, in group- + * based layout, it's the Series object. + * @param it The iterator within Series::iterations pointing to that + * iteration. + * @param iteration The actual Iteration object. + * @return AdvanceStatus + */ + AdvanceStatus + advance( + AdvanceMode mode, + Attributable & file, + iterations_iterator it, + Iteration & iteration ); + + /** + * Whether a step is currently active for this iteration. + * Used for group-based iteration layout, see Series.hpp for + * iteration-based layout. + * Access via stepStatus() method to automatically select the correct + * one among both flags. + */ + std::shared_ptr< StepStatus > m_stepStatus = + std::make_shared< StepStatus >( StepStatus::NoStep ); + static constexpr char const * const BASEPATH = "/data/%T/"; std::shared_ptr< IterationEncoding > m_iterationEncoding; @@ -282,5 +372,141 @@ class Series : public Attributable std::shared_ptr< std::string > m_filenamePrefix; std::shared_ptr< std::string > m_filenamePostfix; std::shared_ptr< int > m_filenamePadding; + + std::shared_ptr< auxiliary::Option< WriteIterations > > m_writeIterations = + std::make_shared< auxiliary::Option< WriteIterations > >(); }; // Series + +/** + * @brief Subclass of Iteration that knows its own index withing the containing + * Series. + */ +class IndexedIteration : public Iteration +{ + friend class SeriesIterator; + +public: + using iterations_t = decltype( Series::iterations ); + using index_t = iterations_t::key_type; + index_t const iterationIndex; + +private: + template< typename Iteration_t > + IndexedIteration( Iteration_t && it, index_t index ) + : Iteration( std::forward< Iteration_t >( it ) ) + , iterationIndex( index ) + { + } +}; + +class SeriesIterator +{ + using iteration_index_t = IndexedIteration::index_t; + + using maybe_series_t = auxiliary::Option< Series * >; + + maybe_series_t m_series; + iteration_index_t m_currentIteration = 0; + + //! construct the end() iterator + SeriesIterator(); + +public: + SeriesIterator( Series * ); + + SeriesIterator & + operator++(); + + IndexedIteration + operator*(); + + bool + operator==( SeriesIterator const & other ) const; + + bool + operator!=( SeriesIterator const & other ) const; + + static SeriesIterator + end(); +}; + +/** + * @brief Reading side of the streaming API. + * + * Create instance via Series::readIterations(). + * For use in a C++11-style foreach loop over iterations. + * Designed to allow reading any kind of Series, streaming and non- + * streaming alike. + * Calling Iteration::close() manually before opening the next iteration is + * encouraged and will implicitly flush all deferred IO actions. + * Otherwise, Iteration::close() will be implicitly called upon + * SeriesIterator::operator++(), i.e. upon going to the next iteration in + * the foreach loop. + * Since this is designed for streaming mode, reopening an iteration is + * not possible once it has been closed. + * + */ +class ReadIterations +{ + friend class Series; + +private: + using iterations_t = decltype( Series::iterations ); + using iterator_t = SeriesIterator; + + Series * m_series; + + ReadIterations( Series * ); + +public: + iterator_t + begin(); + + iterator_t + end(); +}; + +/** Writing side of the streaming API. + * + * Create instance via Series::writeIterations(). + * For use via WriteIterations::operator[](). + * Designed to allow reading any kind of Series, streaming and non- + * streaming alike. Calling Iteration::close() manually before opening + * the next iteration is encouraged and will implicitly flush all + * deferred IO actions. Otherwise, Iteration::close() will be implicitly + * called upon SeriesIterator::operator++(), i.e. upon going to the next + * iteration in the foreach loop. + * + * Since this is designed for streaming mode, reopening an iteration is + * not possible once it has been closed. + * + */ +class WriteIterations : private Container< Iteration, uint64_t > +{ + friend class Series; + +private: + using iterations_t = Container< Iteration, uint64_t >; + struct SharedResources + { + iterations_t iterations; + auxiliary::Option< uint64_t > currentlyOpen; + + SharedResources( iterations_t ); + ~SharedResources(); + }; + + using key_type = typename iterations_t::key_type; + using value_type = typename iterations_t::key_type; + WriteIterations( iterations_t ); + explicit WriteIterations() = default; + //! Index of the last opened iteration + std::shared_ptr< SharedResources > shared; + +public: + mapped_type & + operator[]( key_type const & key ) override; + mapped_type & + operator[]( key_type && key ) override; +}; } // namespace openPMD diff --git a/include/openPMD/Streaming.hpp b/include/openPMD/Streaming.hpp new file mode 100644 index 0000000000..19c6bdd88c --- /dev/null +++ b/include/openPMD/Streaming.hpp @@ -0,0 +1,49 @@ +#pragma once + +#include +#include +#include +#include + +#include "openPMD/Dataset.hpp" +#include "openPMD/benchmark/mpi/BlockSlicer.hpp" +#include + +namespace openPMD +{ +/** + * In step-based mode (i.e. when using the Streaming API), + * indicate whether there are further steps to read or not + * (i.e. the stream is not over or it is). + * + */ +enum class AdvanceStatus : unsigned char +{ + OK, /* stream goes on */ + OVER /* stream is over */ +}; + +/** + * In step-based mode (i.e. when using the Streaming API), + * stepping/advancing through the Series is performed in terms + * of interleaving begin- and end-step calls. + * Distinguish both kinds by using this enum. + * + */ +enum class AdvanceMode : unsigned char +{ + BEGINSTEP, + ENDSTEP +}; + +/** + * Used in step-based mode (i.e. when using the Streaming API) + * to determine whether a step is currently active or not. + * + */ +enum class StepStatus : unsigned char +{ + DuringStep, /* step is currently active */ + NoStep /* no step is currently active */ +}; +} // namespace openPMD diff --git a/include/openPMD/helper/list_series.hpp b/include/openPMD/helper/list_series.hpp index 0c46995dd3..af42bd4b4c 100644 --- a/include/openPMD/helper/list_series.hpp +++ b/include/openPMD/helper/list_series.hpp @@ -39,7 +39,7 @@ namespace helper */ std::ostream & listSeries( - Series const & series, + Series & series, bool const longer = false, std::ostream & out = std::cout ); diff --git a/src/Format.cpp b/src/Format.cpp index 93013e066e..a31d0a89d7 100644 --- a/src/Format.cpp +++ b/src/Format.cpp @@ -53,6 +53,8 @@ namespace openPMD { bp_backend ); } + if (auxiliary::ends_with(filename, ".sst")) + return Format::ADIOS2_SST; if (auxiliary::ends_with(filename, ".json")) return Format::JSON; if (std::string::npos != filename.find('.') /* extension is provided */ ) @@ -69,6 +71,8 @@ namespace openPMD { case Format::ADIOS1: case Format::ADIOS2: return ".bp"; + case Format::ADIOS2_SST: + return ".sst"; case Format::JSON: return ".json"; default: diff --git a/src/IO/ADIOS/ADIOS1IOHandler.cpp b/src/IO/ADIOS/ADIOS1IOHandler.cpp index f5c091e980..15ad1b3c0b 100644 --- a/src/IO/ADIOS/ADIOS1IOHandler.cpp +++ b/src/IO/ADIOS/ADIOS1IOHandler.cpp @@ -132,6 +132,9 @@ ADIOS1IOHandlerImpl::flush() case O::OPEN_PATH: openPath(i.writable, deref_dynamic_cast< Parameter< O::OPEN_PATH > >(i.parameter.get())); break; + case O::CLOSE_PATH: + closePath(i.writable, deref_dynamic_cast< Parameter< O::CLOSE_PATH > >(i.parameter.get())); + break; case O::OPEN_DATASET: openDataset(i.writable, deref_dynamic_cast< Parameter< O::OPEN_DATASET > >(i.parameter.get())); break; @@ -168,6 +171,9 @@ ADIOS1IOHandlerImpl::flush() case O::LIST_ATTS: listAttributes(i.writable, deref_dynamic_cast< Parameter< O::LIST_ATTS > >(i.parameter.get())); break; + case O::ADVANCE: + advance(i.writable, deref_dynamic_cast< Parameter< O::ADVANCE > >(i.parameter.get())); + break; case O::AVAILABLE_CHUNKS: availableChunks(i.writable, deref_dynamic_cast< Parameter< O::AVAILABLE_CHUNKS > >(i.parameter.get())); break; diff --git a/src/IO/ADIOS/ADIOS2IOHandler.cpp b/src/IO/ADIOS/ADIOS2IOHandler.cpp index 72b44a790a..64608191ea 100644 --- a/src/IO/ADIOS/ADIOS2IOHandler.cpp +++ b/src/IO/ADIOS/ADIOS2IOHandler.cpp @@ -29,6 +29,7 @@ #include "openPMD/auxiliary/StringManip.hpp" #include +#include // std::tolower #include #include #include @@ -66,10 +67,11 @@ namespace openPMD ADIOS2IOHandlerImpl::ADIOS2IOHandlerImpl( AbstractIOHandler * handler, MPI_Comm communicator, - nlohmann::json cfg ) + nlohmann::json cfg, + std::string engineType ) : AbstractIOHandlerImplCommon( handler ) - , m_comm{ communicator } , m_ADIOS{ communicator, ADIOS2_DEBUG_MODE } + , m_engineType( std::move( engineType ) ) { init( std::move( cfg ) ); } @@ -78,8 +80,11 @@ ADIOS2IOHandlerImpl::ADIOS2IOHandlerImpl( ADIOS2IOHandlerImpl::ADIOS2IOHandlerImpl( AbstractIOHandler * handler, - nlohmann::json cfg ) - : AbstractIOHandlerImplCommon( handler ), m_ADIOS{ ADIOS2_DEBUG_MODE } + nlohmann::json cfg, + std::string engineType ) + : AbstractIOHandlerImplCommon( handler ) + , m_ADIOS{ ADIOS2_DEBUG_MODE } + , m_engineType( std::move( engineType ) ) { init( std::move( cfg ) ); } @@ -122,6 +127,22 @@ ADIOS2IOHandlerImpl::init( nlohmann::json cfg ) return; } m_config = std::move( cfg[ "adios2" ] ); + auto engineConfig = config( ADIOS2Defaults::str_engine ); + if( !engineConfig.json().is_null() ) + { + auto engineTypeConfig = + config( ADIOS2Defaults::str_type, engineConfig ).json(); + if( !engineTypeConfig.is_null() ) + { + // convert to string + m_engineType = engineTypeConfig; + std::transform( + m_engineType.begin(), + m_engineType.end(), + m_engineType.begin(), + []( unsigned char c ) { return std::tolower( c ); } ); + } + } auto operators = getOperators(); if( operators ) { @@ -181,6 +202,26 @@ ADIOS2IOHandlerImpl::getOperators() return getOperators( m_config ); } +std::string +ADIOS2IOHandlerImpl::fileSuffix() const +{ + // SST engine adds its suffix unconditionally + // so we don't add it + static std::map< std::string, std::string > endings{ + { "sst", "" }, { "staging", "" }, { "bp4", ".bp" }, + { "bp3", ".bp" }, { "file", ".bp" }, { "hdf5", ".h5" } + }; + auto it = endings.find( m_engineType ); + if( it != endings.end() ) + { + return it->second; + } + else + { + return ".adios2"; + } +} + std::future< void > ADIOS2IOHandlerImpl::flush() { @@ -209,9 +250,10 @@ void ADIOS2IOHandlerImpl::createFile( if ( !writable->written ) { std::string name = parameters.name; - if ( !auxiliary::ends_with( name, ".bp" ) ) + std::string suffix( fileSuffix() ); + if( !auxiliary::ends_with( name, suffix ) ) { - name += ".bp"; + name += suffix; } auto res_pair = getPossiblyExisting( name ); @@ -363,9 +405,10 @@ void ADIOS2IOHandlerImpl::openFile( } std::string name = parameters.name; - if ( !auxiliary::ends_with( name, ".bp" ) ) + std::string suffix( fileSuffix() ); + if( !auxiliary::ends_with( name, suffix ) ) { - name += ".bp"; + name += suffix; } auto file = std::get< PE_InvalidatableFile >( getPossiblyExisting( name ) ); @@ -389,7 +432,15 @@ ADIOS2IOHandlerImpl::closeFile( auto it = m_fileData.find( fileIterator->second ); if ( it != m_fileData.end( ) ) { - it->second->flush( ); + /* + * No need to finalize unconditionally, destructor will take care + * of it. + */ + it->second->flush( + []( detail::BufferedActions & ba, adios2::Engine & ) { + ba.finalize(); + }, + /* flushUnconditionally = */ false ); m_fileData.erase( it ); } } @@ -526,7 +577,7 @@ void ADIOS2IOHandlerImpl::listPaths( * from variables and attributes. */ auto & fileData = getFileData( file ); - fileData.getEngine( ); // make sure that the attributes are present + fileData.requireActiveStep(); std::unordered_set< std::string > subdirs; /* @@ -602,8 +653,7 @@ void ADIOS2IOHandlerImpl::listDatasets( */ auto & fileData = getFileData( file ); - fileData.getEngine( ); // make sure that the attributes are present - + fileData.requireActiveStep(); std::map< std::string, adios2::Params > vars = fileData.availableVariables(); @@ -637,7 +687,8 @@ void ADIOS2IOHandlerImpl::listAttributes( attributePrefix = ""; } auto & ba = getFileData( file ); - ba.getEngine( ); // make sure that the attributes are present + ba.requireActiveStep(); // make sure that the attributes are present + auto const & attrs = ba.availableAttributesPrefixed( attributePrefix ); for( auto & rawAttr : attrs ) { @@ -652,6 +703,49 @@ void ADIOS2IOHandlerImpl::listAttributes( } } +void +ADIOS2IOHandlerImpl::advance( + Writable * writable, + Parameter< Operation::ADVANCE > & parameters ) +{ + auto file = m_files[ writable ]; + auto & ba = getFileData( file ); + *parameters.status = ba.advance( parameters.mode ); +} + +void +ADIOS2IOHandlerImpl::closePath( + Writable * writable, + Parameter< Operation::CLOSE_PATH > const & ) +{ + VERIFY_ALWAYS( + writable->written, + "[ADIOS2] Cannot close a path that has not been written yet." ); + if( m_handler->m_backendAccess == Access::READ_ONLY ) + { + // nothing to do + return; + } + auto file = refreshFileFromParent( writable ); + auto & fileData = getFileData( file ); + if( !fileData.optimizeAttributesStreaming ) + { + return; + } + auto position = setAndGetFilePosition( writable ); + auto const positionString = filePositionToString( position ); + VERIFY( + !auxiliary::ends_with( positionString, '/' ), + "[ADIOS2] Position string has unexpected format. This is a bug " + "in the openPMD API." ); + + for( auto const & attr : + fileData.availableAttributesPrefixed( positionString ) ) + { + fileData.m_IO.RemoveAttribute( positionString + '/' + attr ); + } +} + void ADIOS2IOHandlerImpl::availableChunks( Writable * writable, @@ -924,8 +1018,37 @@ namespace detail std::string t = IO.AttributeType( fullName ); if ( !t.empty( ) ) // an attribute is present <=> it has a type { - IO.RemoveAttribute( fullName ); + // don't overwrite attributes if they are equivalent + // overwriting is only legal within the same step + + auto attributeModifiable = [ &filedata, &fullName ]() { + auto it = filedata.uncommittedAttributes.find( fullName ); + return it != filedata.uncommittedAttributes.end(); + }; + if( AttributeTypes< T >::attributeUnchanged( + IO, + fullName, + variantSrc::get< T >( parameters.resource ) ) ) + { + return; + } + else if( attributeModifiable() ) + { + IO.RemoveAttribute( fullName ); + } + else + { + std::cerr << "[Warning][ADIOS2] Cannot modify attribute from " + "previous step: " + << fullName << std::endl; + return; + } } + else + { + filedata.uncommittedAttributes.emplace( fullName ); + } + typename AttributeTypes< T >::Attr attr = AttributeTypes< T >::createAttribute( IO, fullName, variantSrc::get< T >( parameters.resource ) ); @@ -1131,7 +1254,9 @@ namespace detail openDataset( InvalidatableFile file, const std::string & varName, Parameter< Operation::OPEN_DATASET > & parameters ) { - auto & IO = m_impl->getFileData( file ).m_IO; + auto & fileData = m_impl->getFileData( file ); + fileData.requireActiveStep(); + auto & IO = fileData.m_IO; adios2::Variable< T > var = IO.InquireVariable< T >( varName ); if ( !var ) { @@ -1339,15 +1464,20 @@ namespace detail *param.dtype = ret; } - - BufferedActions::BufferedActions( ADIOS2IOHandlerImpl & impl, - InvalidatableFile file ) - : m_file( impl.fullPath( std::move( file ) ) ), - m_IO( impl.m_ADIOS.DeclareIO( std::to_string( impl.nameCounter++ ) ) ), - m_mode( impl.adios2AccessMode( ) ), m_writeDataset( &impl ), - m_readDataset( &impl ), m_attributeReader( ), m_impl( impl ) - { - if ( !m_IO ) + BufferedActions::BufferedActions( + ADIOS2IOHandlerImpl & impl, + InvalidatableFile file ) + : m_file( impl.fullPath( std::move( file ) ) ) + , m_IOName( std::to_string( impl.nameCounter++ ) ) + , m_ADIOS( impl.m_ADIOS ) + , m_IO( impl.m_ADIOS.DeclareIO( m_IOName ) ) + , m_mode( impl.adios2AccessMode() ) + , m_writeDataset( &impl ) + , m_readDataset( &impl ) + , m_attributeReader() + , m_engineType( impl.m_engineType ) + { + if( !m_IO ) { throw std::runtime_error( "[ADIOS2] Internal error: Failed declaring ADIOS2 IO object for file " + @@ -1361,6 +1491,15 @@ namespace detail BufferedActions::~BufferedActions() { + finalize(); + } + + void BufferedActions::finalize() + { + if( finalized ) + { + return; + } // if write accessing, ensure that the engine is opened if( !m_engine && m_mode != adios2::Mode::Read ) { @@ -1368,20 +1507,82 @@ namespace detail } if( m_engine ) { - m_engine.get().Close(); + auto & engine = m_engine.get(); + // might have been closed previously + if( engine ) + { + if( streamStatus == StreamStatus::DuringStep ) + { + engine.EndStep(); + } + engine.Close(); + m_ADIOS.RemoveIO( m_IOName ); + } } + finalized = true; } - void - BufferedActions::configure_IO( ADIOS2IOHandlerImpl & impl ) - { - (void)impl; + void BufferedActions::configure_IO(ADIOS2IOHandlerImpl& impl){ + ( void )impl; + static std::set< std::string > streamingEngines = { + "sst", "insitumpi", "inline", "staging" + }; + static std::set< std::string > fileEngines = { + "bp4", "bp3", "hdf5", "file" + }; + + // set engine type + { + // allow overriding through environment variable + m_engineType = auxiliary::getEnvString( + "OPENPMD_ADIOS2_ENGINE", m_engineType ); + std::transform( + m_engineType.begin(), + m_engineType.end(), + m_engineType.begin(), + []( unsigned char c ) { return std::tolower( c ); } ); + impl.m_engineType = this->m_engineType; + m_IO.SetEngine( m_engineType ); + auto it = streamingEngines.find( m_engineType ); + if( it != streamingEngines.end() ) + { + optimizeAttributesStreaming = true; + streamStatus = StreamStatus::OutsideOfStep; + } + else + { + it = fileEngines.find( m_engineType ); + if( it != fileEngines.end() ) + { + switch( m_mode ) + { + case adios2::Mode::Read: + streamStatus = StreamStatus::Undecided; + delayOpeningTheFirstStep = true; + break; + case adios2::Mode::Write: + streamStatus = StreamStatus::NoStream; + break; + default: + throw std::runtime_error( "Unreachable!" ); + } + optimizeAttributesStreaming = false; + } + else + { + throw std::runtime_error( + "[ADIOS2IOHandler] Unknown engine type. Please choose " + "one out of [sst, staging, bp4, bp3, hdf5, file]" ); + // not listing unsupported engines + } + } + } + + // set engine parameters std::set< std::string > alreadyConfigured; auto engineConfig = impl.config( ADIOS2Defaults::str_engine ); if( !engineConfig.json().is_null() ) { - m_IO.SetEngine( - impl.config( ADIOS2Defaults::str_type, engineConfig ).json() ); auto params = impl.config( ADIOS2Defaults::str_params, engineConfig ); params.declareFullyRead(); @@ -1394,8 +1595,23 @@ namespace detail alreadyConfigured.emplace( it.key() ); } } - alreadyConfigured.emplace( "Engine" ); + auto _useAdiosSteps = + impl.config( ADIOS2Defaults::str_usesteps, engineConfig ); + if( !_useAdiosSteps.json().is_null() && + m_mode != adios2::Mode::Read ) + { + bool tmp = _useAdiosSteps.json(); + if( streamStatus == StreamStatus::OutsideOfStep && + !bool( tmp ) ) + { + throw std::runtime_error( + "Cannot switch off steps for streaming engines." ); + } + streamStatus = bool( tmp ) ? StreamStatus::OutsideOfStep + : StreamStatus::NoStream; + } } + auto shadow = impl.m_config.invertShadow(); if( shadow.size() > 0 ) { @@ -1410,13 +1626,6 @@ namespace detail }; // read parameters from environment - if( notYetConfigured( "Engine" ) ) - { - auto const engine = - auxiliary::getEnvString( "OPENPMD_ADIOS2_ENGINE", "File" ); - m_IO.SetEngine( engine ); - } - if( notYetConfigured( "CollectiveMetadata" ) ) { if( 1 == @@ -1470,10 +1679,64 @@ namespace detail adios2::Engine & BufferedActions::getEngine() { - if ( !m_engine ) + if( !m_engine ) { - m_engine = auxiliary::makeOption( - adios2::Engine( m_IO.Open( m_file, m_mode ) ) ); + switch( m_mode ) + { + case adios2::Mode::Write: + { + bool_representation usesSteps = + streamStatus == StreamStatus::NoStream ? 0 : 1; + m_IO.DefineAttribute< bool_representation >( + ADIOS2Defaults::str_usesstepsAttribute, usesSteps ); + m_engine = auxiliary::makeOption( + adios2::Engine( m_IO.Open( m_file, m_mode ) ) ); + break; + } + case adios2::Mode::Read: + { + m_engine = auxiliary::makeOption( + adios2::Engine( m_IO.Open( m_file, m_mode ) ) ); + switch( streamStatus ) + { + case StreamStatus::Undecided: + { + auto attr = + m_IO.InquireAttribute< bool_representation >( + ADIOS2Defaults::str_usesstepsAttribute ); + if( attr && attr.Data()[ 0 ] == 1 ) + { + if( delayOpeningTheFirstStep ) + { + streamStatus = StreamStatus::Parsing; + } + else + { + m_engine.get().BeginStep(); + streamStatus = StreamStatus::DuringStep; + } + } + else + { + streamStatus = StreamStatus::NoStream; + } + break; + } + case StreamStatus::OutsideOfStep: + m_engine.get().BeginStep(); + streamStatus = StreamStatus::DuringStep; + break; + default: + throw std::runtime_error( + "[ADIOS2] Control flow error!" ); + } + break; + } + default: + throw std::runtime_error( + "[ADIOS2] Invalid ADIOS access mode" ); + } + if( !m_engine ) { throw std::runtime_error( "[ADIOS2] Failed opening Engine." ); @@ -1482,88 +1745,180 @@ namespace detail return m_engine.get(); } + adios2::Engine & BufferedActions::requireActiveStep( ) + { + adios2::Engine & eng = getEngine(); + if( streamStatus == StreamStatus::OutsideOfStep ) + { + m_lastStepStatus = eng.BeginStep(); + streamStatus = StreamStatus::DuringStep; + } + return eng; + } + template < typename BA > void BufferedActions::enqueue( BA && ba ) + { + enqueue< BA >( std::forward< BA >( ba ), m_buffer ); + } + + template < typename BA > void BufferedActions::enqueue( + BA && ba, + decltype( m_buffer ) & buffer ) { using _BA = typename std::remove_reference< BA >::type; - m_buffer.emplace_back( std::unique_ptr< BufferedAction >( + buffer.emplace_back( std::unique_ptr< BufferedAction >( new _BA( std::forward< BA >( ba ) ) ) ); } - void BufferedActions::flush( ) + template< typename F > + void + BufferedActions::flush( F && performPutGets, bool flushUnconditionally ) { - auto & eng = getEngine( ); + if( streamStatus == StreamStatus::StreamOver ) + { + if( flushUnconditionally ) + { + throw std::runtime_error( + "[ADIOS2] Cannot access engine since stream is over." ); + } + return; + } + auto & eng = getEngine(); + /* + * Only open a new step if it is necessary. + */ + if( streamStatus == StreamStatus::OutsideOfStep ) { - for ( auto const & ba : m_buffer ) + if( m_buffer.empty() ) { - ba->run( *this ); + if( flushUnconditionally ) + { + performPutGets( *this, eng ); + } + return; } - // Flush() does not necessarily perform - // deferred actions.... - switch ( m_mode ) + else { - case adios2::Mode::Write: - eng.PerformPuts( ); - break; - case adios2::Mode::Read: - eng.PerformGets( ); - break; - case adios2::Mode::Append: - // TODO order? - eng.PerformGets( ); - eng.PerformPuts( ); - break; - default: - break; + requireActiveStep(); } } - m_buffer.clear( ); - } + for( auto & ba : m_buffer ) + { + ba->run( *this ); + } + + performPutGets( *this, eng ); - void BufferedActions::drop( ) - { m_buffer.clear(); } void - BufferedActions::invalidateAttributesMap() + BufferedActions::flush() { - m_availableAttributes = auxiliary::Option< AttributeMap_t >(); + flush( + []( BufferedActions & ba, adios2::Engine & eng ) { + switch( ba.m_mode ) + { + case adios2::Mode::Write: + eng.PerformPuts(); + break; + case adios2::Mode::Read: + eng.PerformGets(); + break; + case adios2::Mode::Append: + // TODO order? + eng.PerformGets(); + eng.PerformPuts(); + break; + default: + break; + } + }, + /* flushUnconditionally = */ false ); } - BufferedActions::AttributeMap_t const & - BufferedActions::availableAttributes() + AdvanceStatus + BufferedActions::advance( AdvanceMode mode ) { - if( m_availableAttributes ) + if( streamStatus == StreamStatus::Undecided ) { - return m_availableAttributes.get(); + // stream status gets decided on upon opening an engine + getEngine(); } - else + // sic! no else + if( streamStatus == StreamStatus::NoStream ) { - m_availableAttributes = - auxiliary::makeOption( m_IO.AvailableAttributes() ); - return m_availableAttributes.get(); + flush(); + return AdvanceStatus::OK; } - } + switch( mode ) + { + case AdvanceMode::ENDSTEP: + { + /* + * Advance mode write: + * Close the current step, defer opening the new step + * until one is actually needed: + * (1) The engine is accessed in BufferedActions::flush + * (2) A new step is opened before the currently active step + * has seen an access. See the following lines: open the + * step just to skip it again. + */ + if( streamStatus == StreamStatus::OutsideOfStep ) + { + getEngine().BeginStep(); + } + flush( + []( BufferedActions &, adios2::Engine & eng ) { + eng.EndStep(); + }, + /* flushUnconditionally = */ true ); + uncommittedAttributes.clear(); + streamStatus = StreamStatus::OutsideOfStep; + return AdvanceStatus::OK; + } + case AdvanceMode::BEGINSTEP: + { + adios2::StepStatus adiosStatus = m_lastStepStatus; - void - BufferedActions::invalidateVariablesMap() - { - m_availableVariables = auxiliary::Option< AttributeMap_t >(); + // Step might have been opened implicitly already + // by requireActiveStep() + // In that case, streamStatus is DuringStep and Adios + // return status is stored in m_lastStepStatus + if( streamStatus != StreamStatus::DuringStep ) + { + flush( + [ &adiosStatus ]( + BufferedActions &, adios2::Engine & engine ) { + adiosStatus = engine.BeginStep(); + }, + /* flushUnconditionally = */ true ); + } + AdvanceStatus res = AdvanceStatus::OK; + switch( adiosStatus ) + { + case adios2::StepStatus::EndOfStream: + streamStatus = StreamStatus::StreamOver; + res = AdvanceStatus::OVER; + break; + default: + streamStatus = StreamStatus::DuringStep; + res = AdvanceStatus::OK; + break; + } + invalidateAttributesMap(); + invalidateVariablesMap(); + return res; + } + } + throw std::runtime_error( + "Internal error: Advance mode should be explicitly" + " chosen by the front-end." ); } - BufferedActions::AttributeMap_t const & - BufferedActions::availableVariables() + void BufferedActions::drop( ) { - if( m_availableVariables ) - { - return m_availableVariables.get(); - } - else - { - m_availableVariables = - auxiliary::makeOption( m_IO.AvailableVariables() ); - return m_availableVariables.get(); - } + m_buffer.clear(); } static std::vector< std::string > @@ -1612,6 +1967,48 @@ namespace detail *this ); } + void + BufferedActions::invalidateAttributesMap() + { + m_availableAttributes = auxiliary::Option< AttributeMap_t >(); + } + + BufferedActions::AttributeMap_t const & + BufferedActions::availableAttributes() + { + if( m_availableAttributes ) + { + return m_availableAttributes.get(); + } + else + { + m_availableAttributes = + auxiliary::makeOption( m_IO.AvailableAttributes() ); + return m_availableAttributes.get(); + } + } + + void + BufferedActions::invalidateVariablesMap() + { + m_availableVariables = auxiliary::Option< AttributeMap_t >(); + } + + BufferedActions::AttributeMap_t const & + BufferedActions::availableVariables() + { + if( m_availableVariables ) + { + return m_availableVariables.get(); + } + else + { + m_availableVariables = + auxiliary::makeOption( m_IO.AvailableVariables() ); + return m_availableVariables.get(); + } + } + } // namespace detail # if openPMD_HAVE_MPI @@ -1620,9 +2017,10 @@ ADIOS2IOHandler::ADIOS2IOHandler( std::string path, openPMD::Access at, MPI_Comm comm, - nlohmann::json options ) - : AbstractIOHandler( std::move( path ), at, comm ), - m_impl{ this, comm, std::move( options ) } + nlohmann::json options, + std::string engineType ) + : AbstractIOHandler( std::move( path ), at, comm ) + , m_impl{ this, comm, std::move( options ), std::move( engineType ) } { } @@ -1631,9 +2029,10 @@ ADIOS2IOHandler::ADIOS2IOHandler( ADIOS2IOHandler::ADIOS2IOHandler( std::string path, Access at, - nlohmann::json options ) - : AbstractIOHandler( std::move( path ), at ), - m_impl{ this, std::move( options ) } + nlohmann::json options, + std::string engineType ) + : AbstractIOHandler( std::move( path ), at ) + , m_impl{ this, std::move( options ), std::move( engineType ) } { } @@ -1650,8 +2049,8 @@ ADIOS2IOHandler::ADIOS2IOHandler( std::string path, Access at, MPI_Comm comm, - nlohmann::json -) + nlohmann::json, + std::string ) : AbstractIOHandler( std::move( path ), at, comm ) { } @@ -1661,7 +2060,8 @@ ADIOS2IOHandler::ADIOS2IOHandler( ADIOS2IOHandler::ADIOS2IOHandler( std::string path, Access at, - nlohmann::json ) + nlohmann::json, + std::string ) : AbstractIOHandler( std::move( path ), at ) { } diff --git a/src/IO/ADIOS/ParallelADIOS1IOHandler.cpp b/src/IO/ADIOS/ParallelADIOS1IOHandler.cpp index 94e49ddb8d..82cb262da2 100644 --- a/src/IO/ADIOS/ParallelADIOS1IOHandler.cpp +++ b/src/IO/ADIOS/ParallelADIOS1IOHandler.cpp @@ -154,6 +154,9 @@ ParallelADIOS1IOHandlerImpl::flush() case O::OPEN_PATH: openPath(i.writable, deref_dynamic_cast< Parameter< O::OPEN_PATH > >(i.parameter.get())); break; + case O::CLOSE_PATH: + closePath(i.writable, deref_dynamic_cast< Parameter< O::CLOSE_PATH > >(i.parameter.get())); + break; case O::OPEN_DATASET: openDataset(i.writable, deref_dynamic_cast< Parameter< O::OPEN_DATASET > >(i.parameter.get())); break; @@ -190,6 +193,9 @@ ParallelADIOS1IOHandlerImpl::flush() case O::LIST_ATTS: listAttributes(i.writable, deref_dynamic_cast< Parameter< O::LIST_ATTS > >(i.parameter.get())); break; + case O::ADVANCE: + advance(i.writable, deref_dynamic_cast< Parameter< O::ADVANCE > >(i.parameter.get())); + break; case O::AVAILABLE_CHUNKS: availableChunks(i.writable, deref_dynamic_cast< Parameter< O::AVAILABLE_CHUNKS > >(i.parameter.get())); break; diff --git a/src/IO/AbstractIOHandlerHelper.cpp b/src/IO/AbstractIOHandlerHelper.cpp index 7ee072f5dd..a0ae57f2f3 100644 --- a/src/IO/AbstractIOHandlerHelper.cpp +++ b/src/IO/AbstractIOHandlerHelper.cpp @@ -53,7 +53,10 @@ namespace openPMD # endif case Format::ADIOS2: return std::make_shared< ADIOS2IOHandler >( - path, access, comm, std::move( optionsJson ) ); + path, access, comm, std::move( optionsJson ), "bp4" ); + case Format::ADIOS2_SST: + return std::make_shared< ADIOS2IOHandler >( + path, access, comm, std::move( optionsJson ), "sst" ); default: throw std::runtime_error( "Unknown file format! Did you specify a file ending?" ); @@ -82,7 +85,10 @@ namespace openPMD #if openPMD_HAVE_ADIOS2 case Format::ADIOS2: return std::make_shared< ADIOS2IOHandler >( - path, access, std::move( optionsJson ) ); + path, access, std::move( optionsJson ), "bp4" ); + case Format::ADIOS2_SST: + return std::make_shared< ADIOS2IOHandler >( + path, access, std::move( optionsJson ), "sst" ); #endif // openPMD_HAVE_ADIOS2 case Format::JSON: return std::make_shared< JSONIOHandler >( path, access ); diff --git a/src/Iteration.cpp b/src/Iteration.cpp index 3cb9e5bcd1..9ab96804ac 100644 --- a/src/Iteration.cpp +++ b/src/Iteration.cpp @@ -19,7 +19,6 @@ * If not, see . */ #include "openPMD/Iteration.hpp" - #include "openPMD/Dataset.hpp" #include "openPMD/Datatype.hpp" #include "openPMD/Series.hpp" @@ -42,11 +41,12 @@ Iteration::Iteration() setTimeUnitSI(1); } -Iteration::Iteration(Iteration const& i) - : Attributable{i}, - meshes{i.meshes}, - particles{i.particles}, - m_closed{i.m_closed} +Iteration::Iteration( Iteration const & i ) + : Attributable{ i }, + meshes{ i.meshes }, + particles{ i.particles }, + m_closed{ i.m_closed }, + m_stepStatus{ i.m_stepStatus } { IOHandler = i.IOHandler; parent = i.parent; @@ -65,6 +65,7 @@ Iteration& Iteration::operator=(Iteration const& i) IOHandler = i.IOHandler; parent = i.parent; m_closed = i.m_closed; + m_stepStatus = i.m_stepStatus; meshes.IOHandler = IOHandler; meshes.parent = this->m_writable.get(); particles.IOHandler = IOHandler; @@ -105,6 +106,8 @@ Iteration::setTimeUnitSI(double newTimeUnitSI) return *this; } +using iterator_t = Container< Iteration, uint64_t >::iterator; + Iteration & Iteration::close( bool _flush ) { @@ -113,6 +116,7 @@ Iteration::close( bool _flush ) { setAttribute< bool_type >( "closed", 1u ); } + StepStatus flag = getStepStatus(); // update close status switch( *m_closed ) { @@ -141,44 +145,30 @@ Iteration::close( bool _flush ) } if( _flush ) { - /* Find the root point [Series] of this file, - * meshesPath and particlesPath are stored there */ - Writable *w = this->parent; - while( w->parent ) - w = w->parent; - - auto s = dynamic_cast< Series* >( w->attributable ); - if( s == nullptr ) - throw std::runtime_error("[Iteration::close] Series* is a nullptr"); - - // figure out my iteration number - uint64_t index; - bool found = false; - for( auto const & pair : s->iterations ) + if( flag == StepStatus::DuringStep ) { - if( pair.second.m_writable.get() == this->m_writable.get() ) - { - found = true; - index = pair.first; - break; - } + endStep(); + setStepStatus( StepStatus::NoStep ); } - if( !found ) + else { - throw std::runtime_error( - "[Iteration::close] Iteration not found in Series." ); + // flush things manually + Series * s = &auxiliary::deref_dynamic_cast< Series >( + parent->attributable->parent->attributable ); + // figure out my iteration number + auto begin = s->indexOf( *this ); + auto end = begin; + ++end; + + s->flush_impl( begin, end ); } - std::map< uint64_t, Iteration > flushOnlyThisIteration{ - { index, *this } }; - switch( *s->m_iterationEncoding ) + } + else + { + if( flag == StepStatus::DuringStep ) { - using IE = IterationEncoding; - case IE::fileBased: - s->flushFileBased( flushOnlyThisIteration ); - break; - case IE::groupBased: - s->flushGroupBased( flushOnlyThisIteration ); - break; + throw std::runtime_error( "Using deferred Iteration::close " + "unimplemented in auto-stepping mode." ); } } return *this; @@ -209,11 +199,8 @@ Iteration::flushFileBased(std::string const& filename, uint64_t i) { /* Find the root point [Series] of this file, * meshesPath and particlesPath are stored there */ - Writable *w = this->parent; - while( w->parent ) - w = w->parent; - - auto s = dynamic_cast< Series* >( w->attributable ); + Series * s = &auxiliary::deref_dynamic_cast< Series >( + parent->attributable->parent->attributable ); if( s == nullptr ) throw std::runtime_error("[Iteration::flushFileBased] Series* is a nullptr"); @@ -281,10 +268,8 @@ Iteration::flush() { /* Find the root point [Series] of this file, * meshesPath and particlesPath are stored there */ - Writable *w = this->parent; - while( w->parent ) - w = w->parent; - auto s = dynamic_cast< Series* >(w->attributable); + Series * s = &auxiliary::deref_dynamic_cast< Series >( + parent->attributable->parent->attributable ); if( !meshes.empty() || s->containsAttribute("meshesPath") ) { @@ -346,10 +331,8 @@ Iteration::read() /* Find the root point [Series] of this file, * meshesPath and particlesPath are stored there */ - Writable *w = getWritable(this); - while( w->parent ) - w = w->parent; - auto s = dynamic_cast< Series* >(w->attributable); + Series * s = &auxiliary::deref_dynamic_cast< Series >( + parent->attributable->parent->attributable ); Parameter< Operation::LIST_PATHS > pList; std::string version = s->openPMD(); @@ -462,6 +445,109 @@ Iteration::read() readAttributes(); } +AdvanceStatus +Iteration::beginStep() +{ + using IE = IterationEncoding; + auto & series = auxiliary::deref_dynamic_cast< Series >( + parent->attributable->parent->attributable ); + // Initialize file with this to quiet warnings + // The following switch is comprehensive + Attributable * file = this; + switch( *series.m_iterationEncoding ) + { + case IE::fileBased: + file = this; + break; + case IE::groupBased: + file = &series; + break; + } + AdvanceStatus status = series.advance( + AdvanceMode::BEGINSTEP, *file, series.indexOf( *this ), *this ); + if( status != AdvanceStatus::OK ) + { + return status; + } + + // re-read -> new datasets might be available + if( *series.m_iterationEncoding == IE::groupBased && + ( this->IOHandler->m_frontendAccess == Access::READ_ONLY || + this->IOHandler->m_frontendAccess == Access::READ_WRITE ) ) + { + bool previous = series.iterations.written(); + series.iterations.written() = false; + auto oldType = this->IOHandler->m_frontendAccess; + auto newType = + const_cast< Access * >( &this->IOHandler->m_frontendAccess ); + *newType = Access::READ_WRITE; + series.readGroupBased( false ); + *newType = oldType; + series.iterations.written() = previous; + } + + return status; +} + +void +Iteration::endStep() +{ + using IE = IterationEncoding; + auto & series = auxiliary::deref_dynamic_cast< Series >( + parent->attributable->parent->attributable ); + // Initialize file with this to quiet warnings + // The following switch is comprehensive + Attributable * file = this; + switch( *series.m_iterationEncoding ) + { + case IE::fileBased: + file = this; + break; + case IE::groupBased: + file = &series; + break; + } + // @todo filebased check + series.advance( + AdvanceMode::ENDSTEP, *file, series.indexOf( *this ), *this ); +} + +StepStatus +Iteration::getStepStatus() +{ + Series * s = &auxiliary::deref_dynamic_cast< Series >( + parent->attributable->parent->attributable ); + switch( *s->m_iterationEncoding ) + { + using IE = IterationEncoding; + case IE::fileBased: + return *this->m_stepStatus; + case IE::groupBased: + return *s->m_stepStatus; + default: + throw std::runtime_error( "[Iteration] unreachable" ); + } +} + +void +Iteration::setStepStatus( StepStatus status ) +{ + Series * s = &auxiliary::deref_dynamic_cast< Series >( + parent->attributable->parent->attributable ); + switch( *s->m_iterationEncoding ) + { + using IE = IterationEncoding; + case IE::fileBased: + *this->m_stepStatus = status; + break; + case IE::groupBased: + *s->m_stepStatus = status; + break; + default: + throw std::runtime_error( "[Iteration] unreachable" ); + } +} + bool Iteration::dirtyRecursive() const { @@ -503,10 +589,10 @@ Iteration::time< long double >() const; template float Iteration::dt< float >() const; -template -double Iteration::dt< double >() const; -template -long double Iteration::dt< long double >() const; +template double +Iteration::dt< double >() const; +template long double +Iteration::dt< long double >() const; template Iteration& Iteration::setTime< float >(float time); diff --git a/src/Mesh.cpp b/src/Mesh.cpp index 54624e9893..61d1e073f8 100644 --- a/src/Mesh.cpp +++ b/src/Mesh.cpp @@ -323,9 +323,6 @@ Mesh::read() this->at(MeshRecordComponent::SCALAR).read(); } else { - written() = false; - clear_unchecked(); - written() = true; Parameter< Operation::LIST_PATHS > pList; IOHandler->enqueue(IOTask(this, pList)); IOHandler->flush(); @@ -334,6 +331,10 @@ Mesh::read() for( auto const& component : *pList.paths ) { MeshRecordComponent& rc = (*this)[component]; + if ( *rc.hasBeenRead ) + { + continue; + } pOpen.path = component; IOHandler->enqueue(IOTask(&rc, pOpen)); *rc.m_isConstant = true; @@ -347,7 +348,11 @@ Mesh::read() Parameter< Operation::OPEN_DATASET > dOpen; for( auto const& component : *dList.datasets ) { - MeshRecordComponent& rc = (*this)[component]; + MeshRecordComponent & rc = ( *this )[ component ]; + if( *rc.hasBeenRead ) + { + continue; + } dOpen.name = component; IOHandler->enqueue(IOTask(&rc, dOpen)); IOHandler->flush(); diff --git a/src/ParticleSpecies.cpp b/src/ParticleSpecies.cpp index 2340a2d5c9..f9729ac0ae 100644 --- a/src/ParticleSpecies.cpp +++ b/src/ParticleSpecies.cpp @@ -30,10 +30,6 @@ namespace openPMD void ParticleSpecies::read() { - written() = false; - clear_unchecked(); - written() = true; - /* obtain all non-scalar records */ Parameter< Operation::LIST_PATHS > pList; IOHandler->enqueue(IOTask(this, pList)); diff --git a/src/Record.cpp b/src/Record.cpp index c9878baaa5..b9f1171dfb 100644 --- a/src/Record.cpp +++ b/src/Record.cpp @@ -93,9 +93,6 @@ Record::read() this->at(RecordComponent::SCALAR).read(); } else { - written() = false; - clear_unchecked(); - written() = true; Parameter< Operation::LIST_PATHS > pList; IOHandler->enqueue(IOTask(this, pList)); IOHandler->flush(); @@ -104,6 +101,10 @@ Record::read() for( auto const& component : *pList.paths ) { RecordComponent& rc = (*this)[component]; + if ( *rc.hasBeenRead ) + { + continue; + } pOpen.path = component; IOHandler->enqueue(IOTask(&rc, pOpen)); *rc.m_isConstant = true; @@ -117,7 +118,11 @@ Record::read() Parameter< Operation::OPEN_DATASET > dOpen; for( auto const& component : *dList.datasets ) { - RecordComponent& rc = (*this)[component]; + RecordComponent & rc = ( *this )[ component ]; + if( *rc.hasBeenRead ) + { + continue; + } dOpen.name = component; IOHandler->enqueue(IOTask(&rc, dOpen)); IOHandler->flush(); diff --git a/src/RecordComponent.cpp b/src/RecordComponent.cpp index 827c1cced8..a958398c7b 100644 --- a/src/RecordComponent.cpp +++ b/src/RecordComponent.cpp @@ -187,7 +187,12 @@ RecordComponent::flush(std::string const& name) void RecordComponent::read() { + if ( *hasBeenRead ) + { + return; + } readBase(); + *hasBeenRead = true; } void diff --git a/src/Series.cpp b/src/Series.cpp index a7a674ebb9..070e24a0ad 100644 --- a/src/Series.cpp +++ b/src/Series.cpp @@ -358,18 +358,23 @@ Series::backend() const void Series::flush() { - switch( *m_iterationEncoding ) + flush_impl( iterations.begin(), iterations.end() ); +} + +ReadIterations +Series::readIterations() +{ + return { this }; +} + +WriteIterations +Series::writeIterations() +{ + if( !m_writeIterations->has_value() ) { - using IE = IterationEncoding; - case IE::fileBased: - flushFileBased( iterations ); - break; - case IE::groupBased: - flushGroupBased( iterations ); - break; + *m_writeIterations = WriteIterations( this->iterations ); } - - IOHandler->flush(); + return m_writeIterations->get(); } std::unique_ptr< Series::ParsedInput > @@ -500,19 +505,36 @@ Series::initDefaults() setSoftware( "openPMD-api", getVersion() ); } -template< typename IterationsContainer > +std::future< void > +Series::flush_impl( iterations_iterator begin, iterations_iterator end ) +{ + switch( *m_iterationEncoding ) + { + using IE = IterationEncoding; + case IE::fileBased: + flushFileBased( begin, end ); + break; + case IE::groupBased: + flushGroupBased( begin, end ); + break; + } + + return IOHandler->flush(); +} + void -Series::flushFileBased( IterationsContainer && iterationsToFlush ) +Series::flushFileBased( iterations_iterator begin, iterations_iterator end ) { - if( iterationsToFlush.empty() ) + if( end == begin ) throw std::runtime_error( "fileBased output can not be written with no iterations." ); if( IOHandler->m_frontendAccess == Access::READ_ONLY ) - for( auto & i : iterationsToFlush ) + for( auto it = begin; it != end; ++it ) { - bool const dirtyRecursive = i.second.dirtyRecursive(); - if( *i.second.m_closed == Iteration::CloseStatus::ClosedInBackend ) + bool const dirtyRecursive = it->second.dirtyRecursive(); + if( *it->second.m_closed + == Iteration::CloseStatus::ClosedInBackend ) { // file corresponding with the iteration has previously been // closed and fully flushed @@ -535,34 +557,37 @@ Series::flushFileBased( IterationsContainer && iterationsToFlush ) if( dirtyRecursive || this->dirty() ) { // openIteration() will update the close status - openIteration( i.first, i.second ); - i.second.flush(); + openIteration( it->first, it->second ); + it->second.flush(); } - if( *i.second.m_closed == Iteration::CloseStatus::ClosedInFrontend ) + if( *it->second.m_closed + == Iteration::CloseStatus::ClosedInFrontend ) { Parameter< Operation::CLOSE_FILE > fClose; - IOHandler->enqueue( IOTask( &i.second, std::move( fClose ) ) ); - *i.second.m_closed = Iteration::CloseStatus::ClosedInBackend; + IOHandler->enqueue( + IOTask( &it->second, std::move( fClose ) ) ); + *it->second.m_closed = Iteration::CloseStatus::ClosedInBackend; } IOHandler->flush(); } else { bool allDirty = dirty(); - for( auto & i : iterationsToFlush ) + for( auto it = begin; it != end; ++it ) { - bool const dirtyRecursive = i.second.dirtyRecursive(); - if( *i.second.m_closed == Iteration::CloseStatus::ClosedInBackend ) + bool const dirtyRecursive = it->second.dirtyRecursive(); + if( *it->second.m_closed + == Iteration::CloseStatus::ClosedInBackend ) { // file corresponding with the iteration has previously been // closed and fully flushed // verify that there have been no further accesses - if (!i.second.written()) + if (!it->second.written()) { throw std::runtime_error( "[Series] Closed iteration has not been written. This " - "is an internal error."); + "is an internal error." ); } if( dirtyRecursive ) { @@ -588,26 +613,21 @@ Series::flushFileBased( IterationsContainer && iterationsToFlush ) written() = false; iterations.written() = false; - std::stringstream iteration(""); - iteration << std::setw(*m_filenamePadding) - << std::setfill('0') << i.first; - std::string filename = - *m_filenamePrefix + iteration.str() + *m_filenamePostfix; - - dirty() |= i.second.dirty(); - i.second.flushFileBased(filename, i.first); + dirty() |= it->second.dirty(); + std::string filename = iterationFilename( it->first ); + it->second.flushFileBased(filename, it->first); iterations.flush( auxiliary::replace_first(basePath(), "%T/", "")); flushAttributes(); - switch( *i.second.m_closed ) + switch( *it->second.m_closed ) { using CL = Iteration::CloseStatus; case CL::Open: case CL::ClosedTemporarily: - *i.second.m_closed = CL::Open; + *it->second.m_closed = CL::Open; break; default: // keep it @@ -615,11 +635,13 @@ Series::flushFileBased( IterationsContainer && iterationsToFlush ) } } - if( *i.second.m_closed == Iteration::CloseStatus::ClosedInFrontend ) + if( *it->second.m_closed == + Iteration::CloseStatus::ClosedInFrontend ) { Parameter< Operation::CLOSE_FILE > fClose; - IOHandler->enqueue( IOTask( &i.second, std::move( fClose ) ) ); - *i.second.m_closed = Iteration::CloseStatus::ClosedInBackend; + IOHandler->enqueue( + IOTask( &it->second, std::move( fClose ) ) ); + *it->second.m_closed = Iteration::CloseStatus::ClosedInBackend; } IOHandler->flush(); @@ -632,36 +654,32 @@ Series::flushFileBased( IterationsContainer && iterationsToFlush ) } } -template void -Series::flushFileBased< std::map< uint64_t, Iteration > & >( - std::map< uint64_t, Iteration > & ); - -template< typename IterationsContainer > void -Series::flushGroupBased( IterationsContainer && iterationsToFlush ) +Series::flushGroupBased( iterations_iterator begin, iterations_iterator end ) { if( IOHandler->m_frontendAccess == Access::READ_ONLY ) - for( auto & i : iterationsToFlush ) + for( auto it = begin; it != end; ++it ) { - if( *i.second.m_closed == Iteration::CloseStatus::ClosedInBackend ) + if( *it->second.m_closed == + Iteration::CloseStatus::ClosedInBackend ) { // file corresponding with the iteration has previously been // closed and fully flushed // verify that there have been no further accesses - if( i.second.dirtyRecursive() ) + if( it->second.dirtyRecursive() ) { throw std::runtime_error( "[Series] Illegal access to iteration " + - std::to_string( i.first ) + + std::to_string( it->first ) + " that has been closed previously." ); } continue; } - i.second.flush(); - if( *i.second.m_closed == Iteration::CloseStatus::ClosedInFrontend ) + it->second.flush(); + if( *it->second.m_closed == Iteration::CloseStatus::ClosedInFrontend ) { // the iteration has no dedicated file in group-based mode - *i.second.m_closed = Iteration::CloseStatus::ClosedInBackend; + *it->second.m_closed = Iteration::CloseStatus::ClosedInBackend; } IOHandler->flush(); } @@ -674,40 +692,41 @@ Series::flushGroupBased( IterationsContainer && iterationsToFlush ) IOHandler->enqueue(IOTask(this, fCreate)); } - iterations.flush( auxiliary::replace_first( basePath(), "%T/", "" ) ); + iterations.flush(auxiliary::replace_first(basePath(), "%T/", "")); - for( auto & i : iterationsToFlush ) + for( auto it = begin; it != end; ++it ) { - if( *i.second.m_closed == Iteration::CloseStatus::ClosedInBackend ) + if( *it->second.m_closed == + Iteration::CloseStatus::ClosedInBackend ) { // file corresponding with the iteration has previously been // closed and fully flushed // verify that there have been no further accesses - if (!i.second.written()) + if (!it->second.written()) { throw std::runtime_error( "[Series] Closed iteration has not been written. This " - "is an internal error."); + "is an internal error." ); } - if( i.second.dirtyRecursive() ) + if( it->second.dirtyRecursive() ) { throw std::runtime_error( "[Series] Illegal access to iteration " + - std::to_string( i.first ) + + std::to_string( it->first ) + " that has been closed previously." ); } continue; } - if( !i.second.written() ) + if( !it->second.written() ) { - i.second.m_writable->parent = getWritable(&iterations); - i.second.parent = getWritable(&iterations); + it->second.m_writable->parent = getWritable( &iterations ); + it->second.parent = getWritable( &iterations ); } - i.second.flushGroupBased(i.first); - if( *i.second.m_closed == Iteration::CloseStatus::ClosedInFrontend ) + it->second.flushGroupBased(it->first); + if( *it->second.m_closed == Iteration::CloseStatus::ClosedInFrontend ) { // the iteration has no dedicated file in group-based mode - *i.second.m_closed = Iteration::CloseStatus::ClosedInBackend; + *it->second.m_closed = Iteration::CloseStatus::ClosedInBackend; } } @@ -716,10 +735,6 @@ Series::flushGroupBased( IterationsContainer && iterationsToFlush ) } } -template void -Series::flushGroupBased< std::map< uint64_t, Iteration > & >( - std::map< uint64_t, Iteration > & ); - void Series::flushMeshesPath() { @@ -743,7 +758,7 @@ Series::flushParticlesPath() } void -Series::readFileBased() +Series::readFileBased( ) { Parameter< Operation::OPEN_FILE > fOpen; Parameter< Operation::CLOSE_FILE > fClose; @@ -774,36 +789,44 @@ Series::readFileBased() using DT = Datatype; aRead.name = "iterationEncoding"; - IOHandler->enqueue(IOTask(this, aRead)); + IOHandler->enqueue( IOTask( this, aRead ) ); IOHandler->flush(); if( *aRead.dtype == DT::STRING ) { - std::string encoding = Attribute(*aRead.resource).get< std::string >(); + std::string encoding = + Attribute( *aRead.resource ).get< std::string >(); if( encoding == "fileBased" ) *m_iterationEncoding = IterationEncoding::fileBased; else if( encoding == "groupBased" ) { *m_iterationEncoding = IterationEncoding::groupBased; - std::cerr << "Series constructor called with iteration regex '%T' suggests loading a " - << "time series with fileBased iteration encoding. Loaded file is groupBased.\n"; - } else - throw std::runtime_error("Unknown iterationEncoding: " + encoding); - setAttribute("iterationEncoding", encoding); + std::cerr << "Series constructor called with iteration " + "regex '%T' suggests loading a " + << "time series with fileBased iteration " + "encoding. Loaded file is groupBased.\n"; + } + else + throw std::runtime_error( + "Unknown iterationEncoding: " + encoding ); + setAttribute( "iterationEncoding", encoding ); } else - throw std::runtime_error("Unexpected Attribute datatype for 'iterationEncoding'"); + throw std::runtime_error( "Unexpected Attribute datatype " + "for 'iterationEncoding'" ); aRead.name = "iterationFormat"; - IOHandler->enqueue(IOTask(this, aRead)); + IOHandler->enqueue( IOTask( this, aRead ) ); IOHandler->flush(); if( *aRead.dtype == DT::STRING ) { written() = false; - setIterationFormat(Attribute(*aRead.resource).get< std::string >()); + setIterationFormat( + Attribute( *aRead.resource ).get< std::string >() ); written() = true; } else - throw std::runtime_error("Unexpected Attribute datatype for 'iterationFormat'"); + throw std::runtime_error( + "Unexpected Attribute datatype for 'iterationFormat'" ); read(); IOHandler->enqueue(IOTask(this, fClose)); @@ -837,54 +860,51 @@ Series::readFileBased() } void -Series::readGroupBased() +Series::readGroupBased( bool do_init ) { Parameter< Operation::OPEN_FILE > fOpen; fOpen.name = *m_name; IOHandler->enqueue(IOTask(this, fOpen)); IOHandler->flush(); - readBase(); - - using DT = Datatype; - Parameter< Operation::READ_ATT > aRead; - aRead.name = "iterationEncoding"; - IOHandler->enqueue(IOTask(this, aRead)); - IOHandler->flush(); - if( *aRead.dtype == DT::STRING ) + if( do_init ) { - std::string encoding = Attribute(*aRead.resource).get< std::string >(); - if( encoding == "groupBased" ) - *m_iterationEncoding = IterationEncoding::groupBased; - else if( encoding == "fileBased" ) + readBase(); + + using DT = Datatype; + Parameter< Operation::READ_ATT > aRead; + aRead.name = "iterationEncoding"; + IOHandler->enqueue(IOTask(this, aRead)); + IOHandler->flush(); + if( *aRead.dtype == DT::STRING ) { - *m_iterationEncoding = IterationEncoding::fileBased; - std::cerr << "Series constructor called with explicit iteration suggests loading a " - << "single file with groupBased iteration encoding. Loaded file is fileBased.\n"; - } else - throw std::runtime_error("Unknown iterationEncoding: " + encoding); - setAttribute("iterationEncoding", encoding); - } - else - throw std::runtime_error("Unexpected Attribute datatype for 'iterationEncoding'"); + std::string encoding = Attribute(*aRead.resource).get< std::string >(); + if( encoding == "groupBased" ) + *m_iterationEncoding = IterationEncoding::groupBased; + else if( encoding == "fileBased" ) + { + *m_iterationEncoding = IterationEncoding::fileBased; + std::cerr << "Series constructor called with explicit iteration suggests loading a " + << "single file with groupBased iteration encoding. Loaded file is fileBased.\n"; + } else + throw std::runtime_error("Unknown iterationEncoding: " + encoding); + setAttribute("iterationEncoding", encoding); + } + else + throw std::runtime_error("Unexpected Attribute datatype for 'iterationEncoding'"); - aRead.name = "iterationFormat"; - IOHandler->enqueue(IOTask(this, aRead)); - IOHandler->flush(); - if( *aRead.dtype == DT::STRING ) - { - written() = false; - setIterationFormat(Attribute(*aRead.resource).get< std::string >()); - written() = true; + aRead.name = "iterationFormat"; + IOHandler->enqueue(IOTask(this, aRead)); + IOHandler->flush(); + if( *aRead.dtype == DT::STRING ) + { + written() = false; + setIterationFormat(Attribute(*aRead.resource).get< std::string >()); + written() = true; + } + else + throw std::runtime_error("Unexpected Attribute datatype for 'iterationFormat'"); } - else - throw std::runtime_error("Unexpected Attribute datatype for 'iterationFormat'"); - - /* do not use the public checked version - * at this point we can guarantee clearing the container won't break anything */ - written() = false; - iterations.clear_unchecked(); - written() = true; read(); } @@ -975,6 +995,7 @@ Series::read() throw std::runtime_error("Unknown openPMD version - " + version); IOHandler->enqueue(IOTask(&iterations, pOpen)); + readAttributes(); iterations.readAttributes(); /* obtain all paths inside the basepath (i.e. all iterations) */ @@ -985,26 +1006,123 @@ Series::read() for( auto const& it : *pList.paths ) { Iteration& i = iterations[std::stoull(it)]; + if ( i.closedByWriter( ) ) + { + continue; + } pOpen.path = it; IOHandler->enqueue(IOTask(&i, pOpen)); i.read(); } +} - readAttributes(); +std::string +Series::iterationFilename( uint64_t i ) +{ + std::stringstream iteration( "" ); + iteration << std::setw( *m_filenamePadding ) << std::setfill( '0' ) << i; + return *m_filenamePrefix + iteration.str() + *m_filenamePostfix; +} + +Series::iterations_iterator +Series::indexOf( Iteration const & iteration ) +{ + for( auto it = iterations.begin(); it != iterations.end(); ++it ) + { + if( it->second.m_writable.get() == iteration.m_writable.get() ) + { + return it; + } + } + throw std::runtime_error( + "[Iteration::close] Iteration not found in Series." ); +} + +AdvanceStatus +Series::advance( + AdvanceMode mode, + Attributable & file, + iterations_iterator begin, + Iteration & iteration ) +{ + auto end = begin; + ++end; + /* + * @todo By calling flushFileBased/GroupBased, we do not propagate tasks to + * the backend yet. We will append ADVANCE and CLOSE_FILE tasks + * manually. In order to avoid having them automatically appended by + * the flush*Based methods, set CloseStatus to Open for now. + */ + Iteration::CloseStatus oldCloseStatus = *iteration.m_closed; + if( oldCloseStatus == Iteration::CloseStatus::ClosedInFrontend ) + { + *iteration.m_closed = Iteration::CloseStatus::Open; + } + + switch( *m_iterationEncoding ) + { + using IE = IterationEncoding; + case IE::groupBased: + flushGroupBased( begin, end ); + break; + case IE::fileBased: + flushFileBased( begin, end ); + break; + } + *iteration.m_closed = oldCloseStatus; + + Parameter< Operation::ADVANCE > param; + param.mode = mode; + IOTask task( &file, param ); + IOHandler->enqueue( task ); + + + if( oldCloseStatus == Iteration::CloseStatus::ClosedInFrontend && + mode == AdvanceMode::ENDSTEP ) + { + using IE = IterationEncoding; + switch( *m_iterationEncoding ) + { + case IE::fileBased: + { + if( *iteration.m_closed != + Iteration::CloseStatus::ClosedTemporarily ) + { + Parameter< Operation::CLOSE_FILE > fClose; + IOHandler->enqueue( + IOTask( &iteration, std::move( fClose ) ) ); + } + *iteration.m_closed = Iteration::CloseStatus::ClosedInBackend; + break; + } + case IE::groupBased: + { + // We can now put some groups to rest + Parameter< Operation::CLOSE_PATH > fClose; + IOHandler->enqueue( IOTask( &iteration, std::move( fClose ) ) ); + // In group-based iteration layout, files are + // not closed on a per-iteration basis + // We will treat it as such nonetheless + *iteration.m_closed = Iteration::CloseStatus::ClosedInBackend; + } + break; + } + } + + // We cannot call Series::flush now, since the IO handler is still filled + // from calling flush(Group|File)based, but has not been emptied yet + // Do that manually + IOHandler->flush(); + + return *param.status; } void Series::openIteration( uint64_t index, Iteration iteration ) { // open the iteration's file again - std::stringstream nameBuilder( "" ); - nameBuilder << std::setw( *m_filenamePadding ) << std::setfill( '0' ) - << index; - std::string filename = - *m_filenamePrefix + nameBuilder.str() + *m_filenamePostfix; - Parameter< Operation::OPEN_FILE > fOpen; - fOpen.name = filename; + fOpen.name = iterationFilename( index ); IOHandler->enqueue( IOTask( this, fOpen ) ); /* open base path */ @@ -1041,6 +1159,7 @@ namespace case Format::HDF5: case Format::ADIOS1: case Format::ADIOS2: + case Format::ADIOS2_SST: case Format::JSON: return auxiliary::replace_last(filename, suffix(f), ""); default: @@ -1082,6 +1201,16 @@ namespace nameReg += +")" + postfix + ".bp$"; return buildMatcher(nameReg); } + case Format::ADIOS2_SST: + { + std::string nameReg = "^" + prefix + "([[:digit:]]"; + if( padding != 0 ) + nameReg += "{" + std::to_string(padding) + "}"; + else + nameReg += "+"; + nameReg += + ")" + postfix + ".sst$"; + return buildMatcher(nameReg); + } case Format::JSON: { std::string nameReg = "^" + prefix + "([[:digit:]]"; if (padding != 0) @@ -1095,5 +1224,193 @@ namespace return [](std::string const &) -> std::tuple { return std::tuple{false, 0}; }; } } -} // namespace +} // namespace [anonymous] + +SeriesIterator::SeriesIterator() : m_series() +{ +} + +SeriesIterator::SeriesIterator( Series * series ) : m_series( series ) +{ + auto it = series->iterations.begin(); + if( it == series->iterations.end() ) + { + *this = end(); + return; + } + else + { + auto status = it->second.beginStep(); + if( status == AdvanceStatus::OVER ) + { + *this = end(); + return; + } + it->second.setStepStatus( StepStatus::DuringStep ); + } + m_currentIteration = it->first; +} + +SeriesIterator & +SeriesIterator::operator++() +{ + if( !m_series.has_value() ) + { + *this = end(); + return *this; + } + Series & series = *m_series.get(); + auto & iterations = series.iterations; + auto & currentIteration = iterations[ m_currentIteration ]; + if( !currentIteration.closed() ) + { + currentIteration.close(); + } + switch( *series.m_iterationEncoding ) + { + using IE = IterationEncoding; + case IE::groupBased: + { + // since we are in group-based iteration layout, it does not + // matter which iteration we begin a step upon + AdvanceStatus status = currentIteration.beginStep(); + if( status == AdvanceStatus::OVER ) + { + *this = end(); + return *this; + } + currentIteration.setStepStatus( StepStatus::DuringStep ); + break; + } + default: + break; + } + auto it = iterations.find( m_currentIteration ); + auto itEnd = iterations.end(); + if( it == itEnd ) + { + *this = end(); + return *this; + } + ++it; + if( it == itEnd ) + { + *this = end(); + return *this; + } + m_currentIteration = it->first; + switch( *series.m_iterationEncoding ) + { + using IE = IterationEncoding; + case IE::fileBased: + { + auto & iteration = series.iterations[ m_currentIteration ]; + AdvanceStatus status = iteration.beginStep(); + if( status == AdvanceStatus::OVER ) + { + *this = end(); + return *this; + } + iteration.setStepStatus( StepStatus::DuringStep ); + break; + } + default: + break; + } + return *this; +} + +IndexedIteration +SeriesIterator::operator*() +{ + return IndexedIteration( + m_series.get()->iterations[ m_currentIteration ], m_currentIteration ); +} + +bool +SeriesIterator::operator==( SeriesIterator const & other ) const +{ + return this->m_currentIteration == other.m_currentIteration && + this->m_series.has_value() == other.m_series.has_value(); +} + +bool +SeriesIterator::operator!=( SeriesIterator const & other ) const +{ + return !operator==( other ); +} + +SeriesIterator +SeriesIterator::end() +{ + return {}; +} + +ReadIterations::ReadIterations( Series * series ) : m_series( series ) +{ +} + +ReadIterations::iterator_t +ReadIterations::begin() +{ + return iterator_t{ m_series }; +} + +ReadIterations::iterator_t +ReadIterations::end() +{ + return SeriesIterator::end(); +} + +WriteIterations::SharedResources::SharedResources( iterations_t _iterations ) + : iterations( std::move( _iterations ) ) +{ +} + +WriteIterations::SharedResources::~SharedResources() +{ + if( currentlyOpen.has_value() ) + { + auto lastIterationIndex = currentlyOpen.get(); + auto & lastIteration = iterations.at( lastIterationIndex ); + if( !lastIteration.closed() ) + { + lastIteration.close(); + } + } +} + +WriteIterations::WriteIterations( iterations_t iterations ) + : shared{ std::make_shared< SharedResources >( std::move( iterations ) ) } +{ +} + +WriteIterations::mapped_type & +WriteIterations::operator[]( key_type const & key ) +{ + // make a copy + // explicit cast so MSVC can figure out how to do it correctly + return operator[]( static_cast< key_type && >( key_type{ key } ) ); +} +WriteIterations::mapped_type & +WriteIterations::operator[]( key_type && key ) +{ + if( shared->currentlyOpen.has_value() ) + { + auto lastIterationIndex = shared->currentlyOpen.get(); + auto & lastIteration = shared->iterations.at( lastIterationIndex ); + if( lastIterationIndex != key && !lastIteration.closed() ) + { + lastIteration.close(); + } + } + shared->currentlyOpen = key; + auto & res = shared->iterations[ std::move( key ) ]; + if( res.getStepStatus() == StepStatus::NoStep ) + { + res.beginStep(); + res.setStepStatus( StepStatus::DuringStep ); + } + return res; +} } // namespace openPMD diff --git a/src/backend/MeshRecordComponent.cpp b/src/backend/MeshRecordComponent.cpp index 88de3c1de7..af3745f158 100644 --- a/src/backend/MeshRecordComponent.cpp +++ b/src/backend/MeshRecordComponent.cpp @@ -32,6 +32,10 @@ MeshRecordComponent::MeshRecordComponent() void MeshRecordComponent::read() { + if ( *hasBeenRead ) + { + return; + } using DT = Datatype; Parameter< Operation::READ_ATT > aRead; @@ -55,6 +59,7 @@ MeshRecordComponent::read() throw std::runtime_error( "Unexpected Attribute datatype for 'position'"); readBase(); + *hasBeenRead = true; } template< typename T > diff --git a/src/binding/python/Helper.cpp b/src/binding/python/Helper.cpp index d28423f563..7a3707e9b4 100644 --- a/src/binding/python/Helper.cpp +++ b/src/binding/python/Helper.cpp @@ -35,7 +35,7 @@ using namespace openPMD; void init_Helper(py::module &m) { m.def("list_series", - [](Series const & series, bool const longer) { + [](Series & series, bool const longer) { std::stringstream s; helper::listSeries( series, longer, s ); py::print(s.str()); diff --git a/src/binding/python/Series.cpp b/src/binding/python/Series.cpp index 21e25fe46e..7772b89eda 100644 --- a/src/binding/python/Series.cpp +++ b/src/binding/python/Series.cpp @@ -55,6 +55,29 @@ using namespace openPMD; void init_Series(py::module &m) { + + using iterations_key_t = decltype(Series::iterations)::key_type; + py::class_(m, "WriteIterations") + .def("__getitem__", + [](WriteIterations writeIterations, iterations_key_t key){ + return writeIterations[key]; + }, + // keep container alive while iterator exists + py::keep_alive<0, 1>()) + ; + py::class_(m, "IndexedIteration") + .def_readonly( + "iteration_index", &IndexedIteration::iterationIndex) + ; + py::class_(m, "ReadIterations") + .def("__iter__", [](ReadIterations & readIterations) { + return py::make_iterator( + readIterations.begin(), readIterations.end()); + }, + // keep handle alive while iterator exists + py::keep_alive<0, 1>()) + ; + py::class_(m, "Series") .def(py::init(), @@ -161,5 +184,8 @@ void init_Series(py::module &m) { py::return_value_policy::reference, // garbage collection: return value must be freed before Series py::keep_alive<1, 0>()) + .def("read_iterations", &Series::readIterations, py::keep_alive<0, 1>()) + .def("write_iterations", + &Series::writeIterations, py::keep_alive<0, 1>()) ; } diff --git a/src/helper/list_series.cpp b/src/helper/list_series.cpp index 81c9a0fe91..c77983cddf 100644 --- a/src/helper/list_series.cpp +++ b/src/helper/list_series.cpp @@ -37,7 +37,7 @@ namespace helper { std::ostream & listSeries( - Series const & series, + Series & series, bool const longer, std::ostream & out ) @@ -77,21 +77,21 @@ namespace helper if( longer ) out << " all iterations: "; - for( auto const& i : series.iterations ) { + for( auto const& i : series.readIterations() ) { if( longer ) - out << i.first << " "; + out << i.iterationIndex << " "; // find unique record names std::transform( - i.second.meshes.begin(), - i.second.meshes.end(), + i.meshes.begin(), + i.meshes.end(), std::inserter( meshes, meshes.end() ), []( std::pair< std::string, Mesh > const & p ) { return p.first; } ); std::transform( - i.second.particles.begin(), - i.second.particles.end(), + i.particles.begin(), + i.particles.end(), std::inserter( particles, particles.end() ), []( std::pair< std::string, ParticleSpecies > const & p ) { return p.first; } diff --git a/test/ParallelIOTest.cpp b/test/ParallelIOTest.cpp index 4460c019eb..69ce92d3a7 100644 --- a/test/ParallelIOTest.cpp +++ b/test/ParallelIOTest.cpp @@ -524,3 +524,102 @@ TEST_CASE( "close_iteration_test", "[parallel]" ) } } #endif + +#if openPMD_HAVE_ADIOS2 && openPMD_HAVE_MPI + +void +adios2_streaming() +{ + int size{ -1 }; + int rank{ -1 }; + MPI_Comm_size( MPI_COMM_WORLD, &size ); + MPI_Comm_rank( MPI_COMM_WORLD, &rank ); + if( auxiliary::getEnvString( "OPENPMD_BP_BACKEND", "NOT_SET" ) == "ADIOS1" ) + { + // run this test for ADIOS2 only + return; + } + + if( size < 2 || rank > 1 ) + { + return; + } + + constexpr size_t extent = 100; + + if( rank == 0 ) + { + // write + Series writeSeries( + "../samples/adios2_stream.sst", Access::CREATE ); + auto iterations = writeSeries.writeIterations(); + for( size_t i = 0; i < 10; ++i ) + { + auto iteration = iterations[ i ]; + auto E_x = iteration.meshes[ "E" ][ "x" ]; + E_x.resetDataset( + openPMD::Dataset( openPMD::Datatype::INT, { extent } ) ); + std::vector< int > data( extent, i ); + E_x.storeChunk( data, { 0 }, { extent } ); + // we encourage manually closing iterations, but it should + // not matter so let's do the switcharoo for this test + if( i % 2 == 0 ) + { + writeSeries.flush(); + } + else + { + iteration.close(); + } + } + } + else if( rank == 1 ) + { + // read + // it should be possible to select the sst engine via file ending or + // via JSON without difference + std::string options = R"( + { + "adios2": { + "engine": { + "type": "SST" + } + } + } + )"; + + Series readSeries( + "../samples/adios2_stream.bp", Access::READ_ONLY, options ); + + size_t last_iteration_index = 0; + for( auto iteration : readSeries.readIterations() ) + { + auto E_x = iteration.meshes[ "E" ][ "x" ]; + REQUIRE( E_x.getDimensionality() == 1 ); + REQUIRE( E_x.getExtent()[ 0 ] == extent ); + auto chunk = E_x.loadChunk< int >( { 0 }, { extent } ); + // we encourage manually closing iterations, but it should + // not matter so let's do the switcharoo for this test + if( last_iteration_index % 2 == 0 ) + { + readSeries.flush(); + } + else + { + iteration.close(); + } + for( size_t i = 0; i < extent; ++i ) + { + REQUIRE( chunk.get()[ i ] == iteration.iterationIndex ); + } + last_iteration_index = iteration.iterationIndex; + } + REQUIRE( last_iteration_index == 9 ); + } +} + +TEST_CASE( "adios2_streaming", "[pseudoserial][adios2]" ) +{ + adios2_streaming(); +} +#endif \ No newline at end of file diff --git a/test/SerialIOTest.cpp b/test/SerialIOTest.cpp index 8e11624e97..288f80bcf3 100644 --- a/test/SerialIOTest.cpp +++ b/test/SerialIOTest.cpp @@ -339,7 +339,8 @@ TEST_CASE( "close_iteration_throws_test", "[serial]" ) * Test that the openPMD API detects that case and throws. */ { - Series series( "close_iteration_throws_1.bp", Access::CREATE ); + Series series( + "../samples/close_iteration_throws_1.bp", Access::CREATE ); auto it0 = series.iterations[ 0 ]; auto E_x = it0.meshes[ "E" ][ "x" ]; E_x.resetDataset( { Datatype::INT, { 5 } } ); @@ -353,7 +354,8 @@ TEST_CASE( "close_iteration_throws_test", "[serial]" ) REQUIRE_THROWS( series.flush() ); } { - Series series( "close_iteration_throws_2.bp", Access::CREATE ); + Series series( + "../samples/close_iteration_throws_2.bp", Access::CREATE ); auto it0 = series.iterations[ 0 ]; auto E_x = it0.meshes[ "E" ][ "x" ]; E_x.resetDataset( { Datatype::INT, { 5 } } ); @@ -367,7 +369,8 @@ TEST_CASE( "close_iteration_throws_test", "[serial]" ) REQUIRE_THROWS( series.flush() ); } { - Series series( "close_iteration_throws_3.bp", Access::CREATE ); + Series series( + "../samples/close_iteration_throws_3.bp", Access::CREATE ); auto it0 = series.iterations[ 0 ]; auto E_x = it0.meshes[ "E" ][ "x" ]; E_x.resetDataset( { Datatype::INT, { 5 } } ); @@ -1023,7 +1026,7 @@ TEST_CASE( "write_test", "[serial]" ) void test_complex(const std::string & backend) { { - Series o = Series("../samples/serial_write_complex." + backend, AccessType::CREATE); + Series o = Series("../samples/serial_write_complex." + backend, Access::CREATE); o.setAttribute("lifeIsComplex", std::complex(4.56, 7.89)); o.setAttribute("butComplexFloats", std::complex(42.3, -99.3)); if( backend != "bp" ) @@ -1062,7 +1065,7 @@ void test_complex(const std::string & backend) { //! @todo clarify that complex data is not N+1 data in JSON if( backend != "json" ) { - Series i = Series("../samples/serial_write_complex." + backend, AccessType::READ_ONLY); + Series i = Series("../samples/serial_write_complex." + backend, Access::READ_ONLY); REQUIRE(i.getAttribute("lifeIsComplex").get< std::complex >() == std::complex(4.56, 7.89)); REQUIRE(i.getAttribute("butComplexFloats").get< std::complex >() == std::complex(42.3, -99.3)); if( backend != "bp" ) { @@ -2965,4 +2968,188 @@ TEST_CASE( "serial_adios2_json_config", "[serial][adios2]" ) read( "../samples/jsonConfiguredBP3.bp", readConfigBP3 ); read( "../samples/jsonConfiguredBP4.bp", readConfigBP4 ); } + +void +bp4_steps( std::string const & file, std::string const & options_write, std::string const & options_read ) +{ + { + Series writeSeries( file, Access::CREATE, options_write ); + auto iterations = writeSeries.writeIterations(); + for( size_t i = 0; i < 10; ++i ) + { + auto iteration = iterations[ i ]; + auto E_x = iteration.meshes[ "E" ][ "x" ]; + E_x.resetDataset( + openPMD::Dataset( openPMD::Datatype::INT, { 10 } ) ); + std::vector< int > data( 10, i ); + E_x.storeChunk( data, { 0 }, { 10 } ); + iteration.close(); + } + } + + Series readSeries( file, Access::READ_ONLY, options_read ); + + size_t last_iteration_index = 0; + for( auto iteration : readSeries.readIterations() ) + { + auto E_x = iteration.meshes[ "E" ][ "x" ]; + REQUIRE( E_x.getDimensionality() == 1 ); + REQUIRE( E_x.getExtent()[ 0 ] == 10 ); + auto chunk = E_x.loadChunk< int >( { 0 }, { 10 } ); + iteration.close(); // @todo replace with ::close() + for( size_t i = 0; i < 10; ++i ) + { + REQUIRE( chunk.get()[ i ] == iteration.iterationIndex ); + } + last_iteration_index = iteration.iterationIndex; + } + REQUIRE( last_iteration_index == 9 ); +} + +TEST_CASE( "bp4_steps", "[serial][adios2]" ) +{ + std::string useSteps = R"( + { + "adios2": { + "engine": { + "type": "bp4", + "usesteps": true + } + } + } + )"; + std::string dontUseSteps = R"( + { + "adios2": { + "engine": { + "type": "bp4", + "usesteps": false + } + } + } + )"; + // sing the yes no song + bp4_steps( "../samples/bp4steps_yes_yes.bp", useSteps, useSteps ); + bp4_steps( "../samples/bp4steps_no_yes.bp", dontUseSteps, useSteps ); + bp4_steps( "../samples/bp4steps_yes_no.bp", useSteps, dontUseSteps ); + bp4_steps( "../samples/bp4steps_no_no.bp", dontUseSteps, dontUseSteps ); + bp4_steps("../samples/bp4steps_default.bp", "{}", "{}"); +} #endif + +void +serial_iterator( std::string const & file ) +{ + constexpr Extent::value_type extent = 1000; + { + Series writeSeries( file, Access::CREATE ); + auto iterations = writeSeries.writeIterations(); + for( size_t i = 0; i < 10; ++i ) + { + auto iteration = iterations[ i ]; + auto E_x = iteration.meshes[ "E" ][ "x" ]; + E_x.resetDataset( + openPMD::Dataset( openPMD::Datatype::INT, { 1000 } ) ); + std::vector< int > data( 1000, i ); + E_x.storeChunk( data, { 0 }, { 1000 } ); + iteration.close(); + } + } + + Series readSeries( file, Access::READ_ONLY ); + + size_t last_iteration_index = 0; + for( auto iteration : readSeries.readIterations() ) + { + auto E_x = iteration.meshes[ "E" ][ "x" ]; + REQUIRE( E_x.getDimensionality() == 1 ); + REQUIRE( E_x.getExtent()[ 0 ] == extent ); + auto chunk = E_x.loadChunk< int >( { 0 }, { extent } ); + iteration.close(); + for( size_t i = 0; i < extent; ++i ) + { + REQUIRE( chunk.get()[ i ] == iteration.iterationIndex ); + } + last_iteration_index = iteration.iterationIndex; + } + REQUIRE( last_iteration_index == 9 ); +} + +TEST_CASE( "serial_iterator", "[serial][adios2]" ) +{ + for( auto const & t : getFileExtensions() ) + { + serial_iterator( "../samples/serial_iterator_filebased_%T." + t ); + serial_iterator( "../samples/serial_iterator_groupbased." + t ); + } +} + +void +iterate_nonstreaming_series( std::string const & file ) +{ + constexpr size_t extent = 100; + { + Series writeSeries( file, Access::CREATE ); + // use conventional API to write iterations + auto iterations = writeSeries.iterations; + for( size_t i = 0; i < 10; ++i ) + { + auto iteration = iterations[ i ]; + auto E_x = iteration.meshes[ "E" ][ "x" ]; + E_x.resetDataset( + openPMD::Dataset( openPMD::Datatype::INT, { extent } ) ); + std::vector< int > data( extent, i ); + E_x.storeChunk( data, { 0 }, { extent } ); + // we encourage manually closing iterations, but it should not matter + // so let's do the switcharoo for this test + if( i % 2 == 0 ) + { + writeSeries.flush(); + } + else + { + iteration.close(); + } + } + } + + Series readSeries( file, Access::READ_ONLY ); + + size_t last_iteration_index = 0; + // conventionally written Series must be readable with streaming-aware API! + for( auto iteration : readSeries.readIterations() ) + { + auto E_x = iteration.meshes[ "E" ][ "x" ]; + REQUIRE( E_x.getDimensionality() == 1 ); + REQUIRE( E_x.getExtent()[ 0 ] == extent ); + auto chunk = E_x.loadChunk< int >( { 0 }, { extent } ); + // we encourage manually closing iterations, but it should not matter + // so let's do the switcharoo for this test + if( last_iteration_index % 2 == 0 ) + { + readSeries.flush(); + } + else + { + iteration.close(); + } + + for( size_t i = 0; i < extent; ++i ) + { + REQUIRE( chunk.get()[ i ] == iteration.iterationIndex ); + } + last_iteration_index = iteration.iterationIndex; + } + REQUIRE( last_iteration_index == 9 ); +} + +TEST_CASE( "iterate_nonstreaming_series", "[serial][adios2]" ) +{ + for( auto const & t : getFileExtensions() ) + { + iterate_nonstreaming_series( + "../samples/iterate_nonstreaming_series_filebased_%T." + t ); + iterate_nonstreaming_series( + "../samples/iterate_nonstreaming_series_groupbased." + t ); + } +} diff --git a/test/python/unittest/API/APITest.py b/test/python/unittest/API/APITest.py index b039a73b11..aa983b1904 100644 --- a/test/python/unittest/API/APITest.py +++ b/test/python/unittest/API/APITest.py @@ -1460,7 +1460,7 @@ def testFieldRecord(self): def makeCloseIterationRoundTrip(self, file_ending): # write series = io.Series( - "unittest_closeIteration_%T." + file_ending, + "../samples/unittest_closeIteration_%T." + file_ending, io.Access_Type.create ) DS = io.Dataset @@ -1477,7 +1477,7 @@ def makeCloseIterationRoundTrip(self, file_ending): # not supported in ADIOS1: can only open one ADIOS1 series at a time if not is_adios1: read = io.Series( - "unittest_closeIteration_%T." + file_ending, + "../samples/unittest_closeIteration_%T." + file_ending, io.Access_Type.read_only ) it0 = read.iterations[0] @@ -1498,7 +1498,7 @@ def makeCloseIterationRoundTrip(self, file_ending): if not is_adios1: read = io.Series( - "unittest_closeIteration_%T." + file_ending, + "../samples/unittest_closeIteration_%T." + file_ending, io.Access_Type.read_only ) it1 = read.iterations[1] @@ -1515,6 +1515,66 @@ def testCloseIteration(self): for ext in io.file_extensions: self.makeCloseIterationRoundTrip(ext) + def makeIteratorRoundTrip(self, backend, file_ending): + # write + jsonConfig = """ +{ + "adios2": { + "engine": { + "type": "bp4", + "usesteps": true + } + } +} +""" + series = io.Series( + "../samples/unittest_serialIterator." + file_ending, + io.Access_Type.create, + jsonConfig + ) + DS = io.Dataset + data = np.array([2, 4, 6, 8], dtype=np.dtype("int")) + extent = [4] + + for i in range(10): + it = series.write_iterations()[i] + E_x = it.meshes["E"]["x"] + E_x.reset_dataset(DS(np.dtype("int"), extent)) + E_x.store_chunk(data, [0], extent) + it.close() + del it + + del series + + # read + + read = io.Series( + "../samples/unittest_serialIterator." + file_ending, + io.Access_Type.read_only, + jsonConfig + ) + for it in read.read_iterations(): + lastIterationIndex = it.iteration_index + E_x = it.meshes["E"]["x"] + chunk = E_x.load_chunk([0], extent) + it.close() + + for i in range(len(data)): + self.assertEqual(data[i], chunk[i]) + del read + self.assertEqual(lastIterationIndex, 9) + + def testIterator(self): + backend_filesupport = { + 'json': 'json', + 'hdf5': 'h5', + 'adios1': 'bp', + 'adios2': 'bp' + } + for b in io.variants: + if io.variants[b] is True and b in backend_filesupport: + self.makeIteratorRoundTrip(b, backend_filesupport[b]) + def makeAvailableChunksRoundTrip(self, ext): if ext == "h5": return