/****************************************************************************
**  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 <ctype.h>
#include <inttypes.h>
#include <mpi.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <math.h>
#include <time.h>

#include "ompi_partest.h"
#include "sion.h"
#include "sion_debug.h"
#include "sion_printts.h"

#include <omp.h>

#define _PARTEST_SION_INT32 10
#define _PARTEST_SION_INT64 11
#define _PARTEST_DOUBLE 12

/****************************************************************************************************************
 *
 *                      test_paropen_multi_omp
 *
 ***************************************************************************************************************/

/*!\brief test
 *
 * @param[in] *filename					filename of the test file
 * @param[in] *localbuffer 				buffer for written data
 * @param[in] *communicators			struct with all the MPI related stuff
 * @param[in] *options					struct with all the partest specific options
 *
 */

int test_paropen_omp(char *filename, char *localbuffer, _test_communicators *communicators, _test_options *options)
{
  double starttime, gopentime, opentime, unlinktime, gwritetime, writetime, gclosetime, closetime, readtime, greadtime;
  double barr1time, barr2time, barr3time;

  int64_t left;
  int64_t bsumwrote, sumsize, bsumread;
  double checksum_fp, checksum_read_fp;
  int globalrank, sid, i, lstartoffset;
  size_t bwrite, bwrote, btoread, bread;
  int chunkcnt;
  int64_t rchunksize;
  int32_t rfsblksize;
  char cbuffer[2 * MAXCHARLEN];
  char *newfname;
  size_t bytes_in_chunk;
  int myfeof;

  int thread_num = omp_get_thread_num();

  /****************************** WRITE *****************************/

  starttime = MPI_Wtime();
  sid = lsion_paropen_omp(
    filename, (options->use_posix) ? "bw,posix" : "bw", &options->chunksize, &options->fsblksize, &globalrank, NULL, &newfname);
  opentime = MPI_Wtime() - starttime;
  starttime = MPI_Wtime();
  barrier_after_open(communicators->work);
  barr1time = MPI_Wtime() - starttime;

  checksum_fp = 0;

  left = options->totalsize;
  bsumwrote = 0;
  chunkcnt = 0;
  bytes_in_chunk = 0;
  lstartoffset = options->startoffset;
  /* Write until we reach the total size of the data */
  while (left > 0) {
    if (lstartoffset == 0) {
      bwrite = options->bufsize;
    } else {
      bwrite = lstartoffset;
      lstartoffset = 0;
    }
    if (bwrite > left) {
      bwrite = left;
    }

    if (((options->debug && thread_num == 0)) || ((options->Debug && (thread_num == omp_get_num_threads() - 1)))) {
      fprintf(stderr, "timings[t%03d] write %" PRId64 " bytes\n", thread_num, (int64_t)bwrite);
    }

    bytes_in_chunk += bwrite;
    if (bytes_in_chunk > options->chunksize) {
      lsion_ensure_free_space(sid, bwrite);
      bytes_in_chunk = bwrite;
    }

    bwrote = lsion_fwrite(localbuffer, 1, bwrite, sid);

#ifdef CHECKSUM
    if (!options->suppress_checksum) {
      checksum_fp = 0.0;
      for (i = 0; i < bwrote; i++) {
        checksum_fp += (double)localbuffer[i];
      }
    }
#endif

    left -= bwrote;
    bsumwrote += bwrote;
    chunkcnt++;

    if (((options->debug && thread_num == 0)) || ((options->Debug && (thread_num == omp_get_num_threads() - 1)))) {
      fprintf(stderr, "timings[t%03d] wrote (%" PRId64 " bytes) %" PRId64 " bytes (%10.4f MB) (%" PRId64 " left)\n", thread_num,
        (int64_t)bwrote, bsumwrote, bsumwrote / 1024.0 / 1024.0, (int64_t)left);
      fprintf(stderr, "timings[t%03d] after write   position in file= %" PRId64 " \n", thread_num, lsion_get_position(sid));
    }
  }
  lsion_flush(sid);
  writetime = MPI_Wtime() - starttime;

  starttime = MPI_Wtime();
  barrier_after_write(communicators->work);
  barr2time = MPI_Wtime() - starttime;

  starttime = MPI_Wtime();
  lsion_parclose_omp(sid);
  closetime = MPI_Wtime() - starttime;

  starttime = MPI_Wtime();
  barrier_after_close(communicators->work);
  barr3time = MPI_Wtime() - starttime;

  if (writetime == 0) {
    writetime = -1;
  }

  if (options->verbose) {
    sprintf(cbuffer,
      "timings[t%03d] open=%10.6fs write=%10.6fs close=%10.6fs barrier(open=%10.6fs, write=%10.6fs, close=%10.6fs) #chunks=%d "
      "bw=%10.4f MB/s ionode=%d\n",
      thread_num, opentime, writetime, closetime, barr1time, barr2time, barr3time, chunkcnt,
      options->totalsize / 1024.0 / 1024.0 / writetime, communicators->ionode_number);
    collective_print_gather(cbuffer, communicators->work);
  }

  reduce_omp(&bsumwrote, &sumsize, MPI_SUM, _PARTEST_SION_INT64);
  reduce_omp(&opentime, &gopentime, MPI_MAX, _PARTEST_DOUBLE);
  reduce_omp(&closetime, &gclosetime, MPI_MAX, _PARTEST_DOUBLE);
  reduce_omp(&writetime, &gwritetime, MPI_MAX, _PARTEST_DOUBLE);

#pragma omp master
  {
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
    fprintf(stderr, "TOTAL result: open=%10.6fs close=%10.6fs wrote %10.4f MB write=%10.6fs bw=%10.4f MB/s to %d files\n",
      gopentime, gclosetime, 1.0 * sumsize / 1024.0 / 1024.0, gwritetime, 1.0 * sumsize / 1024.0 / 1024.0 / gwritetime,
      options->numfiles);
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
    fprintf(stderr, "*********************************************************************************************\n");
  }
#pragma omp barrier

  /****************************** READ *****************************/

  /* reset localbuffer */
  for (i = 0; i < ((options->totalsize < options->bufsize) ? options->totalsize : options->bufsize); i++) {
    localbuffer[i] = ' ';
  }

  starttime = MPI_Wtime();
  if (options->collectiveopenforread) {
    sid = lsion_paropen_omp(
      filename, (options->use_posix) ? "br,posix" : "br", &options->chunksize, &options->fsblksize, &globalrank, NULL, &newfname);

  } else {
    /* there is some work to for multifile sion file */
    sid = lsion_open_rank(
      filename, (options->use_posix) ? "br,posix" : "br", &rchunksize, &rfsblksize, &communicators->workread_rank, NULL);
  }
  opentime = MPI_Wtime() - starttime;

  starttime = MPI_Wtime();
  barrier_after_open(communicators->workread);
  barr1time = MPI_Wtime() - starttime;

  checksum_read_fp = 0;
  left = options->totalsize;
  bsumread = 0;
  chunkcnt = 0;
  bytes_in_chunk = 0;
  lstartoffset = options->startoffset;

  myfeof = lsion_feof(sid);
  while ((left > 0) && (!myfeof)) {
    if (lstartoffset == 0) {
      btoread = options->bufsize;
    } else {
      btoread = lstartoffset;
      lstartoffset = 0;
    }
    if (btoread > left) {
      btoread = left;
    }

    bytes_in_chunk += btoread;
    if (bytes_in_chunk > options->chunksize) {
      myfeof = lsion_feof(sid);
    }
    if (!myfeof) {
      bread = lsion_fread(localbuffer, 1, btoread, sid);

#ifdef CHECKSUM
      if (!options->suppress_checksum) {
        checksum_read_fp = 0.0;
        for (i = 0; i < bread; i++) {
          checksum_read_fp += (double)localbuffer[i];
        }
      }
#endif

      left -= bread;
      bsumread += bread;
      chunkcnt++;
#pragma omp master
      {
        if (((options->debug && thread_num == 0)) || ((options->Debug && (thread_num == omp_get_num_threads() - 1)))) {
          fprintf(stderr, "timings[t%03d] read (%" PRId64 " bytes) %" PRId64 " bytes (%10.4f MB) (%" PRId64 " left)\n",
            thread_num, (int64_t)bread, bsumread, bsumread / 1024.0 / 1024.0, (int64_t)left);
          fprintf(stderr, "timings[t%03d] after read   position in file= %" PRId64 " restinblock=%" PRId64 "\n", thread_num,
            lsion_get_position(sid), lsion_bytes_avail_in_block(sid));
        }
      }
#pragma omp barrier
    }
  }
  lsion_flush(sid);

  readtime = MPI_Wtime() - starttime;

  starttime = MPI_Wtime();
  barrier_after_read(communicators->workread);
  barr2time = MPI_Wtime() - starttime;

  starttime = MPI_Wtime();
  if (options->collectiveopenforread) {
    lsion_parclose_omp(sid);
  } else {
    lsion_close(sid);
  }
  closetime = MPI_Wtime() - starttime;

  barrier_after_close(communicators->workread);

  if (readtime == 0) {
    readtime = -1;
  }
  if (options->verbose) {
    sprintf(cbuffer,
      "timings[t%03d] open=%10.6fs read=%10.6fs close=%10.6fs barrier(open=%10.6fs, read=%10.6fs, close=%10.6fs) #chunks=%d "
      "br=%10.4f MB/s ionode=%d (check %d)\n",
      thread_num, opentime, readtime, closetime, barr1time, barr2time, barr3time, chunkcnt,
      options->totalsize / 1024.0 / 1024.0 / readtime, communicators->ionode_number,
      (fabs(checksum_fp - checksum_read_fp) < 1e-5));
    collective_print_gather(cbuffer, communicators->workread);
  }

#ifdef CHECKSUM
  if (!options->suppress_checksum) {
    if (fabs(checksum_fp - checksum_read_fp) > 1e-5) {
      fprintf(stderr, "timings[t%03d] ERROR in double checksum  %14.10f==%14.10f, diff=%14.10f\n", thread_num, checksum_fp,
        checksum_read_fp, checksum_fp - checksum_read_fp);
    }
  }
#endif

  reduce_omp(&bsumread, &sumsize, MPI_SUM, _PARTEST_SION_INT64);
  reduce_omp(&opentime, &gopentime, MPI_MAX, _PARTEST_DOUBLE);
  reduce_omp(&closetime, &gclosetime, MPI_MAX, _PARTEST_DOUBLE);
  reduce_omp(&readtime, &greadtime, MPI_MAX, _PARTEST_DOUBLE);

#pragma omp master
  {
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
    fprintf(stderr, "TOTAL result: open=%10.6fs close=%10.6fs read %10.4f MB read=%10.6fs br=%10.4f MB/s from %d files\n",
      gopentime, gclosetime, 1.0 * sumsize / 1024.0 / 1024.0, greadtime, 1.0 * sumsize / 1024.0 / 1024.0 / greadtime,
      options->numfiles);
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
  }
#pragma omp barrier

  if (options->unlink_files) {
    starttime = MPI_Wtime();
    barrier_before_unlink(communicators->workread);
#pragma omp master
    {
      fprintf(stderr, "partest result: unlink file %s ...\n", newfname);
      unlink(newfname);
    }
#pragma omp barrier

    barrier_after_unlink(communicators->workread);
    unlinktime = MPI_Wtime() - starttime;
#pragma omp master
    {
      fprintf(stderr, "partest result:  ultime=%10.6fs unlink %s\n", unlinktime, newfname);
    }
#pragma omp barrier
  }

  return 1;
}
