sion_mpi_gen.c

/****************************************************************************
**  SIONLIB     http://www.fz-juelich.de/jsc/sionlib                       **
*****************************************************************************
**  Copyright (c) 2008-2021                                                **
**  Forschungszentrum Juelich, Juelich Supercomputing Centre               **
**                                                                         **
**  See the file COPYRIGHT in the package base directory for details       **
****************************************************************************/
 
#define _XOPEN_SOURCE 700
 
#ifdef SION_MPI
 
#include <assert.h>
#include <mpi.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
 
#include "sion_const.h"
#include "sion_debug.h"
#include "sion_enums.h"
#include "sion_error_handler.h"
#include "sion_flags.h"
#include "sion_generic.h"
#include "sion_generic_internal.h"
#include "sion_internal.h"
#include "sion_mpi.h"
#include "sion_mpi_cb_gen.h"
#include "sion_mpi_internal_gen.h"
 
int _sion_mpi_api_aid = -1;
 
int sion_parclose_mpi(int sid)
{
  int rc = 0;
 
  DPRINTFP((1, "sion_parclose_mpi", _SION_DEFAULT_RANK, "enter parallel close of sid %d\n", sid));
 
  rc = sion_generic_parclose(sid);
 
  DPRINTFP((1, "sion_parclose_mpi", _SION_DEFAULT_RANK, "leave parallel close of sid %d rc=%d\n", sid, rc));
 
  return rc;
}
 
int sion_parreinit_mpi(int sid, int64_t chunksize)
{
  int rc = 0;
 
  DPRINTFP((1, "sion_parreinit_mpi", _SION_DEFAULT_RANK, "enter parallel reinit of sid %d\n", sid));
 
  rc = sion_generic_parreinit(sid, chunksize);
 
  DPRINTFP((1, "sion_parreinit_mpi", _SION_DEFAULT_RANK, "leave parallel reinit of sid %d rc=%d\n", sid, rc));
 
  return rc;
}
 
int sion_paropen_mapped_mpi(char *fname, const char *file_mode, int *numFiles, MPI_Comm gComm, int *nlocaltasks,
  int **globalranks, int64_t **chunksizes, int **mapping_filenrs, int **mapping_lranks, int32_t *fsblksize, FILE **fileptr)
{
  int sid;
  _sion_flags_store *flags_store = _sion_parse_flags(file_mode);
  if (!flags_store) {
    return _sion_errorprint_mpi(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "sion_paropen_mapped_mpi: could not parse file mode in %s, aborting ...\n", file_mode);
  }
  sid = _sion_paropen_mapped_mpi(fname, flags_store, numFiles, gComm, nlocaltasks, globalranks, chunksizes, mapping_filenrs,
    mapping_lranks, fsblksize, fileptr);
  _sion_flags_destroy_store(&flags_store);
  return sid;
}
 
int _sion_paropen_mapped_mpi(char *fname, const _sion_flags_store *flags_store, int *numFiles, MPI_Comm gComm, int *nlocaltasks,
  int **globalranks, int64_t **chunksizes, int **mapping_filenrs, int **mapping_lranks, int32_t *fsblksize, FILE **fileptr)
{
  int sid = SION_ID_UNDEF;
  int gtasks, gRank;
  _mpi_api_commdata *gen_gcomm;
 
  MPI_Comm_size(gComm, &gtasks);
  MPI_Comm_rank(gComm, &gRank);
 
  DPRINTFP((1, "sion_paropen_mapped_mpi", gRank, "enter parallel open of file %s\n", fname));
 
  /* check parameters */
  if (numFiles == NULL) {
    return _sion_errorprint_mpi(SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_mapped_mpi: No numFiles variable given");
  }
 
  /* register callbacks for generic interface */
  if (_sion_mpi_api_aid < 0) {
    _sion_mpi_api_aid = _sion_register_callbacks_mpi();
  }
 
  if (flags_store->mask & _SION_FMODE_WRITE) {
    /* file mode WRITE */
 
    if (*numFiles <= 0) {
      return _sion_errorprint_mpi(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
        "sion_paropen_mapped_mpi: numFiles variable <= 0 not allowed for mapped files in write mode");
    }
  } else if (flags_store->mask & _SION_FMODE_READ) {
    /* file mode READ */
    /* nothing to do here so far, filenumbers and mapping will be determined by in generic routine */
 
  } else {
    return _sion_errorprint_mpi(SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_mapped_mpi: unknown file mode");
  }
 
  /* create generic communicator container */
  gen_gcomm = malloc(sizeof(_mpi_api_commdata));
  if (gen_gcomm == NULL) {
    return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "cannot allocate mpi internal data structure of size %lu (_mpi_api_commdata), aborting ...\n",
      (unsigned long)sizeof(_mpi_api_commdata));
  }
  gen_gcomm->comm = gComm;
  gen_gcomm->commset = 1;
  gen_gcomm->local = 0;
  gen_gcomm->rank = gRank;
  gen_gcomm->size = gtasks;
  gen_gcomm->lcommgroup = NULL;
 
  // FIXME: re-enable output of file mode
  // DPRINTFP((1, "sion_paropen_mapped_mpi", gRank, "enter parallel open of %d files (current name %s) in %s mode (sid=%d)\n", *numFiles, fname, file_mode, sid));
  DPRINTFP(
    (1, "sion_paropen_mapped_mpi", gRank, "enter parallel open of %d files (current name %s) (sid=%d)\n", *numFiles, fname, sid));
  sid = _sion_generic_paropen_mapped(_sion_mpi_api_aid, fname, flags_store, numFiles, gen_gcomm, gRank, gtasks, nlocaltasks,
    globalranks, chunksizes, mapping_filenrs, mapping_lranks, fsblksize, fileptr);
  // FIXME: re-enable output of file mode
  // DPRINTFP((1, "sion_paropen_mapped_mpi", gRank, "leave parallel open of %d files in %s mode #tasks=%d sid=%d\n", *numFiles, file_mode, *nlocaltasks, sid));
  DPRINTFP(
    (1, "sion_paropen_mapped_mpi", gRank, "leave parallel open of %d files #tasks=%d sid=%d\n", *numFiles, *nlocaltasks, sid));
 
  /* test return code from internal open */
  if (sid == SION_ID_NOT_VALID) {
    return _sion_errorprint_mpi(
      SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_mapped_mpi: invalid return code from internal open %d", sid);
  }
 
  DPRINTFP((1, "sion_paropen_mapped_mpi", gRank, "leave parallel open of file %s sid=%d\n", fname, sid));
 
  return sid;
}
 
int sion_paropen_mapped_mpi_with_options(const char *filename, sion_open_mode mode, int numfiles, MPI_Comm gcomm, int ntasks,
  const int *globalranks, const int *filenumbers, const int *localranks, const sion_mpi_mapped_options *options_)
{
  sion_mpi_mapped_options options = (options_) ? *options_ : SION_MPI_MAPPED_OPTIONS_INIT;
 
  int rank, size;
  MPI_Comm_size(gcomm, &size);
  MPI_Comm_rank(gcomm, &rank);
 
  DPRINTFP((1, "sion_paropen_mapped_mpi", rank, "enter parallel open of file %s\n", filename));
 
  /* register callbacks for generic interface */
  if (_sion_mpi_api_aid < 0) {
    _sion_mpi_api_aid = _sion_register_callbacks_mpi();
  }
 
  if (mode == SION_OPEN_WRITE) {
    /* file mode WRITE */
 
    if (numfiles <= 0) {
      return _sion_errorprint_mpi(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
        "sion_paropen_mapped_mpi: numfiles variable <= 0 not allowed for mapped files in write mode");
    }
  } else if (mode == SION_OPEN_READ) {
    /* file mode READ */
    /* nothing to do here so far, filenumbers and mapping will be determined by in generic routine */
 
  } else {
    return _sion_errorprint_mpi(SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_mapped_mpi: unknown file mode");
  }
 
  /* create generic communicator container */
  _mpi_api_commdata *gen_gcomm = malloc(sizeof(_mpi_api_commdata));
  if (gen_gcomm == NULL) {
    return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "cannot allocate mpi internal data structure of size %lu (_mpi_api_commdata), aborting ...\n",
      (unsigned long)sizeof(_mpi_api_commdata));
  }
  *gen_gcomm = (_mpi_api_commdata){.comm = gcomm, .commset = 1, .local = 0, .rank = rank, .size = size, .lcommgroup = NULL};
 
  // FIXME: re-enable output of file mode
  // DPRINTFP((1, "sion_paropen_mapped_mpi", rank, "enter parallel open of %d files (current name %s) in %s mode (sid=%d)\n", *numFiles, filename, file_mode, sid));
  DPRINTFP((1, "sion_paropen_mapped_mpi", rank, "enter parallel open of %d files (current name %s)\n", numfiles, filename));
  int sid = sion_generic_paropen_mapped_with_options(_sion_mpi_api_aid, filename, mode, numfiles, gen_gcomm, rank, size, ntasks,
    globalranks, filenumbers, localranks, &options.generic_options);
  // FIXME: re-enable output of file mode
  // DPRINTFP((1, "sion_paropen_mapped_mpi", gRank, "leave parallel open of %d files in %s mode #tasks=%d sid=%d\n", *numFiles, file_mode, *nlocaltasks, sid));
  DPRINTFP((1, "sion_paropen_mapped_mpi", rank, "leave parallel open of %d files #tasks=%d sid=%d\n", numfiles, ntasks, sid));
 
  /* test return code from internal open */
  if (sid == SION_ID_NOT_VALID) {
    return _sion_errorprint_mpi(
      SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_mapped_mpi: invalid return code from internal open %d", sid);
  }
 
  DPRINTFP((1, "sion_paropen_mapped_mpi", rank, "leave parallel open of file %s sid=%d\n", filename, sid));
 
  return sid;
}
 
int sion_parclose_mapped_mpi(int sid)
{
  int rc = 0;
 
  DPRINTFP((1, "sion_parclose_mapped_mpi", _SION_DEFAULT_RANK, "enter parallel close of sid %d\n", sid));
 
  rc = sion_generic_parclose_mapped(sid);
 
  DPRINTFP((1, "sion_parclose_mapped_mpi", _SION_DEFAULT_RANK, "leave parallel close of sid %d rc=%d\n", sid, rc));
 
  return rc;
}
 
sion_mpi_options *sion_mpi_options_new()
{
  sion_mpi_options *options = malloc(sizeof(sion_mpi_options));
  if (options) {
    *options = SION_MPI_OPTIONS_INIT;
  }
  return options;
}
 
void sion_mpi_options_delete(sion_mpi_options *options)
{
  free(options);
}
 
void sion_mpi_options_set_chunksize(sion_mpi_options *options, int64_t chunksize)
{
  sion_generic_options_set_chunksize(&options->generic_options, chunksize);
}
 
void sion_mpi_options_set_fsblksize(sion_mpi_options *options, int32_t fsblksize)
{
  sion_generic_options_set_fsblksize(&options->generic_options, fsblksize);
}
 
void sion_mpi_options_set_multifile_number(sion_mpi_options *options, int multifile_number)
{
  options->multifile_mode = _SION_MPI_MULTIFILE_NUMBER;
  options->multifile.number = multifile_number;
}
 
void sion_mpi_options_set_multifile_communicator(sion_mpi_options *options, MPI_Comm multifile_communicator)
{
  options->multifile_mode = _SION_MPI_MULTIFILE_COMMUNICATOR;
  options->multifile.communicator = multifile_communicator;
}
 
void sion_mpi_options_set_keyval_mode(sion_mpi_options *options, sion_keyval_mode keyval_mode)
{
  sion_generic_options_set_keyval_mode(&options->generic_options, keyval_mode);
}
 
void sion_mpi_options_set_buddy(sion_mpi_options *options)
{
  sion_mpi_options_set_buddylevel(options, 1);
}
 
void sion_mpi_options_set_buddylevel(sion_mpi_options *options, int32_t buddylevel)
{
  sion_generic_options_set_buddylevel(&options->generic_options, buddylevel);
}
 
void sion_mpi_options_set_collective(sion_mpi_options *options)
{
  sion_mpi_options_set_collective_size(options, -1);
}
 
void sion_mpi_options_set_collective_size(sion_mpi_options *options, int32_t size)
{
  sion_generic_options_set_collective_size(&options->generic_options, size);
}
 
void sion_mpi_options_set_collective_merge(sion_mpi_options *options)
{
  sion_generic_options_set_collective_merge(&options->generic_options);
}
 
void sion_mpi_options_set_lowlevel_api(sion_mpi_options *options, sion_lowlevel_api lowlevel_api)
{
  sion_generic_options_set_lowlevel_api(&options->generic_options, lowlevel_api);
}
 
void sion_mpi_options_set_endianness(sion_mpi_options *options, sion_endianness endianness)
{
  sion_generic_options_set_endianness(&options->generic_options, endianness);
}
 
sion_mpi_mapped_options *sion_mpi_mapped_options_new()
{
  sion_mpi_mapped_options *options = malloc(sizeof(sion_mpi_mapped_options));
  if (options) {
    *options = SION_MPI_MAPPED_OPTIONS_INIT;
  }
  return options;
}
 
void sion_mpi_mapped_options_delete(sion_mpi_mapped_options *options)
{
  free(options);
}
 
void sion_mpi_mapped_options_set_chunksizes(sion_mpi_mapped_options *options, int64_t *chunksizes)
{
  sion_generic_mapped_options_set_chunksizes(&options->generic_options, chunksizes);
}
 
void sion_mpi_mapped_options_set_fsblksize(sion_mpi_mapped_options *options, int32_t fsblksize)
{
  sion_generic_mapped_options_set_fsblksize(&options->generic_options, fsblksize);
}
 
void sion_mpi_mapped_options_set_keyval_mode(sion_mpi_mapped_options *options, sion_keyval_mode keyval_mode)
{
  sion_generic_mapped_options_set_keyval_mode(&options->generic_options, keyval_mode);
}
 
void sion_mpi_mapped_options_set_lowlevel_api(sion_mpi_mapped_options *options, sion_lowlevel_api lowlevel_api)
{
  sion_generic_mapped_options_set_lowlevel_api(&options->generic_options, lowlevel_api);
}
 
void sion_mpi_mapped_options_set_endianness(sion_mpi_mapped_options *options, sion_endianness endianness)
{
  sion_generic_mapped_options_set_endianness(&options->generic_options, endianness);
}
 
int sion_paropen_mpi(const char *filename, sion_open_mode mode, MPI_Comm communicator, const sion_mpi_options *options_)
{
  sion_mpi_options options = (options_) ? *options_ : SION_MPI_OPTIONS_INIT;
 
  int gsize, grank;
  MPI_Comm_size(communicator, &gsize);
  MPI_Comm_rank(communicator, &grank);
 
  DPRINTFP((1, "sion_paropen_mpi", grank, "enter parallel open of file %s\n", filename));
 
  /* register callbacks for generic interface */
  if (_sion_mpi_api_aid < 0) {
    _sion_mpi_api_aid = _sion_register_callbacks_mpi();
  }
 
  MPI_Comm lcomm;
  int nfiles, filenumber, lrank, lsize;
  if (mode == SION_OPEN_WRITE) {
    switch (options.multifile_mode) {
      int rc;
    case _SION_MPI_MULTIFILE_COMMUNICATOR:
      lcomm = options.multifile.communicator;
      rc = _sion_get_info_from_splitted_comm_mpi(communicator, lcomm, &nfiles, &filenumber, &lrank, &lsize);
      if (rc != SION_SUCCESS) {
        return _sion_errorprint_mpi(
          SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_mpi: error in _sion_get_info_from_splitted_comm_mpi");
      }
      DPRINTFP((1, "sion_paropen_mpi", grank, "%d local communicators found\n", nfiles));
      break;
    case _SION_MPI_MULTIFILE_NUMBER: // FALLTHROUGH
    default:
      nfiles = options.multifile.number;
      rc = _sion_gen_info_from_gcomm_mpi(nfiles, communicator, &filenumber, &lrank, &lsize);
      if (rc != SION_SUCCESS) {
        return _sion_errorprint_mpi(
          SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_mpi: error in _sion_gen_info_from_gcomm_mpi");
      }
      DPRINTFP((1, "sion_paropen_mpi", grank, "Global communicator divided in %d local communicators\n", nfiles));
    }
  } else if (mode == SION_OPEN_READ) {
    if (options.generic_options.buddylevel > 0) {
      // lcomm must be given for buddy checkpointing
      // FIXME
      assert(options.multifile_mode == _SION_MPI_MULTIFILE_COMMUNICATOR);
      lcomm = options.multifile.communicator;
      int rc = _sion_get_info_from_splitted_comm_mpi(communicator, lcomm, &nfiles, &filenumber, &lrank, &lsize);
      if (rc != SION_SUCCESS) {
        return _sion_errorprint_mpi(
          SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_mpi: error in _sion_get_info_from_splitted_comm_mpi");
      }
      DPRINTFP((1, "sion_paropen_mpi", grank, "%d local communicators found\n", nfiles));
    } else {
      lrank = lsize = -1; // will be set by sion_generic_paropen
    }
  } else {
    return _sion_errorprint_mpi(SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_mpi: unknown file mode");
  }
 
  /* create generic communicator container */
  _mpi_api_commdata *gen_gcomm = malloc(sizeof(_mpi_api_commdata));
  if (!gen_gcomm) {
    return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "cannot allocate mpi internal data structure of size %lu (_mpi_api_commdata), aborting ...\n",
      (unsigned long)sizeof(_mpi_api_commdata));
  }
  gen_gcomm->comm = communicator;
  gen_gcomm->commset = 1;
  gen_gcomm->local = 0;
  gen_gcomm->rank = grank;
  gen_gcomm->size = gsize;
  gen_gcomm->lcommgroup = NULL;
 
  /* FIXME: add debug output for flags_store
  DPRINTFP((1, "sion_paropen_mpi", grank, "enter parallel open of %d files (current name %s) in %s mode\n", nfiles, filename, file_mode)); */
  DPRINTFP((1, "sion_paropen_mpi", grank, "enter parallel open of %d files (current name %s)\n", nfiles, filename));
  int sid = sion_generic_paropen(
    _sion_mpi_api_aid, filename, mode, gen_gcomm, grank, gsize, filenumber, nfiles, lrank, lsize, &options.generic_options);
  /* FIXME: add debug output for flags_store
  DPRINTFP((1, "sion_paropen_mpi", grank, "leave parallel open of %d files in %s mode #tasks=%d sid=%d\n", nfiles, file_mode, lSize, sid)); */
  DPRINTFP((1, "sion_paropen_mpi", grank, "leave parallel open of %d files #tasks=%d sid=%d\n", nfiles, lsize, sid));
 
  /* test return code from internal open */
  if (sid == SION_ID_NOT_VALID) {
    return _sion_errorprint_mpi(
      SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_mpi: invalid return code from internal open %d", sid);
  }
 
  DPRINTFP((1, "sion_paropen_mpi", grank, "leave parallel open of file %s sid=%d\n", filename, sid));
 
  return sid;
}
 
/* end of ifdef MPI */
#endif