ParallelGlobal.h

Go to the documentation of this file.

 
 
#ifndef PARALLELGLOBAL_H
#define PARALLELGLOBAL_H
 
#include "../Global/SimulationContext.h"
#include <fstream>
#include <sstream>
#include <filesystem>
#include <iostream>
#include <cstdlib>
#ifdef __APPLE__
#include <mach/mach.h>
#endif
#include <lz4.h>
#include <cstring>
#include <boost/serialization/vector.hpp>
#include <boost/serialization/string.hpp>
 
// ---- Fortran callback declarations (used by both Server and Client) ----
extern "C" void interfc_();
extern "C" void actonce_();
extern "C" void secini_();
extern "C" void actte_();
extern "C" void restartforanewsequence_();
extern "C" void initiationforrestart_();
extern "C" void mutiproactonce_();
extern "C" void setrnode_();
extern "C" { void relap5_(); }
 
namespace fs = std::filesystem;
 
inline std::string restart_path(const std::string& name) {
    return g_sim.config.memory_restart ? (g_sim.config.shm_prefix + name) : name;
}
 
inline long get_rss_mb() {
#if defined(__linux__)
    long rss = 0;
    std::ifstream stat("/proc/self/status");
    std::string line;
    while (std::getline(stat, line)) {
        if (line.rfind("VmRSS:", 0) == 0) {
            std::istringstream iss(line.substr(6));
            iss >> rss;  // in kB
            return rss / 1024;
        }
    }
    return 0;
#elif defined(__APPLE__)
    struct mach_task_basic_info info;
    mach_msg_type_number_t count = MACH_TASK_BASIC_INFO_COUNT;
    if (task_info(mach_task_self(), MACH_TASK_BASIC_INFO,
                  (task_info_t)&info, &count) == KERN_SUCCESS)
        return static_cast<long>(info.resident_size / (1024 * 1024));
    return 0;
#else
    return 0;
#endif
}
 
inline bool should_use_memory_restart() {
    if (!g_sim.config.memory_restart) return false;
    if (g_sim.config.memory_limit_mb == 0) return true;
    return get_rss_mb() < g_sim.config.memory_limit_mb;
}
 
inline std::vector<char> lz4_compress(const char* data, int size) {
    int max_dst = LZ4_compressBound(size);
    std::vector<char> out(sizeof(int32_t) + max_dst);
    int32_t orig = size;
    std::memcpy(out.data(), &orig, sizeof(int32_t));
    int comp_sz = LZ4_compress_default(data, out.data() + sizeof(int32_t), size, max_dst);
    out.resize(sizeof(int32_t) + comp_sz);
    // resize() never shrinks capacity: without shrink_to_fit() a compressed
    // buffer would keep its full LZ4_compressBound allocation and forfeit the
    // RSS benefit of --compress-queue.
    out.shrink_to_fit();
    return out;
}
 
inline std::vector<char> lz4_decompress(const char* data, int size) {
    int32_t orig;
    std::memcpy(&orig, data, sizeof(int32_t));
    std::vector<char> out(orig);
    int n = LZ4_decompress_safe(data + sizeof(int32_t), out.data(), size - (int)sizeof(int32_t), orig);
    if (n < 0 || n != orig) {
        // Feeding a corrupt buffer into the RELAP5 membuf would restart the
        // solver from garbage state, so exit instead.
        std::cerr << "FATAL: LZ4 decompression failed (got " << n << ", expected "
                  << orig << " bytes) — restart buffer is corrupt or was written "
                  << "with a different --compress-queue setting." << std::endl;
        std::exit(1);
    }
    return out;
}
 
inline void cleanup_fort_temps() {
    std::error_code ec;
    for (auto& entry : fs::directory_iterator(".", ec)) {
        auto fname = entry.path().filename().string();
        if (fname.size() > 4 && fname.substr(0, 4) == "fort" && fname != "fort.13")
            fs::remove(entry.path(), ec);
    }
}
 
struct SequenceReport {
    int sequence_id;
    int parent_sequence_id;      
    double branch_time;          
    double end_time;             
    int num_branches_generated;  
    int worker_rank;             
    template<class Archive>
    void serialize(Archive& ar, unsigned int) {
        ar & sequence_id & parent_sequence_id & branch_time
           & end_time & num_branches_generated & worker_rank;
    }
};
 
inline bool viz_outputs_any() { return g_sim.config.viz_output_events || g_sim.config.viz_output_end_states; }
 
struct SequenceOutputEvent {
    double time = 0.0;
    std::string statement;
    int pe_id = -999;  
    template<class Archive>
    void serialize(Archive& ar, unsigned int) { ar & time & statement & pe_id; }
};
 
struct SequenceOutputEndState {
    int sequence_id = -1;
    int type_of_consequence = -1;
    int mother_branch_id = -1;
    std::string consequence_description;
    double time = 0.0;
    double probability = 0.0;
    template<class Archive>
    void serialize(Archive& ar, unsigned int) {
        ar & sequence_id & type_of_consequence & mother_branch_id
           & consequence_description & time & probability;
    }
};
 
struct SequenceOutputData {
    int sequence_id = -1;
    std::vector<SequenceOutputEvent> events;
    SequenceOutputEndState end_state;
    template<class Archive>
    void serialize(Archive& ar, unsigned int) {
        ar & sequence_id & events & end_state;
    }
};
 
namespace PG {
    inline int rank;                
    inline int size;                
    inline bool log = true;         
 
    enum {
        server_branch_this,         
        server_done_simulating,     
        client_simulate_this,       
        client_restart_run,         
        client_shutdown,            
        client_debug_resume,        
        server_branch_meta,         
        client_peer_assign,         
        client_peer_restart,        
        client_send_to_peer,        
        client_run_held,            
        server_send_complete,       
        server_done_with_report,    
        server_output_data,         
        client_checkpoint           
    };
 
}
 
#endif //PARALLELGLOBAL_H