yac_pak.c

Go to the documentation of this file.

// Copyright (c) 2026 The YAC Authors
//
// SPDX-License-Identifier: BSD-3-Clause
 
/*
 * YAC PAK - YAC Performance Analysis Kit
 * Lightweight per-rank event tracing tool
 */
 
#include <mpi.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdarg.h>
#include <sys/stat.h>
#include <unistd.h>
#include <errno.h>
#include <limits.h>
#include <time.h>
 
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
 
#include "yac_config.h"
#include "yac_pak.h"
#include "yac_assert.h"
#include "yac_mpi_common.h"
#include "ensure_array_size.h"
#include "utils_core.h"
#include "utils_common.h"
#include "io_utils.h"
 
/* Environment variables:
 *   YAC_PAK_TRACE=0|1            enable tracing (default: 0)
 *   YAC_PAK_TRACE_LEVEL=0|1|2    coupling trace level (default: 1);
 *                                 level 0: no coupling events (setup events
 *                                          still recorded if TRACE_SETUP=1)
 *                                 level 1: PUT/GET/AGET exchange events
 *                                 level 2: additionally the nested
 *                                          interpolation/compute regions
 *   YAC_PAK_TRACE_SETUP=0|1      also record setup-phase events (default: 0)
 *   YAC_PAK_TRACE_CAPACITY=<n>   event buffer size per rank (default: 4096)
 *                                 buffer grows automatically; aborts on OOM
 *   YAC_PAK_TRACE_DIR=<path>     output directory (default: auto-generated)
 *   YAC_PAK_TRACE_STRIDE=<n>     write only every n-th rank of each component
 *                                 (default: 1) the first rank of a component
 *                                 always writes
 *   YAC_PAK_TRACE_FLUSH_TIME=0|1 include flush_duration_s in header
 *                                 (default: 1)
 *   YAC_PAK_TRACE_CONFIG=0|1     store coupling configuration in trace
 *                                 directory at enddef (default: 0)
 *                                 (only for YAC_PAK_TRACE=1)
 *   YAC_PAK_TRACE_WRITE_MODE=at_finalize|instant
 *                                when event rows are written (default:
 *                                 at_finalize). "instant" writes each row as
 *                                 it occurs and keeps a {..}.open file with
 *                                 the events currently in progress.
 *
 * Output: each rank writes {dir}/{component}_r{rank}.csv
 */
 
#define YAC_PAK_TRACE_ENV            "YAC_PAK_TRACE"
#define YAC_PAK_TRACE_LEVEL_ENV      "YAC_PAK_TRACE_LEVEL"
#define YAC_PAK_TRACE_SETUP_ENV      "YAC_PAK_TRACE_SETUP"
#define YAC_PAK_TRACE_CAPACITY_ENV   "YAC_PAK_TRACE_CAPACITY"
#define YAC_PAK_TRACE_DIR_ENV        "YAC_PAK_TRACE_DIR"
#define YAC_PAK_TRACE_STRIDE_ENV     "YAC_PAK_TRACE_STRIDE"
#define YAC_PAK_TRACE_FLUSH_TIME_ENV "YAC_PAK_TRACE_FLUSH_TIME"
#define YAC_PAK_TRACE_CONFIG_ENV     "YAC_PAK_TRACE_CONFIG"
#define YAC_PAK_TRACE_WRITE_MODE_ENV "YAC_PAK_TRACE_WRITE_MODE"
#define YAC_PAK_TRACE_VERBOSE_ENV    "YAC_PAK_TRACE_VERBOSE"
 
enum {
  YAC_PAK_TRACE_CAPACITY_DEFAULT = 4096,
  YAC_PAK_TRACE_CAPACITY_MIN     = 1,
};
 
typedef enum {
  YAC_PAK_OUTPUT_FILE_PER_RANK
} yac_pak_output_mode_t;
 
typedef enum {
  YAC_PAK_TRACE_WRITE_AT_FINALIZE,
  YAC_PAK_TRACE_WRITE_INSTANT,
} yac_pak_trace_write_mode_t;
 
typedef struct {
  char * name;
  double total;
  int count;
  double start;
  int running;
} yac_pak_timer_t;
 
typedef struct {
  char * field;
  char * component;
  yac_pak_event_type_t type;
  char * grid;
  int put_idx;
  char * datetime;
  int action;
} yac_pak_event_info_t;
 
/* In-progress (pending) event slot */
typedef struct {
  yac_pak_event_info_t info;
  yac_pak_trace_level_t level;
  double tstart;
  int in_use;
} yac_pak_open_event_t;
 
typedef struct {
  char * buf;      
  size_t len;      
  size_t capacity;
} yac_pak_str_t;
 
/* Completed (finished) event record */
typedef struct {
  yac_pak_event_info_t info;
  double tstart;
  double tend;
} yac_pak_trace_entry_t;
 
struct yac_pak {
 
  /* --- general --- */
  MPI_Comm comm;
  int rank;
  int size;
 
  /* --- timers --- */
  yac_pak_timer_t * timers;
  size_t num_timers;
  size_t cap_timers;
 
  /* --- tracing --- */
  yac_pak_output_mode_t output_mode;
  char * output_dir;
  char * config_output_path;
  int trace_stride;
  int trace_flush_time;
  int trace_is_writer;
  int trace_verbose;
 
  yac_pak_trace_write_mode_t trace_write_mode;
 
  /* Only used by write mode "instant": the trace file, kept open for the
   * whole run, and a companion file holding the events currently in
   * progress. Both are NULL in write mode "at_finalize". */
  FILE * trace_file;
  FILE * open_file;
  char * trace_file_path;
  char * open_file_path;
  size_t trace_file_header_len;
 
  yac_pak_open_event_t * pending_events;
  size_t pending_capacity;
 
  yac_pak_trace_entry_t * finished_events;
  size_t finished_count;
  size_t finished_capacity;
 
  int trace_enabled;
  int trace_level;
  int setup_trace_enabled;
 
  char * component_name;
  /* -1 until yac_pak_set_component */
  int comp_rank;
  int comp_size;
 
  double trace_init_wtime;
  double trace_init_epoch;
};
 
/* ------------------------------------------------------------------ */
/* Static helpers                                                     */
/* ------------------------------------------------------------------ */
 
DEF_NAME_TYPE_PAIRS(
  yac_pak_event_type_pairs,
  DEF_NAME_TYPE_PAIR(INIT,     YAC_PAK_EVENT_TYPE_INIT),
  DEF_NAME_TYPE_PAIR(DEF_COMP, YAC_PAK_EVENT_TYPE_DEF_COMP),
  DEF_NAME_TYPE_PAIR(SYNC_DEF, YAC_PAK_EVENT_TYPE_SYNC_DEF),
  DEF_NAME_TYPE_PAIR(ENDDEF,   YAC_PAK_EVENT_TYPE_ENDDEF),
  DEF_NAME_TYPE_PAIR(FINALIZE, YAC_PAK_EVENT_TYPE_FINALIZE),
  DEF_NAME_TYPE_PAIR(GET,      YAC_PAK_EVENT_TYPE_GET),
  DEF_NAME_TYPE_PAIR(AGET,     YAC_PAK_EVENT_TYPE_AGET),
  DEF_NAME_TYPE_PAIR(PUT,      YAC_PAK_EVENT_TYPE_PUT),
  DEF_NAME_TYPE_PAIR(REGION,   YAC_PAK_EVENT_TYPE_REGION),
  DEF_NAME_TYPE_PAIR(UPDATE,   YAC_PAK_EVENT_TYPE_UPDATE),
  DEF_NAME_TYPE_PAIR(WAIT,     YAC_PAK_EVENT_TYPE_WAIT))
 
static char const * yac_pak_event_type_to_str(yac_pak_event_type_t type) {
  char const * name =
    yac_name_type_pair_get_name(
      yac_pak_event_type_pairs,
      (size_t)yac_pak_event_type_pairs_size,
      (int)type);
  return name ? name : "UNKNOWN";
}
 
/* Setup-phase lifecycle events (gated by YAC_PAK_TRACE_SETUP, independently
 * of the coupling trace level). */
static int yac_pak_event_type_is_setup(yac_pak_event_type_t type) {
  return (type == YAC_PAK_EVENT_TYPE_INIT) ||
         (type == YAC_PAK_EVENT_TYPE_DEF_COMP) ||
         (type == YAC_PAK_EVENT_TYPE_SYNC_DEF) ||
         (type == YAC_PAK_EVENT_TYPE_ENDDEF) ||
         (type == YAC_PAK_EVENT_TYPE_FINALIZE);
}
 
DEF_NAME_TYPE_PAIRS(
  yac_pak_trace_write_mode_pairs,
  DEF_NAME_TYPE_PAIR(at_finalize, YAC_PAK_TRACE_WRITE_AT_FINALIZE),
  DEF_NAME_TYPE_PAIR(instant,     YAC_PAK_TRACE_WRITE_INSTANT))
 
static yac_pak_trace_write_mode_t get_trace_write_mode(MPI_Comm comm) {
 
  char * str_value =
    yac_getenv_str(YAC_PAK_TRACE_WRITE_MODE_ENV, "at_finalize", comm);
 
  int const mode =
    yac_name_type_pair_get_type(
      yac_pak_trace_write_mode_pairs,
      (size_t)yac_pak_trace_write_mode_pairs_size, str_value);
 
  YAC_ASSERT_F(
    mode != INT_MAX,
    "\"%s\" is not a valid value for environment variable %s "
    "(expected \"at_finalize\" or \"instant\")",
    str_value, YAC_PAK_TRACE_WRITE_MODE_ENV);
 
  free(str_value);
 
  return (yac_pak_trace_write_mode_t)mode;
}
 
static int directory_exists(char const * path) {
  struct stat st;
  return (stat(path, &st) == 0 && S_ISDIR(st.st_mode));
}
 
static char const * get_component_name(struct yac_pak const * pak) {
  return pak->component_name ? pak->component_name : "yac_pak_unknown";
}
 
/* within the component once it is known, in the communicator before that */
static int compute_is_writer(struct yac_pak const * pak) {
  int const rank = (pak->comp_rank >= 0) ? pak->comp_rank : pak->rank;
  return (rank % pak->trace_stride) == 0;
}
 
static void pak_make_trace_dir(char const * path, int verbose) {
  YAC_ASSERT_F(
    !mkdir(path, 0755), "could not create trace directory '%s': %s",
    path,  strerror(errno));
  if (verbose) fprintf(stderr, "YAC PAK trace: writing to %s/\n", path);
}
 
static char * build_output_path_per_rank(
    struct yac_pak const * pak, char const * suffix, int rank) {
  char const * comp = get_component_name(pak);
  char const * dir = pak->output_dir ? pak->output_dir : ".";
  size_t const len =
    strlen(dir) + 1 + strlen(comp) + 16 + strlen(suffix) + 1;
  char * path = xmalloc(len);
  int const n =
    snprintf(path, len, "%s/%s_r%d%s", dir, comp, rank, suffix);
  YAC_ASSERT_F(
    n >= 0 && (size_t)n < len, "snprintf failed or truncated (n=%d)", n);
  return path;
}
 
/* ------------------------------------------------------------------ */
/* Trace file format                                                  */
/* Single definition of the header and of a row.                      */
/* ------------------------------------------------------------------ */
 
static void str_appendf(yac_pak_str_t * str, char const * fmt, ...)
  __attribute__((format(printf, 2, 3)));
 
static void str_appendf(yac_pak_str_t * str, char const * fmt, ...) {
 
  va_list args, args_size;
  va_start(args, fmt);
  va_copy(args_size, args);
 
  int const n = vsnprintf(NULL, 0, fmt, args_size);
  va_end(args_size);
  YAC_ASSERT(n >= 0, "vsnprintf failed while building a trace string");
 
  ENSURE_ARRAY_SIZE(str->buf, str->capacity, str->len + (size_t)n + 1);
  vsnprintf(str->buf + str->len, (size_t)n + 1, fmt, args);
  str->len += (size_t)n;
 
  va_end(args);
}
 
static void append_trace_header(
  yac_pak_str_t * str, struct yac_pak const * pak, double flush_dur) {
 
  str_appendf(
    str,
    "# yac_pak_version=v" YAC_PAK_VERSION_STRING "\n"
    "# wtime_ref=%.9f\n"
    "# epoch_ref=%.9f\n",
    pak->trace_init_wtime, pak->trace_init_epoch);
  if (pak->trace_flush_time && (flush_dur >= 0.0))
    str_appendf(str, "# flush_duration_s=%.6f\n", flush_dur);
  str_appendf(str, "# trace_stride=%d\n", pak->trace_stride);
  if (pak->comp_size > 0)
    str_appendf(
      str, "# component_rank=%d\n# component_size=%d\n",
      pak->comp_rank, pak->comp_size);
  str_appendf(
    str,
    "rank\tfield\tcomponent\tgrid\ttype"
    "\tput_idx\tdatetime\taction"
    "\ttstart\ttend\tduration\n");
}
 
static void append_trace_row(
  yac_pak_str_t * str, struct yac_pak const * pak,
  yac_pak_trace_entry_t const * entry, int in_progress) {
 
  char const * f  = entry->info.field     ? entry->info.field     : "";
  char const * c  = entry->info.component ? entry->info.component : "";
  char const * g  = entry->info.grid      ? entry->info.grid      : "";
  char const * t  = yac_pak_event_type_to_str(entry->info.type);
  char const * dt = entry->info.datetime  ? entry->info.datetime  : "";
 
  str_appendf(
    str,
    in_progress?
      "%d\t%s\t%s\t%s\t%s\t%d\t%s\t%d\t%.9f\t\t\n":
      "%d\t%s\t%s\t%s\t%s\t%d\t%s\t%d\t%.9f\t%.9f\t%.9f\n",
    pak->rank, f, c, g, t, entry->info.put_idx, dt, entry->info.action,
    entry->tstart, entry->tend, entry->tend - entry->tstart);
}
 
static void write_trace_row(
  FILE * out, struct yac_pak const * pak,
  yac_pak_trace_entry_t const * entry, int in_progress) {
 
  yac_pak_str_t row = {0};
  append_trace_row(&row, pak, entry, in_progress);
  fwrite(row.buf, 1, row.len, out);
  free(row.buf);
}
 
static void write_metadata_json_file(
  struct yac_pak const * pak, double flush_dur_rank0) {
 
  char * trace_metadata_path =
    yac_file_path_join(pak->output_dir, "trace_metadata.json", NULL);
  FILE * trace_metadata_file = fopen(trace_metadata_path, "w");
  YAC_ASSERT_F(
    trace_metadata_file, "could not open '%s': %s",
    trace_metadata_path, strerror(errno));
 
  yac_pak_str_t meta = {0};
  str_appendf(
    &meta,
    "{\n"
    "  \"yac_pak_version\": \"v" YAC_PAK_VERSION_STRING "\",\n"
    "  \"yac_version\": \"v" YAC_VERSION "\",\n"
    "  \"mpi_size\": %d,\n"
    "  \"trace_stride\": %d,\n"
    "  \"init_epoch_sec\": %.9f,\n"
    "  \"wtime_ref_rank0\": %.9f",
    pak->size, pak->trace_stride,
    pak->trace_init_epoch, pak->trace_init_wtime);
  if (pak->trace_flush_time && (flush_dur_rank0 >= 0.0))
    str_appendf(
      &meta, ",\n  \"flush_duration_s_rank0\": %.6f", flush_dur_rank0);
  str_appendf(&meta, "\n}\n");
 
  fwrite(meta.buf, 1, meta.len, trace_metadata_file);
  fclose(trace_metadata_file);
  free(meta.buf);
  free(trace_metadata_path);
}
 
static void yac_pak_setup_trace_dir(struct yac_pak * pak) {
  /* one size for the directory path and for any candidate derived from it,
   * so that the truncation checks below guard the copy back */
  enum {TRACE_DIR_BUF_SIZE = 512};
  char path[TRACE_DIR_BUF_SIZE];
  path[0] = '\0';
 
  if (pak->rank == 0) {
    char const * dir_env = getenv(YAC_PAK_TRACE_DIR_ENV);
    if (dir_env && strlen(dir_env) > 0) {
      YAC_ASSERT_F(
        strlen(dir_env) < sizeof(path),
        "YAC_PAK_TRACE_DIR path \"%s\" exceeds maximum length of %zu",
        dir_env, sizeof(path) - 1);
      strncpy(path, dir_env, sizeof(path) - 1);
      path[sizeof(path) - 1] = '\0';
      if (directory_exists(path)) {
        fprintf(stderr,
          "WARNING: YAC trace: " YAC_PAK_TRACE_DIR_ENV "='%s' already"
          " exists; appending a numeric suffix to avoid overwriting"
          " previous traces.\n",
          path);
        int found = 0;
        for (int s = 1; s <= 99; ++s) {
          char candidate[TRACE_DIR_BUF_SIZE];
          int const n =
            snprintf(candidate, sizeof(candidate), "%s_%d", path, s);
          YAC_ASSERT_F(
            n >= 0 && (size_t)n < sizeof(candidate),
            "candidate path truncated (n=%d)", n);
          if (!directory_exists(candidate)) {
            strncpy(path, candidate, sizeof(path) - 1);
            path[sizeof(path) - 1] = '\0';
            found = 1;
            break;
          }
        }
        YAC_ASSERT_F(
          found, "could not find a unique trace directory (base: '%s')",
          dir_env);
        pak_make_trace_dir(path, pak->trace_verbose);
      } else {
        pak_make_trace_dir(path, pak->trace_verbose);
      }
    } else {
      /* Auto-generate: traces_YYYYMMDD_HHMMSS_N{size}_v{YAC PAK VER} */
      time_t now = time(NULL);
      struct tm const * tm_info = localtime(&now);
      YAC_ASSERT(tm_info != NULL, "localtime() failed");
      char base[256];
      size_t nbase =
        strftime(base, sizeof(base), "traces_%Y%m%d_%H%M%S", tm_info);
      YAC_ASSERT(nbase > 0, "strftime() failed");
      int npath = snprintf(path, sizeof(path), "%s_N%d_v%s",
                     base, pak->size, YAC_PAK_VERSION_STRING);
      YAC_ASSERT_F(
        npath >= 0 && (size_t)npath < sizeof(path),
        "auto-generated path exceeds maximum length of %zu",
        sizeof(path) - 1);
      pak_make_trace_dir(path, pak->trace_verbose);
    }
  }
 
  yac_mpi_call(
    MPI_Bcast(path, (int)sizeof(path), MPI_CHAR, 0, pak->comm),
    pak->comm);
  free(pak->output_dir);
  pak->output_dir = yac_string_dup(path);
}
 
static size_t get_finished_event_idx(struct yac_pak * pak) {
  if (pak->finished_count >= pak->finished_capacity) {
    size_t const old_cap = pak->finished_capacity;
    ENSURE_ARRAY_SIZE(pak->finished_events, pak->finished_capacity,
                      pak->finished_count + 1);
    for (size_t i = old_cap; i < pak->finished_capacity; ++i) {
      pak->finished_events[i].info.field     = NULL;
      pak->finished_events[i].info.component = NULL;
      pak->finished_events[i].info.grid      = NULL;
      pak->finished_events[i].info.datetime  = NULL;
      pak->finished_events[i].tstart = pak->finished_events[i].tend = 0.0;
    }
  }
  return pak->finished_count++;
}
 
static size_t get_open_event_idx(struct yac_pak * pak) {
  size_t open_event_idx = SIZE_MAX;
 
  for (size_t i = 0;
       (i < pak->pending_capacity) && (open_event_idx == SIZE_MAX); ++i)
    if (!pak->pending_events[i].in_use)
      open_event_idx = i;
 
  if (open_event_idx == SIZE_MAX) {
    size_t const old_cap = pak->pending_capacity;
    ENSURE_ARRAY_SIZE(pak->pending_events, pak->pending_capacity,
                      old_cap + 1);
    for (size_t i = old_cap; i < pak->pending_capacity; ++i) {
      pak->pending_events[i].info.field     = NULL;
      pak->pending_events[i].info.component = NULL;
      pak->pending_events[i].info.type      = YAC_PAK_EVENT_TYPE_INIT;
      pak->pending_events[i].info.grid      = NULL;
      pak->pending_events[i].info.datetime  = NULL;
      pak->pending_events[i].info.put_idx   = -1;
      pak->pending_events[i].info.action    = -1;
      pak->pending_events[i].level          = YAC_PAK_TRACE_LEVEL_OFF;
      pak->pending_events[i].tstart         = 0.0;
      pak->pending_events[i].in_use         = 0;
    }
    open_event_idx = old_cap;
  }
 
  return open_event_idx;
}
 
/* ------------------------------------------------------------------ */
/* Write mode "instant"                                               */
/* ------------------------------------------------------------------ */
 
static void rewrite_open_file(struct yac_pak * pak) {
 
  YAC_ASSERT(
    pak->open_file, "no .open file (only write mode \"instant\" has one)");
 
  yac_pak_str_t content = {0};
  append_trace_header(&content, pak, -1.0);
  for (size_t i = 0; i < pak->pending_capacity; ++i) {
    if (!pak->pending_events[i].in_use) continue;
    yac_pak_trace_entry_t const entry =
      {.info   = pak->pending_events[i].info,
       .tstart = pak->pending_events[i].tstart,
       .tend   = 0.0};
    append_trace_row(&content, pak, &entry, 1);
  }
 
  rewind(pak->open_file);
  fwrite(content.buf, 1, content.len, pak->open_file);
  free(content.buf);
  fflush(pak->open_file);
 
  /* Stale rows here would look like a rank being stuck in an event */
  long const len = ftell(pak->open_file);
  YAC_ASSERT_F(
    len >= 0, "could not determine the size of '%s': %s",
    pak->open_file_path, strerror(errno));
  int const rc = ftruncate(fileno(pak->open_file), (off_t)len);
  YAC_ASSERT_F(
    rc == 0, "could not truncate '%s': %s",
    pak->open_file_path, strerror(errno));
}
 
static void write_trace_row_instant(
  struct yac_pak * pak, yac_pak_trace_entry_t const * entry) {
 
  YAC_ASSERT(
    pak->trace_file,
    "no open trace file (only write mode \"instant\" has one)");
  write_trace_row(pak->trace_file, pak, entry, 0);
  fflush(pak->trace_file);
}
 
static void rename_open_file(
  struct yac_pak const * pak, char ** path, char const * suffix) {
 
  char * new_path = build_output_path_per_rank(pak, suffix, pak->rank);
 
  YAC_ASSERT_F(
    rename(*path, new_path) == 0, "could not rename '%s' to '%s': %s",
    *path, new_path, strerror(errno));
  free(*path);
  *path = new_path;
}
 
static void set_component_name(
  struct yac_pak * pak, char const * component) {
 
  if (pak->component_name) return;
  /* setup events (yac_init, yac_enddef, ...) carry no component name */
  if (!component) return;
  YAC_ASSERT(component[0] != '\0', "empty component name is not supported");
 
  pak->component_name = xstrdup(component);
 
  /* only write mode "instant" has files open here and only those still
   * carry the placeholder name */
  if (!pak->trace_file) return;
 
  rename_open_file(pak, &pak->trace_file_path, ".csv");
  rename_open_file(pak, &pak->open_file_path, ".open");
}
 
static void open_instant_files(struct yac_pak * pak) {
 
  pak->trace_file_path = build_output_path_per_rank(pak, ".csv", pak->rank);
  /* "w+": rewrite_instant_header reads the rows back */
  pak->trace_file = fopen(pak->trace_file_path, "w+");
  YAC_ASSERT_F(
    pak->trace_file, "could not open '%s': %s",
    pak->trace_file_path, strerror(errno));
  yac_pak_str_t header = {0};
  append_trace_header(&header, pak, -1.0);
  fwrite(header.buf, 1, header.len, pak->trace_file);
  pak->trace_file_header_len = header.len;
  free(header.buf);
  fflush(pak->trace_file);
 
  pak->open_file_path = build_output_path_per_rank(pak, ".open", pak->rank);
  pak->open_file = fopen(pak->open_file_path, "w");
  YAC_ASSERT_F(
    pak->open_file, "could not open '%s': %s",
    pak->open_file_path, strerror(errno));
  rewrite_open_file(pak);
}
 
/* The header was written before the component was known, so it reports the
   communicator stride and no component. The rows written since are kept. */
static void rewrite_instant_header(struct yac_pak * pak) {
 
  YAC_ASSERT(pak->trace_file, "no open trace file");
 
  long const end = (fflush(pak->trace_file), ftell(pak->trace_file));
  YAC_ASSERT_F(
    end >= 0, "could not determine the size of '%s': %s",
    pak->trace_file_path, strerror(errno));
 
  size_t const rows_len = (size_t)end - pak->trace_file_header_len;
  char * rows = NULL;
  if (rows_len > 0) {
    rows = xmalloc(rows_len);
    YAC_ASSERT_F(
      fseek(pak->trace_file, (long)pak->trace_file_header_len, SEEK_SET) == 0,
      "could not seek in '%s': %s", pak->trace_file_path, strerror(errno));
    YAC_ASSERT_F(
      fread(rows, 1, rows_len, pak->trace_file) == rows_len,
      "could not read back '%s': %s", pak->trace_file_path, strerror(errno));
  }
 
  yac_pak_str_t header = {0};
  append_trace_header(&header, pak, -1.0);
  rewind(pak->trace_file);
  fwrite(header.buf, 1, header.len, pak->trace_file);
  if (rows_len > 0) fwrite(rows, 1, rows_len, pak->trace_file);
  pak->trace_file_header_len = header.len;
  free(header.buf);
  free(rows);
  fflush(pak->trace_file);
 
  YAC_ASSERT_F(
    ftruncate(
      fileno(pak->trace_file), (off_t)(pak->trace_file_header_len + rows_len))
      == 0,
    "could not truncate '%s': %s", pak->trace_file_path, strerror(errno));
}
 
static void discard_instant_files(struct yac_pak * pak) {
 
  if (pak->trace_file) { fclose(pak->trace_file); pak->trace_file = NULL; }
  if (pak->open_file) { fclose(pak->open_file); pak->open_file = NULL; }
  if (pak->trace_file_path) remove(pak->trace_file_path);
  if (pak->open_file_path) remove(pak->open_file_path);
  free(pak->trace_file_path); pak->trace_file_path = NULL;
  free(pak->open_file_path); pak->open_file_path = NULL;
}
 
static void release_open_event(struct yac_pak * pak, size_t handle) {
  pak->pending_events[handle].in_use = 0;
}
 
/* ------------------------------------------------------------------ */
/* Public API                                                          */
/* ------------------------------------------------------------------ */
 
struct yac_pak * yac_pak_new(MPI_Comm comm, int capacity) {
  YAC_ASSERT(comm != MPI_COMM_NULL, "comm must not be MPI_COMM_NULL");
 
  // return NULL if tracing is not enabled
  if (!yac_getenv_bool(YAC_PAK_TRACE_ENV, 0, comm)) return NULL;
 
  struct yac_pak * pak = xmalloc(1 * sizeof(*pak));
 
  pak->comm = comm;
  yac_mpi_call(MPI_Comm_rank(comm, &pak->rank), comm);
  yac_mpi_call(MPI_Comm_size(comm, &pak->size), comm);
 
  pak->num_timers = 0;
  pak->cap_timers = 0;
  pak->timers = NULL;
 
  pak->output_mode = YAC_PAK_OUTPUT_FILE_PER_RANK;
  pak->output_dir = NULL;
  pak->trace_stride = 1;
  pak->trace_flush_time = 1;
  pak->trace_is_writer = 0;
  pak->trace_verbose = 0;
 
  pak->trace_write_mode = YAC_PAK_TRACE_WRITE_AT_FINALIZE;
  pak->trace_file       = NULL;
  pak->open_file        = NULL;
  pak->trace_file_path  = NULL;
  pak->open_file_path   = NULL;
 
  pak->pending_events   = NULL;
  pak->pending_capacity = 0;
 
  pak->finished_events   = NULL;
  pak->finished_count    = 0;
  pak->finished_capacity = 0;
 
  pak->trace_enabled = 1;
  // default 1: YAC_PAK_TRACE=1 records the PUT/GET/AGET events
  pak->trace_level = yac_getenv_int(YAC_PAK_TRACE_LEVEL_ENV, 1, 0, pak->comm);
  pak->setup_trace_enabled =
    yac_getenv_bool(YAC_PAK_TRACE_SETUP_ENV, 0, pak->comm);
 
  pak->component_name = NULL;
  pak->comp_rank = -1;
  pak->comp_size = -1;
 
  pak->trace_flush_time =
    yac_getenv_bool(YAC_PAK_TRACE_FLUSH_TIME_ENV, 1, pak->comm);
  pak->trace_verbose =
    yac_getenv_bool(YAC_PAK_TRACE_VERBOSE_ENV, 0, pak->comm);
 
  // 0 means "no coupling events" (setup-phase events are still
  // recorded if YAC_PAK_TRACE_SETUP=1); 1 = PUT/GET/AGET, 2 = + nested regions.
  YAC_ASSERT_F(
    (pak->trace_level >= 0) && (pak->trace_level <= YAC_PAK_TRACE_LEVEL_MAX),
    YAC_PAK_TRACE_LEVEL_ENV " must be between 0 and %d, but "
    YAC_PAK_TRACE_LEVEL_ENV "=%d was specified",
    YAC_PAK_TRACE_LEVEL_MAX, pak->trace_level);
 
  pak->output_mode = YAC_PAK_OUTPUT_FILE_PER_RANK;
  UNUSED(pak->output_mode);
 
  yac_pak_setup_trace_dir(pak);
 
  pak->config_output_path =
    yac_getenv_bool(YAC_PAK_TRACE_CONFIG_ENV, 0, pak->comm) ?
      yac_file_path_join(pak->output_dir, "coupling_config.yaml", NULL) : NULL;
 
  pak->trace_stride =
    yac_getenv_int(YAC_PAK_TRACE_STRIDE_ENV, 1, 1, pak->comm);
  pak->trace_is_writer = compute_is_writer(pak);
 
  pak->trace_write_mode = get_trace_write_mode(pak->comm);
 
  /* write mode "instant" does not need an event buffer */
  pak->finished_capacity =
    (pak->trace_write_mode == YAC_PAK_TRACE_WRITE_INSTANT)
      ? 0
      : (capacity > 0
           ? (size_t)capacity
           : (size_t)yac_getenv_int(
               YAC_PAK_TRACE_CAPACITY_ENV, YAC_PAK_TRACE_CAPACITY_DEFAULT,
               YAC_PAK_TRACE_CAPACITY_MIN, pak->comm));
  pak->finished_events =
    (pak->finished_capacity > 0)
      ? xcalloc(pak->finished_capacity, sizeof(*pak->finished_events))
      : NULL;
 
  {
    struct timespec ts;
    clock_gettime(CLOCK_REALTIME, &ts);
    pak->trace_init_wtime = MPI_Wtime();
    pak->trace_init_epoch = (double)ts.tv_sec + ts.tv_nsec * 1e-9;
  }
 
  /* Write mode "instant" writes while the run is going, so its files have to
   * exist from the start. The component name is only known once the first
   * event carries one, so they start out under the placeholder name and are
   * renamed later (see set_component_name). */
  if ((pak->trace_write_mode == YAC_PAK_TRACE_WRITE_INSTANT) &&
      pak->trace_is_writer)
    open_instant_files(pak);
 
  if ((pak->trace_write_mode == YAC_PAK_TRACE_WRITE_INSTANT) &&
      (pak->rank == 0))
    write_metadata_json_file(pak, -1.0);
 
  return pak;
}
 
void yac_pak_delete(struct yac_pak * pak) {
  if (!pak) return;
 
  if (pak->trace_enabled) {
    if (pak->pending_events) {
      for (size_t i = 0; i < pak->pending_capacity; ++i)
        if (pak->pending_events[i].in_use)
          yac_pak_trace_stop(pak, (int)i);
    }
    if (yac_getenv_bool("YAC_PAK_TRACE_AUTOFLUSH", 1, pak->comm)) {
      if (pak->finished_count > 0)
        yac_pak_flush_trace(pak);
    }
 
    if (pak->trace_file) {
      fclose(pak->trace_file);
      pak->trace_file = NULL;
      if (pak->trace_verbose && (pak->rank == 0))
        fprintf(stderr,
          "[%s] YAC PAK: per-rank traces written to %s/*.csv\n",
          get_component_name(pak),
          pak->output_dir ? pak->output_dir : ".");
    }
    if (pak->open_file) {
      fclose(pak->open_file);
      pak->open_file = NULL;
      remove(pak->open_file_path);
    }
  }
 
  for (size_t i = 0; i < pak->num_timers; ++i)
    free(pak->timers[i].name);
  free(pak->timers);
 
  free(pak->pending_events);
 
  if (pak->finished_events) {
    for (size_t i = 0; i < pak->finished_count; ++i) {
      free(pak->finished_events[i].info.field);
      free(pak->finished_events[i].info.component);
      free(pak->finished_events[i].info.grid);
      free(pak->finished_events[i].info.datetime);
    }
    free(pak->finished_events);
  }
 
  free(pak->component_name);
  free(pak->output_dir);
  free(pak->config_output_path);
  free(pak->trace_file_path);
  free(pak->open_file_path);
  free(pak);
}
 
/* ------------------------------------------------------------------ */
/* Tracing                                                             */
/* ------------------------------------------------------------------ */
 
int yac_pak_trace_is_enabled(struct yac_pak const * pak) {
 return pak && pak->trace_enabled;
}
 
int yac_pak_setup_trace_is_enabled(struct yac_pak const * pak) {
  return pak && pak->trace_enabled && pak->setup_trace_enabled;
}
 
int yac_pak_region_trace_is_enabled(
  struct yac_pak const * pak, yac_pak_trace_level_t level) {
  // region events are coupling events: gated by the coupling-level
  // threshold (`level` is the event category, not the nesting depth).
  if (!pak || !pak->trace_enabled) return 0;
  // in write mode "instant" a rank that does not write records nothing, so
  // instrumenting it would be pointless (matches yac_pak_trace_start_ex)
  if ((pak->trace_write_mode == YAC_PAK_TRACE_WRITE_INSTANT) &&
      !pak->trace_is_writer) return 0;
  return (int)level <= pak->trace_level;
}
 
char const * yac_pak_get_config_output_path(struct yac_pak const * pak) {
  return pak ? pak->config_output_path : NULL;
}
 
void yac_pak_set_component(
  struct yac_pak * pak, char const * comp_name,
  int comp_rank, int comp_size) {
 
  if (!pak) return;
  YAC_ASSERT_F(
    !pak->component_name,
    "component name was already set to '%s'", pak->component_name);
  YAC_ASSERT(
    comp_name && (comp_name[0] != '\0'),
    "invalid component name (has to be a non-empty string)");
  YAC_ASSERT_F(
    (comp_size > 0) && (comp_rank >= 0) && (comp_rank < comp_size),
    "invalid component rank %d for component size %d", comp_rank, comp_size);
  pak->component_name = yac_string_dup(comp_name);
  pak->comp_rank = comp_rank;
  pak->comp_size = comp_size;
 
  /* the stride basis changes here, so the writing ranks can change */
  int const was_writer = pak->trace_is_writer;
  int const is_writer = compute_is_writer(pak);
  pak->trace_is_writer = is_writer;
 
  /* only "instant" already has files, under the placeholder name */
  if (pak->trace_write_mode == YAC_PAK_TRACE_WRITE_INSTANT) {
    if (was_writer) {
      if (is_writer) {
        // still a writer: rename the files and correct their header
        rename_open_file(pak, &pak->trace_file_path, ".csv");
        rename_open_file(pak, &pak->open_file_path, ".open");
        rewrite_instant_header(pak);
        rewrite_open_file(pak);
      } else {
        // no longer a writer: discard what was written so far
        discard_instant_files(pak);
      }
    } else if (is_writer) {
      // newly a writer: start its files here
      open_instant_files(pak);
    }
  }
}
 
int yac_pak_trace_start_setup(struct yac_pak * pak,
                              char const * name, yac_pak_event_type_t type) {
  // level is ignored for setup-type events (gated by YAC_PAK_TRACE_SETUP)
  return yac_pak_trace_start_ex(
    pak, name, NULL, NULL, type, -1, NULL, -1, YAC_PAK_TRACE_LEVEL_OFF);
}
 
int yac_pak_trace_record(struct yac_pak * pak,
                         char const * field, char const * component,
                         char const * grid, yac_pak_event_type_t type,
                         double tstart, double tend) {
  if (!pak) return -1;
  if (!pak->trace_enabled || !pak->setup_trace_enabled) return -1;
  set_component_name(pak, component);
 
  if (pak->trace_write_mode == YAC_PAK_TRACE_WRITE_INSTANT) {
    if (pak->trace_is_writer) {
      yac_pak_trace_entry_t const entry =
        {.info = {.field     = (char *)field,
                  .component = (char *)component,
                  .type      = type,
                  .grid      = (char *)grid,
                  .put_idx   = -1,
                  .datetime  = NULL,
                  .action    = -1},
         .tstart = tstart,
         .tend   = tend};
      write_trace_row_instant(pak, &entry);
    }
  } else {
    size_t const idx = get_finished_event_idx(pak);
    pak->finished_events[idx].info.field     = yac_string_dup(field);
    pak->finished_events[idx].info.component = yac_string_dup(component);
    pak->finished_events[idx].info.type      = type;
    pak->finished_events[idx].info.grid      = yac_string_dup(grid);
    pak->finished_events[idx].info.put_idx   = -1;
    pak->finished_events[idx].info.datetime  = NULL;
    pak->finished_events[idx].info.action    = -1;
    pak->finished_events[idx].tstart         = tstart;
    pak->finished_events[idx].tend           = tend;
  }
 
  return 0;
}
 
int yac_pak_trace_start_ex(struct yac_pak * pak,
                           char const * field,
                           char const * component,
                           char const * grid,
                           yac_pak_event_type_t type,
                           int put_idx,
                           char const * datetime,
                           int action,
                           yac_pak_trace_level_t level) {
  if (!pak) return -1;
  if (yac_pak_event_type_is_setup(type)) {
    if (!pak->trace_enabled || !pak->setup_trace_enabled) return -1;
  } else if (!yac_pak_region_trace_is_enabled(pak, level)) {
    return -1;
  }
  set_component_name(pak, component);
 
  if ((pak->trace_write_mode == YAC_PAK_TRACE_WRITE_INSTANT) &&
      !pak->trace_is_writer) return -1;
 
  size_t const event_idx = get_open_event_idx(pak);
  pak->pending_events[event_idx].info.field     = yac_string_dup(field);
  pak->pending_events[event_idx].info.component = yac_string_dup(component);
  pak->pending_events[event_idx].info.type      = type;
  pak->pending_events[event_idx].info.grid      = yac_string_dup(grid);
  pak->pending_events[event_idx].info.put_idx   = put_idx;
  pak->pending_events[event_idx].info.datetime  = yac_string_dup(datetime);
  pak->pending_events[event_idx].info.action    = action;
  pak->pending_events[event_idx].level          = level;
  pak->pending_events[event_idx].tstart         = MPI_Wtime();
  pak->pending_events[event_idx].in_use         = 1;
 
  if (pak->trace_write_mode == YAC_PAK_TRACE_WRITE_INSTANT)
    rewrite_open_file(pak);
 
  return (int)event_idx;
}
 
int yac_pak_trace_stop(struct yac_pak * pak, int handle) {
  if (!pak) return -1;
  if (handle < 0) return -1;
  if ((size_t)handle >= pak->pending_capacity) return -1;
  if (!pak->pending_events[handle].in_use) return -1;
  double const tend = MPI_Wtime();
 
  if (pak->trace_write_mode == YAC_PAK_TRACE_WRITE_INSTANT) {
    /* DEF_COMP is open across yac_pak_set_component, which may close it */
    if (pak->trace_file) {
      yac_pak_trace_entry_t const entry =
        {.info   = pak->pending_events[handle].info,
         .tstart = pak->pending_events[handle].tstart,
         .tend   = tend};
      write_trace_row_instant(pak, &entry);
    }
    free(pak->pending_events[handle].info.field);
    free(pak->pending_events[handle].info.component);
    free(pak->pending_events[handle].info.grid);
    free(pak->pending_events[handle].info.datetime);
  } else {
    size_t const idx = get_finished_event_idx(pak);
    pak->finished_events[idx].info   = pak->pending_events[handle].info;
    pak->finished_events[idx].tstart = pak->pending_events[handle].tstart;
    pak->finished_events[idx].tend   = tend;
  }
 
  pak->pending_events[handle].info = (yac_pak_event_info_t){0};
  release_open_event(pak, (size_t)handle);
 
  if ((pak->trace_write_mode == YAC_PAK_TRACE_WRITE_INSTANT) &&
      pak->open_file)
    rewrite_open_file(pak);
 
  return 0;
}
 
int yac_pak_flush_trace(struct yac_pak * pak) {
  if (!pak) return -1;
  if (!pak->finished_events && !pak->trace_enabled) return -1;
  if (pak->finished_count == 0) return 0;
 
  {
    int const should_write = pak->trace_is_writer;
    double rank0_flush_dur = -1.0;
 
    if (should_write) {
      double const t_flush_start = MPI_Wtime();
      yac_pak_str_t rows = {0};
      for (size_t i = 0; i < pak->finished_count; ++i)
        append_trace_row(&rows, pak, &pak->finished_events[i], 0);
      double const flush_dur = MPI_Wtime() - t_flush_start;
      if (pak->rank == 0) rank0_flush_dur = flush_dur;
 
      char * filepath =
        build_output_path_per_rank(pak, ".csv", pak->rank);
      FILE * file_out = fopen(filepath, "w");
      YAC_ASSERT_F(
        file_out, "could not open '%s': %s", filepath,  strerror(errno));
      if (pak->trace_verbose && (pak->rank == 0))
        fprintf(stderr,
          "[%s] YAC PAK: Writing per-rank traces to %s/*.csv\n",
          get_component_name(pak),
          pak->output_dir ? pak->output_dir : ".");
 
      yac_pak_str_t header = {0};
      append_trace_header(&header, pak, flush_dur);
      fwrite(header.buf, 1, header.len, file_out);
      if (rows.len > 0) fwrite(rows.buf, 1, rows.len, file_out);
      fclose(file_out);
      free(header.buf);
      free(rows.buf);
      free(filepath);
    }
 
    /* metadata JSON by rank 0 */
    if (pak->rank == 0 && pak->output_dir)
      write_metadata_json_file(pak, rank0_flush_dur);
  }
 
  for (size_t i = 0; i < pak->finished_count; ++i) {
    free(pak->finished_events[i].info.field);
    free(pak->finished_events[i].info.component);
    free(pak->finished_events[i].info.grid);
    free(pak->finished_events[i].info.datetime);
  }
  pak->finished_count = 0;
  return 0;
}
 
/* ------------------------------------------------------------------ */
/* Performance Timers                                                 */
/* Separate feature from event tracing, not used yet.                 */
/* TODO: Extend and add tests for timer functionality                 */
/* ------------------------------------------------------------------ */
 
int yac_pak_register(struct yac_pak * pak, char const * name) {
  if (!pak) return -1;
  ENSURE_ARRAY_SIZE(pak->timers, pak->cap_timers, pak->num_timers + 1);
  pak->timers[pak->num_timers].name = yac_string_dup(name);
  pak->timers[pak->num_timers].total = 0.0;
  pak->timers[pak->num_timers].count = 0;
  pak->timers[pak->num_timers].start = 0.0;
  pak->timers[pak->num_timers].running = 0;
  return pak->num_timers++;
}
 
int yac_pak_start(struct yac_pak * pak, int idx) {
  if (!pak) return -1;
  if ((size_t)idx >= pak->num_timers) return -1;
  pak->timers[idx].start = MPI_Wtime();
  pak->timers[idx].running = 1;
  return 0;
}
 
int yac_pak_stop(struct yac_pak * pak, int idx) {
  if (!pak) return -1;
  if ((size_t)idx >= pak->num_timers) return -1;
  if (!pak->timers[idx].running) return -1;
  double const dt = MPI_Wtime() - pak->timers[idx].start;
  pak->timers[idx].total += dt;
  pak->timers[idx].count += 1;
  pak->timers[idx].running = 0;
  return 0;
}
 
int yac_pak_report(struct yac_pak * pak, FILE * out) {
  if (!pak) return -1;
  if (!out) out = stdout;
  if (!pak->timers) return -1;
  int comm_size = 1;
  typedef struct { double val; int rank; } double_int_t;
  yac_mpi_call(MPI_Comm_size(pak->comm, &comm_size), pak->comm);
  for (size_t i = 0; i < pak->num_timers; ++i) {
    double const local_total = pak->timers[i].total;
    int const local_count = pak->timers[i].count;
    double global_sum_total = 0.0;
    int global_sum_count = 0;
    double_int_t const local_loc = {local_total, pak->rank};
    double_int_t global_min_loc, global_max_loc;
 
    yac_mpi_call(MPI_Allreduce(&local_total, &global_sum_total,
      1, MPI_DOUBLE, MPI_SUM, pak->comm), pak->comm);
    yac_mpi_call(MPI_Allreduce(&local_loc, &global_min_loc,
      1, MPI_DOUBLE_INT, MPI_MINLOC, pak->comm), pak->comm);
    yac_mpi_call(MPI_Allreduce(&local_loc, &global_max_loc,
      1, MPI_DOUBLE_INT, MPI_MAXLOC, pak->comm), pak->comm);
    yac_mpi_call(MPI_Allreduce(&local_count, &global_sum_count,
      1, MPI_INT, MPI_SUM, pak->comm), pak->comm);
    double const avg = global_sum_total / (double)comm_size;
    if (pak->rank == 0) {
      fprintf(out,
        "Timer '%s': total_sum=%.6f,"
        " min=%.6f(rank %d), max=%.6f(rank %d),"
        " avg(per-rank)=%.6f, total_counts=%d\n",
        pak->timers[i].name, global_sum_total,
        global_min_loc.val, global_min_loc.rank,
        global_max_loc.val, global_max_loc.rank,
        avg, global_sum_count);
    }
  }
  return 0;
}