Implementing WriteBP and DataMan functions.

To do:

Create Format class and assign BP Format to it

Compile and test a Hello DataMan example

examples/hello/dataman/dataman.xml

+<?xml version="1.0"?>
+<adios-config>
+
+  <!--  Changes to ADIOS-1 XML
+        Accommodate for the following extensions and flexiblity:
+        - Multiple engines e.g. SIRIUS can have it's own
+        - Do not bind one transport to a group, i.e. we can reuse a group in multiple I/O 
+          with different transports 
+        - Do not bind one group to each output file, i.e., we can write variables from
+          different groups into a single file at the same step
+   -->
+
+  <!--  Execution Engines: default-engine, sirius, dataman, hdf5
+   -->
+
+
+  <adios-group name="solid">
+    <var name="NX" type="integer"/>
+    <var name="size" type="integer"/>
+    <var name="rank" type="integer"/>
+
+    <global-bounds dimensions="size,NX" offsets="rank,0">
+       <var name="temperature" gwrite="solid.t" type="double"              dimensions="1,NX"/>
+       <var name="pressure"    gwrite="solid.p" type="std::vector<double>" dimensions="1,NX"/>
+    </global-bounds>
+
+    <attribute name="temperature/description" 
+        value="Global array written from 'size' processes over several timesteps" 
+        type="string"/>
+  </adios-group>
+
+
+  <adios-group name="fluid">
+    <var name="NX" type="integer"/>
+    <var name="size" type="integer"/>
+    <var name="rank" type="integer"/>
+
+    <global-bounds dimensions="size,NX" offsets="rank,0">
+       <var name="temperature" gwrite="fluid.t" type="double"              dimensions="1,NX"/>
+       <var name="pressure"    gwrite="fluid.p" type="std::vector<double>" dimensions="1,NX"/>
+    </global-bounds>
+
+    <attribute name="temperature/description" 
+        value="Global array written from 'size' processes over several timesteps" 
+        type="string"/>
+  </adios-group>
+
+  
+  <!--  Associate an engine with a group and parameterize it here
+        The manager will define the transport(s) used in an output 
+  -->  
+  <method name="dataMan2Fermi" group="fluid" engine="DataMan" real_time="yes">
+      <transport="ZeroMQ" target="128.1.1.20">options to ZeroMQ transport</transport>
+      <transport="MDTM" target="128.1.1.1.10">options to MDTM transport</transport>
+      <!--  filenames overwrite the value provided in Open() call -->
+  </method>
+      
+  
+
+
+</adios-config>
+

examples/hello/dataman/helloDataMan.cpp

@@ -29,7 +29,7 @@ int main( int argc, char* argv [] )
     const bool adiosDebug = true;
 
     //Application variable
-    std::vector<double> myInts = { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
+    std::vector<int> myInts = { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
     int myIntsSize = 10;
 
     try
@@ -41,12 +41,19 @@ int main( int argc, char* argv [] )
         adios.DefineVariable( groupName, "myIntsSize", "int" );
         adios.DefineVariable( groupName, "myInts", "double", "myIntsSize" );
 
-        //Define method
+        //Define method...
         const std::string methodName( "DataManSend" );
-        adios.DeclareMethod( methodName, "DataMan" );
-        adios.AddCapsule( methodName, "Heap" );
-        adios.AddTransport( methodName, "POSIX", "fname=myfile.bp" );
-        adios.AddTransport( methodName, "TCP_IP", "fname=myfile.tcp" );
+        adios.DeclareMethod( methodName, "DataMan" ); //2nd item is type and must be supported e.g. Writer (empty default), DataMan, Sirius, etc.
+//        adios.AddTransport( methodName, "ZeroMQ", "format=json", "tcp=128.11.1.1.2", "real_time=yes" );
+//        adios.AddTransport( methodName, "MDTM", "format=otherFormat", "tcp=128.11.1.1.2" );
+//        adios.AddTransport( methodName, "POSIX", "fname=myfile.bp" ); //you can write things to file as well
+
+
+        //this illustrates method uniqueness
+        const std::string methodName2( "DataManSend2" );
+        adios.DeclareMethod( methodName2, "DataMan" ); //should this be variadic?
+        adios.AddTransport( methodName2, "ZeroMQ", "format=json", "tcp=128.11.1.1.2", "realtime_change=yes" );
+
 
         //Create engine handler and Write
         int handler = adios.Open( "myInts.bp", "w", methodName );
@@ -54,8 +61,10 @@ int main( int argc, char* argv [] )
 
         double varDouble = 10.;
         adios.Write<double>( handler, "myIntsSize", &varDouble );
-        adios.Write<double>( handler, "myInts", &myInts.front() );
+        adios.Write<int>( handler, "myInts", &myInts.front() );
         adios.Close( handler );
+
+        int handler2 = adios.Open( "somethingelse.bp", "w", methodName2 );
     }
     catch( std::invalid_argument& e )
     {

include/core/Transport.h

@@ -91,7 +91,7 @@ protected:
     const bool m_DebugMode = false; ///< if true: additional checks and exceptions
     /**
      * Initialize particular derived transport class members
-     * @param arguments particular transport arguments from ADIOS Open variadic function
+     * @param arguments particular transport arguments
      */
     virtual void Init( const std::vector<std::string>& arguments );
 

include/engine/dataman/DataManTemplates.h

+/*
+ * DataManTemplates.h
+ *
+ *  Created on: Jan 18, 2017
+ *      Author: wfg
+ */
+
+#ifndef DATAMANTEMPLATES_H_
+#define DATAMANTEMPLATES_H_
+
+
+namespace adios
+{
+
+template<class T>
+void DataManWrite( Group& group, const std::string variableName, Variable<T>& variable, Heap& capsule )
+{
+    //here write your magic
+}
+
+
+
+
+
+} //end namespace
+
+
+
+#endif /* DATAMANTEMPLATES_H_ */

include/engine/writer/WriterHelper.h → include/engine/writer/WriterTemplates.h

@@ -5,8 +5,8 @@
  *      Author: wfg
  */
 
-#ifndef WRITERHELPER_H_
-#define WRITERHELPER_H_
+#ifndef WRITERTEMPLATES_H_
+#define WRITERTEMPLATES_H_
 
 
 
@@ -14,18 +14,13 @@ namespace adios
 {
 
 template <class T>
-void WriteHelper( Group& group, const std::string variableName, Variable<T>& variable, Heap& capsule )
+void WriterWriteVariable( Group& group, const std::string variableName, Variable<T>& variable, Heap& capsule )
 {
     const auto localDimensions = group.GetDimensions( variable.DimensionsCSV );
     const auto size = GetTotalSize( localDimensions );
     T min, max;
     GetMinMax( variable.Values, size, min, max );
 
-
-
-
-
-
 //
 //    const std::size_t bytesToWrite = localSize * sizeof( double ); //size of values + min + max in bytes
 //    const std::size_t currentSize = m_Data.size(); //0 the first time
@@ -95,4 +90,4 @@ void WriteHelper( Group& group, const std::string variableName, Variable<T>& var
 
 
 
-#endif /* WRITERHELPER_H_ */
+#endif /* WRITERTEMPLATES_H_ */

include/functions/bp1Write.h

@@ -181,36 +181,76 @@ void WriteVariableToBuffers( const Group& group, const std::string variableName,
 
    unsigned int characteristicsCounter = 0; //used for characteristics count, characteristics length will be calculated at the end
 
+   //Get dimensions
+   const std::vector<unsigned long long int> localDimensions = group.GetDimensions( variable.DimensionsCSV );
+
+   //write to metadata characteristic
    //characteristic: dimension
    const std::uint8_t characteristicID = characteristic_dimensions;
    WriteToBuffers( metadataBuffers, &characteristicID, 1, metadataOffset );
+   const std::uint8_t dimensions = localDimensions.size();
+   WriteToBuffers( metadataBuffers, &dimensions, 1, metadataOffset );
+   const std::uint16_t dimensionsLength = dimensions * 24; //24 is from 8 bytes for each: local, global dimension, global offset
+   WriteToBuffers( metadataBuffers, &dimensions, 2, metadataOffset );
 
-   std::size_t dimensions = std::count( variable.DimensionsCSV.begin(), variable.DimensionsCSV.end(), ',' ) + 1;
-   if( dimensions == 1 ) //scalar
+   //dimensions in data buffer
+   if( writeDimensionsInData == true )
    {
-       if( writeDimensionsInData == true )
-       {
-           const char yes = 'y';
-           WriteToBuffers( dataBuffers, &yes, 1, dataOffset );
+       const char yes = 'y';
+       WriteToBuffers( dataBuffers, &yes, 1, dataOffset );
+       //for now only support unsigned long integer value (8 bytes), as in metadata
+       WriteToBuffers( dataBuffers, &dimensions, 1, dataOffset );
+       const std::uint16_t dimensionsLengthInData = dimensions * 27; //27 is from 9 bytes for each: var y/n + local, var y/n + global dimension, var y/n + global offset
+       WriteToBuffers( dataBuffers, &dimensionsLengthInData, 2, dataOffset );
+   }
+   else
+   {
+       const char no = 'n';
+       WriteToBuffers( dataBuffers, &no, 1, dataOffset );
+   }
 
-       }
-       else
+
+   if( variable.GlobalBoundsIndex == -1 ) //local variable
+   {
+       for( unsigned int d = 0; d < (unsigned int)localDimensions.size(); ++d )
        {
-           const char no = 'n';
-           WriteToBuffers( dataBuffers, &no, 1, dataOffset );
+           //metadata
+           WriteToBuffers( metadataBuffers, &localDimensions[d], 8, metadataOffset );
+           metadataOffset += 16; //skipping global dimension(8), global offset (8)
+
+           //data
+           if( writeDimensionsInData == true )
+           {
+               const char no = 'n';
+               WriteToBuffers( dataBuffers, &no, 1, dataOffset );
+               WriteToBuffers( dataBuffers, &localDimensions[d], 8, dataOffset );
+               dataOffset += 18; //skipping var no + global dimension(8), var no + global offset (8)
+           }
        }
    }
-   else //multidimensional array
+   else //global variable
    {
-       const std::vector<unsigned long long int> localDimensions = group.GetDimensions( variable.DimensionsCSV );
-       std::vector<unsigned long long int> globalDimensions;
-       std::vector<unsigned long long int> globalOffsets;
-       if( variable.GlobalBoundsIndex > -1 ) //global variable
+       std::vector<unsigned long long int> globalDimensions = group.GetDimensions( group.m_GlobalBounds[variable.GlobalBoundsIndex].first );
+       std::vector<unsigned long long int> globalOffsets = group.GetDimensions( group.m_GlobalBounds[variable.GlobalBoundsIndex].second );
+
+       for( unsigned int d = 0; d < (unsigned int)localDimensions.size(); ++d )
        {
-           globalDimensions = group.GetDimensions( group.m_GlobalBounds[variable.GlobalBoundsIndex].first );
-           globalOffsets = group.GetDimensions( group.m_GlobalBounds[variable.GlobalBoundsIndex].second );
+           //metadata
+           WriteToBuffers( metadataBuffers, &localDimensions[d], 8, metadataOffset );
+           WriteToBuffers( metadataBuffers, &globalDimensions[d], 8, metadataOffset );
+           WriteToBuffers( metadataBuffers, &globalOffsets[d], 8, metadataOffset );
+
+           //data
+           if( writeDimensionsInData == true )
+           {
+               const char no = 'n';
+               WriteToBuffers( dataBuffers, &no, 1, dataOffset );
+               WriteToBuffers( dataBuffers, &localDimensions[d], 8, dataOffset );
+               WriteToBuffers( dataBuffers, &no, 1, dataOffset );
+               WriteToBuffers( dataBuffers, &localDimensions[d], 8, dataOffset );
+
+           }
        }
-
    }
 
 

include/transport/MdtmMan.h

+/*
+ * MdtmMan.h
+ *
+ *  Created on: Jan 18, 2017
+ *      Author: wfg
+ */
+
+#ifndef MDTMMAN_H_
+#define MDTMMAN_H_
+
+/// \cond EXCLUDE_FROM_DOXYGEN
+#include <thread>
+#include <queue>
+/// \endcond
+
+
+#ifdef HAVE_DATAMAN
+#include "external/json.hpp"
+#endif
+
+#include "core/Transport.h"
+
+
+namespace adios
+{
+
+class MdtmMan : public Transport
+{
+
+public:
+
+
+    MdtmMan( ); ///< default empty constructor
+
+
+    MdtmMan( const std::string localIP, const std::string remoteIP, const std::string mode, const std::string prefix,
+             const int numberOfPipes, const std::vector<int> tolerance, const std::vector<int> priority,
+             MPI_Comm mpiComm, const bool debugMode );
+
+
+    ~MdtmMan( );
+
+
+    void Open( const std::string name, const std::string accessMode ) final;
+
+    void SetBuffer( char* buffer, std::size_t size );
+
+    /**
+     * We can always overload this function in the base class and accept other types of data pointers, e.g. Write( json* );
+     * I'm sticking with char* as it's more general (only C++ libraries, e.g. boost understand std::std::vector, MPI, POSIX, Infiniband use pointer*)
+     * @param buffer
+     * @param size
+     */
+    void Write( const char* buffer, std::size_t size );
+
+    void Flush( ); ///< not sure if this one is needed...
+
+    void Close( );
+
+
+private:
+
+    std::string m_LocalIP; ///< local ip address, can change over time
+    std::string m_RemoteIP; ///<  remote ip address, can change over time
+    std::string m_Mode; ///< send/write, receive/read
+    std::string m_Prefix; ///< prefix given to message
+    int m_NumberOfPipes; ///< should it be unsigned int?
+    std::vector<int> m_Tolerace;
+    std::vector<int> m_Priority;
+
+    /**
+     * Should we change data to char* ?
+     * @param data
+     * @param doid
+     * @param variable
+     * @param dType
+     * @param putShape
+     * @param varShape
+     * @param offset
+     * @param timestep
+     * @param tolerance
+     * @param priority
+     * @return
+     */
+    int Put( const void* data, const std::string doid, const std::string variable, const std::string dType,
+             const std::vector<std::uint64_t>& putShape, const std::vector<uint64_t>& varShape, const std::vector<uint64_t>& offset,
+             const std::uint64_t timestep, const int tolerance, const int priority );
+
+    /**
+     * Should we change data to char* ?
+     * @param data
+     * @param doid
+     * @param variable
+     * @param dType
+     * @param putShape
+     * @param varShape
+     * @param offset
+     * @param timestep
+     * @param tolerance
+     * @param priority
+     * @return
+     */
+    int Get( void* data, const std::string doid, const std::string variable, const std::string dType,
+             const std::vector<std::uint64_t>& putShape, const std::vector<uint64_t>& varShape, const std::vector<uint64_t>& offset,
+             const std::uint64_t timestep, const int tolerance, const int priority );
+
+    /**
+     *
+     * @param data
+     * @param doid
+     * @param variable
+     * @param dType
+     * @param varShape
+     * @param timestep
+     * @return
+     */
+    int Get( void *data, const std::string doid, const std::string variable, const std::string dType,
+             std::vector<std::uint64_t>& varShape, const std::uint64_t timestep );
+
+    /**
+     *
+     * @param jData
+     */
+    void OnReceive( nlohmann::json& jData );
+};
+
+
+} //end namespace
+
+
+
+#endif /* MDTMMAN_H_ */

src/engine/dataman/DataMan.cpp

@@ -5,10 +5,10 @@
  *      Author: wfg
  */
 
-#include "../../include/engine/dataman/DataMan.h"
-
 #include <iostream>
 
+
+#include "engine/dataman/DataMan.h"
 #include "core/Support.h"
 
 //supported capsules
@@ -27,7 +27,7 @@ namespace engine
 
 DataMan::DataMan( const std::string streamName, const std::string accessMode, const MPI_Comm mpiComm,
                   const Method& method, const bool debugMode, const unsigned int cores ):
-    Engine( "SingleBP", streamName, accessMode, mpiComm, method, debugMode, cores, " SingleBP constructor (or call to ADIOS Open).\n" )
+    Engine( "DataMan", streamName, accessMode, mpiComm, method, debugMode, cores, " DataMan constructor (or call to ADIOS Open).\n" )
 {
     Init( );
 }
@@ -187,7 +187,7 @@ void DataMan::InitCapsules( )
     {
         if( m_Method.m_CapsuleParameters.size() > 1 )
         {
-            throw std::invalid_argument( "ERROR: SingleBP engine only allows one heap buffer, in " + m_Name +
+            throw std::invalid_argument( "ERROR: DataMan engine only allows one heap buffer, in " + m_Name +
                                          m_EndMessage );
         }
         else if( m_Method.m_CapsuleParameters.size() == 1 )
@@ -199,7 +199,7 @@ void DataMan::InitCapsules( )
                                 ", in " + m_Name + m_EndMessage );
 
             if( !( itType->second == "Heap" || itType->second == "HEAP" ) )
-                throw std::invalid_argument( "ERROR: SingleBP doesn't support Capsule of buffer type " +
+                throw std::invalid_argument( "ERROR: DataMan doesn't support Capsule of buffer type " +
                                               itType->second + " in " + m_Name + m_EndMessage );
         }
     }

src/engine/writer/Writer.cpp

@@ -8,7 +8,8 @@
 #include <iostream>
 
 #include "engine/writer/Writer.h"
-#include "engine/writer/WriterHelper.h"
+
+#include "../../../include/engine/writer/WriterTemplates.h"
 #include "core/Support.h"
 #include "functions/adiosFunctions.h" //GetTotalSize
 
@@ -26,7 +27,7 @@ namespace adios
 
 
 Writer::Writer( const std::string streamName, const std::string accessMode, const MPI_Comm mpiComm,
-                    const Method& method, const bool debugMode, const unsigned int cores ):
+                const Method& method, const bool debugMode, const unsigned int cores ):
     Engine( "Writer", streamName, accessMode, mpiComm, method, debugMode, cores, " Writer constructor (or call to ADIOS Open).\n" )
 {
     Init( );
@@ -135,9 +136,8 @@ void Writer::Write( const std::string variableName, const unsigned short* values
 void Writer::Write( const std::string variableName, const int* values )
 {
     const unsigned int index = PreSetVariable( *m_Group, variableName, Support::DatatypesAliases.at("int"), " from call to Write int*" );
-    Variable<int>& variable = m_Group->m_Int[index];
-    variable.Values = values;
-    WriteHelper( m_Group, variable, )
+    m_Group->m_Int[index].Values = values;
+    //here call the Template function WriteHelper( m_Group, variable, )
 
 
 }