sion_ompi_gen.c

/****************************************************************************
**  SIONLIB     http://www.fz-juelich.de/jsc/sionlib                       **
*****************************************************************************
**  Copyright (c) 2008-2018                                                **
**  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_error_handler.h"
#include "sion_flags.h"
#include "sion_internal.h"

#ifdef SION_OMPI

#include <omp.h>

#include "sion_generic.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;
}

#define DFUNCTION "sion_paropen_ompi"
int sion_paropen_ompi(const char *fname, const char *file_mode, int *numFiles, MPI_Comm gComm, const MPI_Comm *lComm,
  int64_t *chunksize, int32_t *fsblksize, int *globalrank, FILE **fileptr, char **newfname)
{
  /* gRank and lRank refer to the MPI process rank in the global and local communicator respectively */
  int rc, sid = -1;
  int filenumber, gRank, lRank, lSize, gSize;

  _sion_flags_store *flags_store = NULL;

  _ompi_api_commdata *gen_gcomm;
  _ompi_api_commdata *gen_lcomm = NULL;

  int num_threads, thread_num;
  __ompi_thread_sync *thread_sync;

  thread_num = omp_get_thread_num();

#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);
    num_threads = omp_get_num_threads();

    thread_sync = malloc(sizeof(__ompi_thread_sync));
    if (thread_sync == NULL)
      (_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;
    thread_sync->numFiles = *numFiles;
    __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 *)__ompi_thread_sync_struct;

  DPRINTFP((1, "sion_paropen_ompi", thread_sync->grank_master_ompi + thread_num, "thread %d enters parallel open of file %s\n",
    thread_num, fname));

  /* check parameters */
  if (lComm == NULL) {
    return _sion_errorprint_ompi(SION_ID_NOT_VALID, _SION_ERROR_ABORT, "sion_paropen_ompi: No lComm variable given");
  }
  if (numFiles == NULL) {
    return _sion_errorprint_ompi(SION_ID_NOT_VALID, _SION_ERROR_ABORT, "sion_paropen_ompi: No numFiles variable given");
  }
  flags_store = _sion_parse_flags(file_mode);
  if (!flags_store) {
    return _sion_errorprint_ompi(
      SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_mpi: could not parse file mode in %s, aborting ...\n", file_mode);
  }

/* 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 */
  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 {
    _sion_flags_destroy_store(&flags_store);
    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
  }

  if (flags_store->mask & _SION_FMODE_WRITE) {
    /* file mode WRITE */

    /* create generic local communicator container on each thread */
    if (*numFiles <= 0) {
      gen_lcomm = malloc(sizeof(_ompi_api_commdata));
      if (gen_lcomm != NULL) {
        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 */
      } else {
        _sion_flags_destroy_store(&flags_store);
        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));
      }
    }

#pragma omp master
    {
      if (*numFiles <= 0) {
        /* lComm contains local communicator */

        rc = _sion_get_info_from_splitted_comm_ompi(gComm, *lComm, numFiles, &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");
        }
        DPRINTFP((1, DFUNCTION, gRank, "%d local communicators found\n", *numFiles));

        gen_lcomm->comm = *lComm;

      } else {
        /* number of files is given */
        rc = _sion_gen_info_from_gcomm_ompi(*numFiles, 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");
        }
        DPRINTFP((1, "sion_paropen_ompi", gRank, "Global communicator divided in %d local communicators\n", *numFiles));
      }

      thread_sync->filenumber = filenumber;
      thread_sync->numFiles = *numFiles;
      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);

    } /* OMP MASTER END */

    {
#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;
    *numFiles = thread_sync->numFiles;

    if (gen_lcomm != NULL) {
      gen_lcomm->rank = thread_sync->lrank_master_ompi + thread_num;
      gen_lcomm->size = thread_sync->lsize_ompi;
    }

  } else if (flags_store->mask & _SION_FMODE_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; /* " */
    *numFiles = -1; /* " */

  } else {
    _sion_flags_destroy_store(&flags_store);
    return _sion_errorprint_ompi(SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_ompi: unknown file mode");
  }
  _sion_flags_destroy_store(&flags_store);

  DPRINTFP((1, DFUNCTION, gRank, "enter parallel open of %d files (current name %s) in %s mode\n", *numFiles, fname, file_mode));
  DPRINTFP(
    (2, DFUNCTION, gRank, "enter parallel parameters: grank=%d gsize=%d fnum=%d numfiles=%d lrank=%d lsize=%d chunksize=%d\n",
      gRank, gSize, filenumber, *numFiles, lRank, lSize, (int)*chunksize));
  sid = sion_generic_paropen(_sion_ompi_api_aid, fname, file_mode, chunksize, fsblksize, gen_gcomm, gRank, gSize, &filenumber,
    numFiles, &lRank, &lSize, fileptr, newfname);
  DPRINTFP(
    (1, DFUNCTION, gRank, "leave parallel open of %d files in %s mode #tasks=%d sid=%d\n", *numFiles, file_mode, 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", rc);
  }

  /* return parameter */
  *globalrank = gRank;

  DPRINTFP((1, "sion_paropen_ompi", gRank, "leave parallel open of file %s sid=%d globalrank=%d\n", fname, sid, *globalrank));

  return sid;
}
#undef DFUNCTION

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;
}

/* end of ifdef OMPI */
#endif