Merge pull request #993 from eisenhauer/SstLastTimeoutDiscard

Support discard on writer side, LastAvailable and timeout on reader side

bindings/CXX11/cxx11/Engine.h

@@ -75,7 +75,8 @@ class Engine
      * @param timeoutSeconds
      * @return current step status
      */
-    StepStatus BeginStep(const StepMode mode, const float timeoutSeconds = 0.f);
+    StepStatus BeginStep(const StepMode mode,
+                         const float timeoutSeconds = -1.f);
 
     /**
      * Inspect current logical step

bindings/Fortran/modules/adios2_engine_begin_step_mod.f90

@@ -43,7 +43,7 @@ subroutine adios2_begin_step_mode(engine, step_mode, ierr)
         !local
         integer status
 
-        call adios2_begin_step_f2c(engine%f2c, step_mode, 0._4, status, ierr)
+        call adios2_begin_step_f2c(engine%f2c, step_mode, -1._4, status, ierr)
 
     end subroutine
 
@@ -55,11 +55,11 @@ subroutine adios2_begin_step_default(engine, ierr)
 
         if( engine%mode == adios2_mode_read ) then
             call adios2_begin_step_f2c(engine%f2c, &
-                                       adios2_step_mode_next_available, 0._4, &
+                                       adios2_step_mode_next_available, -1._4, &
                                        status, ierr)
         else
             call adios2_begin_step_f2c(engine%f2c, &
-                                       adios2_step_mode_append, 0._4, &
+                                       adios2_step_mode_append, -1.0_4, &
                                        status, ierr)
         end if
 

bindings/Python/py11Engine.h

@@ -33,7 +33,8 @@ class Engine
 
     ~Engine() = default;
 
-    StepStatus BeginStep(const StepMode mode, const float timeoutSeconds = 0.f);
+    StepStatus BeginStep(const StepMode mode,
+                         const float timeoutSeconds = -1.f);
 
     void Put(core::VariableBase *variable, const pybind11::array &array,
              const Mode launch = Mode::Deferred);

bindings/Python/py11glue.cpp

@@ -334,7 +334,7 @@ PYBIND11_MODULE(adios2, m)
     pybind11::class_<adios2::py11::Engine>(m, "Engine")
         .def("BeginStep", &adios2::py11::Engine::BeginStep,
              pybind11::arg("mode") = adios2::StepMode::NextAvailable,
-             pybind11::arg("timeoutSeconds") = 0.f)
+             pybind11::arg("timeoutSeconds") = -1.f)
 
         .def("Put", (void (adios2::py11::Engine::*)(
                         adios2::core::VariableBase *, const pybind11::array &,

examples/heatTransfer/read/heatRead.cpp

@@ -121,7 +121,7 @@ int main(int argc, char *argv[])
         while (true)
         {
             adios2::StepStatus status =
-                reader.BeginStep(adios2::StepMode::NextAvailable, 0.0f);
+                reader.BeginStep(adios2::StepMode::NextAvailable);
             if (status != adios2::StepStatus::OK)
             {
                 break;

source/adios2/core/Engine.cpp

@@ -37,12 +37,11 @@ StepStatus Engine::BeginStep()
 {
     if (m_OpenMode == Mode::Read)
     {
-        return BeginStep(StepMode::NextAvailable,
-                         std::numeric_limits<float>::max());
+        return BeginStep(StepMode::NextAvailable, -1.0);
     }
     else
     {
-        return BeginStep(StepMode::Append, std::numeric_limits<float>::max());
+        return BeginStep(StepMode::Append, -1.0);
     }
 }
 

source/adios2/core/Engine.h

@@ -90,9 +90,8 @@ class Engine
      * @param timeoutSeconds (not yet implemented)
      * @return current step status
      */
-    virtual StepStatus
-    BeginStep(StepMode mode,
-              const float timeoutSeconds = std::numeric_limits<float>::max());
+    virtual StepStatus BeginStep(StepMode mode,
+                                 const float timeoutSeconds = -1.0);
 
     /**
      * Returns current step information for each engine.

source/adios2/engine/bp3/BP3Reader.h

@@ -39,9 +39,8 @@ class BP3Reader : public Engine
 
     virtual ~BP3Reader() = default;
 
-    StepStatus BeginStep(
-        StepMode mode = StepMode::NextAvailable,
-        const float timeoutSeconds = std::numeric_limits<float>::max()) final;
+    StepStatus BeginStep(StepMode mode = StepMode::NextAvailable,
+                         const float timeoutSeconds = -1.0) final;
 
     size_t CurrentStep() const final;
 

source/adios2/engine/bp3/BP3Writer.h

@@ -38,9 +38,8 @@ class BP3Writer : public core::Engine
 
     ~BP3Writer();
 
-    StepStatus BeginStep(
-        StepMode mode,
-        const float timeoutSeconds = std::numeric_limits<float>::max()) final;
+    StepStatus BeginStep(StepMode mode,
+                         const float timeoutSeconds = -1.0) final;
     size_t CurrentStep() const final;
     void PerformPuts() final;
     void EndStep() final;

source/adios2/engine/dataman/DataManReader.cpp

@@ -59,10 +59,11 @@ StepStatus DataManReader::BeginStep(StepMode stepMode,
         // timeout == std::numeric_limits<float>::max() means there is no
         // timeout, and it should block
         // forever until it receives something.
-        if (timeoutSeconds != std::numeric_limits<float>::max())
+        if (timeoutSeconds >= 0.0)
         {
             if (duration.count() > timeoutSeconds)
             {
+                std::cout << "Dataman Reader timeing out" << std::endl;
                 return StepStatus::NotReady;
             }
         }

source/adios2/engine/dataman/DataManReader.h

@@ -29,9 +29,8 @@ class DataManReader : public DataManCommon
     DataManReader(IO &io, const std::string &name, const Mode mode,
                   MPI_Comm mpiComm);
     virtual ~DataManReader();
-    StepStatus BeginStep(
-        StepMode stepMode,
-        const float timeoutSeconds = std::numeric_limits<float>::max()) final;
+    StepStatus BeginStep(StepMode stepMode,
+                         const float timeoutSeconds = -1.0) final;
     size_t CurrentStep() const final;
     void PerformGets() final;
     void EndStep() final;

source/adios2/engine/dataman/DataManWriter.h

@@ -30,9 +30,8 @@ class DataManWriter : public DataManCommon
                   MPI_Comm mpiComm);
     ~DataManWriter() = default;
 
-    StepStatus BeginStep(
-        StepMode mode,
-        const float timeoutSeconds = std::numeric_limits<float>::max()) final;
+    StepStatus BeginStep(StepMode mode,
+                         const float timeoutSeconds = -1.0) final;
     size_t CurrentStep() const;
     void PerformPuts() final;
     void EndStep() final;

source/adios2/engine/hdf5/HDF5ReaderP.h

@@ -40,9 +40,8 @@ class HDF5ReaderP : public Engine
 
     bool IsValid();
 
-    StepStatus BeginStep(
-        StepMode mode,
-        const float timeoutSeconds = std::numeric_limits<float>::max()) final;
+    StepStatus BeginStep(StepMode mode,
+                         const float timeoutSeconds = -1.0) final;
     size_t CurrentStep() const final;
     void EndStep() final;
 

source/adios2/engine/hdf5/HDF5WriterP.h

@@ -42,9 +42,8 @@ class HDF5WriterP : public Engine
 
     ~HDF5WriterP();
 
-    StepStatus BeginStep(
-        StepMode mode,
-        const float timeoutSeconds = std::numeric_limits<float>::max()) final;
+    StepStatus BeginStep(StepMode mode,
+                         const float timeoutSeconds = -1.0) final;
     void EndStep() final;
 
     void PerformPuts() final;

source/adios2/engine/insitumpi/InSituMPIReader.h

@@ -45,9 +45,8 @@ class InSituMPIReader : public Engine
                     MPI_Comm mpiComm);
 
     ~InSituMPIReader();
-    StepStatus BeginStep(
-        StepMode mode = StepMode::NextAvailable,
-        const float timeoutSeconds = std::numeric_limits<float>::max()) final;
+    StepStatus BeginStep(StepMode mode = StepMode::NextAvailable,
+                         const float timeoutSeconds = -1.0) final;
     void PerformGets() final;
     size_t CurrentStep() const final;
     void EndStep() final;

source/adios2/engine/insitumpi/InSituMPIWriter.h

@@ -43,9 +43,8 @@ class InSituMPIWriter : public Engine
 
     ~InSituMPIWriter();
 
-    StepStatus BeginStep(
-        StepMode mode,
-        const float timeoutSeconds = std::numeric_limits<float>::max()) final;
+    StepStatus BeginStep(StepMode mode,
+                         const float timeoutSeconds = -1.0) final;
     void PerformPuts() final;
     void EndStep() final;
 

source/adios2/engine/skeleton/SkeletonReader.h

@@ -42,9 +42,8 @@ class SkeletonReader : public Engine
                    MPI_Comm mpiComm);
 
     ~SkeletonReader();
-    StepStatus BeginStep(
-        StepMode mode = StepMode::NextAvailable,
-        const float timeoutSeconds = std::numeric_limits<float>::max()) final;
+    StepStatus BeginStep(StepMode mode = StepMode::NextAvailable,
+                         const float timeoutSeconds = -1.0) final;
     void PerformGets() final;
     size_t CurrentStep() const final;
     void EndStep() final;

source/adios2/engine/skeleton/SkeletonWriter.h

@@ -39,9 +39,8 @@ class SkeletonWriter : public Engine
 
     ~SkeletonWriter() = default;
 
-    StepStatus BeginStep(
-        StepMode mode,
-        const float timeoutSeconds = std::numeric_limits<float>::max()) final;
+    StepStatus BeginStep(StepMode mode,
+                         const float timeoutSeconds = -1.0) final;
     size_t CurrentStep() const final;
     void PerformPuts() final;
     void EndStep() final;

source/adios2/engine/sst/SstReader.cpp

@@ -127,6 +127,11 @@ StepStatus SstReader::BeginStep(StepMode mode, const float timeout_sec)
     {
         return StepStatus::EndOfStream;
     }
+    if (result == SstTimeout)
+    {
+        return StepStatus::NotReady;
+    }
+
     if (result != SstSuccess)
     {
         return StepStatus::OtherError;

source/adios2/engine/sst/SstReader.h

@@ -45,9 +45,7 @@ class SstReader : public Engine
     virtual ~SstReader();
 
     StepStatus BeginStep();
-    StepStatus
-    BeginStep(StepMode mode,
-              const float timeoutSeconds = std::numeric_limits<float>::max());
+    StepStatus BeginStep(StepMode mode, const float timeoutSeconds = -1.0);
     size_t CurrentStep() const final;
     void EndStep();
     void PerformGets();

source/adios2/engine/sst/SstWriter.h

@@ -35,9 +35,8 @@ class SstWriter : public Engine
 
     virtual ~SstWriter();
 
-    StepStatus BeginStep(
-        StepMode mode,
-        const float timeoutSeconds = std::numeric_limits<float>::max()) final;
+    StepStatus BeginStep(StepMode mode,
+                         const float timeoutSeconds = -1.0) final;
     void PerformPuts() final;
     void EndStep() final;
     void Flush(const int transportIndex = -1) final;

source/adios2/toolkit/sst/cp/cp_common.c

@@ -931,6 +931,8 @@ SstStream CP_newStream()
     pthread_cond_init(&Stream->DataCondition, NULL);
     Stream->WriterTimestep = -1; // Filled in by ProvideTimestep
     Stream->ReaderTimestep = -1; // first beginstep will get us timestep 0
+    Stream->DiscardPriorTimestep =
+        -1; // Timesteps prior to this discarded/released upon arrival
     if (getenv("SstVerbose"))
     {
         Stream->Verbose = 1;

source/adios2/toolkit/sst/cp/cp_internal.h

@@ -160,6 +160,8 @@ struct _SstStream
     SstFullMetadata CurrentMetadata;
     struct _SstParams *WriterConfigParams;
     void *ParamsBlock;
+    long DiscardPriorTimestep; /* timesteps numerically less than this will be
+                                  discarded with prejudice */
 
     /* reader side marshal info */
     FFSContext ReaderFFSContext;