sion_ompi_gen.c

/****************************************************************************
**  SIONLIB     http://www.fz-juelich.de/jsc/sionlib                       **
*****************************************************************************
**  Copyright (c) 2008-2019                                                **
**  Forschungszentrum Juelich, Juelich Supercomputing Centre               **
**                                                                         **
**  See the file COPYRIGHT in the package base directory for details       **
****************************************************************************/

#define _XOPEN_SOURCE 700

#include <mpi.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>

#include "sion_common.h"
#include "sion_const.h"
#include "sion_debug.h"
#include "sion_enums.h"
#include "sion_error_handler.h"

#ifdef SION_OMPI

#include <omp.h>

#include "sion_generic.h"
#include "sion_ompi.h"
#include "sion_ompi_cb_gen.h"
#include "sion_ompi_internal_gen.h"

int _sion_ompi_api_aid = -1;
static omp_lock_t _sion_ompi_lock_data;

static void *__ompi_thread_sync_struct;

int _sion_ompi_user_lock(void *data)
{
  int rc = SION_SUCCESS;
  omp_set_lock(&_sion_ompi_lock_data);
  return rc;
}
int _sion_ompi_user_unlock(void *data)
{
  int rc = SION_SUCCESS;
  omp_unset_lock(&_sion_ompi_lock_data);
  return rc;
}

int sion_parclose_ompi(int sid)
{
  int rc = 0;

  DPRINTFP((1, "sion_parclose_ompi", _SION_DEFAULT_RANK, "enter parallel close of sid %d\n", sid));

  rc = sion_generic_parclose(sid);

  DPRINTFP((1, "sion_parclose_ompi", _SION_DEFAULT_RANK, "leave parallel close of sid %d rc=%d\n", sid, rc));

  return rc;
}

sion_ompi_options *sion_ompi_options_new()
{
  sion_ompi_options *options = malloc(sizeof(sion_ompi_options));
  if (options) {
    *options = SION_OMPI_OPTIONS_INIT;
  }
  return options;
}

void sion_ompi_options_delete(sion_ompi_options *options)
{
  free(options);
}

void sion_ompi_options_set_chunksize(sion_ompi_options *options, int64_t chunksize)
{
  sion_generic_options_set_chunksize(&options->generic_options, chunksize);
}

void sion_ompi_options_set_fsblksize(sion_ompi_options *options, int32_t fsblksize)
{
  sion_generic_options_set_fsblksize(&options->generic_options, fsblksize);
}

void sion_ompi_options_set_multifile_number(sion_ompi_options *options, int multifile_number)
{
  options->multifile_mode = _SION_OMPI_MULTIFILE_NUMBER;
  options->multifile.number = multifile_number;
}

void sion_ompi_options_set_multifile_communicator(sion_ompi_options *options, MPI_Comm multifile_communicator)
{
  options->multifile_mode = _SION_OMPI_MULTIFILE_COMMUNICATOR;
  options->multifile.communicator = multifile_communicator;
}

void sion_ompi_options_set_keyval_mode(sion_ompi_options *options, sion_keyval_mode keyval_mode)
{
  sion_generic_options_set_keyval_mode(&options->generic_options, keyval_mode);
}

void sion_ompi_options_set_buddy(sion_ompi_options *options)
{
  sion_ompi_options_set_buddylevel(options, 1);
}

void sion_ompi_options_set_buddylevel(sion_ompi_options *options, int32_t buddylevel)
{
  sion_generic_options_set_buddylevel(&options->generic_options, buddylevel);
}

void sion_ompi_options_set_collective(sion_ompi_options *options)
{
  sion_ompi_options_set_collective_size(options, -1);
}

void sion_ompi_options_set_collective_size(sion_ompi_options *options, int32_t size)
{
  sion_generic_options_set_collective_size(&options->generic_options, size);
}

void sion_ompi_options_set_collective_merge(sion_ompi_options *options)
{
  sion_generic_options_set_collective_merge(&options->generic_options);
}

void sion_ompi_options_set_lowlevel_api(sion_ompi_options *options, sion_lowlevel_api lowlevel_api)
{
  sion_generic_options_set_lowlevel_api(&options->generic_options, lowlevel_api);
}

void sion_ompi_options_set_endianness(sion_ompi_options *options, sion_endianness endianness)
{
  sion_generic_options_set_endianness(&options->generic_options, endianness);
}

int sion_paropen_ompi(const char *fname, sion_open_mode mode, MPI_Comm gcomm, const sion_ompi_options *options_)
{
  sion_ompi_options options = (options_) ? *options_ : SION_OMPI_OPTIONS_INIT;

  int thread_num = omp_get_thread_num();
  int num_threads = omp_get_num_threads();

  int grank, gsize;
  __ompi_thread_sync *thread_sync;
#pragma omp master
  {
    _sion_debug_set_query_thread_num_function(omp_get_thread_num);
    _sion_error_set_query_thread_num_function(omp_get_thread_num);
    omp_init_lock(&_sion_ompi_lock_data);
    sion_lock_register_lock_callbacks(_sion_ompi_user_lock, _sion_ompi_user_unlock, &_sion_ompi_lock_data);

    MPI_Comm_size(gcomm, &gsize);
    MPI_Comm_rank(gcomm, &grank);

    thread_sync = malloc(sizeof(__ompi_thread_sync));
    if (!thread_sync) {
      _sion_errorprint_ompi(SION_ID_NOT_VALID, _SION_ERROR_ABORT,
        "sion_paropen_ompi: cannot allocate struct of size %lu (__ompi_thread_sync), aborting...", sizeof(__ompi_thread_sync));
    }

    thread_sync->grank_master_mpi = grank;
    thread_sync->gsize_mpi = gsize;
    thread_sync->grank_master_ompi = _sion_map_rank_mpi_to_ompi(grank, num_threads, thread_num);
    thread_sync->gsize_ompi = _sion_get_size_ompi(gsize, num_threads);
    thread_sync->num_threads = num_threads;
    __ompi_thread_sync_struct = thread_sync;
  }
  /* sync to ensure that info in thread_sync is accessible */
#pragma omp barrier

  /* this is actually not necessary, but it makes for cleaner code by preventing us from doing lots of typecasts */
  thread_sync = __ompi_thread_sync_struct;
  DPRINTFP(
    (1, __func__, thread_sync->grank_master_ompi + thread_num, "thread %d enters parallel open of file %s\n", thread_num, fname));

/* create generic API */
#pragma omp master
  {
    /* register callbacks for generic interface */
    if (_sion_ompi_api_aid < 0) {
      _sion_ompi_api_aid = _sion_register_callbacks_ompi();
    }
  }

  /* create global generic communicator container on all threads */
  _ompi_api_commdata *gen_gcomm = malloc(sizeof(_ompi_api_commdata));
  if (gen_gcomm != NULL) {
    gen_gcomm->commset = 0;
    gen_gcomm->local = 0;
    gen_gcomm->rank = thread_sync->grank_master_ompi + thread_num;
    gen_gcomm->size = thread_sync->gsize_ompi;
    gen_gcomm->num_threads = thread_sync->num_threads;
    gen_gcomm->thread_num = thread_num;
    gen_gcomm->lcommgroup = NULL;
  } else {
    return _sion_errorprint_ompi(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "cannot allocate ompi internal data structure of size %lu (_omp_api_commdata), aborting ...\n",
      (unsigned long)sizeof(_ompi_api_commdata));
  }

/* store MPI communicator in global generic communicator container on master thread */
#pragma omp master
  {
    gen_gcomm->comm = gcomm;
  }

  /* sync to ensure that aid is accessible */
#pragma omp barrier

  _ompi_api_commdata *gen_lcomm = NULL;
  int filenumber, nfiles, lrank, lsize;
  if (mode == SION_OPEN_WRITE) {
    /* create generic local communicator container on each thread */
    if (options.multifile_mode == _SION_OMPI_MULTIFILE_COMMUNICATOR) {
      gen_lcomm = malloc(sizeof(_ompi_api_commdata));
      if (!gen_lcomm) {
        return _sion_errorprint_ompi(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
          "cannot allocate ompi internal data structure of size %lu (_ompi_api_commdata), aborting ...\n",
          (unsigned long)sizeof(_ompi_api_commdata));
      }
      gen_lcomm->commset = 1;
      gen_lcomm->commcreated = 0;
      gen_lcomm->local = 1;
      gen_lcomm->num_threads = gen_gcomm->num_threads;
      gen_lcomm->thread_num = thread_num;
      gen_gcomm->lcommgroup = gen_lcomm; /* store pointer in global comm group */
    }

#pragma omp master
    {
      if (options.multifile_mode == _SION_OMPI_MULTIFILE_COMMUNICATOR) {
        int rc =
          _sion_get_info_from_splitted_comm_ompi(gcomm, options.multifile.communicator, &nfiles, &filenumber, &lrank, &lsize);
        if (rc != SION_SUCCESS) {
          _sion_errorprint_ompi(
            SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_ompi: error in _sion_get_info_from_splitted_comm_ompi");
        }
        gen_lcomm->comm = options.multifile.communicator;
        DPRINTFP((1, __func__, grank, "%d local communicators found\n", nfiles));
      } else {
        /* number of files is given */
        int rc = _sion_gen_info_from_gcomm_ompi(options.multifile.number, gcomm, &filenumber, &lrank, &lsize);
        if (rc != SION_SUCCESS) {
          _sion_errorprint_ompi(
            SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_ompi: error in _sion_gen_info_from_gcomm_ompi");
        }
        nfiles = options.multifile.number;
        DPRINTFP((1, __func__, grank, "Global communicator divided in %d local communicators\n", nfiles));
      }
      thread_sync->filenumber = filenumber;
      thread_sync->numFiles = nfiles;
      thread_sync->lrank_master_mpi = lrank;
      thread_sync->lsize_mpi = lsize;
      thread_sync->lrank_master_ompi = _sion_map_rank_mpi_to_ompi(lrank, num_threads, thread_num);
      thread_sync->lsize_ompi = _sion_get_size_ompi(lsize, num_threads);
    }
#pragma omp barrier

    /* set up parameters of call to generic open (OMPI values) */
    grank = thread_sync->grank_master_ompi + thread_num;
    gsize = thread_sync->gsize_ompi;
    lrank = thread_sync->lrank_master_ompi + thread_num;
    lsize = thread_sync->lsize_ompi;
    filenumber = thread_sync->filenumber;
    nfiles = thread_sync->numFiles;

    if (gen_lcomm) {
      gen_lcomm->rank = lrank;
      gen_lcomm->size = lsize;
    }

  } else if (mode == SION_OPEN_READ) {
    /* file mode READ */
    /* set up parameters of call to generic open (OMPI values) */
    grank = thread_sync->grank_master_ompi + thread_num;
    gsize = thread_sync->gsize_ompi;
    lrank = -1; /* which determined after opening file by sion_generic_paropen */
    lsize = -1; /* " */
    filenumber = -1; /* " */
    nfiles = -1; /* " */

  } else {
    return _sion_errorprint_ompi(SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_ompi: unknown file mode");
  }
  // FIXME
  // DPRINTFP((1, DFUNCTION, grank, "enter parallel open of %d files (current name %s) in %s mode\n", nfiles, fname, file_mode));
  DPRINTFP((1, __func__, grank, "enter parallel open of %d files (current name %s)\n", nfiles, fname));
  DPRINTFP((2, __func__, grank, "enter parallel parameters: grank=%d gsize=%d fnum=%d nfiles=%d lrank=%d lsize=%d\n", grank,
    gsize, filenumber, nfiles, lrank, lsize));
  int sid = sion_generic_paropen(
    _sion_ompi_api_aid, fname, mode, gen_gcomm, grank, gsize, filenumber, nfiles, lrank, lsize, &options.generic_options);
  // DPRINTFP(
  //   (1, __func__, grank, "leave parallel open of %d files in %s mode #tasks=%d sid=%d\n", nfiles, file_mode, lsize, sid));
  DPRINTFP((1, __func__, 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_ompi(
      SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_mpi: invalid return code from internal open %d", sid);
  }
  DPRINTFP((1, __func__, grank, "leave parallel open of file %s sid=%d globalrank=%d\n", fname, sid, grank));
  return sid;
}

/* end of ifdef OMPI */
#endif