sion_omp_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 <omp.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_fd.h"
#include "sion_filedesc.h"
#include "sion_flags.h"
#include "sion_generic.h"
#include "sion_internal.h"
#include "sion_omp_cb_gen.h"
#include "sion_omp_internal_gen.h"

#ifdef SION_OMP

int32_t _sion_omp_api_aid = -1;
static omp_lock_t _sion_omp_lock_data;

int _sion_omp_user_lock(void *data)
{
  int rc = SION_SUCCESS;
  omp_set_lock(&_sion_omp_lock_data);
  return rc;
}
int _sion_omp_user_unlock(void *data)
{
  int rc = SION_SUCCESS;
  omp_unset_lock(&_sion_omp_lock_data);
  return rc;
}

int sion_paropen_omp(const char *fname, const char *file_mode, int64_t *chunksize, int32_t *fsblksize, int *globalrank,
  FILE **fileptr, char **newfname)
{
  int sid = SION_ID_UNDEF;
  int filenumber, num_threads, thread_num;
  int numFiles = 1, gtasks, gRank, lRank, lSize;
  _omp_api_commdata *gen_gcomm;
  _sion_flags_store *flags_store = NULL;

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

#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_omp_lock_data);
    sion_lock_register_lock_callbacks(_sion_omp_user_lock, _sion_omp_user_unlock, &_sion_omp_lock_data);
  }
  {
#pragma omp barrier
  }

  DPRINTFP((1, "sion_paropen_omp", thread_num, "enter parallel open of file %s\n", fname));

  flags_store = _sion_parse_flags(file_mode);
  if (!flags_store) {
    return _sion_errorprint_omp(
      SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_omp: could not parse file mode in %s, aborting ...\n", file_mode);
  }

  if (flags_store->mask & _SION_FMODE_WRITE) {
    /* file mode WRITE */
    *globalrank = thread_num;
  }
  _sion_flags_destroy_store(&flags_store);

#pragma omp master
  {
    /* register callbacks for generic interface */
    if (_sion_omp_api_aid < 0) {
      _sion_omp_api_aid = _sion_register_callbacks_omp();
    }
  }

  /* create generic communicator container */
  gen_gcomm = malloc(sizeof(_omp_api_commdata));
  if (gen_gcomm == NULL) {
    return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "cannot allocate omp internal data structure of size %lu (_omp_api_commdata), aborting ...\n",
      (unsigned long)sizeof(_omp_api_commdata));
  }
  gen_gcomm->commset = 1;
  gen_gcomm->thread_num = thread_num;
  gen_gcomm->num_threads = num_threads;

  /* sync to ensure that aid is accessible */
  _sion_omp_barrier_cb(gen_gcomm);

  lRank = gRank = thread_num;
  lSize = gtasks = num_threads;
  filenumber = 0;
  numFiles = 1;

  DPRINTFP(
    (1, "sion_paropen_omp", gRank, "enter parallel open of %d files (current name %s) in %s mode\n", numFiles, fname, file_mode));
  sid = sion_generic_paropen(_sion_omp_api_aid, fname, file_mode, chunksize, fsblksize, gen_gcomm, gRank, gtasks, &filenumber,
    &numFiles, &lRank, &lSize, fileptr, newfname);
  DPRINTFP((1, "sion_paropen_omp", 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_omp(
      SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_omp: invalid return code from internal open %d", sid);
  }

  DPRINTFP((1, "sion_paropen_omp", thread_num, "leave parallel open of file %s sid=%d\n", fname, sid));

  return sid;
}

int sion_parclose_omp(int sid)
{
  int rc = SION_SUCCESS;
  ONLY_DEBUG(int thread_num;)
  _sion_filedesc *sion_filedesc;

  if ((sid < 0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR) || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    return _sion_errorprint_omp(SION_NOT_SUCCESS, _SION_ERROR_RETURN, "sion_parclose_omp: invalid sion_filedesc %d", sid);
  }

  ONLY_DEBUG(thread_num = omp_get_thread_num();)

  if (omp_get_num_threads() != sion_filedesc->ntasks) {
    return _sion_errorprint_omp(SION_NOT_SUCCESS, _SION_ERROR_RETURN,
      "sion_parclose_omp: invalid number of OpenMP threads, %d <> %d", omp_get_num_threads(), sion_filedesc->ntasks);
  }

  DPRINTFP((1, "sion_parclose_omp", thread_num, "enter parallel close of sid %d\n", sid));
  DPRINTFP((1, "sion_parclose_omp", thread_num, "closing %d file(s)\n", sion_filedesc->nfiles));

  rc = sion_generic_parclose(sid);

  DPRINTFP((1, "sion_parclose_omp", thread_num, "leave parallel close of sid %d rc=%d\n", sid, rc));
  return rc;
}

/* end of ifdef OMP */
#endif