// pi_digits_benchmark.c
// Benchmark pthreads vs fork() for computing digits of pi in independent chunks.

#define _GNU_SOURCE
#include <errno.h>
#include <limits.h>
#include <pthread.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/resource.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/utsname.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>
#include <dirent.h>
#include <signal.h>

typedef enum {
    MODE_THREAD = 0,
    MODE_PROCESS = 1
} run_mode_t;

typedef struct {
    long thread_count;
    long process_count;
    long pi_digits;
    long chunk_size;
} bench_config_t;

typedef struct {
    int worker_id;
    long chunk_start;
    long chunk_count;
    double start_time;
    double end_time;
    int exit_code;
    char output_path[PATH_MAX];
    char stats_path[PATH_MAX];
    pid_t pid;
} worker_info_t;

typedef struct {
    double wall_s;
    double cpu_user_s;
    double cpu_sys_s;
    double cpu_total_s;
    double cpu_pct;
    long max_rss_kb;
    long max_vsz_kb;
    long agg_rss_kb;
    long agg_vsz_kb;
    long agg_pss_kb;
    double mem_snapshot_s;
    size_t output_bytes;
    int exit_code;
} run_metrics_t;

static double now_wall(void) {
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return ts.tv_sec + ts.tv_nsec / 1e9;
}

static void iso_timestamp(char *out, size_t size) {
    time_t now = time(NULL);
    struct tm tm_now;
    localtime_r(&now, &tm_now);
    strftime(out, size, "%Y-%m-%dT%H:%M:%S%z", &tm_now);
}

static void log_event(const char *fmt, ...) {
    char ts[64];
    iso_timestamp(ts, sizeof(ts));
    fprintf(stdout, "[%s] ", ts);

    va_list args;
    va_start(args, fmt);
    vfprintf(stdout, fmt, args);
    va_end(args);

    fputc('\n', stdout);
    fflush(stdout);
}

static long read_env_long(const char *name, long def_value) {
    const char *val = getenv(name);
    if (!val || !*val) {
        return def_value;
    }
    char *end = NULL;
    long v = strtol(val, &end, 10);
    if (end == val || v <= 0) {
        return def_value;
    }
    return v;
}

static int ensure_dir(const char *path) {
    if (mkdir(path, 0755) == 0) {
        return 0;
    }
    if (errno == EEXIST) {
        return 0;
    }
    return -1;
}

static int ensure_dir_recursive(const char *path) {
        char tmp[PATH_MAX];
        size_t len = strlen(path);
        if (len == 0 || len >= sizeof(tmp)) {
                return -1;
        }
        strncpy(tmp, path, sizeof(tmp));
        tmp[sizeof(tmp) - 1] = '\0';
        if (tmp[len - 1] == '/') {
                tmp[len - 1] = '\0';
        }

        for (char *p = tmp + 1; *p; ++p) {
                if (*p == '/') {
                        *p = '\0';
                        if (ensure_dir(tmp) != 0) {
                                return -1;
                        }
                        *p = '/';
                }
        }
        return ensure_dir(tmp);
}

static int remove_dir_tree(const char *path) {
    DIR *dir = opendir(path);
    if (!dir) {
        return 0;
    }
    struct dirent *ent;
    while ((ent = readdir(dir)) != NULL) {
        if (strcmp(ent->d_name, ".") == 0 || strcmp(ent->d_name, "..") == 0) {
            continue;
        }
        char child[PATH_MAX];
        snprintf(child, sizeof(child), "%s/%s", path, ent->d_name);
        unlink(child);
    }
    closedir(dir);
    rmdir(path);
    return 0;
}

static void reap_zombies(void) {
    int status = 0;
    while (waitpid(-1, &status, WNOHANG) > 0) {
    }
}

static void kill_orphan_pids(const char *pid_file) {
    FILE *f = fopen(pid_file, "r");
    if (!f) {
        return;
    }
    pid_t pid = 0;
    while (fscanf(f, "%d", &pid) == 1) {
        if (pid > 1) {
            kill(pid, SIGTERM);
        }
    }
    fclose(f);
    usleep(200000);
    f = fopen(pid_file, "r");
    if (!f) {
        return;
    }
    while (fscanf(f, "%d", &pid) == 1) {
        if (pid > 1) {
            kill(pid, SIGKILL);
        }
    }
    fclose(f);
}

static void cleanup_outputs(const char *out_dir, const char *tmp_dir, const char *pid_file, const char *final_path, const char *workers_csv, const char *results_csv) {
    (void)out_dir;
    log_event("Cleanup start");
    kill_orphan_pids(pid_file);
    reap_zombies();
    remove_dir_tree(tmp_dir);
    unlink(final_path);
    unlink(workers_csv);
    unlink(results_csv);
    unlink(pid_file);
    log_event("Cleanup complete");
}

typedef struct {
    long start;
    long end;
    long index;
    FILE *out;
    char buffer[4096];
    size_t buf_len;
} emit_ctx_t;

static int emit_digit(emit_ctx_t *ctx, int digit) {
    ctx->index++;
    if (ctx->index == 0) {
        return 0;
    }
    long pos = ctx->index - 1;
    if (pos < ctx->start) {
        return 0;
    }
    if (pos >= ctx->end) {
        return 1;
    }
    if (digit < 0 || digit > 9) {
        return 1;
    }
    ctx->buffer[ctx->buf_len++] = (char)('0' + digit);
    if (ctx->buf_len == sizeof(ctx->buffer)) {
        fwrite(ctx->buffer, 1, ctx->buf_len, ctx->out);
        ctx->buf_len = 0;
    }
    return 0;
}

static int flush_emit(emit_ctx_t *ctx) {
    if (ctx->buf_len > 0) {
        if (fwrite(ctx->buffer, 1, ctx->buf_len, ctx->out) != ctx->buf_len) {
            return -1;
        }
    }
    return 0;
}

static int write_pi_chunk(FILE *out, long start, long count) {
    if (count <= 0) {
        return 0;
    }

    long end = start + count;
    if (end <= 0) {
        return 0;
    }

    long total = end;
    long len_calc = (total * 10) / 3 + 1;
    if (len_calc <= 0) {
        return -1;
    }

    size_t len = (size_t)len_calc;
    int *a = calloc(len, sizeof(int));
    if (!a) {
        return -1;
    }
    for (size_t i = 0; i < len; ++i) {
        a[i] = 2;
    }

    int nines = 0;
    int predigit = 0;
    emit_ctx_t ctx = {start, end, -1, out, {0}, 0};

    for (long i = 0; i < total; ++i) {
        int q = 0;
        for (long j = (long)len - 1; j >= 0; --j) {
            int x = 10 * a[j] + q * (int)(j + 1);
            a[j] = x % (int)(2 * j + 1);
            q = x / (int)(2 * j + 1);
        }
        a[0] = q % 10;
        q = q / 10;

        if (q == 9) {
            nines++;
            continue;
        }
        if (q == 10) {
            if (emit_digit(&ctx, predigit + 1)) {
                break;
            }
            for (int k = 0; k < nines; ++k) {
                if (emit_digit(&ctx, 0)) {
                    break;
                }
            }
            predigit = 0;
            nines = 0;
        } else {
            if (emit_digit(&ctx, predigit)) {
                break;
            }
            for (int k = 0; k < nines; ++k) {
                if (emit_digit(&ctx, 9)) {
                    break;
                }
            }
            predigit = q;
            nines = 0;
        }

        if (ctx.index - 1 >= end) {
            break;
        }
    }

    if (ctx.index - 1 < end) {
        emit_digit(&ctx, predigit);
    }

    int rc = flush_emit(&ctx);
    free(a);
    return rc;
}

static int compute_chunk_to_file(const worker_info_t *info) {
    FILE *out = fopen(info->output_path, "w");
    if (!out) {
        return -1;
    }
    int rc = write_pi_chunk(out, info->chunk_start, info->chunk_count);
    fclose(out);
    return rc;
}

static int read_self_status_kb(long *rss_kb, long *vsz_kb) {
    *rss_kb = 0;
    *vsz_kb = 0;
    FILE *f = fopen("/proc/self/status", "r");
    if (!f) {
        return -1;
    }
    char line[256];
    while (fgets(line, sizeof(line), f)) {
        if (sscanf(line, "VmRSS: %ld", rss_kb) == 1) {
            continue;
        }
        if (sscanf(line, "VmSize: %ld", vsz_kb) == 1) {
            continue;
        }
    }
    fclose(f);
    return 0;
}

static long read_self_pss_kb(void) {
    FILE *f = fopen("/proc/self/smaps", "r");
    if (!f) {
        return 0;
    }
    long total = 0;
    char line[256];
    while (fgets(line, sizeof(line), f)) {
        long v = 0;
        if (sscanf(line, "Pss: %ld", &v) == 1) {
            total += v;
        }
    }
    fclose(f);
    return total;
}

static int write_worker_stats(const char *path) {
    long rss_kb = 0;
    long vsz_kb = 0;
    long pss_kb = 0;
    if (read_self_status_kb(&rss_kb, &vsz_kb) != 0) {
        return -1;
    }
    pss_kb = read_self_pss_kb();
    FILE *f = fopen(path, "w");
    if (!f) {
        return -1;
    }
    fprintf(f, "%ld,%ld,%ld\n", rss_kb, vsz_kb, pss_kb);
    fclose(f);
    return 0;
}

static void *thread_worker(void *arg) {
    worker_info_t *info = (worker_info_t *)arg;
    int rc = compute_chunk_to_file(info);
    info->exit_code = rc == 0 ? 0 : 1;
    return NULL;
}

static int run_thread_mode(const bench_config_t *cfg, const char *tmp_dir, const char *workers_csv, worker_info_t *workers, run_metrics_t *metrics) {
    pthread_t *threads = calloc((size_t)cfg->thread_count, sizeof(pthread_t));
    if (!threads) {
        return 1;
    }
    bool *created = calloc((size_t)cfg->thread_count, sizeof(bool));
    if (!created) {
        free(threads);
        return 1;
    }

    log_event("Threaded benchmark start (workers=%ld)", cfg->thread_count);

    for (long i = 0; i < cfg->thread_count; ++i) {
        workers[i].worker_id = (int)i;
        workers[i].chunk_start = i * cfg->chunk_size;
        workers[i].chunk_count = cfg->chunk_size;
        if (workers[i].chunk_start >= cfg->pi_digits) {
            workers[i].chunk_count = 0;
        } else if (workers[i].chunk_start + workers[i].chunk_count > cfg->pi_digits) {
            workers[i].chunk_count = cfg->pi_digits - workers[i].chunk_start;
        }
        snprintf(workers[i].output_path, sizeof(workers[i].output_path), "%s/worker_%06ld.txt", tmp_dir, i);
        workers[i].stats_path[0] = '\0';

        log_event("Worker start (thread id=%ld chunk_start=%ld chunk_size=%ld)", i, workers[i].chunk_start, workers[i].chunk_count);
        workers[i].start_time = now_wall();
        int rc = pthread_create(&threads[i], NULL, thread_worker, &workers[i]);
        if (rc != 0) {
            errno = rc;
            perror("pthread_create");
            workers[i].exit_code = 1;
            created[i] = false;
        } else {
            created[i] = true;
        }
    }

    for (long i = 0; i < cfg->thread_count; ++i) {
        if (created[i]) {
            pthread_join(threads[i], NULL);
        }
        workers[i].end_time = now_wall();
        log_event("Worker complete (thread id=%ld duration_s=%.6f exit_code=%d)", i, workers[i].end_time - workers[i].start_time, workers[i].exit_code);
    }

    FILE *wcsv = fopen(workers_csv, "w");
    if (wcsv) {
        fprintf(wcsv, "worker_id,mode,chunk_start,chunk_size,duration_s,exit_code\n");
        for (long i = 0; i < cfg->thread_count; ++i) {
            fprintf(wcsv, "%ld,thread,%ld,%ld,%.6f,%d\n",
                i, workers[i].chunk_start, workers[i].chunk_count,
                workers[i].end_time - workers[i].start_time, workers[i].exit_code);
        }
        fclose(wcsv);
    }

    free(created);
    free(threads);
    metrics->exit_code = 0;
    log_event("Threaded benchmark complete");
    return 0;
}

static int run_process_mode(const bench_config_t *cfg, const char *tmp_dir, const char *workers_csv, const char *pid_file, worker_info_t *workers, run_metrics_t *metrics) {
    log_event("Process benchmark start (workers=%ld)", cfg->process_count);

    FILE *pf = fopen(pid_file, "w");
    if (!pf) {
        perror("fopen pid_file");
        return 1;
    }

    for (long i = 0; i < cfg->process_count; ++i) {
        workers[i].worker_id = (int)i;
        workers[i].chunk_start = i * cfg->chunk_size;
        workers[i].chunk_count = cfg->chunk_size;
        if (workers[i].chunk_start >= cfg->pi_digits) {
            workers[i].chunk_count = 0;
        } else if (workers[i].chunk_start + workers[i].chunk_count > cfg->pi_digits) {
            workers[i].chunk_count = cfg->pi_digits - workers[i].chunk_start;
        }
        snprintf(workers[i].output_path, sizeof(workers[i].output_path), "%s/worker_%06ld.txt", tmp_dir, i);
        snprintf(workers[i].stats_path, sizeof(workers[i].stats_path), "%s/worker_%06ld.stat", tmp_dir, i);

        log_event("Worker start (process id=%ld chunk_start=%ld chunk_size=%ld)", i, workers[i].chunk_start, workers[i].chunk_count);
        workers[i].start_time = now_wall();
        pid_t pid = fork();
        if (pid < 0) {
            perror("fork");
            workers[i].exit_code = 1;
            continue;
        }
        if (pid == 0) {
            int rc = compute_chunk_to_file(&workers[i]);
            if (rc == 0) {
                if (write_worker_stats(workers[i].stats_path) != 0) {
                    rc = 1;
                }
            }
            _exit(rc == 0 ? 0 : 1);
        }
        workers[i].pid = pid;
        fprintf(pf, "%d\n", pid);
        fflush(pf);
    }
    fclose(pf);

    for (long i = 0; i < cfg->process_count; ++i) {
        if (workers[i].pid <= 0) {
            continue;
        }
        int status = 0;
        pid_t waited = waitpid(workers[i].pid, &status, 0);
        workers[i].end_time = now_wall();
        if (waited < 0) {
            perror("waitpid");
            workers[i].exit_code = 1;
        } else if (WIFEXITED(status)) {
            workers[i].exit_code = WEXITSTATUS(status);
        } else {
            workers[i].exit_code = 1;
        }
        log_event("Worker complete (process id=%ld duration_s=%.6f exit_code=%d)", i, workers[i].end_time - workers[i].start_time, workers[i].exit_code);
    }

    FILE *wcsv = fopen(workers_csv, "w");
    if (wcsv) {
        fprintf(wcsv, "worker_id,mode,chunk_start,chunk_size,duration_s,exit_code\n");
        for (long i = 0; i < cfg->process_count; ++i) {
            fprintf(wcsv, "%ld,process,%ld,%ld,%.6f,%d\n",
                i, workers[i].chunk_start, workers[i].chunk_count,
                workers[i].end_time - workers[i].start_time, workers[i].exit_code);
        }
        fclose(wcsv);
    }

    metrics->exit_code = 0;
    log_event("Process benchmark complete");
    return 0;
}

static int concatenate_outputs(const bench_config_t *cfg, const char *tmp_dir, const char *final_path, long worker_count) {
    log_event("Aggregation start");

    FILE *out = fopen(final_path, "w");
    if (!out) {
        perror("fopen final output");
        return 1;
    }

    char buf[4096];
    for (long i = 0; i < worker_count; ++i) {
        char path[PATH_MAX];
        snprintf(path, sizeof(path), "%s/worker_%06ld.txt", tmp_dir, i);
        FILE *in = fopen(path, "r");
        if (!in) {
            continue;
        }
        size_t n = 0;
        while ((n = fread(buf, 1, sizeof(buf), in)) > 0) {
            fwrite(buf, 1, n, out);
        }
        fclose(in);
    }

    fclose(out);
    log_event("Aggregation complete");
    return 0;
}

static size_t file_size(const char *path) {
    struct stat st;
    if (stat(path, &st) != 0) {
        return 0;
    }
    return (size_t)st.st_size;
}

static void collect_metrics(run_metrics_t *metrics, double start_wall, double end_wall) {
    struct rusage self_ru;
    struct rusage child_ru;
    memset(&self_ru, 0, sizeof(self_ru));
    memset(&child_ru, 0, sizeof(child_ru));
    getrusage(RUSAGE_SELF, &self_ru);
    getrusage(RUSAGE_CHILDREN, &child_ru);

    double self_user = self_ru.ru_utime.tv_sec + self_ru.ru_utime.tv_usec / 1e6;
    double self_sys = self_ru.ru_stime.tv_sec + self_ru.ru_stime.tv_usec / 1e6;
    double child_user = child_ru.ru_utime.tv_sec + child_ru.ru_utime.tv_usec / 1e6;
    double child_sys = child_ru.ru_stime.tv_sec + child_ru.ru_stime.tv_usec / 1e6;

    metrics->wall_s = end_wall - start_wall;
    metrics->cpu_user_s = self_user + child_user;
    metrics->cpu_sys_s = self_sys + child_sys;
    metrics->cpu_total_s = metrics->cpu_user_s + metrics->cpu_sys_s;
    metrics->cpu_pct = metrics->wall_s > 0.0 ? (100.0 * metrics->cpu_total_s / metrics->wall_s) : 0.0;
    metrics->max_rss_kb = self_ru.ru_maxrss;

    metrics->max_vsz_kb = 0;
    read_self_status_kb(&metrics->max_rss_kb, &metrics->max_vsz_kb);
}

static void aggregate_worker_memory(const bench_config_t *cfg, run_mode_t mode, const worker_info_t *workers, run_metrics_t *metrics) {
    double t0 = now_wall();
    long total_rss = 0;
    long total_vsz = 0;
    long total_pss = 0;

    long self_rss = 0;
    long self_vsz = 0;
    read_self_status_kb(&self_rss, &self_vsz);
    total_rss += self_rss;
    total_vsz += self_vsz;
    total_pss += read_self_pss_kb();

    if (mode == MODE_PROCESS) {
        for (long i = 0; i < cfg->process_count; ++i) {
            if (workers[i].stats_path[0] == '\0') {
                continue;
            }
            long rss_kb = 0;
            long vsz_kb = 0;
            long pss_kb = 0;
            FILE *f = fopen(workers[i].stats_path, "r");
            if (f && fscanf(f, "%ld,%ld,%ld", &rss_kb, &vsz_kb, &pss_kb) == 3) {
                fclose(f);
                total_rss += rss_kb;
                total_vsz += vsz_kb;
                total_pss += pss_kb;
            } else if (f) {
                fclose(f);
            }
        }
    }

    metrics->agg_rss_kb = total_rss;
    metrics->agg_vsz_kb = total_vsz;
    metrics->agg_pss_kb = total_pss;
    metrics->mem_snapshot_s = now_wall() - t0;
}

static void write_results_csv(const char *results_csv, const char *run_id, const char *mode, const bench_config_t *cfg, const run_metrics_t *metrics) {
    FILE *csv = fopen(results_csv, "w");
    if (!csv) {
        perror("fopen results_csv");
        return;
    }
    char ts[64];
    iso_timestamp(ts, sizeof(ts));
    char host[128] = "unknown";
    char kernel[128] = "unknown";
    gethostname(host, sizeof(host) - 1);
    struct utsname uts;
    if (uname(&uts) == 0) {
        strncpy(kernel, uts.release, sizeof(kernel) - 1);
        kernel[sizeof(kernel) - 1] = '\0';
    }

    fprintf(csv, "run_id,timestamp,host,kernel,mode,worker_count,pi_digits,chunk_size,exit_code,wall_s,cpu_user_s,cpu_sys_s,cpu_pct,max_rss_kb,max_vsz_kb,agg_rss_kb,agg_vsz_kb,agg_pss_kb,mem_snapshot_s,final_output_bytes\n");
    long workers = (strcmp(mode, "thread") == 0) ? cfg->thread_count : cfg->process_count;
    fprintf(csv, "%s,%s,%s,%s,%s,%ld,%ld,%ld,%d,%.6f,%.6f,%.6f,%.2f,%ld,%ld,%ld,%ld,%ld,%.6f,%zu\n",
        run_id, ts, host, kernel, mode, workers, cfg->pi_digits, cfg->chunk_size,
        metrics->exit_code, metrics->wall_s, metrics->cpu_user_s, metrics->cpu_sys_s,
        metrics->cpu_pct, metrics->max_rss_kb, metrics->max_vsz_kb,
        metrics->agg_rss_kb, metrics->agg_vsz_kb, metrics->agg_pss_kb, metrics->mem_snapshot_s, metrics->output_bytes);
    fclose(csv);
}

static void usage(const char *prog) {
    fprintf(stderr, "Usage: %s --mode thread|process [--out-dir PATH]\n", prog);
}

int main(int argc, char **argv) {
    run_mode_t mode = MODE_THREAD;
    const char *mode_name = "thread";
    const char *out_dir = "./pi_digits_run";

    for (int i = 1; i < argc; ++i) {
        if (strcmp(argv[i], "--mode") == 0 && i + 1 < argc) {
            mode_name = argv[++i];
        } else if (strncmp(argv[i], "--mode=", 7) == 0) {
            mode_name = argv[i] + 7;
        } else if (strcmp(argv[i], "--out-dir") == 0 && i + 1 < argc) {
            out_dir = argv[++i];
        } else if (strncmp(argv[i], "--out-dir=", 10) == 0) {
            out_dir = argv[i] + 10;
        } else if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) {
            usage(argv[0]);
            return 1;
        } else {
            usage(argv[0]);
            return 1;
        }
    }

    if (strcmp(mode_name, "thread") == 0) {
        mode = MODE_THREAD;
    } else if (strcmp(mode_name, "process") == 0) {
        mode = MODE_PROCESS;
    } else {
        usage(argv[0]);
        return 1;
    }

    bench_config_t cfg;
    cfg.thread_count = read_env_long("THREAD_COUNT", 4);
    cfg.process_count = read_env_long("PROCESS_COUNT", 4);
    cfg.pi_digits = read_env_long("PI_DIGITS", 2000);
    cfg.chunk_size = read_env_long("CHUNK_SIZE", 500);

    if (cfg.chunk_size <= 0) {
        cfg.chunk_size = cfg.pi_digits;
    }

    log_event("Benchmark start (mode=%s)", mode_name);

    if (ensure_dir_recursive(out_dir) != 0) {
        perror("mkdir out_dir");
        return 1;
    }

    char tmp_dir[PATH_MAX];
    char final_path[PATH_MAX];
    char workers_csv[PATH_MAX];
    char results_csv[PATH_MAX];
    char pid_file[PATH_MAX];
    char run_id[64];

    snprintf(run_id, sizeof(run_id), "%ld", (long)time(NULL));
    snprintf(tmp_dir, sizeof(tmp_dir), "%s/worker_tmp", out_dir);
    snprintf(final_path, sizeof(final_path), "%s/pi_digits.txt", out_dir);
    snprintf(workers_csv, sizeof(workers_csv), "%s/workers.csv", out_dir);
    snprintf(results_csv, sizeof(results_csv), "%s/results.csv", out_dir);
    snprintf(pid_file, sizeof(pid_file), "%s/workers.pids", out_dir);

    cleanup_outputs(out_dir, tmp_dir, pid_file, final_path, workers_csv, results_csv);

    if (ensure_dir_recursive(tmp_dir) != 0) {
        perror("mkdir tmp_dir");
        return 1;
    }

    long worker_count = (mode == MODE_THREAD) ? cfg.thread_count : cfg.process_count;
    worker_info_t *workers = calloc((size_t)worker_count, sizeof(worker_info_t));
    if (!workers) {
        perror("calloc workers");
        return 1;
    }

    run_metrics_t metrics;
    memset(&metrics, 0, sizeof(metrics));

    double t0 = now_wall();
    int rc = 0;
    if (mode == MODE_THREAD) {
        rc = run_thread_mode(&cfg, tmp_dir, workers_csv, workers, &metrics);
    } else {
        rc = run_process_mode(&cfg, tmp_dir, workers_csv, pid_file, workers, &metrics);
    }

    if (rc != 0) {
        metrics.exit_code = 1;
    }

    double t_calc = now_wall();
    collect_metrics(&metrics, t0, t_calc);
    aggregate_worker_memory(&cfg, mode, workers, &metrics);

    int agg_rc = concatenate_outputs(&cfg, tmp_dir, final_path, worker_count);
    if (agg_rc != 0) {
        metrics.exit_code = 1;
    }

    remove_dir_tree(tmp_dir);
    unlink(pid_file);
    metrics.output_bytes = file_size(final_path);

    write_results_csv(results_csv, run_id, mode_name, &cfg, &metrics);

    log_event("Benchmark end (mode=%s wall_s=%.6f cpu_pct=%.2f output_bytes=%zu)",
        mode_name, metrics.wall_s, metrics.cpu_pct, metrics.output_bytes);

    free(workers);
    return metrics.exit_code;
}