diff --git a/docs/source/details/backendconfig.rst b/docs/source/details/backendconfig.rst index 5d8f59c4e1..411b38cd12 100644 --- a/docs/source/details/backendconfig.rst +++ b/docs/source/details/backendconfig.rst @@ -1,7 +1,7 @@ .. _backendconfig: -Backend-Specific Configuration -============================== +JSON configuration +================== While the openPMD API intends to be a backend-*independent* implementation of the openPMD standard, it is sometimes useful to pass configuration parameters to the specific backend in use. :ref:`For each backend `, configuration options can be passed via a JSON-formatted string or via environment variables. @@ -33,6 +33,19 @@ For a consistent user interface, backends shall follow the following rules: Backends should define clearly which keys are applicable to datasets and which are not. +Backend-independent JSON configuration +-------------------------------------- + +The key ``defer_iteration_parsing`` can be used to optimize the process of opening an openPMD Series. +By default, a Series is parsed eagerly, i.e. opening a Series implies reading all available iterations. +Especially when a Series has many iterations, this can be a costly operation and users may wish to defer parsing of iterations to a later point adding ``{"defer_iteration_parsing": true}`` to their JSON configuration. + +When parsing non-eagerly, each iteration needs to be explicitly opened with ``Iteration::open()`` before accessing. +(Notice that ``Iteration::open()`` is generally recommended to be used in parallel contexts to avoid parallel file accessing hazards). +Using the Streaming API (i.e. ``SeriesImpl::readIteration()``) will do this automatically. +Parsing eagerly might be very expensive for a Series with many iterations, but will avoid bugs by forgotten calls to ``Iteration::open()``. +In complex environments, calling ``Iteration::open()`` on an already open environment does no harm (and does not incur additional runtime cost for additional ``open()`` calls). + Configuration Structure per Backend ----------------------------------- diff --git a/include/openPMD/IO/AbstractIOHandlerHelper.hpp b/include/openPMD/IO/AbstractIOHandlerHelper.hpp index f93962c506..42b03be068 100644 --- a/include/openPMD/IO/AbstractIOHandlerHelper.hpp +++ b/include/openPMD/IO/AbstractIOHandlerHelper.hpp @@ -20,10 +20,10 @@ */ #pragma once + #include "openPMD/config.hpp" #include "openPMD/IO/AbstractIOHandler.hpp" - namespace openPMD { #if openPMD_HAVE_MPI @@ -36,15 +36,18 @@ namespace openPMD * @param comm MPI communicator used for IO. * @param options JSON-formatted option string, to be interpreted by * the backend. + * @tparam JSON Substitute for nlohmann::json. Templated to avoid + including nlohmann::json in a .hpp file. * @return Smart pointer to created IOHandler. */ +template< typename JSON > std::shared_ptr< AbstractIOHandler > createIOHandler( std::string path, Access access, Format format, MPI_Comm comm, - std::string const & options = "{}" ); + JSON options ); #endif /** Construct an appropriate specific IOHandler for the desired IO mode. @@ -56,12 +59,22 @@ createIOHandler( * @param format Format describing the IO backend of the desired handler. * @param options JSON-formatted option string, to be interpreted by * the backend. + * @tparam JSON Substitute for nlohmann::json. Templated to avoid + including nlohmann::json in a .hpp file. * @return Smart pointer to created IOHandler. */ +template< typename JSON > std::shared_ptr< AbstractIOHandler > createIOHandler( std::string path, Access access, Format format, - std::string const & options = "{}" ); + JSON options = JSON() ); + +// version without configuration to use in AuxiliaryTest +std::shared_ptr< AbstractIOHandler > +createIOHandler( + std::string path, + Access access, + Format format ); } // namespace openPMD diff --git a/include/openPMD/Iteration.hpp b/include/openPMD/Iteration.hpp index 2a09f520dc..45fe494ef9 100644 --- a/include/openPMD/Iteration.hpp +++ b/include/openPMD/Iteration.hpp @@ -20,6 +20,7 @@ */ #pragma once +#include "openPMD/auxiliary/Option.hpp" #include "openPMD/auxiliary/Variant.hpp" #include "openPMD/backend/Attributable.hpp" #include "openPMD/backend/Container.hpp" @@ -153,10 +154,35 @@ class Iteration : public LegacyAttributable private: Iteration(); + struct DeferredParseAccess + { + std::string index; + bool fileBased = false; + std::string filename; + }; + void flushFileBased(std::string const&, uint64_t); void flushGroupBased(uint64_t); void flush(); + void deferParseAccess( DeferredParseAccess ); + /* + * Control flow for read(), readFileBased(), readGroupBased() and + * read_impl(): + * read() is called as the entry point. File-based and group-based + * iteration layouts need to be parsed slightly differently: + * In file-based iteration layout, each iteration's file also contains + * attributes for the /data group. In group-based layout, those have + * already been parsed during opening of the Series. + * Hence, read() will call either readFileBased() or readGroupBased() to + * allow for those different control flows. + * Finally, read_impl() is called which contains the common parsing + * logic for an iteration. + * + */ void read(); + void readFileBased( std::string filePath, std::string const & groupPath ); + void readGroupBased( std::string const & groupPath ); + void read_impl( std::string const & groupPath ); /** * @brief Whether an iteration has been closed yet. @@ -164,11 +190,12 @@ class Iteration : public LegacyAttributable */ enum class CloseStatus { + ParseAccessDeferred, //!< The reader has not yet parsed this iteration Open, //!< Iteration has not been closed ClosedInFrontend, /*!< Iteration has been closed, but task has not yet been propagated to the backend */ - ClosedInBackend, /*!< Iteration has been closed and task has been - propagated to the backend */ + ClosedInBackend, /*!< Iteration has been closed and task has been + propagated to the backend */ ClosedTemporarily /*!< Iteration has been closed internally and may be reopened later */ }; @@ -194,6 +221,11 @@ class Iteration : public LegacyAttributable std::shared_ptr< StepStatus > m_stepStatus = std::make_shared< StepStatus >( StepStatus::NoStep ); + std::shared_ptr< auxiliary::Option< DeferredParseAccess > > + m_deferredParseAccess = + std::make_shared< auxiliary::Option< DeferredParseAccess > >( + auxiliary::Option< DeferredParseAccess >() ); + /** * @brief Begin an IO step on the IO file (or file-like object) * containing this iteration. In case of group-based iteration @@ -252,6 +284,13 @@ class Iteration : public LegacyAttributable * @param w The Writable representing the parent. */ virtual void linkHierarchy(Writable& w); + + /** + * @brief Access an iteration in read mode that has potentially not been + * parsed yet. + * + */ + void runDeferredParseAccess(); }; // Iteration extern template diff --git a/include/openPMD/Series.hpp b/include/openPMD/Series.hpp index 39463f79b5..a0c5a264ee 100644 --- a/include/openPMD/Series.hpp +++ b/include/openPMD/Series.hpp @@ -66,10 +66,6 @@ namespace internal */ class SeriesData : public AttributableData { - friend class openPMD::SeriesImpl; - friend class openPMD::Iteration; - friend class openPMD::Series; - public: explicit SeriesData() = default; @@ -83,8 +79,8 @@ class SeriesData : public AttributableData Container< Iteration, uint64_t > iterations{}; -OPENPMD_private : auxiliary::Option< WriteIterations > m_writeIterations; + auxiliary::Option< std::string > m_overrideFilebasedFilename; std::string m_name; std::string m_filenamePrefix; std::string m_filenamePostfix; @@ -99,6 +95,7 @@ OPENPMD_private : * one among both flags. */ StepStatus m_stepStatus = StepStatus::NoStep; + bool m_parseLazily = false; }; // SeriesData class SeriesInternal; @@ -337,6 +334,7 @@ class SeriesImpl : public AttributableImpl void flushMeshesPath(); void flushParticlesPath(); void readFileBased( ); + void readOneIterationFileBased( std::string const & filePath ); /** * Note on re-parsing of a Series: * If init == false, the parsing process will seek for new @@ -347,7 +345,6 @@ class SeriesImpl : public AttributableImpl */ void readGroupBased( bool init = true ); void readBase(); - void read(); std::string iterationFilename( uint64_t i ); void openIteration( uint64_t index, Iteration iteration ); @@ -386,6 +383,9 @@ namespace internal class SeriesInternal : public SeriesData, public SeriesImpl { friend struct SeriesShared; + friend class openPMD::Iteration; + friend class openPMD::Series; + friend class openPMD::Writable; public: #if openPMD_HAVE_MPI @@ -409,6 +409,9 @@ class SeriesInternal : public SeriesData, public SeriesImpl * * Entry point and common link between all iterations of particle and mesh data. * + * An instance can be created either directly via the given constructors or via + * the SeriesBuilder class. + * * @see https://github.com/openPMD/openPMD-standard/blob/latest/STANDARD.md#hierarchy-of-the-data-file * @see https://github.com/openPMD/openPMD-standard/blob/latest/STANDARD.md#iterations-and-time-series */ @@ -426,6 +429,15 @@ class Series : public SeriesImpl std::string const & options = "{}" ); #endif + /** + * @brief Construct a new Series + * + * @param filepath The backend will be determined by the filepath extension. + * @param at Access mode. + * @param options Advanced backend configuration via JSON. + * May be specified as a JSON-formatted string directly, or as a path + * to a JSON textfile, prepended by an at sign '@'. + */ Series( std::string const & filepath, Access at, diff --git a/src/IO/AbstractIOHandlerHelper.cpp b/src/IO/AbstractIOHandlerHelper.cpp index 5a37ec6fe7..ba8b887cef 100644 --- a/src/IO/AbstractIOHandlerHelper.cpp +++ b/src/IO/AbstractIOHandlerHelper.cpp @@ -32,15 +32,16 @@ namespace openPMD { #if openPMD_HAVE_MPI + template<> std::shared_ptr< AbstractIOHandler > - createIOHandler( + createIOHandler< nlohmann::json >( std::string path, Access access, Format format, MPI_Comm comm, - std::string const & options ) + nlohmann::json options ) { - nlohmann::json optionsJson = auxiliary::parseOptions( options, comm ); + (void) options; switch( format ) { case Format::HDF5: @@ -53,13 +54,13 @@ namespace openPMD # endif case Format::ADIOS2: return std::make_shared< ADIOS2IOHandler >( - path, access, comm, std::move( optionsJson ), "bp4" ); + path, access, comm, std::move( options ), "bp4" ); case Format::ADIOS2_SST: return std::make_shared< ADIOS2IOHandler >( - path, access, comm, std::move( optionsJson ), "sst" ); + path, access, comm, std::move( options ), "sst" ); case Format::ADIOS2_SSC: return std::make_shared< ADIOS2IOHandler >( - path, access, comm, std::move( optionsJson ), "ssc" ); + path, access, comm, std::move( options ), "ssc" ); default: throw std::runtime_error( "Unknown file format! Did you specify a file ending?" ); @@ -67,14 +68,15 @@ namespace openPMD } #endif + template<> std::shared_ptr< AbstractIOHandler > - createIOHandler( + createIOHandler< nlohmann::json >( std::string path, Access access, Format format, - std::string const & options ) + nlohmann::json options ) { - nlohmann::json optionsJson = auxiliary::parseOptions( options ); + (void) options; switch( format ) { case Format::HDF5: @@ -88,13 +90,13 @@ namespace openPMD #if openPMD_HAVE_ADIOS2 case Format::ADIOS2: return std::make_shared< ADIOS2IOHandler >( - path, access, std::move( optionsJson ), "bp4" ); + path, access, std::move( options ), "bp4" ); case Format::ADIOS2_SST: return std::make_shared< ADIOS2IOHandler >( - path, access, std::move( optionsJson ), "sst" ); + path, access, std::move( options ), "sst" ); case Format::ADIOS2_SSC: return std::make_shared< ADIOS2IOHandler >( - path, access, std::move( optionsJson ), "ssc" ); + path, access, std::move( options ), "ssc" ); #endif // openPMD_HAVE_ADIOS2 case Format::JSON: return std::make_shared< JSONIOHandler >( path, access ); @@ -103,4 +105,11 @@ namespace openPMD "Unknown file format! Did you specify a file ending?" ); } } + + std::shared_ptr< AbstractIOHandler > + createIOHandler( std::string path, Access access, Format format ) + { + return createIOHandler( + std::move( path ), access, format, nlohmann::json::object() ); + } } // namespace openPMD diff --git a/src/Iteration.cpp b/src/Iteration.cpp index e448bf0696..3cf0f14eb9 100644 --- a/src/Iteration.cpp +++ b/src/Iteration.cpp @@ -106,11 +106,12 @@ Iteration::close( bool _flush ) *m_closed = CloseStatus::ClosedInBackend; } break; + case CloseStatus::ParseAccessDeferred: case CloseStatus::ClosedInBackend: // just keep it like it is + // (this means that closing an iteration that has not been parsed + // yet keeps it re-openable) break; - default: - throw std::runtime_error( "Unreachable!" ); } if( _flush ) { @@ -145,6 +146,11 @@ Iteration::close( bool _flush ) Iteration & Iteration::open() { + if( *m_closed == CloseStatus::ParseAccessDeferred ) + { + *m_closed = CloseStatus::Open; + } + runDeferredParseAccess(); internal::SeriesInternal * s = &retrieveSeries(); // figure out my iteration number auto begin = s->indexOf( *this ); @@ -163,6 +169,7 @@ Iteration::closed() const { switch( *m_closed ) { + case CloseStatus::ParseAccessDeferred: case CloseStatus::Open: /* * Temporarily closing a file is something that the openPMD API @@ -291,9 +298,53 @@ Iteration::flush() } } -void -Iteration::read() +void Iteration::deferParseAccess( DeferredParseAccess dr ) +{ + *m_deferredParseAccess = + auxiliary::makeOption< DeferredParseAccess >( std::move( dr ) ); +} + +void Iteration::read() { + if( !m_deferredParseAccess->has_value() ) + { + return; + } + auto const & deferred = m_deferredParseAccess->get(); + if( deferred.fileBased ) + { + readFileBased( deferred.filename, deferred.index ); + } + else + { + readGroupBased( deferred.index ); + } + // reset this thing + *m_deferredParseAccess = auxiliary::Option< DeferredParseAccess >(); +} + +void Iteration::readFileBased( + std::string filePath, std::string const & groupPath ) +{ + auto & series = retrieveSeries(); + + series.readOneIterationFileBased( filePath ); + + read_impl( groupPath ); +} + +void Iteration::readGroupBased( std::string const & groupPath ) +{ + + read_impl(groupPath ); +} + +void Iteration::read_impl( std::string const & groupPath ) +{ + Parameter< Operation::OPEN_PATH > pOpen; + pOpen.path = groupPath; + IOHandler()->enqueue( IOTask( this, pOpen ) ); + using DT = Datatype; Parameter< Operation::READ_ATT > aRead; @@ -356,7 +407,6 @@ Iteration::read() if( hasMeshes ) { - Parameter< Operation::OPEN_PATH > pOpen; pOpen.path = s->meshesPath(); IOHandler()->enqueue(IOTask(&meshes, pOpen)); @@ -416,7 +466,6 @@ Iteration::read() if( hasParticles ) { - Parameter< Operation::OPEN_PATH > pOpen; pOpen.path = s->particlesPath(); IOHandler()->enqueue(IOTask(&particles, pOpen)); @@ -571,6 +620,28 @@ Iteration::linkHierarchy(Writable& w) particles.linkHierarchy(this->writable()); } +void Iteration::runDeferredParseAccess() +{ + if( IOHandler()->m_frontendAccess == Access::CREATE ) + { + return; + } + auto oldAccess = IOHandler()->m_frontendAccess; + auto newAccess = + const_cast< Access * >( &IOHandler()->m_frontendAccess ); + *newAccess = Access::READ_WRITE; + try + { + read(); + } + catch( ... ) + { + *newAccess = oldAccess; + throw; + } + *newAccess = oldAccess; +} + template float Iteration::time< float >() const; template double diff --git a/src/ReadIterations.cpp b/src/ReadIterations.cpp index c9229459e7..af5e2d1fa5 100644 --- a/src/ReadIterations.cpp +++ b/src/ReadIterations.cpp @@ -41,7 +41,43 @@ SeriesIterator::SeriesIterator( Series series ) } else { - auto status = it->second.beginStep(); + auto openIteration = [ &it ]() + { + /* + * @todo + * Is that really clean? + * Use case: See Python ApiTest testListSeries: + * Call listSeries twice. + */ + if( *it->second.m_closed != + Iteration::CloseStatus::ClosedInBackend ) + { + it->second.open(); + } + }; + AdvanceStatus status{}; + switch( series.iterationEncoding() ) + { + case IterationEncoding::fileBased: + /* + * The file needs to be accessed before beginning a step upon it. + * In file-based iteration layout it maybe is not accessed yet, + * so do that now. There is only one step per file, so beginning + * the step after parsing the file is ok. + */ + openIteration(); + status = it->second.beginStep(); + break; + default: + /* + * In group-based iteration layout, we have definitely already had + * access to the file until now. Better to begin a step right away, + * otherwise we might get another step's data. + */ + status = it->second.beginStep(); + openIteration(); + break; + } if( status == AdvanceStatus::OVER ) { *this = end(); @@ -98,6 +134,10 @@ SeriesIterator & SeriesIterator::operator++() return *this; } m_currentIteration = it->first; + if( *it->second.m_closed != Iteration::CloseStatus::ClosedInBackend ) + { + it->second.open(); + } switch( series.iterationEncoding() ) { using IE = IterationEncoding; diff --git a/src/Series.cpp b/src/Series.cpp index 09e22e40c1..99bc10ba5f 100644 --- a/src/Series.cpp +++ b/src/Series.cpp @@ -20,6 +20,7 @@ */ #include "openPMD/auxiliary/Date.hpp" #include "openPMD/auxiliary/Filesystem.hpp" +#include "openPMD/auxiliary/JSON.hpp" #include "openPMD/auxiliary/StringManip.hpp" #include "openPMD/IO/AbstractIOHandler.hpp" #include "openPMD/IO/AbstractIOHandlerHelper.hpp" @@ -49,6 +50,21 @@ namespace */ std::string cleanFilename(std::string const &filename, Format f); + /** Compound return type for regex matching of filenames */ + struct Match + { + bool isContained{}; //! pattern match successful + int padding{}; //! number of zeros used for padding of iteration + uint64_t iteration{}; //! iteration found in regex pattern (default: 0) + + // support for std::tie + operator std::tuple< bool &, int &, uint64_t & >() + { + return std::tuple< bool &, int &, uint64_t & >{ + isContained, padding, iteration }; + } + }; + /** Create a functor to determine if a file can be of a format and matches an iterationEncoding, given the filename on disk. * * @param prefix String containing head (i.e. before %T) of desired filename without filename extension. @@ -59,7 +75,7 @@ namespace * bool is True if file could be of type f and matches the iterationEncoding. False otherwise. * int is the amount of padding present in the iteration number %T. Is 0 if bool is False. */ - std::function(std::string const &)> + std::function matcher(std::string const &prefix, int padding, std::string const &postfix, Format f); } // namespace [anonymous] @@ -394,9 +410,9 @@ SeriesImpl::parseInput(std::string filepath) return input; } -void -SeriesImpl::init(std::shared_ptr< AbstractIOHandler > ioHandler, - std::unique_ptr< SeriesImpl::ParsedInput > input) +void SeriesImpl::init( + std::shared_ptr< AbstractIOHandler > ioHandler, + std::unique_ptr< SeriesImpl::ParsedInput > input ) { auto & series = get(); writable().IOHandler = ioHandler; @@ -486,6 +502,11 @@ SeriesImpl::flushFileBased( iterations_iterator begin, iterations_iterator end ) if( IOHandler()->m_frontendAccess == Access::READ_ONLY ) for( auto it = begin; it != end; ++it ) { + if( *it->second.m_closed + == Iteration::CloseStatus::ParseAccessDeferred ) + { + continue; + } bool const dirtyRecursive = it->second.dirtyRecursive(); if( *it->second.m_closed == Iteration::CloseStatus::ClosedInBackend ) @@ -530,6 +551,11 @@ SeriesImpl::flushFileBased( iterations_iterator begin, iterations_iterator end ) bool allDirty = dirty(); for( auto it = begin; it != end; ++it ) { + if( *it->second.m_closed + == Iteration::CloseStatus::ParseAccessDeferred ) + { + continue; + } bool const dirtyRecursive = it->second.dirtyRecursive(); if( *it->second.m_closed == Iteration::CloseStatus::ClosedInBackend ) @@ -615,6 +641,11 @@ SeriesImpl::flushGroupBased( iterations_iterator begin, iterations_iterator end if( IOHandler()->m_frontendAccess == Access::READ_ONLY ) for( auto it = begin; it != end; ++it ) { + if( *it->second.m_closed + == Iteration::CloseStatus::ParseAccessDeferred ) + { + continue; + } if( *it->second.m_closed == Iteration::CloseStatus::ClosedInBackend ) { @@ -651,6 +682,11 @@ SeriesImpl::flushGroupBased( iterations_iterator begin, iterations_iterator end for( auto it = begin; it != end; ++it ) { + if( *it->second.m_closed + == Iteration::CloseStatus::ParseAccessDeferred ) + { + continue; + } if( *it->second.m_closed == Iteration::CloseStatus::ClosedInBackend ) { @@ -716,7 +752,6 @@ SeriesImpl::readFileBased( ) { auto & series = get(); Parameter< Operation::OPEN_FILE > fOpen; - Parameter< Operation::CLOSE_FILE > fClose; Parameter< Operation::READ_ATT > aRead; if( !auxiliary::directory_exists(IOHandler()->directory) ) @@ -727,74 +762,21 @@ SeriesImpl::readFileBased( ) series.m_filenamePostfix, series.m_format); bool isContained; int padding; + uint64_t iterationIndex; std::set< int > paddings; for( auto const& entry : auxiliary::list_directory(IOHandler()->directory) ) { - std::tie(isContained, padding) = isPartOfSeries(entry); + std::tie(isContained, padding, iterationIndex) = isPartOfSeries(entry); if( isContained ) { + Iteration & i = series.iterations[ iterationIndex ]; + i.deferParseAccess( + { std::to_string( iterationIndex ), true, entry } ); // TODO skip if the padding is exact the number of chars in an iteration? paddings.insert(padding); - - fOpen.name = entry; - IOHandler()->enqueue(IOTask(this, fOpen)); - IOHandler()->flush(); - series.iterations.parent() = getWritable(this); - - readBase(); - - using DT = Datatype; - aRead.name = "iterationEncoding"; - IOHandler()->enqueue( IOTask( this, aRead ) ); - IOHandler()->flush(); - if( *aRead.dtype == DT::STRING ) - { - std::string encoding = - Attribute( *aRead.resource ).get< std::string >(); - if( encoding == "fileBased" ) - series.m_iterationEncoding = IterationEncoding::fileBased; - else if( encoding == "groupBased" ) - { - series.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 ); - } - 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; - } - else - throw std::runtime_error( - "Unexpected Attribute datatype for 'iterationFormat'" ); - - read(); - IOHandler()->enqueue(IOTask(this, fClose)); - IOHandler()->flush(); } } - for( auto & iteration : series.iterations ) - { - *iteration.second.m_closed = Iteration::CloseStatus::ClosedTemporarily; - } - if( series.iterations.empty() ) { /* Frontend access type might change during SeriesImpl::read() to allow parameter modification. @@ -805,6 +787,35 @@ SeriesImpl::readFileBased( ) std::cerr << "No matching iterations found: " << name() << std::endl; } + auto readIterationEagerly = []( Iteration & iteration ) + { + iteration.runDeferredParseAccess(); + Parameter< Operation::CLOSE_FILE > fClose; + iteration.IOHandler()->enqueue( IOTask( &iteration, fClose ) ); + iteration.IOHandler()->flush(); + *iteration.m_closed = Iteration::CloseStatus::ClosedTemporarily; + }; + if( series.m_parseLazily ) + { + for( auto & iteration : series.iterations ) + { + *iteration.second.m_closed = + Iteration::CloseStatus::ParseAccessDeferred; + } + // open the last iteration, just to parse Series attributes + auto getLastIteration = series.iterations.end(); + getLastIteration--; + auto & lastIteration = getLastIteration->second; + readIterationEagerly( lastIteration ); + } + else + { + for( auto & iteration : series.iterations ) + { + readIterationEagerly( iteration.second ); + } + } + if( paddings.size() == 1u ) series.m_filenamePadding = *paddings.begin(); @@ -815,6 +826,73 @@ SeriesImpl::readFileBased( ) "Please specify '%0T' or open as read-only."); } +void SeriesImpl::readOneIterationFileBased( std::string const & filePath ) +{ + auto & series = get(); + + Parameter< Operation::OPEN_FILE > fOpen; + Parameter< Operation::READ_ATT > aRead; + + fOpen.name = filePath; + IOHandler()->enqueue(IOTask(this, fOpen)); + IOHandler()->flush(); + series.iterations.parent() = getWritable(this); + + readBase(); + + using DT = Datatype; + aRead.name = "iterationEncoding"; + IOHandler()->enqueue( IOTask( this, aRead ) ); + IOHandler()->flush(); + if( *aRead.dtype == DT::STRING ) + { + std::string encoding = + Attribute( *aRead.resource ).get< std::string >(); + if( encoding == "fileBased" ) + series.m_iterationEncoding = IterationEncoding::fileBased; + else if( encoding == "groupBased" ) + { + series.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 ); + } + 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; + } + else + throw std::runtime_error( + "Unexpected Attribute datatype for 'iterationFormat'" ); + + Parameter< Operation::OPEN_PATH > pOpen; + std::string version = openPMD(); + if( version == "1.0.0" || version == "1.0.1" || version == "1.1.0" ) + pOpen.path = auxiliary::replace_first(basePath(), "/%T/", ""); + else + throw std::runtime_error("Unknown openPMD version - " + version); + IOHandler()->enqueue(IOTask(&series.iterations, pOpen)); + + readAttributes(); + series.iterations.readAttributes(); +} + void SeriesImpl::readGroupBased( bool do_init ) { @@ -843,6 +921,12 @@ SeriesImpl::readGroupBased( bool do_init ) series.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"; + /* + * We'll want the openPMD API to continue series.m_name to open + * the file instead of piecing the name together via + * prefix-padding-postfix things. + */ + series.m_overrideFilebasedFilename = series.m_name; } else throw std::runtime_error("Unknown iterationEncoding: " + encoding); setAttribute("iterationEncoding", encoding); @@ -863,7 +947,56 @@ SeriesImpl::readGroupBased( bool do_init ) throw std::runtime_error("Unexpected Attribute datatype for 'iterationFormat'"); } - read(); + Parameter< Operation::OPEN_PATH > pOpen; + std::string version = openPMD(); + if( version == "1.0.0" || version == "1.0.1" || version == "1.1.0" ) + pOpen.path = auxiliary::replace_first(basePath(), "/%T/", ""); + else + throw std::runtime_error("Unknown openPMD version - " + version); + IOHandler()->enqueue(IOTask(&series.iterations, pOpen)); + + readAttributes(); + series.iterations.readAttributes(); + + /* obtain all paths inside the basepath (i.e. all iterations) */ + Parameter< Operation::LIST_PATHS > pList; + IOHandler()->enqueue(IOTask(&series.iterations, pList)); + IOHandler()->flush(); + + for( auto const & it : *pList.paths ) + { + uint64_t index = std::stoull( it ); + if( series.iterations.contains( index ) ) + { + // maybe re-read + auto & i = series.iterations.at( index ); + if( i.closedByWriter() ) + { + continue; + } + if( *i.m_closed != Iteration::CloseStatus::ParseAccessDeferred ) + { + pOpen.path = it; + IOHandler()->enqueue( IOTask( &i, pOpen ) ); + i.read(); + } + } + else + { + // parse for the first time, resp. delay the parsing process + Iteration & i = series.iterations[ std::stoull( it ) ]; + i.deferParseAccess( { it, false, "" } ); + if( !series.m_parseLazily ) + { + i.runDeferredParseAccess(); + *i.m_closed = Iteration::CloseStatus::Open; + } + else + { + *i.m_closed = Iteration::CloseStatus::ParseAccessDeferred; + } + } + } } void @@ -942,43 +1075,14 @@ SeriesImpl::readBase() } } -void -SeriesImpl::read() -{ - auto & series = get(); - Parameter< Operation::OPEN_PATH > pOpen; - std::string version = openPMD(); - if( version == "1.0.0" || version == "1.0.1" || version == "1.1.0" ) - pOpen.path = auxiliary::replace_first(basePath(), "/%T/", ""); - else - throw std::runtime_error("Unknown openPMD version - " + version); - IOHandler()->enqueue(IOTask(&series.iterations, pOpen)); - - readAttributes(); - series.iterations.readAttributes(); - - /* obtain all paths inside the basepath (i.e. all iterations) */ - Parameter< Operation::LIST_PATHS > pList; - IOHandler()->enqueue(IOTask(&series.iterations, pList)); - IOHandler()->flush(); - - for( auto const& it : *pList.paths ) - { - Iteration& i = series.iterations[std::stoull(it)]; - if ( i.closedByWriter( ) ) - { - continue; - } - pOpen.path = it; - IOHandler()->enqueue(IOTask(&i, pOpen)); - i.read(); - } -} - std::string SeriesImpl::iterationFilename( uint64_t i ) { auto & series = get(); + if( series.m_overrideFilebasedFilename.has_value() ) + { + return series.m_overrideFilebasedFilename.get(); + } std::stringstream iteration( "" ); iteration << std::setw( series.m_filenamePadding ) << std::setfill( '0' ) << i; @@ -1132,6 +1236,7 @@ SeriesImpl::openIteration( uint64_t index, Iteration iteration ) throw std::runtime_error( "[Series] Detected illegal access to iteration that " "has been closed previously." ); + case CL::ParseAccessDeferred: case CL::Open: case CL::ClosedTemporarily: *iteration.m_closed = CL::Open; @@ -1139,11 +1244,28 @@ SeriesImpl::openIteration( uint64_t index, Iteration iteration ) case CL::ClosedInFrontend: // just keep it like it is break; - default: - throw std::runtime_error( "Unreachable!" ); + } +} + +namespace +{ +template< typename T > +void getJsonOption( + nlohmann::json const & config, std::string const & key, T & dest ) +{ + if( config.contains( key ) ) + { + dest = config.at( key ).get< T >(); } } +void parseJsonOptions( + internal::SeriesData & series, nlohmann::json const & options ) +{ + getJsonOption( options, "defer_iteration_parsing", series.m_parseLazily ); +} +} + namespace internal { #if openPMD_HAVE_MPI @@ -1156,9 +1278,11 @@ SeriesInternal::SeriesInternal( static_cast< internal::SeriesData * >( this ), static_cast< internal::AttributableData * >( this ) } { + nlohmann::json optionsJson = auxiliary::parseOptions( options, comm ); + parseJsonOptions( *this, optionsJson ); auto input = parseInput( filepath ); - auto handler = - createIOHandler( input->path, at, input->format, comm, options ); + auto handler = createIOHandler( + input->path, at, input->format, comm, std::move( optionsJson ) ); init( handler, std::move( input ) ); } #endif @@ -1169,8 +1293,11 @@ SeriesInternal::SeriesInternal( static_cast< internal::SeriesData * >( this ), static_cast< internal::AttributableData * >( this ) } { + nlohmann::json optionsJson = auxiliary::parseOptions( options ); + parseJsonOptions( *this, optionsJson ); auto input = parseInput( filepath ); - auto handler = createIOHandler( input->path, at, input->format, options ); + auto handler = createIOHandler( + input->path, at, input->format, std::move( optionsJson ) ); init( handler, std::move( input ) ); } @@ -1210,7 +1337,9 @@ Series::Series( #endif Series::Series( - std::string const & filepath, Access at, std::string const & options ) + std::string const & filepath, + Access at, + std::string const & options) : SeriesImpl{ nullptr, nullptr } , m_series{ std::make_shared< internal::SeriesInternal >( filepath, at, options ) } @@ -1254,19 +1383,19 @@ namespace } } - std::function(std::string const &)> + std::function buildMatcher(std::string const ®exPattern) { std::regex pattern(regexPattern); - return [pattern](std::string const &filename) -> std::tuple { + return [pattern](std::string const &filename) -> Match { std::smatch regexMatches; bool match = std::regex_match(filename, regexMatches, pattern); int padding = match ? regexMatches[1].length() : 0; - return std::tuple{match, padding}; + return {match, padding, match ? std::stoull(regexMatches[1]) : 0}; }; } - std::function(std::string const &)> + std::function matcher(std::string const &prefix, int padding, std::string const &postfix, Format f) { switch (f) { case Format::HDF5: { @@ -1318,8 +1447,8 @@ namespace return buildMatcher(nameReg); } default: - return [](std::string const &) -> std::tuple { return std::tuple{false, 0}; }; + return [](std::string const &) -> Match { return {false, 0, 0}; }; } } - } // namespace [anonymous] - } // namespace openPMD +} // namespace [anonymous] +} // namespace openPMD diff --git a/src/binding/python/Series.cpp b/src/binding/python/Series.cpp index 62686ac326..b8bb2e8298 100644 --- a/src/binding/python/Series.cpp +++ b/src/binding/python/Series.cpp @@ -81,13 +81,15 @@ void init_Series(py::module &m) { py::class_(m, "Series") .def(py::init(), - py::arg("filepath"), py::arg("access"), py::arg("options") = "{}") + py::arg("filepath"), + py::arg("access"), + py::arg("options") = "{}") #if openPMD_HAVE_MPI .def(py::init([]( std::string const& filepath, Access at, py::object &comm, - std::string const& options){ + std::string const& options ){ //! TODO perform mpi4py import test and check min-version //! careful: double MPI_Init risk? only import mpi4py.MPI? //! required C-API init? probably just checks: @@ -133,7 +135,7 @@ void init_Series(py::module &m) { "(Mismatched MPI at compile vs. runtime?)"); } - return new Series(filepath, at, *mpiCommPtr, options); + return new Series(filepath, at, *mpiCommPtr, options ); }), py::arg("filepath"), py::arg("access"), diff --git a/test/ParallelIOTest.cpp b/test/ParallelIOTest.cpp index fe6f63f6f1..4a4e0d9f25 100644 --- a/test/ParallelIOTest.cpp +++ b/test/ParallelIOTest.cpp @@ -691,7 +691,11 @@ file_based_write_read( std::string file_ending ) // check non-collective, parallel read { - Series read( name, Access::READ_ONLY, MPI_COMM_WORLD ); + Series read( + name, + Access::READ_ONLY, + MPI_COMM_WORLD, + "{\"defer_iteration_parsing\": true}" ); Iteration it = read.iterations[ 30 ]; it.open(); // collective if( mpi_rank == 0 ) // non-collective branch @@ -903,7 +907,9 @@ adios2_streaming() )"; Series readSeries( - "../samples/adios2_stream.bp", Access::READ_ONLY, options ); + "../samples/adios2_stream.sst", + Access::READ_ONLY, + "{\"defer_iteration_parsing\": true}" ); size_t last_iteration_index = 0; for( auto iteration : readSeries.readIterations() ) diff --git a/test/SerialIOTest.cpp b/test/SerialIOTest.cpp index d3682ad3c6..f7981a4271 100644 --- a/test/SerialIOTest.cpp +++ b/test/SerialIOTest.cpp @@ -191,9 +191,12 @@ write_and_read_many_iterations( std::string const & ext ) { } // ~Series intentionally not yet called - Series read(filename, Access::READ_ONLY); - for (auto iteration : read.iterations) { - // std::cout << "Reading iteration " << iteration.first << std::endl; + Series read( filename, Access::READ_ONLY, "{\"defer_iteration_parsing\": true}" ); + for( auto iteration : read.iterations ) + { + iteration.second.open(); + // std::cout << "Reading iteration " << iteration.first << + // std::endl; auto E_x = iteration.second.meshes["E"]["x"]; auto chunk = E_x.loadChunk({0}, {10}); iteration.second.close(); @@ -3480,12 +3483,13 @@ iterate_nonstreaming_series( std::string const & file ) } } - Series readSeries( file, Access::READ_ONLY ); + Series readSeries( file, Access::READ_ONLY, "{\"defer_iteration_parsing\": true}" ); size_t last_iteration_index = 0; // conventionally written Series must be readable with streaming-aware API! for( auto iteration : readSeries.readIterations() ) { + // ReadIterations takes care of Iteration::open()ing iterations auto E_x = iteration.meshes[ "E" ][ "x" ]; REQUIRE( E_x.getDimensionality() == 1 ); REQUIRE( E_x.getExtent()[ 0 ] == extent ); @@ -3656,3 +3660,54 @@ TEST_CASE( "extend_dataset", "[serial]" ) // extendDataset( "h5" ); #endif } + + +void deferred_parsing( std::string const & extension ) +{ + std::string const basename = "../samples/lazy_parsing/lazy_parsing_"; + // create a single iteration + { + Series series( basename + "%T." + extension, Access::CREATE ); + std::vector< float > buffer( 20 ); + std::iota( buffer.begin(), buffer.end(), 0.f ); + auto dataset = series.iterations[ 1000 ].meshes[ "E" ][ "x" ]; + dataset.resetDataset( { Datatype::FLOAT, { 20 } } ); + dataset.storeChunk( buffer, { 0 }, { 20 } ); + series.flush(); + } + // create some empty pseudo files + // if the reader tries accessing them it's game over + { + for( size_t i = 0; i < 1000; i += 100 ) + { + std::ofstream file; + file.open( basename + std::to_string( i ) + "." + extension ); + file.close(); + } + } + { + Series series( + basename + "%T." + extension, + Access::READ_ONLY, + "{\"defer_iteration_parsing\": true}" ); + auto dataset = series.iterations[ 1000 ] + .open() + .meshes[ "E" ][ "x" ] + .loadChunk< float >( { 0 }, { 20 } ); + series.flush(); + for( size_t i = 0; i < 20; ++i ) + { + REQUIRE( + std::abs( dataset.get()[ i ] - float( i ) ) <= + std::numeric_limits< float >::epsilon() ); + } + } +} + +TEST_CASE( "deferred_parsing", "[serial]" ) +{ + for( auto const & t : testedFileExtensions() ) + { + deferred_parsing( t ); + } +} diff --git a/test/python/unittest/API/APITest.py b/test/python/unittest/API/APITest.py index 390462d216..7f37ff64e5 100644 --- a/test/python/unittest/API/APITest.py +++ b/test/python/unittest/API/APITest.py @@ -1602,6 +1602,7 @@ def makeIteratorRoundTrip(self, backend, file_ending): # write jsonConfig = """ { + "defer_iteration_parsing": true, "adios2": { "engine": { "type": "bp4", @@ -1682,9 +1683,11 @@ def makeAvailableChunksRoundTrip(self, ext): read = io.Series( name, - io.Access_Type.read_only + io.Access_Type.read_only, + options='{"defer_iteration_parsing": true}' ) + read.iterations[0].open() chunks = read.iterations[0].meshes["E"]["x"].available_chunks() chunks = sorted(chunks, key=lambda chunk: chunk.offset) for chunk in chunks: