sion_omp_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 <omp.h>
#include <stdint.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"
#include "sion_fd.h"
#include "sion_filedesc.h"
#include "sion_generic.h"
#include "sion_omp.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_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;
}

sion_omp_options *sion_omp_options_new()
{
  sion_omp_options *options = malloc(sizeof(sion_omp_options));
  if (options) {
    *options = SION_OMP_OPTIONS_INIT;
  }
  return options;
}

void sion_omp_options_delete(sion_omp_options *options)
{
  free(options);
}

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

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

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

void sion_omp_options_set_collective(sion_omp_options *options)
{
  sion_omp_options_set_collective_size(options, -1);
}

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

void sion_omp_options_set_collective_merge(sion_omp_options *options)
{
  sion_generic_options_set_collective_merge(&options->generic_options);
}

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

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

int sion_paropen_omp(const char *fname, sion_open_mode mode, const sion_omp_options *options_)
{
  sion_omp_options options = (options_) ? *options_ : SION_OMP_OPTIONS_INIT;

  int thread_num = omp_get_thread_num();
  int 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, __func__, thread_num, "enter parallel open of file %s\n", fname));

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

  /* create generic communicator container */
  _omp_api_commdata *gen_gcomm = malloc(sizeof(_omp_api_commdata));
  if (!gen_gcomm) {
    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);

  // FIXME
  // DPRINTFP((1, "sion_paropen_omp", thread_num, "enter parallel open of 1 file (current name %s) in %s mode\n", fname, file_mode));
  DPRINTFP((1, "sion_paropen_omp", thread_num, "enter parallel open of 1 file (current name %s)\n", fname));
  int sid = sion_generic_paropen(
    _sion_omp_api_aid, fname, mode, gen_gcomm, thread_num, num_threads, 0, 1, thread_num, num_threads, &options.generic_options);
  // DPRINTFP((1, "sion_paropen_omp", thread_num, "leave parallel open of 1 file in %s mode #tasks=%d sid=%d\n", file_mode,
  //   num_threads, sid));
  DPRINTFP((1, "sion_paropen_omp", thread_num, "leave parallel open of 1 file #tasks=%d sid=%d\n", num_threads, 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;
}

/* end of ifdef OMP */
#endif