BP1Writer.h

/*
 * Distributed under the OSI-approved Apache License, Version 2.0.  See
 * accompanying file Copyright.txt for details.
 *
 * BP1.h
 *
 *  Created on: Jan 24, 2017
 *      Author: wfg
 */

#ifndef BP1WRITER_H_
#define BP1WRITER_H_

/// \cond EXCLUDE_FROM_DOXYGEN
#include <algorithm> //std::count, std::copy, std::for_each
#include <cmath>     //std::ceil
#include <cstring>   //std::memcpy

#include "utilities/format/bp1/BP1Base.h"
#include "utilities/format/bp1/BP1Structs.h"
/// \endcond

#include "capsule/heap/STLVector.h"
#include "core/Variable.h"
#include "functions/adiosFunctions.h"
#include "functions/adiosTemplates.h"

namespace adios
{
namespace format
{

class BP1Writer : public BP1Base
{

public:
    unsigned int m_Threads = 1; ///< thread operations in large array (min,max)
    unsigned int m_Verbosity = 0; ///< statistics verbosity, only 0 is supported
    float m_GrowthFactor = 1.5;   ///< memory growth factor
    const std::uint8_t m_Version = 3; ///< BP format version

    /**
     * Calculates the Process Index size in bytes according to the BP format,
     * including list of method with no parameters (for now)
     * @param name
     * @param timeStepName
     * @param numberOfTransports
     * @return size of pg index
     */
    std::size_t GetProcessGroupIndexSize(
        const std::string name, const std::string timeStepName,
        const std::size_t numberOfTransports) const noexcept;

    /**
     * Writes a process group index PGIndex and list of methods (from
     * transports),
     * done at Open or aggregation of new time step
     * Version that operates on a single heap buffer and metadataset.
     * @param isFortran
     * @param name
     * @param processID
     * @param transports
     * @param buffer
     * @param metadataSet
     */
    void WriteProcessGroupIndex(
        const bool isFortran, const std::string name,
        const std::uint32_t processID,
        const std::vector<std::shared_ptr<Transport>> &transports,
        capsule::STLVector &heap, BP1MetadataSet &metadataSet) const noexcept;

    /**
     * Returns the estimated variable index size
     * @param group
     * @param variableName
     * @param variable
     * @param verbosity
     * @return variable index size
     */
    template <class T>
    size_t GetVariableIndexSize(const Variable<T> &variable) const noexcept
    {
        // size_t indexSize = varEntryLength + memberID + lengthGroupName +
        // groupName + lengthVariableName + lengthOfPath + path + datatype
        size_t indexSize = 23; // without characteristics
        indexSize += variable.m_Name.size();

        // characteristics 3 and 4, check variable number of dimensions
        const std::size_t dimensions =
            variable.DimensionsSize(); // commas in CSV + 1
        indexSize += 28 * dimensions;  // 28 bytes per dimension
        indexSize += 1;                // id

        // characteristics, offset + payload offset in data
        indexSize += 2 * (1 + 8);
        // characteristic 0, if scalar add value, for now only allowing string
        if (dimensions == 1)
        {
            indexSize += sizeof(T);
            indexSize += 1; // id
            // must have an if here
            indexSize += 2 + variable.m_Name.size();
            indexSize += 1; // id
        }

        // characteristic statistics
        if (m_Verbosity == 0) // default, only min and max
        {
            indexSize += 2 * (sizeof(T) + 1);
            indexSize += 1 + 1; // id
        }

        return indexSize + 12; /// extra 12 bytes in case of attributes
                               // need to add transform characteristics
    }

    /**
     * Version for primitive types (except std::complex<T>)
     * @param variable
     * @param heap
     * @param metadataSet
     */
    template <class T>
    void WriteVariableMetadata(const Variable<T> &variable,
                               capsule::STLVector &heap,
                               BP1MetadataSet &metadataSet) const noexcept
    {
        Stats<T> stats = GetStats(variable);
        WriteVariableMetadataCommon(variable, stats, heap, metadataSet);
    }

    /**
     * Overloaded version for std::complex<T> variables
     * @param variable
     * @param heap
     * @param metadataSet
     */
    template <class T>
    void WriteVariableMetadata(const Variable<std::complex<T>> &variable,
                               capsule::STLVector &heap,
                               BP1MetadataSet &metadataSet) const noexcept
    {
        Stats<T> stats = GetStats(variable);
        WriteVariableMetadataCommon(variable, stats, heap, metadataSet);
    }

    /**
     * Expensive part this is only for heap buffers need to adapt to vector of
     * capsules
     * @param variable
     * @param buffer
     */
    template <class T>
    void WriteVariablePayload(const Variable<T> &variable,
                              capsule::STLVector &heap,
                              const unsigned int nthreads = 1) const noexcept
    {
        // EXPENSIVE part, might want to use threads if large, serial for now
        InsertToBuffer(heap.m_Data, variable.m_AppValues, variable.TotalSize());
        heap.m_DataAbsolutePosition += variable.PayLoadSize();
    }

    void Advance(BP1MetadataSet &metadataSet, capsule::STLVector &buffer);

    /**
     * Function that sets metadata (if first close) and writes to a single
     * transport
     * @param metadataSet current rank metadata set
     * @param heap contains data buffer
     * @param transport does a write after data and metadata is setup
     * @param isFirstClose true: metadata has been set and aggregated
     * @param doAggregation true: for N-to-M, false: for N-to-N
     */
    void Close(BP1MetadataSet &metadataSet, capsule::STLVector &heap,
               Transport &transport, bool &isFirstClose,
               const bool doAggregation) const noexcept;

    /**
     * Writes the ADIOS log information (buffering, open, write and close) for a
     * rank process
     * @param rank current rank
     * @param metadataSet contains Profile info for buffering
     * @param transports  contains Profile info for transport open, writes and
     * close
     * @return string for this rank that will be aggregated into profiling.log
     */
    std::string GetRankProfilingLog(
        const int rank, const BP1MetadataSet &metadataSet,
        const std::vector<std::shared_ptr<Transport>> &transports) const
        noexcept;

private:
    template <class T, class U>
    void WriteVariableMetadataCommon(const Variable<T> &variable,
                                     Stats<U> &stats, capsule::STLVector &heap,
                                     BP1MetadataSet &metadataSet) const noexcept
    {
        stats.TimeIndex = metadataSet.TimeStep;

        // Get new Index or point to existing index
        bool isNew = true; // flag to check if variable is new
        BP1Index &varIndex =
            GetBP1Index(variable.m_Name, metadataSet.VarsIndices, isNew);
        stats.MemberID = varIndex.MemberID;

        // write metadata header in data and extract offsets
        stats.Offset = heap.m_DataAbsolutePosition;
        WriteVariableMetadataInData(variable, stats, heap);
        stats.PayloadOffset = heap.m_DataAbsolutePosition;

        // write to metadata  index
        WriteVariableMetadataInIndex(variable, stats, isNew, varIndex);

        ++metadataSet.DataPGVarsCount;
    }

    template <class T, class U>
    void WriteVariableMetadataInData(const Variable<T> &variable,
                                     const Stats<U> &stats,
                                     capsule::STLVector &heap) const noexcept
    {
        auto &buffer = heap.m_Data;

        const std::size_t varLengthPosition =
            buffer.size(); // capture initial position for variable length
        buffer.insert(buffer.end(), 8, 0);              // skip var length (8)
        InsertToBuffer(buffer, &stats.MemberID);        // memberID
        WriteNameRecord(variable.m_Name, buffer);       // variable name
        buffer.insert(buffer.end(), 2, 0);              // skip path
        const std::uint8_t dataType = GetDataType<T>(); // dataType
        InsertToBuffer(buffer, &dataType);
        constexpr char no = 'n'; // isDimension
        InsertToBuffer(buffer, &no);

        // write variable dimensions
        const std::uint8_t dimensions = variable.m_LocalDimensions.size();
        InsertToBuffer(buffer, &dimensions); // count
        std::uint16_t dimensionsLength =
            27 *
            dimensions; // 27 is from 9 bytes for each: var y/n + local, var
                        // y/n + global dimension, var y/n + global offset,
                        // changed for characteristic
        InsertToBuffer(buffer, &dimensionsLength); // length
        WriteDimensionsRecord(buffer, variable.m_LocalDimensions,
                              variable.m_GlobalDimensions, variable.m_Offsets,
                              18, true);

        // CHARACTERISTICS
        WriteVariableCharacteristics(variable, stats, buffer, true);

        // Back to varLength including payload size
        const std::uint64_t varLength = buffer.size() - varLengthPosition +
                                        variable.PayLoadSize() -
                                        8; // remove its own size
        CopyToBufferPosition(buffer, varLengthPosition, &varLength); // length

        heap.m_DataAbsolutePosition +=
            buffer.size() - varLengthPosition; // update absolute position to be
                                               // used as payload position
    }

    template <class T, class U>
    void WriteVariableMetadataInIndex(const Variable<T> &variable,
                                      const Stats<U> &stats, const bool isNew,
                                      BP1Index &index) const noexcept
    {
        auto &buffer = index.Buffer;

        if (isNew == true) // write variable header (might be shared with
                           // attributes index)
        {
            buffer.insert(buffer.end(), 4, 0); // skip var length (4)
            InsertToBuffer(buffer, &stats.MemberID);
            buffer.insert(buffer.end(), 2, 0); // skip group name
            WriteNameRecord(variable.m_Name, buffer);
            buffer.insert(buffer.end(), 2, 0); // skip path

            const std::uint8_t dataType = GetDataType<T>();
            InsertToBuffer(buffer, &dataType);

            // Characteristics Sets Count in Metadata
            index.Count = 1;
            InsertToBuffer(buffer, &index.Count);
        }
        else // update characteristics sets count
        {
            const std::size_t characteristicsSetsCountPosition =
                15 + variable.m_Name.size();
            ++index.Count;
            CopyToBufferPosition(buffer, characteristicsSetsCountPosition,
                                 &index.Count); // test
        }

        WriteVariableCharacteristics(variable, stats, buffer);
    }

    template <class T, class U>
    void WriteVariableCharacteristics(const Variable<T> &variable,
                                      const Stats<U> &stats,
                                      std::vector<char> &buffer,
                                      const bool addLength = false) const
        noexcept
    {
        const std::size_t characteristicsCountPosition =
            buffer.size(); // very important to track as writer is going back to
                           // this position
        buffer.insert(buffer.end(), 5,
                      0); // skip characteristics count(1) + length (4)
        std::uint8_t characteristicsCounter = 0;

        // DIMENSIONS
        std::uint8_t characteristicID = characteristic_dimensions;
        InsertToBuffer(buffer, &characteristicID);
        const std::uint8_t dimensions = variable.m_LocalDimensions.size();

        if (addLength == true)
        {
            const std::int16_t lengthOfDimensionsCharacteristic =
                24 * dimensions +
                3; // 24 = 3 local, global, offset x 8 bytes/each
            InsertToBuffer(buffer, &lengthOfDimensionsCharacteristic);
        }

        InsertToBuffer(buffer, &dimensions); // count
        const std::uint16_t dimensionsLength = 24 * dimensions;
        InsertToBuffer(buffer, &dimensionsLength); // length
        WriteDimensionsRecord(buffer, variable.m_LocalDimensions,
                              variable.m_GlobalDimensions, variable.m_Offsets,
                              16, addLength);
        ++characteristicsCounter;

        // VALUE for SCALAR or STAT min, max for ARRAY
        WriteBoundsRecord(variable.m_IsScalar, stats, buffer,
                          characteristicsCounter, addLength);
        // TIME INDEX
        WriteCharacteristicRecord(characteristic_time_index, stats.TimeIndex,
                                  buffer, characteristicsCounter, addLength);

        if (addLength == false) // only in metadata offset and payload offset
        {
            WriteCharacteristicRecord(characteristic_offset, stats.Offset,
                                      buffer, characteristicsCounter);
            WriteCharacteristicRecord(characteristic_payload_offset,
                                      stats.PayloadOffset, buffer,
                                      characteristicsCounter);
        }
        // END OF CHARACTERISTICS

        // Back to characteristics count and length
        CopyToBufferPosition(buffer, characteristicsCountPosition,
                             &characteristicsCounter); // count (1)
        const std::uint32_t characteristicsLength =
            buffer.size() - characteristicsCountPosition - 4 -
            1; // remove its own length (4 bytes) + characteristic counter ( 1
               // byte
               // )
        CopyToBufferPosition(buffer, characteristicsCountPosition + 1,
                             &characteristicsLength); // length
    }

    /**
     * Writes from &buffer[position]:  [2
     * bytes:string.length()][string.length():
     * string.c_str()]
     * @param name
     * @param buffer
     * @param position
     */
    void WriteNameRecord(const std::string name,
                         std::vector<char> &buffer) const noexcept;

    /**
     * Write a dimension record for a global variable used by
     * WriteVariableCommon
     * @param buffer
     * @param position
     * @param localDimensions
     * @param globalDimensions
     * @param offsets
     * @param addType true: for data buffers, false: for metadata buffer and
     * data
     * characteristic
     */
    void WriteDimensionsRecord(std::vector<char> &buffer,
                               const std::vector<std::size_t> &localDimensions,
                               const std::vector<std::size_t> &globalDimensions,
                               const std::vector<std::size_t> &offsets,
                               const unsigned int skip,
                               const bool addType = false) const noexcept;

    /**
     * GetStats for primitive types except std::complex<T> types
     * @param variable
     * @return stats
     */
    template <class T>
    Stats<T> GetStats(const Variable<T> &variable) const noexcept
    {
        Stats<T> stats;
        const std::size_t valuesSize = variable.TotalSize();

        if (m_Verbosity == 0)
        {
            if (valuesSize >=
                10000000) // ten million? this needs actual results
                          // //here we can make decisions for threads
                          // based on valuesSize
                GetMinMax(variable.m_AppValues, valuesSize, stats.Min,
                          stats.Max,
                          m_Threads); // here we can add cores from constructor
            else
                GetMinMax(variable.m_AppValues, valuesSize, stats.Min,
                          stats.Max);
        }
        return stats;
    }

    /**
     * GetStats for std::complex<T> types
     * @param variable
     * @return stats
     */
    template <class T>
    Stats<T> GetStats(const Variable<std::complex<T>> &variable) const noexcept
    {
        Stats<T> stats;
        const std::size_t valuesSize = variable.TotalSize();

        if (m_Verbosity == 0)
        {
            if (valuesSize >=
                10000000) // ten million? this needs actual results
                          // //here we can make decisions for threads
                          // based on valuesSize
                GetMinMax(variable.m_AppValues, valuesSize, stats.Min,
                          stats.Max, m_Threads);
            else
                GetMinMax(variable.m_AppValues, valuesSize, stats.Min,
                          stats.Max);
        }
        return stats;
    }

    template <class T>
    void WriteBoundsRecord(const bool isScalar, const Stats<T> &stats,
                           std::vector<char> &buffer,
                           std::uint8_t &characteristicsCounter,
                           const bool addLength) const noexcept
    {
        if (isScalar == true)
        {
            WriteCharacteristicRecord(
                characteristic_value, stats.Min, buffer, characteristicsCounter,
                addLength); // stats.min = stats.max = value
            return;
        }

        if (m_Verbosity == 0) // default verbose
        {
            WriteCharacteristicRecord(characteristic_min, stats.Min, buffer,
                                      characteristicsCounter, addLength);
            WriteCharacteristicRecord(characteristic_max, stats.Max, buffer,
                                      characteristicsCounter, addLength);
        }
    }

    /**
     * Write a characteristic value record to buffer
     * @param id
     * @param value
     * @param buffers
     * @param positions
     * @param characvteristicsCounter to be updated by 1
     * @param addLength true for data, false for metadata
     */
    template <class T>
    void WriteCharacteristicRecord(const std::uint8_t characteristicID,
                                   const T &value, std::vector<char> &buffer,
                                   std::uint8_t &characteristicsCounter,
                                   const bool addLength = false) const noexcept
    {
        const std::uint8_t id = characteristicID;
        InsertToBuffer(buffer, &id);

        if (addLength == true)
        {
            const std::uint16_t lengthOfCharacteristic = sizeof(T); // id
            InsertToBuffer(buffer, &lengthOfCharacteristic);
        }

        InsertToBuffer(buffer, &value);
        ++characteristicsCounter;
    }

    /**
     * Returns corresponding index of type BP1Index, if doesn't exists creates a
     * new one.
     * Used for variables and attributes
     * @param name variable or attribute name to look for index
     * @param indices look up hash table of indices
     * @param isNew true: index is newly created, false: index already exists in
     * indices
     * @return reference to BP1Index in indices
     */
    BP1Index &GetBP1Index(const std::string name,
                          std::unordered_map<std::string, BP1Index> &indices,
                          bool &isNew) const noexcept;

    /**
     * Flattens the data and fills the pg length, vars count, vars length and
     * attributes
     * @param metadataSet
     * @param buffer
     */
    void FlattenData(BP1MetadataSet &metadataSet,
                     capsule::STLVector &buffer) const noexcept;

    /**
     * Flattens the metadata indices into a single metadata buffer in capsule
     * @param metadataSet
     * @param buffer
     */
    void FlattenMetadata(BP1MetadataSet &metadataSet,
                         capsule::STLVector &buffer) const noexcept;
};

} // end namespace format
} // end namespace adios

#endif /* BP1WRITER_H_ */