Client.cpp

Go to the documentation of this file.

//
// Created by atezbas on 10/21/24.
//
 
#include "Client.h"
#include "../Global/RestartTrim.h"
#include "../Global/SharedConfig.h"
#include "../Solver/RelapSolver.h"
#include "../Solver/RestApiSolver.h"
#include "../Global/KeyParameterHistory.h"
#include "../Scheduler/AnalysisInfoContainer.h"
#include <csignal>
#include <unistd.h>
#include <fstream>
#include <unordered_set>
#include "../Global/SimulationContext.h"
#include "../Global/CheckedIO.h"
#include "../Global/PEOutput.h"
using namespace std;
 
 
// SEQ-FP: log sequence fingerprint after a sequence completes (MPI version).
static void log_seq_fingerprint_mpi(int rank, double term_time) {
    if (!getenv("ADSIDAC_BRANCH_DIAG")) return;
    // Compute ancestry path by walking mother chain
    std::string ancestry;
    if (g_sim.seq_restart_cid == 0) {
        ancestry = "root";
    } else {
        std::vector<std::string> parts;
        int cur_cid = g_sim.seq_restart_cid;
        int cur_bp = g_sim.seq_restart_branch_num;
        while (cur_cid != 0) {
            BranchCluster *bc = g_sim.Branch_center.findBrnachCluster(cur_cid);
            if (!bc) {
                fprintf(stderr, "[SEQ-FP] WALK FAIL: CID=%d not found, %zu clusters. CIDs:", cur_cid, g_sim.Branch_center.branch_cluster.size());
                for (auto& c : g_sim.Branch_center.branch_cluster) fprintf(stderr, " %d", c->getIDOfThisBranchCluster());
                fprintf(stderr, "\n");
                break;
            }
            char buf[64];
            snprintf(buf, sizeof(buf), "%.4f:%d", bc->getTimeAtWhichThisEventHappen(), cur_bp);
            parts.emplace_back(buf);
            int mother_cid = bc->getIDOfMotherCluster();
            if (mother_cid == 0) break;
            BranchCluster *mother_bc = g_sim.Branch_center.findBrnachCluster(mother_cid);
            if (!mother_bc) break;
            int mother_bp = 0;
            for (int i = 0; i < static_cast<int>(mother_bc->all_branch.size()); ++i) {
                if (mother_bc->all_branch[i]->getIDOfTheChildBranchCluster() == cur_cid) {
                    mother_bp = i; break;
                }
            }
            cur_bp = mother_bp;
            cur_cid = mother_cid;
        }
        for (int i = static_cast<int>(parts.size()) - 1; i >= 0; --i) {
            if (!ancestry.empty()) ancestry += ">";
            ancestry += parts[i];
        }
    }
    std::string events;
    for (auto& [t, n] : g_sim.current_seq_events) {
        if (!events.empty()) events += ",";
        char buf[64];
        snprintf(buf, sizeof(buf), "%.4f:%d", t, n);
        events += buf;
    }
    if (events.empty()) events = "none";
    fprintf(stderr, "[SEQ-FP] rank=%d ancestry=%s events=%s term=%.4f restart_cid=%d restart_bn=%d\n",
            rank, ancestry.c_str(), events.c_str(), term_time, g_sim.seq_restart_cid, g_sim.seq_restart_branch_num);
}
 
static SequenceOutputData extract_sequence_outputs(int seq_id) {
    SequenceOutputData data;
    data.sequence_id = seq_id;
    if (g_sim.config.viz_output_events) {
        for (const auto& rec : g_sim.Aux_info.getRecords()) {
            SequenceOutputEvent ev;
            ev.time = rec->getTime();
            ev.statement = rec->getStatement();
            ev.pe_id = rec->getAssociatedBranchID();
            data.events.push_back(std::move(ev));
        }
    }
    if (g_sim.config.viz_output_end_states && !g_sim.Branch_center.end_state.empty()) {
        auto* es = g_sim.Branch_center.end_state.back().get();
        data.end_state.sequence_id = seq_id;
        data.end_state.type_of_consequence = static_cast<int>(es->getTypeOfConsequence());
        data.end_state.mother_branch_id = es->getIDOfTheMotherBranch();
        data.end_state.consequence_description = es->getConsequenceDescription();
        data.end_state.time = es->getTimeAtWhichThisEventHappen();
        data.end_state.probability = es->getProbability();
    }
    return data;
}
 
static void save_sequence_outputs(int seq_id) {
    // Set the ES ID so output files use the global sequence ID
    if (!g_sim.Branch_center.end_state.empty()) {
        g_sim.Branch_center.end_state.back()->setIDOfThisEndState(seq_id);
        g_sim.Branch_center.end_state.back()->saveInfoToFile();
    }
    g_sim.Branch_center.setIDOfRecentGeneratedES(seq_id);
 
    g_sim.KPS.outputHistoryToFile(seq_id, g_sim.SScheduler.getKPSDirectory());
    g_sim.Aux_info.saveInfoToFile(g_sim.SScheduler.getESDirectory(), seq_id);
    for (auto& bc : g_sim.Branch_center.branch_cluster)
        bc->savePENodeToFile();
}
 
 
// Path of the run-wide restart base. Lives in the project directory, so it
// survives a checkpoint/resume in the same project (which the config
// fingerprint already requires to be unchanged).
static std::string restart_base_path() {
    return g_sim.SScheduler.getProjectDirectory() + "/restart_base.bin";
}
 
bool Client::ensure_base_cache() {
    if (!base_cache_.empty()) return true;
    std::vector<char> buf;
    if (read_file_verified(restart_base_path(), buf) && !buf.empty()) {
        base_cache_ = std::move(buf);
        return true;
    }
    return false;
}
 
void Client::restore_buffer(std::vector<char> raw) {
    if (adsidac::has_complete_restart(raw)) {
        // Self-contained (base included). Remember the base for later transfers
        // that omit it; this is how a rank warms its cache without extra I/O.
        if (base_cache_.empty()) base_cache_ = adsidac::extract_header_and_base(raw);
    } else {
        // Base-less transfer: splice the run-wide base back on. RELAP aborts in
        // fabend if the first restart record it sees is not a complete one.
        if (!ensure_base_cache()) {
            std::cerr << "FATAL: rank " << rank << " received a base-less restart buffer "
                      << "and has no cached base (" << restart_base_path() << " missing) — "
                      << "restoring it would abend in RELAP. Aborting." << std::endl;
            MPI_Abort(MPI_COMM_WORLD, 1);
        }
        auto spliced = adsidac::splice_base(base_cache_, raw);
        if (spliced.empty()) {
            std::cerr << "FATAL: rank " << rank << " could not splice the restart base "
                      << "onto a transfer — aborting." << std::endl;
            MPI_Abort(MPI_COMM_WORLD, 1);
        }
        raw = std::move(spliced);
    }
    solver_->restore(raw);
}
 
void Client::restore_relap_state(const std::string& file_path, bool clear_buf) {
    if (!restart_info_->relap_restart_buf.empty()) {
        auto& buf = restart_info_->relap_restart_buf;
        if (g_sim.config.compress_queue) {
            restore_buffer(lz4_decompress(buf.data(), (int)buf.size()));
        } else {
            restore_buffer(buf);
        }
        if (clear_buf) {
            buf.clear();
            buf.shrink_to_fit();
        }
    } else {
        std::vector<char> buf;
        if (!read_file_verified(file_path, buf) || buf.empty()) {
            // Restoring from a missing/short restart file means simulating a
            // different state than the branch point, so abort the job.
            std::cerr << "FATAL: cannot restore RELAP5 state from " << file_path
                      << " — aborting." << std::endl;
            MPI_Abort(MPI_COMM_WORLD, 1);
        }
        if (g_sim.config.compress_queue) {
            restore_buffer(lz4_decompress(buf.data(), static_cast<int>(buf.size())));
        } else {
            restore_buffer(std::move(buf));
        }
    }
}
 
void Client::complete_sequence(int seq_id) {
    log_seq_fingerprint_mpi(rank, solver_->currentTime());
    save_sequence_outputs(seq_id);
    if (viz_outputs_any()) {
        auto od = extract_sequence_outputs(seq_id);
        PU::send_object(od, 0, PG::server_output_data, MPI_COMM_WORLD);
    }
    SequenceReport report{};
    report.sequence_id = seq_id;
    report.parent_sequence_id = parent_seq_id_;
    report.branch_time = branch_time_;
    report.end_time = solver_->currentTime();
    report.num_branches_generated = branches_this_seq_;
    report.worker_rank = rank;
    MPI_Send(&report, sizeof(SequenceReport), MPI_BYTE, 0, PG::server_done_with_report, MPI_COMM_WORLD);
    PU::print_message(rank, "Sent the finished signal to Server", PG::log);
}
 
Client::Client(const int rank)
{
    this->rank = rank;
    local_branch_counter_ = 0;
 
    // Create solver based on --solver flag (same logic as serial path)
    if (!g_sim.config.solver_url.empty()) {
        solver_ = std::make_unique<RestApiSolver>(g_sim.config.solver_url);
    } else {
        solver_ = std::make_unique<RelapSolver>();
    }
 
    g_sim.SScheduler.initiationForSimulation();
 
    // Redirect output directories to per-rank local paths (cwd is rank_N/)
    std::string local_out = fs::current_path().string() + "/Output/";
    g_sim.SScheduler.setOutputDirectory(local_out);
    g_sim.SScheduler.setESDirectory(local_out + "ES/");
    g_sim.SScheduler.setPEDirectory(local_out + "PE/");
    g_sim.SScheduler.setKPSDirectory(local_out + "KPS/");
 
    g_sim.Plant.initiationForSimulation();
    g_sim.Translator.initiationForSimulation();
    g_sim.Branch_center.initiationForSimulation();
    // The shared procedures master must exist before the first snapshot is
    // deserialized (peer task, checkpoint gather, or resume): overlay loads
    // seed definitions from it. Idempotent; onFirstInit reaches the same load
    // again on the engine's normal path.
    g_sim.Indicator.readInfoFromFile();  // procedures parser needs the panel maps
    // Procedures BEFORE the crew, mirroring AdsidacCallback::onFirstInit.
    // The crew's reason machine resolves Response_Procedure_Step pointers
    // through All_procedure.procedure_map during its own init, so the map
    // must be populated first; otherwise the lookup fails with
    // "Could not find this procedure step" and the RM node gets no pointer.
    // Procedures init does not depend on the crew knowledge bases, only on the
    // panel maps read just above.
    g_sim.All_procedure.initiationForSimulation();
    adsidac::set_shared_procedures(&g_sim.All_procedure);
    adsidac::set_shared_control_panel(&g_sim.Indicator);
    g_sim.Decision_Maker.initiationForSimulation();
    g_sim.Action_Taker.initiationForSimulation();
    adsidac::set_shared_kb(sim::Operator::ODM, &g_sim.Decision_Maker.knowledge_base);
    adsidac::set_shared_kb(sim::Operator::OAT, &g_sim.Action_Taker.knowledge_base);
    adsidac::set_shared_diagnosis_matrix(sim::Operator::ODM, &g_sim.Decision_Maker.diagnosis_matrix);
    adsidac::set_shared_diagnosis_matrix(sim::Operator::OAT, &g_sim.Action_Taker.diagnosis_matrix);
}
 
int Client::Work()
{
    // Ignore SIGINT on client ranks — server handles the interrupt menu
    signal(SIGINT, SIG_IGN);
    // Same for SIGUSR1, the HPC wall-time checkpoint request. Only rank 0
    // installs a handler (Server.cpp), but mpirun forwards the signal to EVERY
    // rank, and the default action would terminate a client (exit 138 =
    // 128+SIGUSR1). The server orchestrates the checkpoint and drains the
    // clients over MPI; the clients only need to survive the signal.
    signal(SIGUSR1, SIG_IGN);
    while (true)
    {
        PU::print_message(rank, "Checking for signals", PG::log);
        // Blocking MPI_Probe busy-polls in OpenMPI, so an idle client would
        // occupy a full core and take cycles from ranks running RELAP5.
        // Mirror the server's Iprobe + usleep idle loop (1 ms poll).
        int msg_ready = 0;
        MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &msg_ready, &status_client);
        while (!msg_ready) {
            usleep(1000);
            MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &msg_ready, &status_client);
        }
        if (status_client.MPI_TAG == PG::client_simulate_this)
        {
            MPI_Recv(nullptr, 0, MPI_INT, status_client.MPI_SOURCE, PG::client_simulate_this, MPI_COMM_WORLD, &status_client);
            PU::print_message(rank, "Received a new sequence to simulate", PG::log);
            parent_seq_id_ = -1;
            branch_time_ = 0.0;
            branches_this_seq_ = 0;
            g_sim.seq_restart_cid = 0; g_sim.seq_restart_branch_num = 0;
            g_sim.current_seq_events.clear();
            execute_relap();
            complete_sequence(0);
        }
        else if (status_client.MPI_TAG == PG::client_restart_run)
        {
            fs::remove("out.o");
 
            restart_info_ = std::make_shared<RestartNodeInfo>();
            PU::receive_object(*restart_info_, 0, PG::client_restart_run, MPI_COMM_WORLD);
            // Latch the resumed run's branch-ID epoch (rides in every
            // RestartNodeInfo; 0 for fresh runs).
            if (restart_info_->id_epoch_base > g_sim.id_epoch_base)
                g_sim.id_epoch_base = restart_info_->id_epoch_base;
 
            branch_center_ = std::make_shared<BranchRecord>();
            PU::receive_object(*branch_center_, 0, PG::client_restart_run, MPI_COMM_WORLD);
 
            int path_len;
            MPI_Recv(&path_len, 1, MPI_INT, 0, PG::client_restart_run,
                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            std::vector<char> path_buf(path_len + 1, '\0');
            MPI_Recv(path_buf.data(), path_len, MPI_CHAR, 0, PG::client_restart_run,
                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            std::string server_file_path(path_buf.data());
 
            int seq_id;
            MPI_Recv(&seq_id, 1, MPI_INT, 0, PG::client_restart_run, MPI_COMM_WORLD,
                     MPI_STATUS_IGNORE);
 
            restore_relap_state(server_file_path, /*clear_buf=*/true);
 
            int branch_index;
            MPI_Recv(&branch_index, 1, MPI_INT, 0, PG::client_restart_run, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
 
            parallel_restartforanewsequence_(branch_index);
            solver_->setInputDeck("indeck.i");
            solver_->setNewRestart(false);
 
            execute_relap();
            complete_sequence(seq_id);
        }
 
        else if (status_client.MPI_TAG == PG::client_peer_restart)
        {
            int source_rank = status_client.MPI_SOURCE;
            fs::remove("out.o");
 
            restart_info_ = std::make_shared<RestartNodeInfo>();
            PU::receive_object(*restart_info_, source_rank, PG::client_peer_restart, MPI_COMM_WORLD);
            // Latch the branch-ID epoch from the donor's snapshot.
            if (restart_info_->id_epoch_base > g_sim.id_epoch_base)
                g_sim.id_epoch_base = restart_info_->id_epoch_base;
            branch_center_ = std::make_shared<BranchRecord>();
            PU::receive_object(*branch_center_, source_rank, PG::client_peer_restart, MPI_COMM_WORLD);
 
            int path_len;
            MPI_Recv(&path_len, 1, MPI_INT, source_rank, PG::client_peer_restart,
                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            std::vector<char> path_buf(path_len + 1, '\0');
            MPI_Recv(path_buf.data(), path_len, MPI_CHAR, source_rank, PG::client_peer_restart,
                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            std::string file_path(path_buf.data());
 
            int seq_id;
            MPI_Recv(&seq_id, 1, MPI_INT, source_rank, PG::client_peer_restart,
                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
 
            restore_relap_state(file_path, /*clear_buf=*/true);
 
            int branch_index;
            MPI_Recv(&branch_index, 1, MPI_INT, source_rank, PG::client_peer_restart,
                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            MPI_Recv(&parent_seq_id_, 1, MPI_INT, source_rank, PG::client_peer_restart,
                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
 
            PU::print_message(rank, "[PEER] Received from C[" + std::to_string(source_rank)
                + "] seq=" + std::to_string(seq_id) + " bi=" + std::to_string(branch_index), PG::log);
 
            // Extract branch_time before parallel_restartforanewsequence_ moves clusters
            int cid = restart_info_->getCorrespondingBranchClusterID();
            BranchCluster* bc = branch_center_->findBrnachCluster(cid);
            branch_time_ = bc ? bc->getTimeAtWhichThisEventHappen() : 0.0;
            branches_this_seq_ = 0;
 
            parallel_restartforanewsequence_(branch_index);
            solver_->setInputDeck("indeck.i");
            solver_->setNewRestart(false);
 
            execute_relap();
            complete_sequence(seq_id);
        }
 
        else if (status_client.MPI_TAG == PG::client_send_to_peer)
        {
            int local_id, target_rank, branch_index, seq_id, parent_seq_id;
            MPI_Recv(&local_id, 1, MPI_INT, 0, PG::client_send_to_peer, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            MPI_Recv(&target_rank, 1, MPI_INT, 0, PG::client_send_to_peer, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            MPI_Recv(&branch_index, 1, MPI_INT, 0, PG::client_send_to_peer, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            MPI_Recv(&seq_id, 1, MPI_INT, 0, PG::client_send_to_peer, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            MPI_Recv(&parent_seq_id, 1, MPI_INT, 0, PG::client_send_to_peer, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
 
            auto it = held_data_.find(local_id);
            if (it == held_data_.end()) {
                // The server already marked the receiver busy for this transfer;
                // continuing without sending would leave it stranded and the
                // job would hang at termination. This is an invariant
                // violation, so abort the job.
                std::cerr << "FATAL: rank " << rank << " was asked to forward held branch data "
                          << "for lid=" << local_id << " but holds none — aborting." << std::endl;
                MPI_Abort(MPI_COMM_WORLD, 1);
            } else {
                auto& held = it->second;
                PU::print_message(rank, "[PEER] Deferred send lid=" + std::to_string(local_id)
                    + " -> C[" + std::to_string(target_rank) + "] bi="
                    + std::to_string(branch_index) + " seq=" + std::to_string(seq_id), PG::log);
 
                PU::send_object(*held.rinfo, target_rank, PG::client_peer_restart, MPI_COMM_WORLD);
                PU::send_object(*held.branch_record, target_rank, PG::client_peer_restart, MPI_COMM_WORLD);
                int path_len = static_cast<int>(held.file_path.size());
                MPI_Send(&path_len, 1, MPI_INT, target_rank, PG::client_peer_restart, MPI_COMM_WORLD);
                MPI_Send(held.file_path.c_str(), path_len, MPI_CHAR,
                         target_rank, PG::client_peer_restart, MPI_COMM_WORLD);
                MPI_Send(&seq_id, 1, MPI_INT, target_rank, PG::client_peer_restart, MPI_COMM_WORLD);
                MPI_Send(&branch_index, 1, MPI_INT, target_rank, PG::client_peer_restart, MPI_COMM_WORLD);
                MPI_Send(&parent_seq_id, 1, MPI_INT, target_rank, PG::client_peer_restart, MPI_COMM_WORLD);
 
                held.ref_count--;
                if (held.ref_count <= 0) held_data_.erase(it);
            }
            MPI_Send(nullptr, 0, MPI_INT, 0, PG::server_send_complete, MPI_COMM_WORLD);
        }
 
        else if (status_client.MPI_TAG == PG::client_run_held)
        {
            int local_id, branch_index, seq_id;
            MPI_Recv(&local_id, 1, MPI_INT, 0, PG::client_run_held, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            MPI_Recv(&branch_index, 1, MPI_INT, 0, PG::client_run_held, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            MPI_Recv(&seq_id, 1, MPI_INT, 0, PG::client_run_held, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            MPI_Recv(&parent_seq_id_, 1, MPI_INT, 0, PG::client_run_held, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
 
            auto it = held_data_.find(local_id);
            if (it == held_data_.end()) {
                PU::print_message(rank, "[PEER] ERROR: no held data for lid=" + std::to_string(local_id), PG::log);
                SequenceReport err_report{};
                err_report.sequence_id = seq_id;
                err_report.parent_sequence_id = parent_seq_id_;
                err_report.worker_rank = rank;
                MPI_Send(&err_report, sizeof(SequenceReport), MPI_BYTE, 0, PG::server_done_with_report, MPI_COMM_WORLD);
                continue;
            }
 
            PU::print_message(rank, "[PEER] Self-run lid=" + std::to_string(local_id)
                + " bi=" + std::to_string(branch_index) + " seq=" + std::to_string(seq_id), PG::log);
 
            fs::remove("out.o");
            auto& held = it->second;
            restart_info_ = held.rinfo;
 
            if (held.ref_count <= 1) {
                branch_center_ = std::move(held.branch_record);
            } else {
                branch_center_ = std::make_shared<BranchRecord>();
                branch_center_->copyBranchRecord(*held.branch_record);
            }
 
            restore_relap_state(held.file_path, /*clear_buf=*/held.ref_count <= 1);
 
            // Extract branch_time before parallel_restartforanewsequence_ moves data
            int cid = restart_info_->getCorrespondingBranchClusterID();
            BranchCluster* bc = branch_center_->findBrnachCluster(cid);
            branch_time_ = bc ? bc->getTimeAtWhichThisEventHappen() : 0.0;
            branches_this_seq_ = 0;
 
            held.ref_count--;
            if (held.ref_count <= 0) held_data_.erase(it);
 
            parallel_restartforanewsequence_(branch_index);
            solver_->setInputDeck("indeck.i");
            solver_->setNewRestart(false);
 
            execute_relap();
            complete_sequence(seq_id);
        }
 
        else if (status_client.MPI_TAG == PG::client_checkpoint)
        {
            MPI_Recv(nullptr, 0, MPI_INT, 0, PG::client_checkpoint, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            // Ensure all held rinfo have embedded RELAP5 restart data.
            // Adaptive memory mode may have written restart data to file instead
            // of keeping it in relap_restart_buf; read it back before sending.
            std::vector<int> inflated_lids;
            for (auto& [lid, held] : held_data_) {
                if (held.rinfo->relap_restart_buf.empty() && !held.file_path.empty()) {
                    if (!read_file_verified(held.file_path, held.rinfo->relap_restart_buf)) {
                        // A checkpoint with a missing/short blob is corrupt by
                        // construction and would fail only at resume, so abort now.
                        std::cerr << "FATAL: cannot embed restart data for lid=" << lid
                                  << " from " << held.file_path << " — aborting." << std::endl;
                        MPI_Abort(MPI_COMM_WORLD, 1);
                    }
                    inflated_lids.push_back(lid);
                    PU::print_message(rank, "[CHECKPOINT] Read "
                        + std::to_string(held.rinfo->relap_restart_buf.size())
                        + " bytes from " + held.file_path, PG::log);
                }
            }
            // Send held_data_ to server for checkpoint
            int count = static_cast<int>(held_data_.size());
            MPI_Send(&count, 1, MPI_INT, 0, PG::client_checkpoint, MPI_COMM_WORLD);
            for (auto& [lid, held] : held_data_) {
                MPI_Send(&lid, 1, MPI_INT, 0, PG::client_checkpoint, MPI_COMM_WORLD);
                PU::send_object(*held.rinfo, 0, PG::client_checkpoint, MPI_COMM_WORLD);
                PU::send_object(*held.branch_record, 0, PG::client_checkpoint, MPI_COMM_WORLD);
            }
            // Buffers inflated from file above are needed only for the
            // checkpoint payload. The file stays on disk for any later peer
            // transfer or self-run, so drop the in-memory copy.
            for (int lid : inflated_lids) {
                auto& buf = held_data_[lid].rinfo->relap_restart_buf;
                buf.clear();
                buf.shrink_to_fit();
            }
            PU::print_message(rank, "[CHECKPOINT] Sent " + std::to_string(count) + " held entries to server", PG::log);
            // Continue Work() loop — will get next task or shutdown
        }
 
        else if (status_client.MPI_TAG == PG::client_shutdown)
        {
            MPI_Recv(nullptr, 0, MPI_INT, status_client.MPI_SOURCE, PG::client_shutdown, MPI_COMM_WORLD, &status_client);
            PU::print_message(rank, "Shutting down", PG::log);
            cleanup_fort_temps();
            // File-mode holds write restart_<lid>_cid<N>.r transfer files into
            // the rank dir. One file serves every consumer of its branch point
            // and receivers read it asynchronously with no ack, so no mid-run
            // owner can delete it safely. Shutdown means the server has
            // confirmed global completion, so every read is done and the
            // files are swept here (one restart image per branch point).
            {
                std::error_code ec;
                for (auto& entry : fs::directory_iterator(".", ec)) {
                    auto fname = entry.path().filename().string();
                    if (fname.rfind("restart_", 0) == 0 &&
                        fname.size() > 2 && fname.compare(fname.size() - 2, 2, ".r") == 0)
                        fs::remove(entry.path(), ec);
                }
            }
            if (g_sim.config.merge_mode == 0) {
                // Worker self-merge, then rendezvous with the server's merge
                // barrier. Every rank parses the same argv, so merge_mode is
                // globally consistent and the barrier count matches.
                self_merge_outputs();
                MPI_Barrier(MPI_COMM_WORLD);
            }
            return 0;
        }
        // break;
    }
    // return 0;
}
 
void Client::self_merge_outputs()
{
    // Worker self-merge (--merge=now): every rank moves its own
    // unique-ID outputs into the shared Output/ tree and stages its append-file
    // sources under parts/ for the server's ordered concatenation. ES/KPS names
    // key on the server-minted global sequence id, so cross-rank collisions are
    // impossible; per-file PE nodes are left for rank 0, whose ordered renames
    // preserve the highest-rank-wins semantics for lineage re-saves.
    std::error_code ec;
    const fs::path shared_out = fs::path(g_sim.SScheduler.getProjectDirectory()) / "Output";
    const fs::path local_out = "Output";   // cwd = rank_N
    if (!fs::exists(local_out, ec)) return;
    static const std::vector<std::string> append_files = {
        "Scenario info.txt", "Procedure data.txt", "Scan data.txt",
        adsidac::pe::aggregate_basename()};
    const std::string suffix = ".rank_" + std::to_string(rank);
    std::unordered_set<std::string> made_dirs;
    auto move_file = [&](const fs::path& src, const fs::path& dest) {
        if (made_dirs.insert(dest.parent_path().string()).second)
            fs::create_directories(dest.parent_path(), ec);
        std::error_code mec;
        fs::rename(src, dest, mec);
        if (mec) {
            mec.clear();
            fs::copy(src, dest, fs::copy_options::overwrite_existing, mec);
            if (!mec) fs::remove(src, mec);
        }
    };
    // Append sources are collected first and staged after the walk: compacting
    // the PE aggregate creates and renames a temp file, which must not happen
    // under the feet of the directory iterator.
    std::vector<std::pair<fs::path, fs::path>> append_moves;
    for (auto& entry : fs::recursive_directory_iterator(local_out, ec)) {
        if (ec) break;
        if (!entry.is_regular_file()) continue;
        const fs::path rel = fs::relative(entry.path(), local_out);
        const std::string fname = entry.path().filename().string();
        bool is_append = false;
        for (auto& af : append_files)
            if (fname == af) { is_append = true; break; }
        if (is_append)
            append_moves.emplace_back(entry.path(),
                shared_out / rel.parent_path() / "parts" / (fname + suffix));
        else if (rel.begin()->string() == "PE")
            continue;   // per-file PE winner selection stays with rank 0
        else
            move_file(entry.path(), shared_out / rel);
    }
    for (auto& [src, dest] : append_moves) {
        // Rank-local compaction: last-wins over this rank's own re-saves cuts
        // the finalize's read volume ~4.5x without changing what collect()
        // resolves (see PEOutput.h).
        if (src.filename() == adsidac::pe::aggregate_basename())
            adsidac::pe::compact_aggregate(src);
        move_file(src, dest);
    }
    // err messages.txt lives in the rank cwd, outside Output/, so the walk
    // above does not see it; stage it with the other parts.
    if (fs::exists("err messages.txt", ec))
        move_file("err messages.txt", shared_out / "ES" / "parts" / ("err messages.txt" + suffix));
}
 
void Client::parallel_restartforanewsequence_(int branch_index)
{
    // 1) Find target cluster in received branch_center_ using CID from restart_info_
    int cid = restart_info_->getCorrespondingBranchClusterID();
    BranchCluster *bc = branch_center_->findBrnachCluster(cid);
    long n_count = bc->getNCountForRestartRELAP5();
    PU::print_message(rank, "CID=" + std::to_string(cid)
        + " branch_index=" + std::to_string(branch_index)
        + " n_count=" + std::to_string(n_count), PG::log);
 
    // 2) Generate indeck.i
    string content = "100 restart  transnt\n101 run\n103 ";
    content += to_string(n_count);
    content += "\n.";
    ofstream fout("indeck.i", ios::out);
    fout << content;
    fout.close();
 
    // 3) Transfer branch_center_ to GLOBAL Branch_center
    //    Delete all existing clusters in global
    g_sim.Branch_center.branch_cluster.clear();
 
    //    Move clusters to global (branch_center_ is consumed after this)
    g_sim.Branch_center.branch_cluster = std::move(branch_center_->branch_cluster);
 
    //    Use rank-based offset for branch IDs: each rank gets its own ID space
    //    (rank * 100000) within the current ID epoch. Fresh runs have epoch 0;
    //    resumed runs get an epoch above every phase-1 identity so new IDs
    //    can never collide with identities preserved inside merged PE outputs.
    int rank_base = g_sim.id_epoch_base + rank * 100000;
    if (g_sim.Branch_center.getIDOfRecentGeneratedBranch() < rank_base)
        g_sim.Branch_center.setIDOfRecentGeneratedBranch(rank_base);
    g_sim.Branch_center.setIDOfRecentGeneratedES(
        branch_center_->getIDOfRecentGeneratedES());
    g_sim.Branch_center.setIDOfRecentGeneratedBranchCluster(
        branch_center_->getIDOfRecentGeneratedBranchCluster());
    g_sim.Branch_center.setIDOfMotherCluster(
        branch_center_->getIDOfMotherCluster());
    g_sim.Branch_center.setIDOfMotherBranch(
        branch_center_->getIDOfMotherBranch());
    g_sim.Branch_center.setRecentUnconditionalBranchProbability(
        branch_center_->getRecentUnconditionalBranchProbability());
 
    // 4) Set up for initiationforrestart_()
    //    Re-find target cluster in global (pointers were moved)
    BranchCluster *target = g_sim.Branch_center.findBrnachCluster(cid);
 
    //    Set target's Current_branch_num = branch_index - 1
    //    (initiationforrestart_() will advance it to branch_index)
    target->setCurrentBranchNum(branch_index - 1);
 
    //    Branch ID assignment and mother linkage are handled by initiationforrestart_()
    //    (called via interfc_() when fresh_restart is true). No pre-assignment needed here.
 
    //    Mark ALL other clusters as fully consumed
    for (auto &gc : g_sim.Branch_center.branch_cluster) {
        if (gc.get() != target) {
            gc->setCurrentBranchNum(static_cast<int>(gc->all_branch.size()) - 1);
        }
    }
 
    //    Share RestartNodeInfo pointer directly — no copy needed (immutable after creation)
    g_sim.restart_info.clear();
    g_sim.restart_info.push_back(restart_info_);
 
    // 5) Set global fresh_restart = true
    g_sim.fresh_restart = true;
 
    // 6) Clear KPS and Aux_info
    g_sim.KPS.clearKeyParameterHistory();
    g_sim.Aux_info.clearAnalysisInfoContainerForRestart();
}
 
void Client::execute_relap()
{
    solver_->setThermoCacheMode(g_sim.config.thermo_cache_mode);
    // Always use membuf for RELAP5 I/O — adaptive decision only affects
    // whether the extracted buffer is kept in RAM or dumped to file.
    solver_->setMembufMode(1);
    solver_->runSequence(callback_);
}
 
void Client::hold_branch_data(int cid, std::shared_ptr<RestartNodeInfo> rinfo,
                              BranchRecord& br, const std::string& path, int refs)
{
    // Deep-copy BranchRecord via copyBranchRecord (global may change before consumption)
    auto br_copy = std::make_shared<BranchRecord>();
    br_copy->copyBranchRecord(br);
    held_data_[cid] = {rinfo, br_copy, path, refs};
    PU::print_message(rank, "[PEER] Holding data for CID=" + std::to_string(cid)
        + " refs=" + std::to_string(refs)
        + " held_total=" + std::to_string(held_data_.size()), PG::log);
}