ParallelUtilities.cpp

Go to the documentation of this file.

//
// Created by atezbas on 10/21/24.
//
 
#include "ParallelUtilities.h"
#include <memory>
#include <boost/iostreams/device/array.hpp>
#include <boost/iostreams/device/back_inserter.hpp>
#include <boost/iostreams/stream.hpp>
using namespace std;
 
// Function to get a consistent color based on rank
std::ostream& ParallelUtilities::get_random_color(std::ostream& stream, const int rank) {
    // Use function pointers to termcolor manipulators
    static std::vector<std::ostream& (*)(std::ostream&)> colors = {
        termcolor::green,
        termcolor::yellow,
        termcolor::blue,
        termcolor::magenta,
        termcolor::cyan,
        termcolor::white,
        termcolor::bright_green,
        termcolor::bright_yellow,
        termcolor::bright_blue,
        termcolor::bright_magenta,
        termcolor::bright_cyan,
        termcolor::bright_white
    };
 
    // Map the rank to one of the colors
    return colors[rank % colors.size()](stream);
}
 
// Function to print colored message based on rank
void ParallelUtilities::print_message(const int rank, const std::string& message, const bool log) {
    if (!log) {
        return; // Do not print if log is false
    }
 
    if (rank == 0) {
        std::cout << termcolor::colorize << termcolor::red << "S[" << rank << "] - " << message << termcolor::reset << std::endl;
    } else {
        // Apply the color to std::cout
        std::cout << termcolor::colorize;
        get_random_color(std::cout, rank);
        // Continue with the message
        std::cout << termcolor::colorize << "C[" << rank << "] - " << message << termcolor::reset << std::endl;
    }
}
 
//
// void parallel_restartforanewsequence_() {    //reload resrat info
//     bool fresh_restart{false};
//     bool found_restart_point = false;
//     long n_count;
//     string content = "100 restart  transnt\n101 run\n103 ";
//     vector<BranchCluster *>::reverse_iterator bc_ritr, bc_ritr_temp;
//     bc_ritr = Branch_center.branch_cluster.rbegin();
//     //identify restart branch cluster and branch
//     while (bc_ritr != Branch_center.branch_cluster.rend()
//            && !found_restart_point) {
//         if ((*bc_ritr)->getCurrentBranchNum()
//             < ((*bc_ritr)->all_branch.size() - 1)) {
//             cout << "[getCurrentBranchNum()] " << (*bc_ritr)->getCurrentBranchNum() << endl;
//             cout << "[all_branch.size()] " << (*bc_ritr)->all_branch.size() - 1 << endl;
//             n_count = (*bc_ritr)->getNCountForRestartRELAP5();
//             fresh_restart = true;
//             found_restart_point = true;
//         } else {
//             (*bc_ritr)->savePENodeToFile();
//             bc_ritr_temp = bc_ritr;
//             if ((*bc_ritr)->getIDOfThisBranchCluster() == 0) {//the final branchCluster; all branches have been simulated
//                 r5pars_.terminateall = true;
//             } else {//current branch cluser is no longer needed; remove memory
//                 bc_ritr++;
//             }
//             delete *bc_ritr_temp;
//             Branch_center.branch_cluster.pop_back();
//         }
//     }
//     if (found_restart_point) {    // more sequence left to be simulated
//         //generate RELAP restart file
//         content += to_string(n_count);
//         content += "\n.";
//         //
//         ofstream fout("indeck.i", ios::out);
//         fout << content;
//         fout.close();
//     } else if (r5pars_.terminateall) {
//         cout << "All sequences have been simulated" << endl;
//     } else {        //all seuences have been sinulated
//         ofstream fout("err messages.txt", ios::app);
//         cout << "restartForANewSequence: could not find the restart point."
//              << endl;
//         fout << r5pars_.timesno
//              << "  restartForANewSequence: could not find the restart point.\n";
//         fout.close();
//     }
//     /*!
//      * dump all key parameter data of current sequence into ../output/KPS/KPSFolder_x/ZKPS_y.txt
//      */
// //    KPS.outputHistoryToFile();
//     /*!
//      * dump all activities records of current sequence into ../output/ES/scenario info.txt
//      */
// //    Aux_info.saveInfoToFile();
//     KPS.clearKeyParameterHistory(); /*!< release non-necessary memory */
//     Aux_info.clearAnalysisInfoContainerForRestart(); /*!< release non-necessary memory */
//     //Added by Yuandan LI
// //    Decision_Maker.m_RM->SaveSequenceInfoToFile();
// }
 
 
// Binary string restart information exchange
template<typename T>
std::string ParallelUtilities::serialize_object(const T& obj) {
    // An uncaught throw between the size and payload sends would leave the
    // matched receiver blocked forever, so serialize fully (and abort the job
    // on failure) before any bytes hit the wire.
    try {
        std::string bytes;
        {
            // The archive writes straight into the string: no ostringstream
            // and no final .str() copy of the full payload.
            boost::iostreams::stream<boost::iostreams::back_insert_device<std::string>> os(bytes);
            boost::archive::binary_oarchive oa(os);
            oa << obj;
        }   // archive and stream flush on destruction
        // Object streams are LZ4-framed on the wire (the 4-byte original-size
        // header rides inside the blob); receive_object strips the layer.
        // onBranchingPoint calls this once per branch point and sends the
        // same bytes to every receiver.
        // Only payloads up to WIRE_LZ4_MAX (4 MiB) are compressed: small
        // structured payloads such as state-only crew snapshots and branch
        // records. Larger streams are dominated by the RELAP restart image,
        // whose encoding belongs to the queue layer (--compress-queue,
        // opt-in); compressing them here would put the compressor on the
        // dispatch critical path. They ship raw behind a negative-size header.
        static constexpr size_t WIRE_LZ4_MAX = size_t(4) << 20;
        if (bytes.size() <= WIRE_LZ4_MAX) {
            auto packed = lz4_compress(bytes.data(), static_cast<int>(bytes.size()));
            return std::string(packed.data(), packed.size());
        }
        std::string framed(sizeof(int32_t) + bytes.size(), '\0');
        const int32_t raw_marker = -static_cast<int32_t>(bytes.size());
        std::memcpy(&framed[0], &raw_marker, sizeof(int32_t));
        std::memcpy(&framed[sizeof(int32_t)], bytes.data(), bytes.size());
        return framed;
    }
    catch (const std::exception& ex) {
        std::cerr << "FATAL: serialization failed: " << ex.what() << std::endl;
        MPI_Abort(MPI_COMM_WORLD, 1);
    }
    return {};  // unreachable — MPI_Abort does not return
}
 
void ParallelUtilities::send_bytes(const std::string& bytes, const int dest_rank, const int tag, MPI_Comm comm) {
    // Send the size of the serialized payload, then the payload: the wire
    // framing every receive_object expects.
    int size = static_cast<int>(bytes.size());
    MPI_Send(&size, 1, MPI_INT, dest_rank, tag, comm);
    MPI_Send(bytes.data(), size, MPI_CHAR, dest_rank, tag, comm);
}
 
template<typename T>
void ParallelUtilities::send_object(const T& obj, int dest_rank, int tag, MPI_Comm comm) {
    send_bytes(serialize_object(obj), dest_rank, tag, comm);
}
 
template<typename T>
void ParallelUtilities::receive_object(T& obj, int src_rank, int tag, MPI_Comm comm) {
    // Receive the size of the serialized payload
    int size = 0;
    MPI_Recv(&size, 1, MPI_INT, src_rank, tag, comm, MPI_STATUS_IGNORE);
 
    // Receive into an uninitialized buffer and deserialize in place: no
    // zero-fill and no intermediate std::string or istringstream copy.
    std::unique_ptr<char[]> buffer(new char[static_cast<size_t>(size)]);
    MPI_Recv(buffer.get(), size, MPI_CHAR, src_rank, tag, comm, MPI_STATUS_IGNORE);
 
    try {
        // Strip the wire layer added by serialize_object. A negative size
        // header marks an uncompressed frame (oversized streams, i.e. RELAP
        // restart images, ship raw); deserialize those in place, zero-copy.
        int32_t head = 0;
        std::memcpy(&head, buffer.get(), sizeof(int32_t));
        if (head < 0) {
            const size_t raw_sz = static_cast<size_t>(-static_cast<int64_t>(head));
            boost::iostreams::stream<boost::iostreams::array_source> is(
                buffer.get() + sizeof(int32_t), raw_sz);
            boost::archive::binary_iarchive ia(is);
            ia >> obj;
        } else {
            auto raw = lz4_decompress(buffer.get(), size);
            buffer.reset();
            boost::iostreams::stream<boost::iostreams::array_source> is(raw.data(), raw.size());
            boost::archive::binary_iarchive ia(is);
            ia >> obj;
        }
    }
    catch (const std::exception& ex) {
        // A half-deserialized object would make the client simulate from torn
        // state and report success, so abort the job instead.
        std::cerr << "FATAL: deserialization failed (tag " << tag << " from rank "
                  << src_rank << "): " << ex.what() << std::endl;
        MPI_Abort(comm, 1);
    }
}
 
// Explicit instantiation for specific types
template void ParallelUtilities::send_object<RestartNodeInfo>(const RestartNodeInfo& obj, int dest_rank, int tag, MPI_Comm comm);
template void ParallelUtilities::send_object<BranchRecord>(const BranchRecord& obj, int dest_rank, int tag, MPI_Comm comm);
template std::string ParallelUtilities::serialize_object<RestartNodeInfo>(const RestartNodeInfo& obj);
template std::string ParallelUtilities::serialize_object<BranchRecord>(const BranchRecord& obj);
//
template void ParallelUtilities::receive_object<RestartNodeInfo>(RestartNodeInfo& obj, int src_rank, int tag, MPI_Comm comm);
template void ParallelUtilities::receive_object<BranchRecord>(BranchRecord& obj, int src_rank, int tag, MPI_Comm comm);
//
template void ParallelUtilities::send_object<SequenceOutputData>(const SequenceOutputData& obj, int dest_rank, int tag, MPI_Comm comm);
template void ParallelUtilities::receive_object<SequenceOutputData>(SequenceOutputData& obj, int src_rank, int tag, MPI_Comm comm);
 
 
// TODO: add file-based and string stream information exchange