Modified file transports to enable write loop in size > 2Gb

Checking std::ios_base::failure exceptions
IOChrono is off by default, each owner turns it on explicitly
Modified FileDescriptor and FilePointer destructors
Added Tests BPWriteRead for stdio and fstream
Removed IO AddTransport with vector (shoud have been there?) to enable
AddTransport in Tests

examples/experimental/multistep/reader_allsteps.cpp

@@ -115,7 +115,7 @@ int main(int argc, char *argv[])
             bpReaderSettings.SetEngine(
                 "ADIOS1Reader"); // BP is the default engine
             // see only one step at a time
-            bpReaderSettings.SetParameters({"OpenAsFile=yes"});
+            bpReaderSettings.SetParameters({{"OpenAsFile", "yes"}});
         }
 
         // Create engine smart pointer due to polymorphism,

examples/experimental/multistep/writer_multistep.cpp

@@ -74,11 +74,15 @@ int main(int argc, char *argv[])
             bpWriterSettings.AddTransport("file"
 #ifdef ADIOS2_HAVE_MPI
                                           ,
-                                          { "library=MPI-IO" }
+                                          {
+                                              {
+                                                  "library", "MPI-IO"
+                                              }
+                                          }
 #endif
                                           );
             // Passing parameters to the engine
-            bpWriterSettings.SetParameters({"have_metadata_file=yes"});
+            bpWriterSettings.SetParameters({{"have_metadata_file", "yes"}});
             // number of aggregators
             // bpWriterSettings.SetParameters("Aggregation", (nproc + 1) / 2);
         }

examples/hello/bpWriter/helloBPWriter_nompi.cpp

@@ -29,6 +29,7 @@ int main(int argc, char *argv[])
         /*** IO class object: settings and factory of Settings: Variables,
          * Parameters, Transports, and Execution: Engines */
         adios2::IO &bpIO = adios.DeclareIO("BPFile_N2N");
+        bpIO.AddTransport("File", {{"Library", "stdio"}});
 
         /** global array: name, { shape (total dimensions) }, { start (local) },
          * { count (local) }, all are constant dimensions */

source/adios2/ADIOSTypes.h

@@ -155,6 +155,9 @@ constexpr size_t DefaultInitialBufferSize(16384);
 constexpr size_t DefaultMaxBufferSize(16777216);
 /** default buffer growth factor (from STL vector = 2.) */
 constexpr float DefaultBufferGrowthFactor(2.);
+/** default size for writing/reading files using POSIX/fstream/stdio write
+ *  1Gb - 1Kb (tolerance)*/
+constexpr size_t DefaultMaxFileBatchSize(1024 * 1024 * 1024 - 1024);
 
 // adios alias values and types
 constexpr bool DebugON = true;

source/adios2/core/IO.cpp

@@ -47,17 +47,17 @@ void IO::SetParameters(const Params &parameters) { m_Parameters = parameters; }
 
 const Params &IO::GetParameters() const { return m_Parameters; }
 
-unsigned int IO::AddTransport(const std::string type,
-                              const std::vector<std::string> &parametersVector)
-{
-    Params parametersMap(BuildParametersMap(parametersVector, m_DebugMode));
-    return AddTransportCommon(type, parametersMap);
-}
-
 unsigned int IO::AddTransport(const std::string type, const Params &parameters)
 {
     Params parametersMap(parameters);
-    return AddTransportCommon(type, parametersMap);
+    if (m_DebugMode)
+    {
+        CheckTransportType(type);
+    }
+
+    parametersMap["transport"] = type;
+    m_TransportsParameters.push_back(parametersMap);
+    return static_cast<unsigned int>(m_TransportsParameters.size() - 1);
 }
 
 VariableCompound &IO::GetVariableCompound(const std::string &name)
@@ -235,18 +235,6 @@ std::shared_ptr<Engine> IO::Open(const std::string &name,
 }
 
 // PRIVATE Functions
-unsigned int IO::AddTransportCommon(const std::string type, Params &parameters)
-{
-    if (m_DebugMode)
-    {
-        CheckTransportType(type);
-    }
-
-    parameters["transport"] = type;
-    m_TransportsParameters.push_back(parameters);
-    return static_cast<unsigned int>(m_TransportsParameters.size() - 1);
-}
-
 unsigned int IO::GetVariableIndex(const std::string &name) const
 {
     if (m_DebugMode)

source/adios2/core/IO.h

@@ -93,14 +93,11 @@ public:
     const Params &GetParameters() const;
 
     /**
-     * Adds a transport and its parameters for the method
-     * @param type must be a supported transport type under /include/transport
-     * @param args list of parameters for a transport with format
-     * "parameter1=value1", ..., "parameterN=valueN"
+     * Adds a transport and its parameters for the IO Engine
+     * @param type must be a supported transport type
+     * @param params acceptable parameters for a particular transport
+     * @return
      */
-    unsigned int AddTransport(const std::string type,
-                              const std::vector<std::string> &paramsVector);
-
     unsigned int AddTransport(const std::string type,
                               const Params &params = Params());
 
@@ -251,14 +248,6 @@ private:
 
     std::set<std::string> m_EngineNames;
 
-    /**
-     * Called from AddTransport overloads
-     * @param type
-     * @param parameters
-     * @return transport index
-     */
-    unsigned int AddTransportCommon(const std::string type, Params &parameters);
-
     /** Gets the internal reference to a variable map for type T
      *  This function is specialized in IO.tcc */
     template <class T>

source/adios2/toolkit/format/bp1/BP1Base.cpp

@@ -22,6 +22,8 @@ BP1Base::BP1Base(MPI_Comm mpiComm, const bool debugMode)
 : m_HeapBuffer(debugMode), m_BP1Aggregator(mpiComm, debugMode),
   m_DebugMode(debugMode)
 {
+    // default
+    m_Profiler.IsActive = true;
 }
 
 void BP1Base::InitParameters(const Params &parameters)

source/adios2/toolkit/profiling/iochrono/IOChrono.h

@@ -43,8 +43,8 @@ struct IOChrono
     /** Create byte tracking counter for each process*/
     std::unordered_map<std::string, size_t> Bytes;
 
-    /** flag to determine if IOChrono object is used */
-    bool IsActive = true;
+    /** flag to determine if IOChrono object is being used */
+    bool IsActive = false;
 };
 
 } // end namespace profiling

source/adios2/toolkit/transport/Transport.cpp

@@ -25,6 +25,8 @@ Transport::Transport(const std::string type, const std::string library,
 
 void Transport::InitProfiler(const OpenMode openMode, const TimeUnit timeUnit)
 {
+    m_Profiler.IsActive = true;
+
     m_Profiler.Timers.emplace(std::make_pair(
         "open", profiling::Timer("open", TimeUnit::Microseconds, m_DebugMode)));
 
@@ -51,8 +53,6 @@ void Transport::InitProfiler(const OpenMode openMode, const TimeUnit timeUnit)
     m_Profiler.Timers.emplace(
         "close",
         profiling::Timer("close", TimeUnit::Microseconds, m_DebugMode));
-
-    m_Profiler.IsActive = true;
 }
 
 void Transport::SetBuffer(char * /*buffer*/, size_t /*size*/)

source/adios2/toolkit/transport/file/FileDescriptor.cpp

@@ -18,6 +18,7 @@
 
 /// \cond EXCLUDE_FROM_DOXYGEN
 #include <ios> //std::ios_base::failure
+#include <iostream>
 /// \endcond
 
 namespace adios2
@@ -32,7 +33,7 @@ FileDescriptor::FileDescriptor(MPI_Comm mpiComm, const bool debugMode)
 
 FileDescriptor::~FileDescriptor()
 {
-    if (m_FileDescriptor != -1)
+    if (m_IsOpen)
     {
         close(m_FileDescriptor);
     }
@@ -40,6 +41,15 @@ FileDescriptor::~FileDescriptor()
 
 void FileDescriptor::Open(const std::string &name, const OpenMode openMode)
 {
+    if (m_DebugMode)
+    {
+        if (name.empty())
+        {
+            throw std::invalid_argument(
+                "ERROR: file name is empty, in call to FilePointer Open\n");
+        }
+    }
+
     m_Name = name;
     m_OpenMode = openMode;
 
@@ -91,37 +101,34 @@ void FileDescriptor::Open(const std::string &name, const OpenMode openMode)
     {
         if (m_FileDescriptor == -1)
         {
-            throw std::ios_base::failure(
-                "ERROR: couldn't open file " + m_Name +
-                ", from call to Open in FileDescriptor transport using "
-                "POSIX open. Does file exists?\n");
+            throw std::ios_base::failure("ERROR: couldn't open file " + m_Name +
+                                         ", check permissions or existence, in "
+                                         "call to FileDescriptor Open\n");
         }
     }
 
     m_IsOpen = true;
 }
 
-void FileDescriptor::Write(const char *buffer, std::size_t size)
+void FileDescriptor::Write(const char *buffer, size_t size)
 {
-    if (m_Profiler.IsActive)
-    {
-        m_Profiler.Timers.at("write").Resume();
-    }
-
-    auto writtenSize = write(m_FileDescriptor, buffer, size);
+    auto lf_Write = [&](const char *buffer, size_t size) {
 
-    if (m_Profiler.IsActive)
-    {
-        m_Profiler.Timers.at("write").Pause();
-    }
+        if (m_Profiler.IsActive)
+        {
+            m_Profiler.Timers.at("write").Resume();
+        }
+        auto writtenSize = write(m_FileDescriptor, buffer, size);
+        if (m_Profiler.IsActive)
+        {
+            m_Profiler.Timers.at("write").Pause();
+        }
 
-    if (m_DebugMode)
-    {
         if (writtenSize == -1)
         {
-            throw std::ios_base::failure(
-                "ERROR: couldn't write to file " + m_Name +
-                ", in call to POSIX FileDescriptor write\n");
+            throw std::ios_base::failure("ERROR: couldn't write to file " +
+                                         m_Name +
+                                         ", in call to FileDescriptor Write\n");
         }
 
         if (static_cast<size_t>(writtenSize) != size)
@@ -129,8 +136,26 @@ void FileDescriptor::Write(const char *buffer, std::size_t size)
             throw std::ios_base::failure(
                 "ERROR: written size + " + std::to_string(writtenSize) +
                 " is not equal to intended size " + std::to_string(size) +
-                " in file " + m_Name + ", in call to POSIX write\n");
+                " in file " + m_Name + ", in call to FileDescriptor Write\n");
+        }
+    };
+
+    if (size > DefaultMaxFileBatchSize)
+    {
+        const size_t batches = size / DefaultMaxFileBatchSize;
+        const size_t remainder = size % DefaultMaxFileBatchSize;
+
+        size_t position = 0;
+        for (size_t b = 0; b < batches; ++b)
+        {
+            lf_Write(&buffer[position], DefaultMaxFileBatchSize);
+            position += DefaultMaxFileBatchSize;
         }
+        lf_Write(&buffer[position], remainder);
+    }
+    else
+    {
+        lf_Write(buffer, size);
     }
 }
 
@@ -143,24 +168,21 @@ void FileDescriptor::Close()
         m_Profiler.Timers.at("close").Resume();
     }
 
-    int status = close(m_FileDescriptor);
+    const int status = close(m_FileDescriptor);
 
     if (m_Profiler.IsActive)
     {
         m_Profiler.Timers.at("close").Pause();
     }
 
-    if (m_DebugMode)
+    if (status == -1)
     {
-        if (status == -1)
-        {
-            throw std::ios_base::failure("ERROR: couldn't close file " +
-                                         m_Name + ", in call to POSIX write\n");
-        }
+        throw std::ios_base::failure("ERROR: couldn't close file " + m_Name +
+                                     ", in call to FileDescriptor Close\n");
     }
 
     m_IsOpen = false;
 }
 
 } // end namespace transport
-} // namespace adios
+} // end namespace adios

source/adios2/toolkit/transport/file/FilePointer.cpp

@@ -26,7 +26,7 @@ FilePointer::FilePointer(MPI_Comm mpiComm, const bool debugMode)
 
 FilePointer::~FilePointer()
 {
-    if (m_File)
+    if (m_IsOpen)
     {
         fclose(m_File);
     }
@@ -34,73 +34,136 @@ FilePointer::~FilePointer()
 
 void FilePointer::Open(const std::string &name, const OpenMode openMode)
 {
+    if (m_DebugMode)
+    {
+        if (name.empty())
+        {
+            throw std::invalid_argument(
+                "ERROR: file name is empty, in call to FilePointer Open\n");
+        }
+    }
+
     m_Name = name;
     m_OpenMode = openMode;
 
     if (m_OpenMode == OpenMode::Write)
     {
-        m_File = fopen(name.c_str(), "w");
+        m_File = std::fopen(name.c_str(), "w");
     }
     else if (m_OpenMode == OpenMode::Append)
     {
-        m_File = fopen(name.c_str(), "a"); // need to change
+        // need to change when implemented
+        m_File = std::fopen(name.c_str(), "a");
     }
     else if (m_OpenMode == OpenMode::Read)
     {
-        m_File = fopen(name.c_str(), "r");
+        m_File = std::fopen(name.c_str(), "r");
     }
 
-    if (m_DebugMode)
+    if (std::ferror(m_File))
     {
-        if (m_File == nullptr)
-        {
-            throw std::ios_base::failure(
-                "ERROR: couldn't open file " + name +
-                ", "
-                "in call to Open from stdio.h FilePointer* transport\n");
-        }
+        throw std::ios_base::failure("ERROR: couldn't open file " + name +
+                                     ", "
+                                     "in call to FilePointer Open\n");
     }
+
     m_IsOpen = true;
 }
 
 void FilePointer::SetBuffer(char *buffer, size_t size)
 {
-    int status = setvbuf(m_File, buffer, _IOFBF, size);
+    const int status = std::setvbuf(m_File, buffer, _IOFBF, size);
 
-    if (m_DebugMode)
+    if (!status)
     {
-        if (status == 1)
-        {
-            throw std::ios_base::failure(
-                "ERROR: could not set buffer in rank " +
-                std::to_string(m_RankMPI) + "\n");
-        }
+        throw std::ios_base::failure(
+            "ERROR: could not set FILE* buffer in file " + m_Name +
+            ", in call to FilePointer SetBuffer\n");
     }
 }
 
 void FilePointer::Write(const char *buffer, size_t size)
 {
-    fwrite(buffer, sizeof(char), size, m_File);
+    auto lf_Write = [&](const char *buffer, size_t size) {
 
-    if (m_DebugMode)
-    {
-        if (ferror(m_File))
+        if (m_Profiler.IsActive)
+        {
+            m_Profiler.Timers.at("write").Resume();
+        }
+        auto writtenSize = std::fwrite(buffer, sizeof(char), size, m_File);
+
+        if (m_Profiler.IsActive)
+        {
+            m_Profiler.Timers.at("write").Pause();
+        }
+
+        if (std::ferror(m_File))
+        {
+            throw std::ios_base::failure("ERROR: couldn't write to file " +
+                                         m_Name + ", in call to FILE* Write\n");
+        }
+
+        if (writtenSize != size)
         {
             throw std::ios_base::failure(
-                "ERROR: couldn't write to file " + m_Name +
-                ", in call to FilePointer* transport write\n");
+                "ERROR: written size + " + std::to_string(writtenSize) +
+                " is not equal to intended size " + std::to_string(size) +
+                " in file " + m_Name + ", in call to FilePointer Write\n");
+        }
+    };
+
+    if (size > DefaultMaxFileBatchSize)
+    {
+        const size_t batches = size / DefaultMaxFileBatchSize;
+        const size_t remainder = size % DefaultMaxFileBatchSize;
+
+        size_t position = 0;
+        for (size_t b = 0; b < batches; ++b)
+        {
+            lf_Write(&buffer[position], DefaultMaxFileBatchSize);
+            position += DefaultMaxFileBatchSize;
         }
+        lf_Write(&buffer[position], remainder);
+    }
+    else
+    {
+        lf_Write(buffer, size);
     }
 }
 
-void FilePointer::Flush() { fflush(m_File); }
+void FilePointer::Flush()
+{
+    const int status = std::fflush(m_File);
+
+    if (status == EOF)
+    {
+        throw std::ios_base::failure("ERROR: couldn't flush file " + m_Name +
+                                     ", in call to FilePointer Flush\n");
+    }
+}
 
 void FilePointer::Close()
 {
-    fclose(m_File);
+    if (m_Profiler.IsActive)
+    {
+        m_Profiler.Timers.at("close").Resume();
+    }
+
+    const int status = std::fclose(m_File);
+
+    if (m_Profiler.IsActive)
+    {
+        m_Profiler.Timers.at("close").Pause();
+    }
+
+    if (status == EOF)
+    {
+        throw std::ios_base::failure("ERROR: couldn't close file " + m_Name +
+                                     ", in call to FilePointer Write\n");
+    }
 
     m_IsOpen = false;
 }
 
 } // end namespace transport
-} // namespace adios
+} // end namespace adios

source/adios2/toolkit/transport/file/FilePointer.h

@@ -11,9 +11,8 @@
 #ifndef ADIOS2_TOOLKIT_TRANSPORT_FILE_FILEPOINTER_H_
 #define ADIOS2_TOOLKIT_TRANSPORT_FILE_FILEPOINTER_H_
 
-#include <stdio.h> // FILE*
+#include <cstdio> // FILE*
 
-#include "adios2/ADIOSConfig.h"
 #include "adios2/toolkit/transport/Transport.h"
 
 namespace adios2
@@ -40,13 +39,13 @@ public:
 
     void Write(const char *buffer, size_t size) final;
 
-    void Flush();
+    void Flush() final;
 
-    void Close();
+    void Close() final;
 
 private:
     /** C File pointer */
-    FILE *m_File = nullptr; // NULL or nullptr?
+    std::FILE *m_File = nullptr; // NULL or nullptr?
 };
 
 } // end namespace transport

source/adios2/toolkit/transport/file/FileStream.cpp

@@ -26,6 +26,15 @@ FileStream::FileStream(MPI_Comm mpiComm, const bool debugMode)
 
 void FileStream::Open(const std::string &name, const OpenMode openMode)
 {
+    if (m_DebugMode)
+    {
+        if (name.empty())
+        {
+            throw std::invalid_argument(
+                "ERROR: file name is empty, in call to FilePointer Open\n");
+        }
+    }
+
     m_Name = name;
     m_OpenMode = openMode;
 
@@ -43,72 +52,87 @@ void FileStream::Open(const std::string &name, const OpenMode openMode)
         m_FileStream.open(name, std::fstream::in);
     }
 
-    if (m_DebugMode)
+    if (!m_FileStream)
     {
-        if (!m_FileStream)
-        {
-            throw std::ios_base::failure("ERROR: couldn't open file " + name +
-                                         ", in call to FileStream Open\n");
-        }
+        throw std::ios_base::failure("ERROR: couldn't open file " + m_Name +
+                                     ", in call to FileStream Open\n");
     }
+
     m_IsOpen = true;
 }
 
 void FileStream::SetBuffer(char *buffer, size_t size)
 {
     m_FileStream.rdbuf()->pubsetbuf(buffer, size);
-    if (m_DebugMode)
+
+    if (!m_FileStream)
     {
-        if (!m_FileStream)
-        {
-            throw std::ios_base::failure("ERROR: couldn't SetBuffer to file " +
-                                         m_Name +
-                                         ", in call to FileStream SetBuffer\n");
-        }
+        throw std::ios_base::failure("ERROR: couldn't set buffer in file " +
+                                     m_Name +
+                                     ", in call to FileStream SetBuffer\n");
     }
 }
 
 void FileStream::Write(const char *buffer, size_t size)
 {
-    m_FileStream.write(buffer, size);
+    auto lf_Write = [&](const char *buffer, size_t size) {
+
+        if (m_Profiler.IsActive)
+        {
+            m_Profiler.Timers.at("write").Resume();
+        }
+        m_FileStream.write(buffer, static_cast<std::streamsize>(size));
+
+        if (m_Profiler.IsActive)
+        {
+            m_Profiler.Timers.at("write").Pause();
+        }
 
-    if (m_DebugMode)
-    {
         if (!m_FileStream)
         {
             throw std::ios_base::failure("ERROR: couldn't write to file " +
-                                         m_Name +
-                                         ", in call to FileStream write\n");
+                                         m_Name + ", in call to FILE* Write\n");
+        }
+    };
+
+    if (size > DefaultMaxFileBatchSize)
+    {
+        const size_t batches = size / DefaultMaxFileBatchSize;
+        const size_t remainder = size % DefaultMaxFileBatchSize;
+
+        size_t position = 0;
+        for (size_t b = 0; b < batches; ++b)
+        {
+            lf_Write(&buffer[position], DefaultMaxFileBatchSize);
+            position += DefaultMaxFileBatchSize;
         }
+        lf_Write(&buffer[position], remainder);
+    }
+    else
+    {
+        lf_Write(buffer, size);
     }
 }
 
 void FileStream::Flush()
 {
     m_FileStream.flush();
-    if (m_DebugMode)
+    if (!m_FileStream)
     {
-        if (!m_FileStream)
-        {
-            throw std::ios_base::failure("ERROR: couldn't flush to file " +
-                                         m_Name +
-                                         ", in call to FileStream Flush\n");
-        }
+        throw std::ios_base::failure("ERROR: couldn't flush to file " + m_Name +
+                                     ", in call to FileStream Flush\n");
     }
 }
 
 void FileStream::Close()
 {
     m_FileStream.close();
-    if (m_DebugMode)
+    if (!m_FileStream)
     {
-        if (!m_FileStream)
-        {
-            throw std::ios_base::failure("ERROR: couldn't close file " +
-                                         m_Name +
-                                         ", in call to FileStream Close\n");
-        }
+        throw std::ios_base::failure("ERROR: couldn't close file " + m_Name +
+                                     ", in call to FileStream Close\n");
     }
+
     m_IsOpen = false;
 }
 

testing/adios2/engine/bp/CMakeLists.txt

@@ -9,6 +9,15 @@ if(NOT ADIOS2_HAVE_MPI)
 
   add_executable(TestBPWriteRead TestBPWriteRead.cpp)
   target_link_libraries(TestBPWriteRead adios2 gtest adios1::adios)
+  
+  add_executable(TestBPWriteReadstdio TestBPWriteReadstdio.cpp)
+  target_link_libraries(TestBPWriteReadstdio adios2 gtest adios1::adios)
+
+  add_executable(TestBPWriteReadfstream TestBPWriteReadfstream.cpp)
+  target_link_libraries(TestBPWriteReadfstream adios2 gtest adios1::adios)
+  
 
   gtest_add_tests(TARGET TestBPWriteRead)
+  gtest_add_tests(TARGET TestBPWriteReadstdio)
+  gtest_add_tests(TARGET TestBPWriteReadfstream)
 endif()