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

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <ctype.h>
#include <unistd.h>
#include <mpi.h>
#include <time.h>
#include <math.h>

#include "sion.h"
#include "partest.h"

int
test_single_mpi(char *filename,
                int numfiles,
                char *localbuffer,
                sion_int64 bufsize,
                sion_int64 totalsize,
                sion_int64 chunksize,
                sion_int32 fsblksize, 
		int verbose, 
		int debug, 
		int Debug, 
		_test_communicators *communicators, 
		int collectiveread, 
		int idletask, 
		int unlink_files) {

  char     *ptr;
  int       i;
  char      fname[FNAMELEN];
  FILE     *fp;

  double    gstarttime, starttime, gopentime, opentime, gopencreatetime, opencreatetime, unlinktime, gwritetime, writetime, gclosetime, closetime, readtime, greadtime, gunlinktime;
  double    barr0time,barr1time, barr2time, barr3time;

  sion_int64 left;
  sion_int64 bsumwrote, sumsize, bsumread;
  char      cbuffer[MAXCHARLEN];
  size_t    bwrite, bwrote, btoread, bread, chunkcnt;
  double    checksum_fp, checksum_read_fp;

  /* not working task ? */
  if (communicators->work_size == -1) {
/*     printf("wf: not working %d\n",communicators->all_rank); */
    return (0);
  }

  sprintf(fname, "%s.%06d", filename, communicators->work_rank);
/*   printf("on rank %3d WRITE: filename %s\n",communicators->all_rank,fname);  */

  /*                                                                      */ DPRINTFTS(communicators->all_rank, "start");

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

  /*                                                                      */ DPRINTFTS(communicators->all_rank, "[W]before open create");

  barrier_after_start(communicators->work);

  /* FIRST OPEN to create files */
  /*                                    */ starttime = MPI_Wtime();
  fp = _sion_create_and_open_file_for_write(fname);
  if (!fp) {
    fprintf(stderr, "cannot open %s for writing, aborting ...\n", fname);
    MPI_Abort(communicators->all, 1);
  }
  /*                                    */ opencreatetime = MPI_Wtime() - starttime;
  /*                                    */ starttime = MPI_Wtime();
  barrier_after_open(communicators->work);
  /*                                    */ barr0time = MPI_Wtime() - starttime;
  /*                                                                      */ DPRINTFTS(communicators->all_rank, "[W]after open create");


  fclose(fp);

  barrier_after_open(communicators->work);

  /* SECOND REOPEN of existing files */
  /*                                                                      */ DPRINTFTS(communicators->all_rank, "[W]before open");
  /*                                    */ starttime = MPI_Wtime();
  fp = _sion_open_existing_file_for_write(fname);
  if (!fp) {
    fprintf(stderr, "cannot open %s for writing, aborting ...\n", fname);
    MPI_Abort(communicators->all, 1);
  }
  /*                                    */ opentime = MPI_Wtime() - starttime;

  /*                                    */ starttime = MPI_Wtime();
  barrier_after_open(communicators->work);
  /*                                    */ barr1time = MPI_Wtime() - starttime;
  /*                                                                      */ DPRINTFTS(communicators->all_rank, "[W]after open");
  /*                                    */ gstarttime = starttime = MPI_Wtime();

  ptr = localbuffer;
  left = totalsize;
  bsumwrote = 0;
  chunkcnt = 0;
  bwrote = 0;

  while (left > 0) {
    bwrite = bufsize;
    if (bwrite > left)
      bwrite = left;

    if (((debug && communicators->all_rank == 0)) || ((Debug && communicators->all_rank == communicators->all_size))) {
      fprintf(stderr, "timings[%03d] write %lld bytes\n", communicators->all_rank, (sion_int64) bwrite);
    }

    bwrote = fwrite(localbuffer, 1, bwrite, fp);
    left -= (sion_int64) bwrote;
    bsumwrote += (sion_int64) bwrote;
    chunkcnt++;

#ifdef CHECKSUM
    checksum_fp=0.0;
    for (i = 0; i < bwrote; i++)
      checksum_fp += (double) localbuffer[i];
#endif

    if (((debug && communicators->all_rank == 0)) || ((Debug && communicators->all_rank == communicators->all_size))) {
      fprintf(stderr, "timings[%03d] wrote (%lld bytes) %lld bytes (%10.4f MB) (%lld left)\n",
              communicators->all_rank, (sion_int64) bwrote, bsumwrote, bsumwrote / 1024.0 / 1024.0, (sion_int64) left);

    }
  }

  /*                                    */ writetime = MPI_Wtime() - starttime;
  /*                                    */ starttime = MPI_Wtime();
  barrier_after_write(communicators->work);
  /*                                    */ barr2time = MPI_Wtime() - starttime;
  /*                                    */ gwritetime = MPI_Wtime() - gstarttime;
  /*                                    */ starttime = MPI_Wtime();
  fclose(fp);
  /*                                    */ closetime = MPI_Wtime() - starttime;
  /*                                                                      */ DPRINTFTS(communicators->all_rank, "[W]before close");
  /*                                    */ starttime = MPI_Wtime();
  barrier_after_close(communicators->work);
  /*                                    */ barr3time = MPI_Wtime() - starttime;
  /*                                                                      */ DPRINTFTS(communicators->all_rank, "[W]after close");

  if (verbose) {
    sprintf(cbuffer,
            "timings[%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 create=%10.6fs\n",
            communicators->all_rank, opentime, writetime, closetime, barr1time, barr2time, barr3time, chunkcnt,
            totalsize / 1024.0 / 1024.0 / writetime, communicators->ionode_number, opencreatetime);
    collective_print_gather(cbuffer, communicators->work);
  }

  /*                                                                      */ DPRINTFTS(communicators->all_rank, "before red.");
  MPI_Reduce(&bsumwrote, &sumsize, 1, SION_MPI_INT64, MPI_SUM, 0, communicators->work);
  MPI_Reduce(&opentime, &gopentime, 1, MPI_DOUBLE, MPI_MAX, 0, communicators->work);
  MPI_Reduce(&opencreatetime, &gopencreatetime, 1, MPI_DOUBLE, MPI_MAX, 0, communicators->work);
  MPI_Reduce(&closetime, &gclosetime, 1, MPI_DOUBLE, MPI_MAX, 0, communicators->work);

  /*                                                                      */ DPRINTFTS(communicators->all_rank, "after red.");
  if (communicators->work_rank == 0) {
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
    fprintf(stderr, "TOTAL result: open=%10.6fs close=%10.6fs wrote %10.4f MB write+barrier=%10.6fs bw=%10.4f MB/s to %d files, create=%10.6fs\n",
            gopentime, gclosetime, 1.0 * sumsize / 1024.0 / 1024.0, gwritetime, 1.0 * sumsize / 1024.0 / 1024.0 / gwritetime, communicators->work_size,gopencreatetime);
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
  }

  if (communicators->work_rank == 0)
    fprintf(stderr, "*********************************************************************************************\n");

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

  /*                                    */ DPRINTFTS(communicators->all_rank, "[R]before open");
  /*                                    */ starttime = MPI_Wtime();

  sprintf(fname, "%s.%06d", filename, communicators->workread_rank);
/*   printf("on rank %3d READ: filename %s\n",communicators->all_rank,fname);  */

  fp = _sion_open_file_for_read(fname);
  if (!fp) {
    fprintf(stderr, "cannot open %s for writing, aborting ...\n", fname);
    MPI_Abort(communicators->all, 1);
  }
  /*                                    */ opentime = MPI_Wtime() - starttime;

  /*                                    */ starttime = MPI_Wtime();
  barrier_after_open(communicators->work);
  /*                                    */ barr1time = MPI_Wtime() - starttime;
  /*                                                                      */ DPRINTFTS(communicators->all_rank, "[R]after open");
  /*                                    */ gstarttime = starttime = MPI_Wtime();


  ptr = localbuffer;
  left = totalsize;
  bsumread = 0;
  chunkcnt = 0;

  while (left > 0) {
    btoread = bufsize;
    if (btoread > left)
      btoread = left;

    bread = fread(localbuffer, 1, btoread, fp);

    left -= bread;
    bsumread += bread;
    chunkcnt++;

#ifdef CHECKSUM
    checksum_read_fp=0.0;
    for (i = 0; i < bread; i++)
      checksum_read_fp += (double) localbuffer[i];
#endif

    if (((debug && communicators->all_rank == 0)) || ((Debug && communicators->all_rank == communicators->all_size))) {
      fprintf(stderr, "timings[%03d] read (%lld bytes) %lld bytes (%10.4f MB) (%lld left)\n", communicators->all_rank, (sion_int64) bread, bsumread,
              bsumread / 1024.0 / 1024.0, (sion_int64) left);
    }
  }
  /*                                    */ readtime = MPI_Wtime() - starttime;
  /*                                    */ starttime = MPI_Wtime();
  barrier_after_read(communicators->work);
  /*                                    */ barr2time = MPI_Wtime() - starttime;
  /*                                    */ greadtime = MPI_Wtime() - gstarttime;

  /*                                    */ starttime = MPI_Wtime();
  fclose(fp);
  /*                                    */ closetime = MPI_Wtime() - starttime;
  barrier_after_close(communicators->work);

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

    collective_print_gather(cbuffer, communicators->work);

  }

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

  /*                                                                      */ DPRINTFTS(communicators->all_rank, "before red.");
  MPI_Reduce(&bsumread, &sumsize, 1, SION_MPI_INT64, MPI_SUM, 0, communicators->work);
  MPI_Reduce(&opentime, &gopentime, 1, MPI_DOUBLE, MPI_MAX, 0, communicators->work);
  MPI_Reduce(&closetime, &gclosetime, 1, MPI_DOUBLE, MPI_MAX, 0, communicators->work);
  /*                                                                      */ DPRINTFTS(communicators->all_rank, "after red.");
  if (communicators->work_rank == 0) {
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
    fprintf(stderr, "TOTAL result: open=%10.6fs close=%10.6fs read %10.4f MB read+barrier=%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, communicators->work_size);
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
  }


  if(unlink_files) {
    /*                                    */ starttime = MPI_Wtime();
    barrier_before_unlink(communicators->work);
    unlink(fname);
    barrier_after_unlink(communicators->work);
    /*                                    */ unlinktime = MPI_Wtime() - starttime;
    MPI_Reduce(&unlinktime, &gunlinktime, 1, MPI_DOUBLE, MPI_MAX, 0, communicators->work);
    if (communicators->work_rank == 0) {
      fprintf(stderr, "partest result:  ultime=%10.6fs unlink %s\n", gunlinktime, fname);
    }
  }

  return (1);
}
