OutputMerge.h

Go to the documentation of this file.

 
//
// PURPOSE. Tree compute and output consolidation need not share one wall
// clock. Inside an allocation the merge runs in parallel with the workers'
// self-merge (--merge=now). With these helpers it can also run as a separate
// `--merge-only` process against a project whose run ended with
// `--merge=defer`, or whose merge was interrupted.
//
// Everything is MPI-free by design: `--merge-only` runs serially (with an
// optional thread pool over rank directories), and the serial-only build
// (ENABLE_MPI=OFF) compiles this header without the parallelexec target.
// All functions report filesystem errors on stderr or ignore them; none throws
// on a failed move. They assume no run and no other merge is writing the
// project at the same time.
//
// Layout contracts preserved (see docs/developer/output-layouts.md):
// ES/KPS keep their `<Kind>Folder_<id/1000>/` buckets and exact file names;
// the finalized PE layout is a single `PEFolder_0/` with dense sequential ids
// and no -999 sentinels; the three analysis logs are per-sequence blocks in
// any rank order; the PE aggregate destination is append-only because it
// accumulates across checkpoint phases.
//
#ifndef ADSIDAC_OUTPUTMERGE_H
#define ADSIDAC_OUTPUTMERGE_H
 
#include <algorithm>
#include <atomic>
#include <filesystem>
#include <fstream>
#include <iostream>
#include <limits>
#include <set>
#include <sstream>
#include <string>
#include <thread>
#include <unordered_set>
#include <utility>
#include <vector>
 
#include "PEOutput.h"
 
namespace adsidac::merge {
 
namespace fs = std::filesystem;
 
inline const char* manifest_basename() { return "MERGE_PENDING.txt"; }
 
inline std::vector<std::string> append_basenames() {
    return {"Scenario info.txt", "Procedure data.txt", "Scan data.txt",
            adsidac::pe::aggregate_basename()};
}
 
inline bool is_append_basename(const std::string& fname) {
    for (const auto& af : append_basenames())
        if (fname == af) return true;
    return false;
}
 
inline std::vector<int> discover_ranks(const std::string& project_dir) {
    std::vector<int> ranks;
    std::error_code ec;
    for (auto& e : fs::directory_iterator(project_dir, ec)) {
        if (!e.is_directory()) continue;
        const std::string n = e.path().filename().string();
        if (n.rfind("rank_", 0) != 0) continue;
        try {
            const int r = std::stoi(n.substr(5));
            if (r > 0) ranks.push_back(r);
        } catch (...) {}
    }
    std::sort(ranks.begin(), ranks.end());
    return ranks;
}
 
inline void move_file(const fs::path& src, const fs::path& dest,
                      std::unordered_set<std::string>& made_dirs) {
    std::error_code ec;
    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)
            std::cerr << "[MERGE] failed to move " << src << ": " << mec.message() << std::endl;
        else
            fs::remove(src, mec);
    }
}
 
inline void move_rank_es_kps(const std::string& project_dir, int r) {
    std::error_code ec;
    const fs::path rank_out = fs::path(project_dir) / ("rank_" + std::to_string(r)) / "Output";
    const fs::path shared_out = fs::path(project_dir) / "Output";
    if (!fs::exists(rank_out, ec)) return;
    std::unordered_set<std::string> made_dirs;
    for (const char* kind : {"ES", "KPS"}) {
        const fs::path src_root = rank_out / kind;
        if (!fs::exists(src_root, ec)) continue;
        for (auto& entry : fs::recursive_directory_iterator(src_root, ec)) {
            if (ec) break;
            if (!entry.is_regular_file()) continue;
            const std::string fname = entry.path().filename().string();
            if (is_append_basename(fname)) continue;
            move_file(entry.path(), shared_out / kind / fs::relative(entry.path(), src_root),
                      made_dirs);
        }
    }
}
 
inline void move_rank_pe_files(const std::string& project_dir, const std::vector<int>& ranks) {
    std::error_code ec;
    const fs::path shared_pe = fs::path(project_dir) / "Output" / "PE";
    std::unordered_set<std::string> made_dirs;
    for (int r : ranks) {
        const fs::path src_root = fs::path(project_dir) / ("rank_" + std::to_string(r))
                                  / "Output" / "PE";
        if (!fs::exists(src_root, ec)) continue;
        for (auto& entry : fs::recursive_directory_iterator(src_root, ec)) {
            if (ec) break;
            if (!entry.is_regular_file()) continue;
            const std::string fname = entry.path().filename().string();
            if (is_append_basename(fname)) continue;
            move_file(entry.path(), shared_pe / fs::relative(entry.path(), src_root), made_dirs);
        }
    }
}
 
inline void concat_appends_from_ranks(const std::string& project_dir,
                                      const std::vector<int>& ranks) {
    std::error_code ec;
    const fs::path shared_out = fs::path(project_dir) / "Output";
    for (const auto& base : append_basenames()) {
        const bool is_agg = (base == adsidac::pe::aggregate_basename());
        const char* kind = is_agg ? "PE" : "ES";
        auto src_of = [&](int r) -> fs::path {
            const fs::path in_rank = fs::path(project_dir) / ("rank_" + std::to_string(r))
                                     / "Output" / kind / base;
            std::error_code e2;
            if (fs::exists(in_rank, e2)) return in_rank;
            const fs::path staged = shared_out / kind / "parts" / (base + ".rank_" + std::to_string(r));
            if (fs::exists(staged, e2)) return staged;
            return {};
        };
        bool any = false;
        for (int r : ranks)
            if (!src_of(r).empty()) { any = true; break; }
        if (!any) continue;
        fs::create_directories(shared_out / kind, ec);
        std::ofstream out(shared_out / kind / base,
                          std::ios::binary | (is_agg ? std::ios::app : std::ios::trunc));
        size_t agg_raw = 0, agg_unique = 0;
        for (int r : ranks) {
            const fs::path src = src_of(r);
            if (src.empty()) continue;
            if (is_agg) {
                // Rank-local compaction before streaming: last-wins over this
                // rank's own re-saves, so the finalize reads fewer records with
                // identical resolution (see PEOutput.h). A part staged by a
                // --merge=now worker is already compacted; re-compacting is a
                // cheap no-op.
                const auto [raw, uniq] = adsidac::pe::compact_aggregate(src);
                agg_raw += raw; agg_unique += uniq;
            }
            std::ifstream in(src, std::ios::binary);
            if (in) { out << in.rdbuf(); in.close(); fs::remove(src, ec); }
        }
        if (is_agg && agg_raw > agg_unique)
            std::cout << "[MERGE] compacted PE aggregates: " << agg_raw << " -> "
                      << agg_unique << " records across " << ranks.size()
                      << " rank dirs" << std::endl;
    }
    for (const char* kind : {"ES", "PE"}) {
        const fs::path parts = shared_out / kind / "parts";
        std::error_code e2;
        if (fs::exists(parts, e2) && fs::is_empty(parts, e2))
            fs::remove(parts, e2);
    }
    // Per-rank "err messages.txt" lives in the rank cwd (including the
    // server's), outside Output/; fold it in. Like the logs above, a rank's
    // copy may instead sit staged under parts/ if a --merge=now run ended
    // between staging and concat. Both locations are read for every rank.
    {
        std::ofstream out(shared_out / "ES" / "err messages.txt",
                          std::ios::binary | std::ios::app);
        std::vector<int> with_zero = ranks;
        with_zero.insert(with_zero.begin(), 0);
        for (int r : with_zero) {
            for (const fs::path src : {
                     fs::path(project_dir) / ("rank_" + std::to_string(r)) / "err messages.txt",
                     shared_out / "ES" / "parts" / ("err messages.txt.rank_" + std::to_string(r))}) {
                std::ifstream in(src, std::ios::binary);
                if (in) { out << in.rdbuf(); in.close(); fs::remove(src, ec); }
            }
        }
    }
}
 
inline size_t build_es_summary(const std::string& dest_es) {
    std::string summary_path = dest_es + "ES_summary.txt";
    std::ofstream summary(summary_path, std::ios::out);
    summary << "ES#\tProbability\tEnd_time\tConsequence#\tDescription\n";
 
    std::vector<std::pair<int, fs::path>> es_files;
    std::error_code ec;
    if (fs::exists(dest_es, ec))
        for (auto& entry : fs::recursive_directory_iterator(dest_es, ec)) {
            std::string fname = entry.path().filename().string();
            if (fname.substr(0, 7) == "ZESNode" && fname.back() == 't') {
                auto start = fname.find('_') + 1;
                auto end = fname.find('.');
                int id = 0;
                try { id = std::stoi(fname.substr(start, end - start)); }
                catch (...) { continue; }
                es_files.push_back({id, entry.path()});
            }
        }
    std::sort(es_files.begin(), es_files.end());
 
    for (auto& [id, path] : es_files) {
        std::ifstream fin(path);
        std::string line;
        if (std::getline(fin, line)) {
            // ZESNode format: "ES_N time N mother_id 0 3096 0 probability"
            std::istringstream iss(line);
            std::string es_label;
            double time_val, probability;
            int es_id, mother_id, z1, z2, z3;
            iss >> es_label >> time_val >> es_id >> mother_id >> z1 >> z2 >> z3 >> probability;
            std::string consequence_line, description;
            if (std::getline(fin, consequence_line)) {
                std::istringstream css(consequence_line);
                std::string tag;
                int cons_type;
                css >> tag >> cons_type;
                if (std::getline(fin, description)) {
                    if (!description.empty() && description.front() == '"') description.erase(0, 1);
                    if (!description.empty() && description.back() == '"') description.pop_back();
                }
                summary << id << '\t' << probability << '\t' << time_val << '\t'
                        << cons_type << '\t' << description << '\n';
            }
        }
    }
    return es_files.size();
}
 
inline void write_manifest(const std::string& project_dir, int ranks, bool aggregate,
                           long sequences, unsigned long long fingerprint) {
    std::error_code ec;
    const fs::path out = fs::path(project_dir) / "Output";
    fs::create_directories(out, ec);
    std::ofstream m(out / manifest_basename(), std::ios::trunc);
    m << "# adsidac deferred merge v1\n"
      << "ranks " << ranks << "\n"
      << "aggregate_pe " << (aggregate ? 1 : 0) << "\n"
      << "sequences " << sequences << "\n"
      << "config_fingerprint 0x" << std::hex << fingerprint << std::dec << "\n";
}
 
inline bool manifest_exists(const std::string& project_dir) {
    std::error_code ec;
    return fs::exists(fs::path(project_dir) / "Output" / manifest_basename(), ec);
}
 
inline bool manifest_aggregate(const std::string& project_dir, bool fallback) {
    std::ifstream m(fs::path(project_dir) / "Output" / manifest_basename());
    std::string key;
    int val = 0;
    while (m >> key) {
        if (key == "aggregate_pe" && (m >> val)) return val != 0;
        m.ignore(std::numeric_limits<std::streamsize>::max(), '\n');
    }
    return fallback;
}
 
inline bool looks_finalized(const std::string& pe_dir) {
    std::error_code ec;
    if (!fs::exists(pe_dir, ec)) return false;
    long nonneg = 0;
    int max_id = -1;
    bool has_zero = false, has_negative = false, any = false;
    for (auto& entry : fs::recursive_directory_iterator(pe_dir, ec)) {
        if (!entry.is_regular_file()) continue;
        const std::string fname = entry.path().filename().string();
        std::vector<int> ids;
        if (fname == adsidac::pe::aggregate_basename()) {
            std::vector<std::pair<int, std::string>> recs;
            adsidac::pe::parse_aggregate(entry.path(), recs);
            for (auto& [id, c] : recs) ids.push_back(id);
        } else if (fname.size() > 7 && fname.compare(0, 7, "ZPENode") == 0
                   && fname.back() == 't') {
            try {
                ids.push_back(std::stoi(fname.substr(fname.find('_') + 1,
                                                     fname.find('.') - fname.find('_') - 1)));
            } catch (...) { continue; }
        }
        for (int id : ids) {
            any = true;
            if (id > max_id) max_id = id;
            if (id == 0) has_zero = true;
            if (id < 0) has_negative = true;
            if (id >= 0) ++nonneg;
        }
    }
    return any && has_zero && !has_negative && nonneg == static_cast<long>(max_id) + 1;
}
 
inline bool ranks_hold_outputs(const std::string& project_dir, const std::vector<int>& ranks) {
    std::error_code ec;
    for (int r : ranks) {
        const fs::path out = fs::path(project_dir) / ("rank_" + std::to_string(r)) / "Output";
        if (!fs::exists(out, ec)) continue;
        for (auto& entry : fs::recursive_directory_iterator(out, ec)) {
            if (ec) break;
            if (!entry.is_regular_file()) continue;
            const std::string f = entry.path().filename().string();
            if (f.rfind("ZESNode", 0) == 0 || f.rfind("ZKPS", 0) == 0
                || f.rfind("ZDIA", 0) == 0 || f.rfind("ZPIF", 0) == 0
                || f.rfind("ZPENode", 0) == 0 || is_append_basename(f))
                return true;
        }
    }
    return false;
}
 
inline std::pair<size_t, size_t> merge_project(const std::string& project_dir,
                                               bool aggregate, int threads) {
    const std::vector<int> ranks = discover_ranks(project_dir);
    const std::string dest_es = (fs::path(project_dir) / "Output" / "ES").string() + "/";
    const std::string dest_pe = (fs::path(project_dir) / "Output" / "PE").string() + "/";
 
    if (threads < 1) threads = 1;
    std::atomic<size_t> next{0};
    auto worker = [&] {
        for (size_t i = next.fetch_add(1); i < ranks.size(); i = next.fetch_add(1))
            move_rank_es_kps(project_dir, ranks[i]);
    };
    if (threads == 1 || ranks.size() <= 1) {
        worker();
    } else {
        std::vector<std::thread> pool;
        const int n = std::min<int>(threads, static_cast<int>(ranks.size()));
        pool.reserve(n);
        for (int t = 0; t < n; ++t) pool.emplace_back(worker);
        for (auto& th : pool) th.join();
    }
    move_rank_pe_files(project_dir, ranks);
    concat_appends_from_ranks(project_dir, ranks);
    const size_t seqs = build_es_summary(dest_es);
    const size_t pe_count = adsidac::pe::finalize_sequential(dest_pe, aggregate);
    std::error_code ec;
    fs::remove(fs::path(project_dir) / "Output" / manifest_basename(), ec);
    return {seqs, pe_count};
}
 
} // namespace adsidac::merge
 
#endif // ADSIDAC_OUTPUTMERGE_H