/*
 * Distributed under the OSI-approved Apache License, Version 2.0.  See
 * accompanying file Copyright.txt for details.
 *
 * BP4Writer.cpp
 *
 *  Created on: Aug 1, 2018
 *      Author: Lipeng Wan wanl@ornl.gov
 */

#include "BP4Writer.h"
#include "BP4Writer.tcc"

#include "adios2/common/ADIOSMacros.h"
#include "adios2/core/IO.h"
#include "adios2/helper/adiosFunctions.h" //CheckIndexRange
#include "adios2/toolkit/profiling/taustubs/tautimer.hpp"
#include "adios2/toolkit/transport/file/FileFStream.h"

#include <ctime>
#include <iostream>

namespace adios2
{
namespace core
{
namespace engine
{

BP4Writer::BP4Writer(IO &io, const std::string &name, const Mode mode,
                     helper::Comm comm)
: Engine("BP4Writer", io, name, mode, std::move(comm)), m_BP4Serializer(m_Comm),
  m_FileDataManager(m_Comm), m_FileMetadataManager(m_Comm),
  m_FileMetadataIndexManager(m_Comm), m_FileDrainer()
{
    TAU_SCOPED_TIMER("BP4Writer::Open");
    m_IO.m_ReadStreaming = false;
    m_EndMessage = " in call to IO Open BP4Writer " + m_Name + "\n";

    Init();
}

StepStatus BP4Writer::BeginStep(StepMode mode, const float timeoutSeconds)
{
    TAU_SCOPED_TIMER("BP4Writer::BeginStep");
    m_BP4Serializer.m_DeferredVariables.clear();
    m_BP4Serializer.m_DeferredVariablesDataSize = 0;
    m_IO.m_ReadStreaming = false;
    return StepStatus::OK;
}

size_t BP4Writer::CurrentStep() const
{
    return m_BP4Serializer.m_MetadataSet.CurrentStep;
}

void BP4Writer::PerformPuts()
{
    TAU_SCOPED_TIMER("BP4Writer::PerformPuts");
    if (m_BP4Serializer.m_DeferredVariables.empty())
    {
        return;
    }

    m_BP4Serializer.ResizeBuffer(m_BP4Serializer.m_DeferredVariablesDataSize,
                                 "in call to PerformPuts");

    for (const std::string &variableName : m_BP4Serializer.m_DeferredVariables)
    {
        const std::string type = m_IO.InquireVariableType(variableName);
        if (type == "compound")
        {
            // not supported
        }
#define declare_template_instantiation(T)                                      \
    else if (type == helper::GetType<T>())                                     \
    {                                                                          \
        Variable<T> &variable = FindVariable<T>(                               \
            variableName, "in call to PerformPuts, EndStep or Close");         \
        PerformPutCommon(variable);                                            \
    }

        ADIOS2_FOREACH_PRIMITIVE_STDTYPE_1ARG(declare_template_instantiation)
#undef declare_template_instantiation
    }
    m_BP4Serializer.m_DeferredVariables.clear();
}

void BP4Writer::EndStep()
{
    TAU_SCOPED_TIMER("BP4Writer::EndStep");
    if (m_BP4Serializer.m_DeferredVariables.size() > 0)
    {
        PerformPuts();
    }

    // true: advances step
    m_BP4Serializer.SerializeData(m_IO, true);

    const size_t currentStep = CurrentStep();
    const size_t flushStepsCount = m_BP4Serializer.m_Parameters.FlushStepsCount;

    if (currentStep % flushStepsCount == 0)
    {
        Flush();
    }
}

void BP4Writer::Flush(const int transportIndex)
{
    TAU_SCOPED_TIMER("BP4Writer::Flush");
    DoFlush(false, transportIndex);
    m_BP4Serializer.ResetBuffer(m_BP4Serializer.m_Data);

    if (m_BP4Serializer.m_Parameters.CollectiveMetadata)
    {
        WriteCollectiveMetadataFile();
    }
}

// PRIVATE
void BP4Writer::Init()
{
    InitParameters();
    InitTransports();
    InitBPBuffer();
}

#define declare_type(T)                                                        \
    void BP4Writer::DoPut(Variable<T> &variable,                               \
                          typename Variable<T>::Span &span,                    \
                          const size_t bufferID, const T &value)               \
    {                                                                          \
        TAU_SCOPED_TIMER("BP4Writer::Put");                                    \
        PutCommon(variable, span, bufferID, value);                            \
    }

ADIOS2_FOREACH_PRIMITIVE_STDTYPE_1ARG(declare_type)
#undef declare_type

#define declare_type(T)                                                        \
    void BP4Writer::DoPutSync(Variable<T> &variable, const T *data)            \
    {                                                                          \
        PutSyncCommon(variable, variable.SetBlockInfo(data, CurrentStep()));   \
        variable.m_BlocksInfo.pop_back();                                      \
    }                                                                          \
    void BP4Writer::DoPutDeferred(Variable<T> &variable, const T *data)        \
    {                                                                          \
        PutDeferredCommon(variable, data);                                     \
    }

ADIOS2_FOREACH_STDTYPE_1ARG(declare_type)
#undef declare_type

void BP4Writer::InitParameters()
{
    m_BP4Serializer.Init(m_IO.m_Parameters, "in call to BP4::Open to write");
    m_WriteToBB = !(m_BP4Serializer.m_Parameters.BurstBufferPath.empty());
    m_DrainBB = m_WriteToBB && m_BP4Serializer.m_Parameters.BurstBufferDrain;
}

void BP4Writer::InitTransports()
{
    // TODO need to add support for aggregators here later
    if (m_IO.m_TransportsParameters.empty())
    {
        Params defaultTransportParameters;
        defaultTransportParameters["transport"] = "File";
        m_IO.m_TransportsParameters.push_back(defaultTransportParameters);
    }

    // only consumers will interact with transport managers
    m_BBName = m_Name;
    if (m_WriteToBB)
    {
        m_BBName = m_BP4Serializer.m_Parameters.BurstBufferPath +
                   PathSeparator + m_Name;
    }

    if (m_BP4Serializer.m_Aggregator.m_IsConsumer)
    {
        // Names passed to IO AddTransport option with key "Name"
        const std::vector<std::string> transportsNames =
            m_FileDataManager.GetFilesBaseNames(m_BBName,
                                                m_IO.m_TransportsParameters);

        // /path/name.bp.dir/name.bp.rank
        m_SubStreamNames = m_BP4Serializer.GetBPSubStreamNames(transportsNames);
        if (m_DrainBB)
        {
            const std::vector<std::string> drainTransportNames =
                m_FileDataManager.GetFilesBaseNames(
                    m_Name, m_IO.m_TransportsParameters);
            m_DrainSubStreamNames =
                m_BP4Serializer.GetBPSubStreamNames(drainTransportNames);
            /* start up BB thread */
            m_FileDrainer.SetVerbose(
                m_BP4Serializer.m_Parameters.BurstBufferVerbose,
                m_BP4Serializer.m_RankMPI);
            m_FileDrainer.Start();
        }
    }

    /* Create the directories either on target or burst buffer if used */
    m_BP4Serializer.m_Profiler.Start("mkdir");
    m_FileDataManager.MkDirsBarrier(m_SubStreamNames,
                                    m_BP4Serializer.m_Parameters.NodeLocal ||
                                        m_WriteToBB);
    if (m_DrainBB)
    {
        /* Create the directories on target anyway by main thread */
        m_FileDataManager.MkDirsBarrier(m_DrainSubStreamNames,
                                        m_BP4Serializer.m_Parameters.NodeLocal);
    }
    m_BP4Serializer.m_Profiler.Stop("mkdir");

    if (m_BP4Serializer.m_Aggregator.m_IsConsumer)
    {
        if (m_BP4Serializer.m_Parameters.AsyncTasks)
        {
            for (size_t i = 0; i < m_IO.m_TransportsParameters.size(); ++i)
            {
                m_IO.m_TransportsParameters[i]["asynctasks"] = "true";
            }
        }

        m_FileDataManager.OpenFiles(m_SubStreamNames, m_OpenMode,
                                    m_IO.m_TransportsParameters,
                                    m_BP4Serializer.m_Profiler.m_IsActive);

        if (m_DrainBB)
        {
            for (const auto &name : m_DrainSubStreamNames)
            {
                m_FileDrainer.AddOperationOpen(name, m_OpenMode);
            }
        }
    }

    if (m_BP4Serializer.m_RankMPI == 0)
    {
        const std::vector<std::string> transportsNames =
            m_FileMetadataManager.GetFilesBaseNames(
                m_BBName, m_IO.m_TransportsParameters);

        m_MetadataFileNames =
            m_BP4Serializer.GetBPMetadataFileNames(transportsNames);

        m_FileMetadataManager.OpenFiles(m_MetadataFileNames, m_OpenMode,
                                        m_IO.m_TransportsParameters,
                                        m_BP4Serializer.m_Profiler.m_IsActive);

        m_MetadataIndexFileNames =
            m_BP4Serializer.GetBPMetadataIndexFileNames(transportsNames);

        m_FileMetadataIndexManager.OpenFiles(
            m_MetadataIndexFileNames, m_OpenMode, m_IO.m_TransportsParameters,
            m_BP4Serializer.m_Profiler.m_IsActive);

        if (m_DrainBB)
        {
            const std::vector<std::string> drainTransportNames =
                m_FileDataManager.GetFilesBaseNames(
                    m_Name, m_IO.m_TransportsParameters);
            m_DrainMetadataFileNames =
                m_BP4Serializer.GetBPMetadataFileNames(drainTransportNames);
            m_DrainMetadataIndexFileNames =
                m_BP4Serializer.GetBPMetadataIndexFileNames(
                    drainTransportNames);

            for (const auto &name : m_DrainMetadataFileNames)
            {
                m_FileDrainer.AddOperationOpen(name, m_OpenMode);
            }
            for (const auto &name : m_DrainMetadataIndexFileNames)
            {
                m_FileDrainer.AddOperationOpen(name, m_OpenMode);
            }
        }
    }
}

void BP4Writer::InitBPBuffer()
{
    if (m_OpenMode == Mode::Append)
    {
        // TODO: Get last pg timestep and update timestep counter in
        format::BufferSTL preMetadataIndex;
        size_t preMetadataIndexFileSize;

        if (m_BP4Serializer.m_RankMPI == 0)
        {
            preMetadataIndexFileSize =
                m_FileMetadataIndexManager.GetFileSize(0);
            preMetadataIndex.m_Buffer.resize(preMetadataIndexFileSize);
            preMetadataIndex.m_Buffer.assign(preMetadataIndex.m_Buffer.size(),
                                             '\0');
            preMetadataIndex.m_Position = 0;
            m_FileMetadataIndexManager.ReadFile(
                preMetadataIndex.m_Buffer.data(), preMetadataIndexFileSize);
        }
        m_Comm.BroadcastVector(preMetadataIndex.m_Buffer);
        preMetadataIndexFileSize = preMetadataIndex.m_Buffer.size();
        if (preMetadataIndexFileSize > 0)
        {
            size_t position = 0;
            position += 28;
            const uint8_t endianness =
                helper::ReadValue<uint8_t>(preMetadataIndex.m_Buffer, position);
            bool IsLittleEndian = true;
            IsLittleEndian = (endianness == 0) ? true : false;
            if (helper::IsLittleEndian() != IsLittleEndian)
            {
                throw std::runtime_error(
                    "ERROR: previous run generated BigEndian bp file, "
                    "this version of ADIOS2 wasn't compiled "
                    "with the cmake flag -DADIOS2_USE_Endian_Reverse=ON "
                    "explicitly, in call to Open\n");
            }
            const size_t pos_last_step = preMetadataIndexFileSize - 64;
            position = pos_last_step;
            const uint64_t lastStep = helper::ReadValue<uint64_t>(
                preMetadataIndex.m_Buffer, position, IsLittleEndian);
            m_BP4Serializer.m_MetadataSet.TimeStep +=
                static_cast<uint32_t>(lastStep);
            m_BP4Serializer.m_MetadataSet.CurrentStep += lastStep;

            if (m_BP4Serializer.m_Aggregator.m_IsConsumer)
            {
                m_BP4Serializer.m_PreDataFileLength =
                    m_FileDataManager.GetFileSize(0);
            }

            if (m_BP4Serializer.m_RankMPI == 0)
            {
                // Get the size of existing metadata file
                m_BP4Serializer.m_PreMetadataFileLength =
                    m_FileMetadataManager.GetFileSize(0);
            }
        }
    }

    if (m_BP4Serializer.m_PreDataFileLength == 0)
    {
        /* This is a new file.
         * Make headers in data buffer and metadata buffer (but do not write
         * them yet so that Open() can stay free of writing to disk)
         */
        if (m_BP4Serializer.m_RankMPI == 0)
        {
            m_BP4Serializer.MakeHeader(m_BP4Serializer.m_Metadata, "Metadata",
                                       false);
            m_BP4Serializer.MakeHeader(m_BP4Serializer.m_MetadataIndex,
                                       "Index Table", true);
        }
        if (m_BP4Serializer.m_Aggregator.m_IsConsumer)
        {
            m_BP4Serializer.MakeHeader(m_BP4Serializer.m_Data, "Data", false);
        }
    }
    else
    {
        if (m_BP4Serializer.m_RankMPI == 0)
        {
            // Set the flag in the header of metadata index table to 1 again
            // to indicate a new run begins
            UpdateActiveFlag(true);
        }
    }

    m_BP4Serializer.PutProcessGroupIndex(
        m_IO.m_Name, m_IO.m_HostLanguage,
        m_FileDataManager.GetTransportsTypes());
}

void BP4Writer::DoFlush(const bool isFinal, const int transportIndex)
{
    if (m_BP4Serializer.m_Aggregator.m_IsActive)
    {
        AggregateWriteData(isFinal, transportIndex);
    }
    else
    {
        WriteData(isFinal, transportIndex);
    }
}

void BP4Writer::DoClose(const int transportIndex)
{
    TAU_SCOPED_TIMER("BP4Writer::Close");
    if (m_BP4Serializer.m_DeferredVariables.size() > 0)
    {
        PerformPuts();
    }

    DoFlush(true, transportIndex);

    if (m_BP4Serializer.m_Aggregator.m_IsConsumer)
    {
        m_FileDataManager.CloseFiles(transportIndex);
        // Delete files from temporary storage if draining was on
        if (m_DrainBB)
        {
            for (const auto &name : m_SubStreamNames)
            {
                m_FileDrainer.AddOperationDelete(name);
            }
        }
    }

    if (m_BP4Serializer.m_Parameters.CollectiveMetadata &&
        m_FileDataManager.AllTransportsClosed())
    {
        WriteCollectiveMetadataFile(true);
    }

    if (m_BP4Serializer.m_Profiler.m_IsActive &&
        m_FileDataManager.AllTransportsClosed())
    {
        // std::cout << "write profiling file!" << std::endl;
        WriteProfilingJSONFile();
    }
    if (m_BP4Serializer.m_Aggregator.m_IsActive)
    {
        m_BP4Serializer.m_Aggregator.Close();
    }

    if (m_BP4Serializer.m_RankMPI == 0)
    {
        // Update the active flag in index to indicate current run is over.
        UpdateActiveFlag(false);

        // close metadata file
        m_FileMetadataManager.CloseFiles();

        // close metadata index file
        m_FileMetadataIndexManager.CloseFiles();

        // Delete metadata files from temporary storage if draining was on
        if (m_DrainBB)
        {
            for (const auto &name : m_MetadataFileNames)
            {
                m_FileDrainer.AddOperationDelete(name);
            }
            for (const auto &name : m_MetadataIndexFileNames)
            {
                m_FileDrainer.AddOperationDelete(name);
            }
            const std::vector<std::string> transportsNames =
                m_FileDataManager.GetFilesBaseNames(
                    m_BBName, m_IO.m_TransportsParameters);
            for (const auto &name : transportsNames)
            {
                m_FileDrainer.AddOperationDelete(name);
            }
        }
    }

    if (m_BP4Serializer.m_Aggregator.m_IsConsumer && m_DrainBB)
    {
        /* Signal the BB thread that no more work is coming */
        m_FileDrainer.Finish();
    }
    // m_BP4Serializer.DeleteBuffers();
}

void BP4Writer::WriteProfilingJSONFile()
{
    TAU_SCOPED_TIMER("BP4Writer::WriteProfilingJSONFile");
    auto transportTypes = m_FileDataManager.GetTransportsTypes();
    auto transportProfilers = m_FileDataManager.GetTransportsProfilers();

    auto transportTypesMD = m_FileMetadataManager.GetTransportsTypes();
    auto transportProfilersMD = m_FileMetadataManager.GetTransportsProfilers();

    transportTypes.insert(transportTypes.end(), transportTypesMD.begin(),
                          transportTypesMD.end());

    transportProfilers.insert(transportProfilers.end(),
                              transportProfilersMD.begin(),
                              transportProfilersMD.end());

    const std::string lineJSON(m_BP4Serializer.GetRankProfilingJSON(
                                   transportTypes, transportProfilers) +
                               ",\n");

    const std::vector<char> profilingJSON(
        m_BP4Serializer.AggregateProfilingJSON(lineJSON));

    if (m_BP4Serializer.m_RankMPI == 0)
    {
        // std::cout << "write profiling file!" << std::endl;
        if (m_DrainBB)
        {
            auto bpTargetNames = m_BP4Serializer.GetBPBaseNames({m_Name});
            std::string targetProfiler(bpTargetNames[0] + "/profiling.json");
            m_FileDrainer.AddOperationWrite(
                targetProfiler, profilingJSON.size(), profilingJSON.data());
        }
        else
        {
            transport::FileFStream profilingJSONStream(m_Comm);
            auto bpBaseNames = m_BP4Serializer.GetBPBaseNames({m_BBName});
            profilingJSONStream.Open(bpBaseNames[0] + "/profiling.json",
                                     Mode::Write);
            profilingJSONStream.Write(profilingJSON.data(),
                                      profilingJSON.size());
            profilingJSONStream.Close();
        }
    }
}

/*write the content of metadata index file*/
void BP4Writer::PopulateMetadataIndexFileContent(
    format::BufferSTL &b, const uint64_t currentStep, const uint64_t mpirank,
    const uint64_t pgIndexStart, const uint64_t variablesIndexStart,
    const uint64_t attributesIndexStart, const uint64_t currentStepEndPos,
    const uint64_t currentTimeStamp)
{
    TAU_SCOPED_TIMER("BP4Writer::PopulateMetadataIndexFileContent");
    auto &buffer = b.m_Buffer;
    auto &position = b.m_Position;
    helper::CopyToBuffer(buffer, position, &currentStep);
    helper::CopyToBuffer(buffer, position, &mpirank);
    helper::CopyToBuffer(buffer, position, &pgIndexStart);
    helper::CopyToBuffer(buffer, position, &variablesIndexStart);
    helper::CopyToBuffer(buffer, position, &attributesIndexStart);
    helper::CopyToBuffer(buffer, position, &currentStepEndPos);
    helper::CopyToBuffer(buffer, position, &currentTimeStamp);
    position += 8;
}

void BP4Writer::UpdateActiveFlag(const bool active)
{
    const char activeChar = (active ? '\1' : '\0');
    m_FileMetadataIndexManager.WriteFileAt(
        &activeChar, 1, m_BP4Serializer.m_ActiveFlagPosition, 0);
    m_FileMetadataIndexManager.FlushFiles();
    m_FileMetadataIndexManager.SeekToFileEnd();
    if (m_DrainBB)
    {
        for (int i = 0; i < m_MetadataIndexFileNames.size(); ++i)
        {
            m_FileDrainer.AddOperationWriteAt(
                m_DrainMetadataIndexFileNames[i],
                m_BP4Serializer.m_ActiveFlagPosition, 1, &activeChar);
            m_FileDrainer.AddOperationSeekEnd(m_DrainMetadataIndexFileNames[i]);
        }
    }
}

void BP4Writer::WriteCollectiveMetadataFile(const bool isFinal)
{

    TAU_SCOPED_TIMER("BP4Writer::WriteCollectiveMetadataFile");

    if (isFinal && m_BP4Serializer.m_MetadataSet.DataPGCount == 0)
    {
        // If data pg count is zero, it means all metadata
        // has already been written, don't need to write it again.
        return;
    }
    m_BP4Serializer.AggregateCollectiveMetadata(
        m_Comm, m_BP4Serializer.m_Metadata, true);

    if (m_BP4Serializer.m_RankMPI == 0)
    {

        m_FileMetadataManager.WriteFiles(
            m_BP4Serializer.m_Metadata.m_Buffer.data(),
            m_BP4Serializer.m_Metadata.m_Position);
        m_FileMetadataManager.FlushFiles();

        if (m_DrainBB)
        {
            for (int i = 0; i < m_MetadataFileNames.size(); ++i)
            {
                m_FileDrainer.AddOperationCopy(
                    m_MetadataFileNames[i], m_DrainMetadataFileNames[i],
                    m_BP4Serializer.m_Metadata.m_Position);
            }
        }

        std::time_t currentTimeStamp = std::time(nullptr);

        std::vector<size_t> timeSteps;
        timeSteps.reserve(
            m_BP4Serializer.m_MetadataIndexTable[m_BP4Serializer.m_RankMPI]
                .size());
        for (auto const &pair :
             m_BP4Serializer.m_MetadataIndexTable[m_BP4Serializer.m_RankMPI])
        {
            timeSteps.push_back(pair.first);
        }
        std::sort(timeSteps.begin(), timeSteps.end());

        size_t rowsInMetadataIndexTable = timeSteps.size() + 1;
        m_BP4Serializer.m_MetadataIndex.Resize(rowsInMetadataIndexTable * 64,
                                               "BP4 Index Table");
        for (auto const &t : timeSteps)
        {
            const uint64_t pgIndexStartMetadataFile =
                m_BP4Serializer
                    .m_MetadataIndexTable[m_BP4Serializer.m_RankMPI][t][0] +
                m_BP4Serializer.m_MetadataSet.MetadataFileLength +
                m_BP4Serializer.m_PreMetadataFileLength;
            const uint64_t varIndexStartMetadataFile =
                m_BP4Serializer
                    .m_MetadataIndexTable[m_BP4Serializer.m_RankMPI][t][1] +
                m_BP4Serializer.m_MetadataSet.MetadataFileLength +
                m_BP4Serializer.m_PreMetadataFileLength;
            const uint64_t attrIndexStartMetadataFile =
                m_BP4Serializer
                    .m_MetadataIndexTable[m_BP4Serializer.m_RankMPI][t][2] +
                m_BP4Serializer.m_MetadataSet.MetadataFileLength +
                m_BP4Serializer.m_PreMetadataFileLength;
            const uint64_t currentStepEndPosMetadataFile =
                m_BP4Serializer
                    .m_MetadataIndexTable[m_BP4Serializer.m_RankMPI][t][3] +
                m_BP4Serializer.m_MetadataSet.MetadataFileLength +
                m_BP4Serializer.m_PreMetadataFileLength;
            PopulateMetadataIndexFileContent(
                m_BP4Serializer.m_MetadataIndex, t, m_BP4Serializer.m_RankMPI,
                pgIndexStartMetadataFile, varIndexStartMetadataFile,
                attrIndexStartMetadataFile, currentStepEndPosMetadataFile,
                currentTimeStamp);
        }

        m_FileMetadataIndexManager.WriteFiles(
            m_BP4Serializer.m_MetadataIndex.m_Buffer.data(),
            m_BP4Serializer.m_MetadataIndex.m_Position);
        m_FileMetadataIndexManager.FlushFiles();

        m_BP4Serializer.m_MetadataSet.MetadataFileLength +=
            m_BP4Serializer.m_Metadata.m_Position;

        if (m_DrainBB)
        {
            for (int i = 0; i < m_MetadataIndexFileNames.size(); ++i)
            {
                m_FileDrainer.AddOperationWrite(
                    m_DrainMetadataIndexFileNames[i],
                    m_BP4Serializer.m_MetadataIndex.m_Position,
                    m_BP4Serializer.m_MetadataIndex.m_Buffer.data());
            }
        }
    }
    /*Clear the local indices buffer at the end of each step*/
    m_BP4Serializer.ResetBuffer(m_BP4Serializer.m_Metadata, true);

    /* clear the metadata index buffer*/
    m_BP4Serializer.ResetBuffer(m_BP4Serializer.m_MetadataIndex, true);

    /* reset the metadata index table*/
    m_BP4Serializer.ResetMetadataIndexTable();
    m_BP4Serializer.ResetAllIndices();
}

void BP4Writer::WriteData(const bool isFinal, const int transportIndex)
{
    TAU_SCOPED_TIMER("BP4Writer::WriteData");
    size_t dataSize;

    // write data without footer
    if (isFinal)
    {
        dataSize = m_BP4Serializer.CloseData(m_IO);
    }
    else
    {
        dataSize = m_BP4Serializer.CloseStream(m_IO, false);
    }

    m_FileDataManager.WriteFiles(m_BP4Serializer.m_Data.m_Buffer.data(),
                                 dataSize, transportIndex);

    m_FileDataManager.FlushFiles(transportIndex);
    if (m_DrainBB)
    {
        for (int i = 0; i < m_SubStreamNames.size(); ++i)
        {
            m_FileDrainer.AddOperationCopy(m_SubStreamNames[i],
                                           m_DrainSubStreamNames[i], dataSize);
        }
    }
}

void BP4Writer::AggregateWriteData(const bool isFinal, const int transportIndex)
{
    TAU_SCOPED_TIMER("BP4Writer::AggregateWriteData");
    m_BP4Serializer.CloseStream(m_IO, false);
    size_t totalBytesWritten = 0;

    // async?
    for (int r = 0; r < m_BP4Serializer.m_Aggregator.m_Size; ++r)
    {
        aggregator::MPIAggregator::ExchangeRequests dataRequests =
            m_BP4Serializer.m_Aggregator.IExchange(m_BP4Serializer.m_Data, r);

        aggregator::MPIAggregator::ExchangeAbsolutePositionRequests
            absolutePositionRequests =
                m_BP4Serializer.m_Aggregator.IExchangeAbsolutePosition(
                    m_BP4Serializer.m_Data, r);

        if (m_BP4Serializer.m_Aggregator.m_IsConsumer)
        {
            const format::Buffer &bufferSTL =
                m_BP4Serializer.m_Aggregator.GetConsumerBuffer(
                    m_BP4Serializer.m_Data);
            if (bufferSTL.m_Position > 0)
            {
                m_FileDataManager.WriteFiles(
                    bufferSTL.Data(), bufferSTL.m_Position, transportIndex);

                m_FileDataManager.FlushFiles(transportIndex);

                totalBytesWritten += bufferSTL.m_Position;
            }
        }

        m_BP4Serializer.m_Aggregator.WaitAbsolutePosition(
            absolutePositionRequests, r);

        m_BP4Serializer.m_Aggregator.Wait(dataRequests, r);
        m_BP4Serializer.m_Aggregator.SwapBuffers(r);
    }

    if (m_DrainBB)
    {
        for (int i = 0; i < m_SubStreamNames.size(); ++i)
        {
            m_FileDrainer.AddOperationCopy(m_SubStreamNames[i],
                                           m_DrainSubStreamNames[i],
                                           totalBytesWritten);
        }
    }

    m_BP4Serializer.UpdateOffsetsInMetadata();

    if (isFinal) // Write metadata footer
    {
        m_BP4Serializer.m_Aggregator.Close();
    }

    m_BP4Serializer.m_Aggregator.ResetBuffers();
}

#define declare_type(T, L)                                                     \
    T *BP4Writer::DoBufferData_##L(const size_t payloadPosition,               \
                                   const size_t bufferID) noexcept             \
    {                                                                          \
        return BufferDataCommon<T>(payloadPosition, bufferID);                 \
    }

ADIOS2_FOREACH_PRIMITVE_STDTYPE_2ARGS(declare_type)
#undef declare_type

} // end namespace engine
} // end namespace core
} // end namespace adios2
