io_utils.c

Go to the documentation of this file.

// Copyright (c) 2024 The YAC Authors
//
// SPDX-License-Identifier: BSD-3-Clause
 
#include <string.h>
#include <stdio.h>
#include <dirent.h>
#include <sys/stat.h>
#include <unistd.h>
#include "yac_mpi_internal.h"
#include "io_utils.h"
#include "math.h"
 
#ifdef YAC_NETCDF_ENABLED
#include <netcdf.h>
#endif
 
#define IO_RANK_LIST_STR "YAC_IO_RANK_LIST"
#define IO_MAX_NUM_RANKS_STR "YAC_IO_MAX_NUM_RANKS"
#define IO_RANK_EXCLUDE_LIST_STR "YAC_IO_RANK_EXCLUDE_LIST"
#define IO_MAX_NUM_RANKS_PER_NODE "YAC_IO_MAX_NUM_RANKS_PER_NODE"
#define DEFAULT_MAX_NUM_IO_RANK_PER_NODE (1)
 
static inline int compare_int(const void * a, const void * b) {
 
  int const * a_ = a, * b_ = b;
 
  return (*a_ > *b_) - (*b_ > *a_);
}
 
static void read_rank_list(
  char const * env_name, MPI_Comm comm, size_t * num_ranks_, int ** ranks_) {
 
  int rank;
  yac_mpi_call(MPI_Comm_rank(comm, &rank), comm);
 
  size_t num_ranks = 0;
  int * ranks = NULL;
 
  // environment is only checked on rank 0, results are broadcasted
  // to other processes
  if (rank == 0) {
 
    // check whether the user provided a list of ranks for IO
    char * rank_list_str = getenv(env_name);
    if ((rank_list_str != NULL) && (rank_list_str[0] != '\0')) {
 
      int temp_num_ranks = 1;
      for (char * curr_char = rank_list_str + 1; *curr_char != '\0'; ++curr_char)
        if (*curr_char == ',') ++temp_num_ranks;
 
      char * rank_list_copy = xstrdup(rank_list_str);
      num_ranks = 0;
      int * world_ranks = xmalloc(temp_num_ranks * sizeof(*world_ranks));
      int world_size;
      yac_mpi_call(MPI_Comm_size(MPI_COMM_WORLD, &world_size), MPI_COMM_WORLD);
 
      // parse rank list
      char * rank_str = strtok(rank_list_copy, ",");
      while (rank_str != NULL) {
        int curr_rank = atoi(rank_str);
        YAC_ASSERT_F(curr_rank >= 0, "\"%s\" is not a valid rank", rank_str);
        YAC_ASSERT_F(
          curr_rank < world_size, "rank %d exceeds size of MPI_COMM_WORLD (%d)",
          curr_rank,  world_size);
        world_ranks[num_ranks++] = curr_rank;
        rank_str = strtok(NULL, ",");
      }
      free(rank_list_copy);
 
      // sort ranks
      qsort(world_ranks, num_ranks, sizeof(*world_ranks), compare_int);
 
      // remove duplicated ranks
      yac_remove_duplicates_int(world_ranks, &num_ranks);
 
      // translate rank list from MPI_COMM_WORLD to comm
      ranks = xmalloc(num_ranks * sizeof(ranks));
      MPI_Group world_group, comm_group;
      yac_mpi_call(
        MPI_Comm_group(MPI_COMM_WORLD, &world_group), MPI_COMM_WORLD);
      yac_mpi_call(MPI_Comm_group(comm, &comm_group), comm);
      yac_mpi_call(
        MPI_Group_translate_ranks(
          world_group, (int)num_ranks, world_ranks,
          comm_group, ranks), MPI_COMM_WORLD);
      yac_mpi_call(MPI_Group_free(&comm_group), comm);
      yac_mpi_call(MPI_Group_free(&world_group), MPI_COMM_WORLD);
      free(world_ranks);
 
      // remove ranks, which are not avaible in comm
      size_t new_num_ranks = 0;
      for (size_t i = 0; i < num_ranks; ++i) {
        if (ranks[i] != MPI_UNDEFINED) {
          if (i != new_num_ranks)
            ranks[new_num_ranks] = ranks[i];
          ++new_num_ranks;
        }
      }
      num_ranks = new_num_ranks;
 
      // sort ranks
      qsort(ranks, num_ranks, sizeof(*ranks), compare_int);
    }
 
    // broadcast ranks
    yac_mpi_call(
      MPI_Bcast(&num_ranks, 1, YAC_MPI_SIZE_T, 0, comm), comm);
    if (num_ranks > 0) {
      ranks = xrealloc(ranks, num_ranks * sizeof(*ranks));
      yac_mpi_call(
        MPI_Bcast(ranks, (int)num_ranks, MPI_INT, 0, comm), comm);
    } else {
      free(ranks);
    }
  } else {
 
    // receive ranks from root
    yac_mpi_call(
      MPI_Bcast(&num_ranks, 1, YAC_MPI_SIZE_T, 0, comm), comm);
    if (num_ranks > 0) {
      ranks = xmalloc(num_ranks * sizeof(*ranks));
      yac_mpi_call(
        MPI_Bcast(ranks, (int)num_ranks, MPI_INT, 0, comm), comm);
    }
  }
  *num_ranks_ = num_ranks;
  *ranks_ = ranks;
}
 
static void read_io_rank_list(
  MPI_Comm comm, size_t * num_io_ranks, int ** io_ranks) {
 
  read_rank_list(IO_RANK_LIST_STR, comm, num_io_ranks, io_ranks);
}
 
static void check_io_max_num_ranks_per_node(
  MPI_Comm comm, size_t * num_io_ranks_, int ** io_ranks_) {
 
  int rank, size;
  yac_mpi_call(MPI_Comm_rank(comm, &rank), comm);
  yac_mpi_call(MPI_Comm_size(comm, &size), comm);
 
  int max_num_io_rank_per_node;
  // get the maximum number of io ranks per node from the environment variable
  max_num_io_rank_per_node =
    yac_getenv_int(
      IO_MAX_NUM_RANKS_PER_NODE, DEFAULT_MAX_NUM_IO_RANK_PER_NODE, 1, comm);
 
  // if there is no limit on the number of io ranks per node
  if (max_num_io_rank_per_node == -1) return;
 
  // create one communicator per node
  MPI_Comm node_comm;
  yac_mpi_call(
    MPI_Comm_split_type(
      comm, MPI_COMM_TYPE_SHARED, rank, MPI_INFO_NULL, &node_comm), comm);
 
  // determine rank within node comm
  int node_rank;
  yac_mpi_call(MPI_Comm_rank(node_comm, &node_rank), node_comm);
 
  // translate io ranks to node ranks
  int * node_ranks = xmalloc(*num_io_ranks_ * sizeof(*node_ranks));
  MPI_Group io_group, node_group;
  yac_mpi_call(MPI_Comm_group(comm, &io_group), comm);
  yac_mpi_call(MPI_Comm_group(node_comm, &node_group), node_comm);
  yac_mpi_call(
    MPI_Group_translate_ranks(
      io_group, (int)*num_io_ranks_, *io_ranks_,
      node_group, node_ranks), comm);
  yac_mpi_call(MPI_Group_free(&node_group), node_comm);
  yac_mpi_call(MPI_Group_free(&io_group), comm);
  yac_mpi_call(MPI_Comm_free(&node_comm), comm);
 
  // remove MPI_UNDEFINED entries
  size_t num_node_io_ranks = 0;
  for (size_t i = 0; i < *num_io_ranks_; ++i) {
    if (node_ranks[i] != MPI_UNDEFINED) {
      if (i != num_node_io_ranks)
        node_ranks[num_node_io_ranks] = node_ranks[i];
      ++num_node_io_ranks;
    }
  }
 
  // sort ranks
  qsort(node_ranks, num_node_io_ranks, sizeof(*node_ranks), compare_int);
 
  // check whether local process is within the first n io ranks on its node
  int local_is_io_rank = 0;
  for (size_t i = 0;
       (i < num_node_io_ranks) &&
       (i < (size_t)max_num_io_rank_per_node) && !local_is_io_rank; ++i)
    if (node_ranks[i] == node_rank) local_is_io_rank = 1;
  free(node_ranks);
 
  // determine io ranks on all nodes
  int * is_io_rank = xmalloc((size_t)size * sizeof(*is_io_rank));
  yac_mpi_call(
    MPI_Allgather(
      &local_is_io_rank, 1, MPI_INT, is_io_rank, 1, MPI_INT, comm), comm);
 
  // compress is_io_rank into io_ranks
  size_t num_io_ranks = 0;
  for (int i = 0; i < size; ++i)
    if (is_io_rank[i]) is_io_rank[num_io_ranks++] = i;
 
  free(*io_ranks_);
  *num_io_ranks_ = num_io_ranks;
  *io_ranks_ = xrealloc(is_io_rank, num_io_ranks * sizeof(**io_ranks_));
}
 
static void check_io_max_num_ranks(
  MPI_Comm comm, size_t * num_io_ranks, int ** io_ranks) {
 
  int rank;
  yac_mpi_call(MPI_Comm_rank(comm, &rank), comm);
 
  int max_num_ranks = yac_getenv_int(IO_MAX_NUM_RANKS_STR, -1, 1, comm);
 
  // if there is no limit on the number of io ranks
  if (max_num_ranks == -1) return;
 
  if ((max_num_ranks > 0) && ((size_t)max_num_ranks < *num_io_ranks)) {
    *io_ranks =
      xrealloc(*io_ranks, (size_t)max_num_ranks * sizeof(**io_ranks));
    *num_io_ranks = (size_t)max_num_ranks;
  }
}
 
static void check_io_rank_exclude_list(
  MPI_Comm comm, size_t * num_io_ranks, int ** io_ranks) {
 
  // read in the list of ranks which are not to be used for io
  size_t num_io_ranks_excluded;
  int * io_ranks_excluded;
  read_rank_list(
    IO_RANK_EXCLUDE_LIST_STR, comm, &num_io_ranks_excluded, &io_ranks_excluded);
 
  // if there are ranks to be excluded
  if (num_io_ranks_excluded > 0) {
 
    // sort ranks
    qsort(*io_ranks, *num_io_ranks, sizeof(**io_ranks), compare_int);
 
    // match exclude list with io rank list
    size_t new_num_io_ranks = 0;
    for (size_t i = 0, j = 0; i < *num_io_ranks; ++i) {
 
      while ((j < num_io_ranks_excluded) &&
             (io_ranks_excluded[j] < (*io_ranks)[i])) ++j;
 
      if ((j >= num_io_ranks_excluded) ||
          ((*io_ranks)[i] != io_ranks_excluded[j])) {
 
        if (i != new_num_io_ranks)
          (*io_ranks)[new_num_io_ranks] = (*io_ranks)[i];
        ++new_num_io_ranks;
      }
    }
 
    if (new_num_io_ranks != *num_io_ranks) {
      *io_ranks = xrealloc(*io_ranks, new_num_io_ranks * sizeof(**io_ranks));
      *num_io_ranks = new_num_io_ranks;
    }
  }
 
  free(io_ranks_excluded);
}
 
void yac_get_io_ranks(
  MPI_Comm comm, int * local_is_io_, int ** io_ranks_, int * num_io_ranks_) {
 
  int rank, size;
  yac_mpi_call(MPI_Comm_rank(comm, &rank), comm);
  yac_mpi_call(MPI_Comm_size(comm, &size), comm);
 
  size_t num_io_ranks = 0;
  int * io_ranks = NULL;
 
  // check environment for io rank list
  read_io_rank_list(comm, &num_io_ranks, &io_ranks);
 
  // if no rank list was provided -> generate default rank list (all processes)
  if (num_io_ranks == 0) {
    num_io_ranks = (size_t)size;
    io_ranks = xmalloc(num_io_ranks * sizeof(*io_ranks));
    for (int i = 0; i < size; ++i) io_ranks[i] = i;
  }
 
  // check whether we have to exclude some ranks
  check_io_rank_exclude_list(comm, &num_io_ranks, &io_ranks);
 
  // check for the maximum number of io ranks per node
  check_io_max_num_ranks_per_node(comm, &num_io_ranks, &io_ranks);
 
  // check maximum number of io ranks
  check_io_max_num_ranks(comm, &num_io_ranks, &io_ranks);
 
  YAC_ASSERT(num_io_ranks > 0, "could not determine io ranks");
 
  int local_is_io = 0;
  for (size_t i = 0; (i < num_io_ranks) && !local_is_io; ++i)
    if (io_ranks[i] == rank) local_is_io = 1;
 
  *local_is_io_ = local_is_io;
  *io_ranks_ = io_ranks;
  *num_io_ranks_ = (int)num_io_ranks;
}
 
void yac_nc_open(const char * path, int omode, int * ncidp) {
 
#ifndef YAC_NETCDF_ENABLED
 
  UNUSED(path);
  UNUSED(omode);
  UNUSED(ncidp);
  die("ERROR(yac_nc_open): YAC is built without the NetCDF support");
#else
 
  YAC_ASSERT_F(yac_file_exists(path), "file \"%s\" does not exist",  path);
  YAC_HANDLE_ERROR(nc_open(path, omode, ncidp));
#endif
}
 
void yac_nc_create(const char * path, int cmode, int * ncidp) {
 
#ifndef YAC_NETCDF_ENABLED
 
  UNUSED(path);
  UNUSED(cmode);
  UNUSED(ncidp);
  die("ERROR(yac_nc_create): YAC is built without the NetCDF support");
#else
 
  int status = nc_create(path, cmode, ncidp);
  YAC_ASSERT_F(
    status == NC_NOERR,
    "failed to create file \"%s\" (NetCDF error message: \"%s\")",
    path,  nc_strerror(status));
#endif
}
 
void yac_nc_inq_dimid(int ncid, char const * name, int * dimidp) {
 
#ifndef YAC_NETCDF_ENABLED
 
  UNUSED(ncid);
  UNUSED(name);
  UNUSED(dimidp);
  die("ERROR(yac_nc_inq_dimid): YAC is built without the NetCDF support");
#else
 
  int status = nc_inq_dimid(ncid, name, dimidp);
 
  if (status == NC_EBADDIM) {
    // GCOVR_EXCL_START
    size_t pathlen;
    YAC_HANDLE_ERROR(nc_inq_path(ncid, &pathlen, NULL));
    char * path = xmalloc(pathlen * sizeof(*path));
    YAC_HANDLE_ERROR(nc_inq_path(ncid, NULL, path));
    YAC_ASSERT_F(
      0, "dimension \"%s\" could not be found in file \"%s\"",  name,  path);
    // GCOVR_EXCL_STOP
  } else YAC_HANDLE_ERROR(status);
#endif
}
 
void yac_nc_inq_varid(int ncid, char const * name, int * varidp) {
 
#ifndef YAC_NETCDF_ENABLED
 
  UNUSED(ncid);
  UNUSED(name);
  UNUSED(varidp);
  die("ERROR(yac_nc_inq_varid): YAC is built without the NetCDF support");
#else
 
  int status = nc_inq_varid(ncid, name, varidp);
 
  if (status == NC_ENOTVAR) {
    // GCOVR_EXCL_START
    size_t pathlen;
    YAC_HANDLE_ERROR(nc_inq_path(ncid, &pathlen, NULL));
    char * path = xmalloc(pathlen * sizeof(*path));
    YAC_HANDLE_ERROR(nc_inq_path(ncid, NULL, path));
    YAC_ASSERT_F(
      0, "variable \"%s\" could not be found in file \"%s\"",  name,  path);
    // GCOVR_EXCL_STOP
  } else YAC_HANDLE_ERROR(status);
#endif
}
 
int yac_file_exists(const char * filename) {
  struct stat buffer;
  return !stat(filename,&buffer);
}
 
int yac_file_contains(char const *path, char const *needle) {
 
  YAC_ASSERT(path != NULL, "path must not be NULL");
  YAC_ASSERT(path[0] != '\0', "path must not be an empty string");
 
  FILE *f = fopen(path, "r");
 
  YAC_ASSERT_F(f, "failed to open file \"%s\" for reading", path);
  YAC_ASSERT(needle != NULL, "needle must not be NULL");
  YAC_ASSERT(needle[0] != '\0', "needle must not be an empty string");
 
  char *line = NULL;
  size_t line_cap = 0;
  int found = 0;
 
  while (!found && (getline(&line, &line_cap, f) != -1)) {
    found = strstr(line, needle) != NULL;
  }
 
  YAC_ASSERT_F(!ferror(f), "failed to read file \"%s\"", path);
 
  fclose(f);
  free(line);
 
  return found;
}
 
char * yac_file_path_join(
  char const *path_prefix, char const *path_suffix, char const *extension) {
 
  YAC_ASSERT(path_prefix != NULL, "path_prefix must not be NULL");
  YAC_ASSERT(path_prefix[0] != '\0', "path_prefix must not be empty");
  YAC_ASSERT(path_suffix != NULL, "path_suffix must not be NULL");
  YAC_ASSERT(path_suffix[0] != '\0', "path_suffix must not be empty");
 
  extension = extension  ? extension : "";
 
  size_t len_prefix = strlen(path_prefix);
  size_t len_suffix = strlen(path_suffix);
  size_t len_extension = strlen(extension);
  size_t len = len_prefix + len_suffix + len_extension + 2;
  char * path = xmalloc(len * sizeof(*path));
 
  int nw = snprintf(path, len, "%s/%s%s", path_prefix, path_suffix, extension);
 
  YAC_ASSERT_F(
    (nw >= 0) && ((size_t)nw < len),
    "failed to construct path from \"%s\", \"%s\", [and \"%s\"]",
    path_prefix, path_suffix, extension);
 
  return path;
}
 
static void remove_directory_recursive(char const *path) {
 
  YAC_ASSERT(path != NULL, "path must not be NULL");
  YAC_ASSERT(path[0] != '\0', "path must not be an empty string");
 
  DIR *dir = opendir(path);
 
  YAC_ASSERT_F(dir != NULL, "failed to open directory \"%s\"", path);
 
  struct dirent *entry;
 
  while ((entry = readdir(dir)) != NULL) {
 
    // skip "." and ".." entries
    if (!strcmp(entry->d_name, ".") ||
        !strcmp(entry->d_name, "..")) {
      continue;
    }
 
    char * child = yac_file_path_join(path, entry->d_name, NULL);
 
    struct stat statbuf;
 
    YAC_ASSERT_F(stat(child, &statbuf) == 0, "failed to stat \"%s\"", child);
 
    // if the entry is a directory, recursively remove it,
    // otherwise remove the file
    if (S_ISDIR(statbuf.st_mode)) {
 
      remove_directory_recursive(child);
 
    } else {
 
      int status = unlink(child);
 
      YAC_ASSERT_F(status == 0, "failed to remove file \"%s\"", child);
    }
 
    free(child);
  }
 
  int status = closedir(dir);
  YAC_ASSERT_F(status == 0, "failed to close directory \"%s\"", path);
 
  status = rmdir(path);
  YAC_ASSERT_F(status == 0, "failed to remove directory \"%s\"", path);
}
 
void yac_remove_directory(char const *path) {
 
  YAC_ASSERT(path != NULL, "path must not be NULL");
  YAC_ASSERT(path[0] != '\0', "path must not be an empty string");
  YAC_ASSERT_F(
    yac_file_exists(path), "directory \"%s\" does not exist", path);
 
  remove_directory_recursive(path);
}