/****************************************************************************
**  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 <mpi.h>
#include <time.h>
#include <math.h>

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


#ifdef _SION_BGP
/* #include <mpix.h> */
#include <common/bgp_personality.h>
#include <common/bgp_personality_inlines.h>
#endif

#ifdef _SION_XT
#include <unistd.h>
#endif

#ifdef _SION_AIX
#include <unistd.h>
#endif

#ifdef _SION_LINUX
#include <unistd.h>
#endif

/*!
 * \file partest.c
 * \brief Benchmark/Test Tool
 * \author Wolfgang Frings
 * \author Ventsislav Petkov
 */



void usage(char *name)
{
  fprintf(stderr, "Usage: %s options\n\nwith the following optional options (default values in parathesis):\n\n", name);

  fprintf(stderr, " Sion File Settings:\n");
  fprintf(stderr, "  [-f filename]          filename of direct access file \n");
  fprintf(stderr, "  [-n <number of files>] number of files file \n");
  fprintf(stderr, "  [-r <chunksize>]       sion chunk size in bytes \n");
  fprintf(stderr, "  [-R <chunksize>]       sion chunk size in MBytes \n");
  fprintf(stderr, "  [-q <fsblksize>]       size of filesystem blocks in bytes \n");
  fprintf(stderr, "  [-Q <fsblksize>]       size of filesystem blocks in MBytes \n");

  fprintf(stderr, " Test Configuration:\n");
  fprintf(stderr, "  [-F <factor>]          factor for random size (0.0 to 1.0, def: 0.0) \n");
  fprintf(stderr, "  [-b <bufsize>]         size of blocks written in ONE fwrite in bytes \n");
  fprintf(stderr, "  [-B <bufsize>]         size of blocks written in ONE fwrite in MBytes \n");
  fprintf(stderr, "  [-g <globaltotalsize>] global total size of data written in bytes \n");
  fprintf(stderr, "  [-G <globaltotalsize>] global total size of data written in GBytes \n");
  fprintf(stderr, "  [-s <totalsize>]       total size of data written by each processor in bytes \n");
  fprintf(stderr, "  [-S <totalsize>]       total size of data written by each processor in MBytes \n");
  fprintf(stderr, "  [-i <task#>]           Task that will be idle: it will not write any data!! \n");
  fprintf(stderr, "  [-T <type>]            type of test (0): w/o collective read; (2): MPI IO\n");
  fprintf(stderr, "  [-v]                   verbose print info for each task\n");
  fprintf(stderr, "  [-d]                   debug task 0\n");
  fprintf(stderr, "  [-D]                   debug task n\n");
  fprintf(stderr, "  [-Z <offset>]          shift tasks numbering for reading by offset to ommit data caching of file-system (0) \n");
  fprintf(stderr, "  [-o <bytes>]           start offset, write <bytes> first before using blksize (0) \n");

  fprintf(stderr, "  Blue Gene/L, Blue Gene/P:\n");
  fprintf(stderr, "  [-P]                   order tasks by BG I/O-node\n");
  fprintf(stderr, "  [-p <numtasks>]        number of tasks per BG I/O-node\n");

  fprintf(stderr, "  MPI-IO, GPFS options:\n");
  fprintf(stderr, "  [-w <1|0>]             Hint MPI-IO, IBM, Large Block IO\n");
  fprintf(stderr, "  [-W <size>]            Hint MPI-IO, IBM, IO bufsize in KB\n");
  fprintf(stderr, "  [-x <1|0>]             Hint MPI-IO, IBM, sparse access\n");
  exit(1);
}


int barrier_after_start(MPI_Comm comm)
{
  MPI_Barrier(comm);
  return (1);
}

int barrier_after_malloc(MPI_Comm comm)
{
  MPI_Barrier(comm);
  return (1);
}

int barrier_after_open(MPI_Comm comm)
{
  MPI_Barrier(comm);
  return (1);
}

int barrier_after_write(MPI_Comm comm)
{
  MPI_Barrier(comm);
  return (1);
}

int barrier_after_read(MPI_Comm comm)
{
  MPI_Barrier(comm);
  return (1);
}

int barrier_after_close(MPI_Comm comm)
{
  MPI_Barrier(comm);
  return (1);
}

int barrier_before_unlink(MPI_Comm comm)
{
  MPI_Barrier(comm);
  return (1);
}

int barrier_after_unlink(MPI_Comm comm)
{
  MPI_Barrier(comm);
  return (1);
}

int collective_print_gather(char *cbuffer, MPI_Comm comm)
{
  int       rank, size, p;
  char     *lbuffer;

  lbuffer = (char *) malloc(MAXCHARLEN * MAXPE);

  MPI_Comm_size(comm, &size);
  MPI_Comm_rank(comm, &rank);

  MPI_Gather(cbuffer, MAXCHARLEN, MPI_CHAR, lbuffer, MAXCHARLEN, MPI_CHAR, 0, comm);

  if (rank == 0) {

    for (p = 0; p < size; p++) {
      /*       fprintf(stderr,"%06d:%s\n",p,lbuffer+p*MAXCHARLEN);  */
      fprintf(stderr, lbuffer + p * MAXCHARLEN);
    }
  }

  return (1);
}

int collective_print(char *cbuffer, MPI_Comm comm)
{
  int       rank, size, p;
  int       dummy = 0;
  char      lbuffer[MAXCHARLEN];
  MPI_Status status;

  MPI_Comm_size(comm, &size);
  MPI_Comm_rank(comm, &rank);
  if (rank == 0) {
/*     fprintf(stderr,"%06d:%s",0,cbuffer); */
    fprintf(stderr, cbuffer);

    for (p = 1; p < size; p++) {
      if (p > 0) {
        MPI_Send(&dummy, 1, MPI_INT, p, WAKEUP, comm);
        MPI_Recv(lbuffer, MAXCHARLEN, MPI_CHAR, p, COLPRINT, comm, &status);
/*      fprintf(stderr,"%06d:%s",p,lbuffer); */
        if (strlen(lbuffer) > 0)
          fprintf(stderr, lbuffer);
      }
    }

  }
  else {

    MPI_Recv(&dummy, 1, MPI_INT, 0, WAKEUP, comm, &status);
    MPI_Send(cbuffer, MAXCHARLEN, MPI_CHAR, 0, COLPRINT, comm);

  }
  return (1);
}

int       split_communicator(_test_communicators * communicators, int bluegene, int bluegene_np, int numfiles, int read_task_offset, int verbose);




int main(int argc, char **argv)
{
  int       rank, size, fsblksize, i;
  sion_int64 ltotalsize, totalsize = 0, chunksize, bufsize, globalsize = 0;
  char     *localbuffer, filename[FNAMELEN];
  time_t    t;
  int       verbose = 0;
  int       debug = 0;
  int       Debug = 0;
  double    factor = 0.0;
  int       type = 0;
  int       bluegene = 0;
  int       bluegene_np = 0;
  int       collectiveread = 1;
  int       idletask = -99;
  char      sIdletask[6];
  int       numfiles = 1;
  sion_int64 commwork_size64 = 1;
  int       mpiio_lb = -1, mpiio_bs = -1, mpiio_sa = -1;
  int       read_task_offset = 0;
  int       startoffset=0;
  int       unlink_files=0; 

  /* communicators */
  _test_communicators communicators;

  MPI_Init(&argc, &argv);
  MPI_Comm_size(MPI_COMM_WORLD, &size);
  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
  /*                                                                      */ DPRINTFTS(rank, "after MPI_Init");

  /*   srand ( time(NULL)*rank ); */

  /*   printf("starting partest %02d of %02d\n", rank,size); */

  bufsize = 10 MB;
  totalsize = 20 MB;

  chunksize = bufsize;
  fsblksize = 2.0 MB;           /* IBM GPFS */
  strcpy(filename, "partest_parfile.dat");

  if (rank == 0) {
    /* parse command line */
    i = 1;
    while (i < argc) {
      if (argv[i][0] == '-') {
        switch (argv[i][1]) {
        case 'f':
          strcpy(filename, argv[++i]);
          break;
        case 'F':
          factor = atof(argv[++i]);
          break;
        case 'n':
          numfiles = atoi(argv[++i]);
          break;
        case 'b':
          bufsize = (sion_int64) atoi(argv[++i]);
          break;
        case 'B':
          bufsize = (sion_int64) atoi(argv[++i]) MB;
          break;
        case 'g':
          globalsize = (sion_int64) atoi(argv[++i]);
          break;
        case 'G':
          globalsize = (sion_int64) atoi(argv[++i]) * 1024 MB;
          break;
        case 's':
          totalsize = (sion_int64) atoi(argv[++i]);
          break;
        case 'S':
          totalsize = (sion_int64) atoi(argv[++i]) MB;
          break;
        case 'r':
          chunksize = (sion_int64) atoi(argv[++i]);
          break;
        case 'R':
          chunksize = (sion_int64) atoi(argv[++i]) MB;
          break;
        case 'q':
          fsblksize = atoi(argv[++i]);
          break;
        case 'Q':
          fsblksize = atof(argv[++i]) MB;
          break;
        case 'o':
          startoffset = (sion_int64) atoi(argv[++i]);
          break;
        case 'P':
          bluegene = 1;
          break;
        case 'p':
          bluegene_np = atoi(argv[++i]);
          break;
        case 'T':
          type = atoi(argv[++i]);
          break;
        case 'w':
          mpiio_lb = atoi(argv[++i]);
          break;
        case 'W':
          mpiio_bs = atoi(argv[++i]);
          break;
        case 'x':
          mpiio_sa = atoi(argv[++i]);
          break;
        case 'Z':
          read_task_offset = atoi(argv[++i]);
          break;
        case 'X':
          unlink_files = atoi(argv[++i]);
          break;
        case 'd':
          debug++;
          break;
        case 'D':
          Debug++;
          break;
        case 'v':
          verbose++;
          break;
        default:
          printf("Arg default: %s\n", argv[i]);
          usage(argv[0]);
        }
      }
      else {
        printf("Arg error: %s\n", argv[i]);
        usage(argv[0]);
      }
      i++;
    }
  }

  MPI_Bcast(filename, FNAMELEN, MPI_CHAR, 0, MPI_COMM_WORLD);
  MPI_Bcast(&factor, 1, MPI_DOUBLE, 0, MPI_COMM_WORLD);

  MPI_Bcast(&bufsize, 1, SION_MPI_INT64, 0, MPI_COMM_WORLD);
  MPI_Bcast(&totalsize, 1, SION_MPI_INT64, 0, MPI_COMM_WORLD);
  MPI_Bcast(&globalsize, 1, SION_MPI_INT64, 0, MPI_COMM_WORLD);
  MPI_Bcast(&chunksize, 1, SION_MPI_INT64, 0, MPI_COMM_WORLD);
  MPI_Bcast(&fsblksize, 1, SION_MPI_INT32, 0, MPI_COMM_WORLD);

  MPI_Bcast(&type, 1, MPI_INT, 0, MPI_COMM_WORLD);
  MPI_Bcast(&verbose, 1, MPI_INT, 0, MPI_COMM_WORLD);
  MPI_Bcast(&bluegene, 1, MPI_INT, 0, MPI_COMM_WORLD);
  MPI_Bcast(&bluegene_np, 1, MPI_INT, 0, MPI_COMM_WORLD);
  MPI_Bcast(&idletask, 1, MPI_INT, 0, MPI_COMM_WORLD);
  MPI_Bcast(&unlink_files, 1, MPI_INT, 0, MPI_COMM_WORLD);

  MPI_Bcast(&numfiles, 1, MPI_INT, 0, MPI_COMM_WORLD);
  MPI_Bcast(&read_task_offset, 1, MPI_INT, 0, MPI_COMM_WORLD);


  /* adjust communicators */
  communicators.all = MPI_COMM_WORLD;
  MPI_Comm_size(MPI_COMM_WORLD, &communicators.all_size);
  MPI_Comm_rank(MPI_COMM_WORLD, &communicators.all_rank);
  split_communicator(&communicators, bluegene, bluegene_np, numfiles, read_task_offset, verbose);

  /* determine global and local size of data to be written and read */
  commwork_size64 = (sion_int64) communicators.work_size;
  if (globalsize > 0) {
    totalsize = (sion_int64) globalsize / commwork_size64;
  }
  else {
    commwork_size64 = 1;
    globalsize = totalsize * size;
  }

  if(fsblksize<0) fsblksize=-1; 

  sprintf(sIdletask, "%d", idletask);

  if ( (communicators.work_size>0) && (communicators.work_rank==0) ) {
    time(&t);
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
    fprintf(stderr, "SION parallel file I/O benchmark 'partest': start at %s", ctime(&t));
    fprintf(stderr, "partest Number of tasks that will use the file tasks:       running on %d tasks\n", size);
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
#ifndef     CHECKSUM
    fprintf(stderr, "partest parameter:   CHECKSUM DISABLED!\n\n");
#endif
    fprintf(stderr, "partest parameter:   (-f)    datafile                 = %s\n", filename);
    fprintf(stderr, "partest parameter:   (-n)    number of files          = %d\n", numfiles);
    fprintf(stderr, "partest parameter:   (-F)    random factor            = %13.4f\n", factor);
    fprintf(stderr, "partest parameter:   (-X)    remove files after test  = %d\n", unlink_files);
    fprintf(stderr, "partest parameter:   (-b/-B) local buffer size / task = %15lld bytes %10.3f MB\n", bufsize, bufsize / (1.0 MB));
    fprintf(stderr, "partest parameter:   (-g/-G) global total data size   = %15lld bytes %10.3f GB\n", globalsize, globalsize / (1024.0 MB));
    fprintf(stderr, "partest parameter:   (-s/-S) total data size / task   = %15lld bytes %10.3f MB\n", totalsize, totalsize / (1.0 MB));
    fprintf(stderr, "partest parameter:   (-r/-R) sion chunk size          = %15lld bytes %10.3f MB\n", chunksize, chunksize / (1.0 MB));
    fprintf(stderr, "partest parameter:   (-Q)    fs block size            = %15d bytes %10.3f MB\n", fsblksize, fsblksize / (1.0 MB));
    if (type == 0)
      fprintf(stderr, "partest parameter:   (-T)    test type                = %d (sion, collective read)\n", type);
    if (type == 1)
      fprintf(stderr, "partest parameter:   (-T)    test type                = %d (sion, independant read)\n", type);
    if (type == 2)
      fprintf(stderr, "partest parameter:   (-T)    test type                = %d (mpi-io)\n", type);
    if (type == 3)
      fprintf(stderr, "partest parameter:   (-T)    test type                = %d (single files)\n", type);
    fprintf(stderr, "partest parameter:   (-i)    idle task                = %s\n", (idletask < 0) ? "None" : sIdletask);
    fprintf(stderr, "partest parameter:   (-Z)    read task offset         = %d\n", read_task_offset);
    fprintf(stderr, "partest parameter:   (-o)    start offset bytes       = %d\n", startoffset);
    fprintf(stderr, "partest parameter:   (-v)    verbose                  = %d\n", verbose);
    fprintf(stderr, "partest parameter:   (-d)    debug                    = %d\n", debug);
    fprintf(stderr, "partest parameter:   (-D)    Debug                    = %d\n", Debug);
    fprintf(stderr, "partest parameter:   (-P)    Blue Gene, I/O nodes     = %d\n", bluegene);
    fprintf(stderr, "partest parameter:   ()      Blue Gene: tasks/IO-node = %d\n", bluegene_np);
    fprintf(stderr, "partest parameter:   (-w)    MPI-IO, IBM, Large Block IO = %d\n", mpiio_lb);
    fprintf(stderr, "partest parameter:   (-W)    MPI-IO, IBM, IO bufsize     = %d KB\n", mpiio_bs);
    fprintf(stderr, "partest parameter:   (-x)    MPI-IO, IBM, sparse access  = %d\n", mpiio_sa);
    fprintf(stderr, "partest parameter:   (  )    commwork_size64             = %lld\n", commwork_size64);
  }

  barrier_after_start(MPI_COMM_WORLD);

  if ( (communicators.work_size>0) && (communicators.work_rank==0) ) {
    fprintf(stderr, "partest parameter:   (  )    comm(all)                   = %d of %d\n", communicators.all_rank, communicators.all_size);
    fprintf(stderr, "partest parameter:   (  )    comm(work)                  = %d of %d\n", communicators.work_rank, communicators.work_size);
    fprintf(stderr, "partest parameter:   (  )    comm(local)                 = %d of %d\n", communicators.local_rank, communicators.local_size);
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
  }

  barrier_after_start(MPI_COMM_WORLD);
  /*                                                                      */ DPRINTFTS(rank, "after pstart");
  /* Init the local buffer that will be written to the file */
  localbuffer = (char *) malloc(bufsize);

  srand(time(NULL));
  for (i = 0; i < (bufsize / sizeof(int)); i++)
    localbuffer[i] = (char) rand() % 256;

/*   memset (localbuffer, 'a'+rank%26, bufsize);  */
  memset (localbuffer, 1, bufsize); 
  barrier_after_malloc(MPI_COMM_WORLD);
  /*                                                                      */ DPRINTFTS(rank, "after malloc");


  ltotalsize = totalsize + (sion_int64) (factor * totalsize * rand() / RAND_MAX);
  /*                                                                      */ DPRINTFTS(rank, "before scall2");
  if (rank == 0) {
    fprintf(stderr, "partest parameter:   (  )    ltotalsize                   = %lld\n", ltotalsize);
  }
  barrier_after_malloc(MPI_COMM_WORLD);
  if (type == 0) {
    collectiveread = 1;
    test_paropen_multi_mpi(filename, numfiles, localbuffer, bufsize, ltotalsize, chunksize, fsblksize, startoffset, verbose, debug, Debug,
                           &communicators, collectiveread, idletask, unlink_files);

  }
  else if (type == 1) {
    collectiveread = 0;
    test_paropen_multi_mpi(filename, numfiles, localbuffer, bufsize, ltotalsize, chunksize, fsblksize, startoffset, verbose, debug, Debug,
                             &communicators, collectiveread, idletask, unlink_files);
  }
  else if (type == 2) {
    collectiveread = 1;
    test_mpiio_multi_mpi(filename, numfiles, localbuffer, bufsize, ltotalsize, chunksize, fsblksize, verbose, debug, Debug,
                         &communicators, collectiveread, idletask, mpiio_lb, mpiio_bs, mpiio_sa, unlink_files);
  }
  else if (type == 3) {
    collectiveread = 1;
    test_single_mpi(filename, numfiles, localbuffer, bufsize, ltotalsize, chunksize, fsblksize, verbose, debug, Debug,
                    &communicators, collectiveread, idletask, unlink_files);
  }

  /*                                                                      */ DPRINTFTS(rank, "before MPI_Finalize");
  barrier_after_malloc(MPI_COMM_WORLD);

  if (rank == 0) {
    time(&t);
    fprintf(stderr, "SION parallel file I/O benchmark 'partest': end at %s\n", ctime(&t));
  }

  MPI_Finalize();
  return (0);
}

/****************************************************************************************************************
 *
 *                      test_paropen_multi_mpi
 *
 ***************************************************************************************************************/

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

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

  sion_int64 left;
  sion_int64 bsumwrote, sumsize, bsumread;
  double    checksum_fp, checksum_read_fp;
  int       globalrank, sid, i, lstartoffset;
  size_t    bwrite, bwrote, btoread, bread;
  int       chunkcnt, size1;
  sion_int64 rchunksize;
  sion_int32 rfsblksize;
  FILE     *fp;
  char      cbuffer[MAXCHARLEN];
  char      *newfname;
  size_t    bytes_in_chunk; 
  int       myfeof;

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

  /* not working task ? */
  if (communicators->work_size == -1) {
    return (0);
  }

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

  /*                                                                      */ DPRINTFTS(communicators->all_rank, "[W]before open");
  /*                                    */ starttime = MPI_Wtime();
  /* numfiles>=1 -> sion will split the communicator  */
  /* numfiles<=0 -> communicators->local contain correct local communicator, computed by split_communicator  */
  sid = sion_paropen_mpi(filename, "bw", &numfiles, communicators->work, &communicators->local, &chunksize, &fsblksize, &globalrank, &fp, &newfname);
  /*                                    */ 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();

  /* local is computed again by sion_paropen_mpi if numfiles >=1 */
  MPI_Comm_size(communicators->local, &communicators->local_size);
  MPI_Comm_rank(communicators->local, &communicators->local_rank);

  checksum_fp = 0;

  /*   if(communicators->all_rank<1) fprintf(stderr,"timings[%03d] before wrloop position in file= %lld \n",communicators->all_rank,sion_get_position(sid));  */
  left = totalsize;
  bsumwrote = 0;
  chunkcnt = 0;
  bytes_in_chunk=0;
  lstartoffset=startoffset;
  /* Write until we reach the total size of the data */
  if (communicators->work_rank != idletask) {   /* only for debugging */
    while (left > 0) {
      if(lstartoffset==0) bwrite = bufsize;
      else {
	bwrite = lstartoffset; lstartoffset=0;
      }
      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);
      }
      
      bytes_in_chunk+=bwrite;
      if(bytes_in_chunk>chunksize) {
	sion_ensure_free_space(sid, bwrite);
	bytes_in_chunk=bwrite;
      }
      bwrote = fwrite(localbuffer, 1, bwrite, fp);

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

/*       fprintf(stderr,"timings[%03d] wrote (%lld bytes) total %lld bytes (%10.4f MB) (%lld left)\n",communicators->all_rank,(sion_int64) bwrote,bsumwrote,bsumwrote/1024.0/1024.0,(sion_int64) left); */

      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);
        fprintf(stderr, "timings[%03d] after write   position in file= %lld \n", communicators->all_rank, sion_get_position(sid));
      }
    }
  }

  /*                                    */ writetime = MPI_Wtime() - starttime;
  /*                                    */ starttime = MPI_Wtime();
  barrier_after_write(communicators->work);
  /*                                    */ barr2time = MPI_Wtime() - starttime;
  /*                                    */ gwritetime = MPI_Wtime() - gstarttime;

  /*                                    */ starttime = MPI_Wtime();
  sion_parclose_mpi(sid);
  /*                                    */ 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 (writetime == 0)
    writetime = -1;

  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\n",
            communicators->all_rank, opentime, writetime, closetime, barr1time, barr2time, barr3time, chunkcnt,
            totalsize / 1024.0 / 1024.0 / writetime, communicators->ionode_number);
    collective_print_gather(cbuffer, communicators->work);
  }

  if (numfiles >= 1) {
    sumsize = 0;
    MPI_Reduce(&bsumwrote, &sumsize, 1, SION_MPI_INT64, MPI_SUM, 0, communicators->local);
    if (communicators->local_rank == 0) {
      fprintf(stderr, "partest result: local totalsize=%10.4f MB  wrote %10.4f MB to %s all_rank=%d\n", totalsize / 1024.0 / 1024.0,
              1.0 * sumsize / 1024.0 / 1024.0, newfname, communicators->all_rank);
    }
  }

  /*                                                                      */ 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(&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\n",
            gopentime, gclosetime, 1.0 * sumsize / 1024.0 / 1024.0, gwritetime, 1.0 * sumsize / 1024.0 / 1024.0 / gwritetime, numfiles);
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
  }

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

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

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

  /*                                                                      */ DPRINTFTS(communicators->all_rank, "[R]before open");
  /*                                    */ starttime = MPI_Wtime();
  if (collectiveread) {
    /* commlocal and numfiles will be read from sion file */
    sid = sion_paropen_mpi(filename,
                           "br", &numfiles, communicators->workread, &communicators->local, &chunksize, &fsblksize, &globalrank, &fp, &newfname);
    /* local is computed again by sion_paropen_mpi if numfiles >=1 */
    MPI_Comm_size(communicators->local, &communicators->local_size);
    MPI_Comm_rank(communicators->local, &communicators->local_rank);

  }
  else {
    /* there is some work to for multifile sion file */
    sid = sion_open_rank(filename, "br", &rchunksize, &rfsblksize, &communicators->workread_rank, &fp);
  }
  /*                                    */ opentime = MPI_Wtime() - starttime;

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


  checksum_read_fp = 0;
  left = totalsize;
  bsumread = 0;
  chunkcnt = 0;
  bytes_in_chunk = 0;
  lstartoffset=startoffset;

  if (communicators->workread_rank != idletask) {   /* only for debugging */
    myfeof=sion_feof(sid);
    while ((left > 0) && (!myfeof)) {

      if(lstartoffset==0) btoread = bufsize;
      else {
	btoread = lstartoffset; lstartoffset=0;
      }
      if (btoread > left)
        btoread = left;

      bytes_in_chunk+=btoread;
      if(bytes_in_chunk>chunksize) {
	myfeof=sion_feof(sid);
      }
      if(!myfeof) {
	bread = fread(localbuffer, 1, btoread, fp);
	
	
#ifdef CHECKSUM
	checksum_read_fp=0.0;
	for (i = 0; i < bread; i++)
	  checksum_read_fp += (double) localbuffer[i];
#endif
	
	left -= bread;
	bsumread += bread;
	chunkcnt++;
	
	if (((debug && communicators->all_rank == 0)) || ((Debug && communicators->all_rank == size1))) {
	  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);
	  fprintf(stderr, "timings[%03d] after read   position in file= %lld restinblock=%lld\n",
		  communicators->all_rank, sion_get_position(sid), sion_bytes_avail_in_block(sid));
	}
      }
    }
  }

  /*                                    */ readtime = MPI_Wtime() - starttime;

  /*                                    */ starttime = MPI_Wtime();
  barrier_after_read(communicators->workread);
  /*                                    */ barr2time = MPI_Wtime() - starttime;
  /*                                    */ greadtime = MPI_Wtime() - gstarttime;

  /*                                    */ starttime = MPI_Wtime();
  if (collectiveread) {
    sion_parclose_mpi(sid);
  }
  else {
    sion_close(sid);
  }
  /*                                    */ closetime = MPI_Wtime() - starttime;
  /*                                                                      */ DPRINTFTS(communicators->all_rank, "[R]before close");
  barrier_after_close(communicators->work);
  /*                                                                      */ DPRINTFTS(communicators->all_rank, "[R]after close");

  if (readtime == 0)
    readtime = -1;
  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->local_rank,
            checksum_fp, checksum_read_fp, checksum_fp - checksum_read_fp);
  }
#endif

  if (numfiles >= 1) {
    MPI_Reduce(&bsumread, &sumsize, 1, SION_MPI_INT64, MPI_SUM, 0, communicators->local);
    if (communicators->local_rank == 0) {
      fprintf(stderr, "partest result: read  %10.4f MB from %s\n", 1.0 * sumsize / 1024.0 / 1024.0, newfname);
    }
  }

  /*                                                                      */ 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, numfiles);
    fprintf(stderr, "------------------------------------------------------------------------------------------\n");
  }
  
  if(unlink_files) {
    /*                                    */ starttime = MPI_Wtime();
    barrier_before_unlink(communicators->work);
    if (communicators->local_rank == 0) {
      fprintf(stderr, "partest result: unlink file %s ...\n", newfname);
      unlink(newfname);
    }
    barrier_after_unlink(communicators->work);
    /*                                    */ unlinktime = MPI_Wtime() - starttime;
    if (communicators->local_rank == 0) {
      fprintf(stderr, "partest result:  ultime=%10.6fs unlink %s\n", unlinktime, newfname);
    }
  }


  return (1);
}

int split_communicator(_test_communicators * communicators, int bluegene, int bluegene_np, int numfiles, int read_task_offset, int verbose)
{
  int       proc_per_file;

  communicators->work_size = communicators->work_rank = -2;
  communicators->local_size = communicators->local_rank = -2;



#ifdef _SION_BGP
  if (bluegene) {               /* order MPI-tasks by I/O-node */
    _BGP_Personality_t personality;
    MPI_Comm  commSame, commDiff, commTemp;
    int       sizeSame, sizeDiff;
    int       rankSame, rankDiff;
    char      location[128];
    unsigned  procid, x, y, z, t;
    char      cbuffer[MAXCHARLEN];

    /* get location information */
    Kernel_GetPersonality(&personality, sizeof(personality));
    BGP_Personality_getLocationString(&personality, location);
    procid = Kernel_PhysicalProcessorID();
    MPIX_rank2torus(communicators->all_rank, &x, &y, &z, &t);

    /* task of communicator working with different I/O-nodes */
    MPIX_Pset_diff_comm_create(&commDiff);
    MPI_Comm_size(commDiff, &sizeDiff);
    MPI_Comm_rank(commDiff, &rankDiff);
    communicators->ionode_number = rankDiff;

    /* communicator consists of all task working with the same I/O-node */
    MPIX_Pset_same_comm_create(&commSame);
    MPI_Comm_size(commSame, &sizeSame);
    MPI_Comm_rank(commSame, &rankSame);

    /*  if -p not specified all proc will write! */
    if (bluegene_np == 0) {
      bluegene_np = sizeSame;
    }

    /*  Get a communicator with all writing tasks => new global communicator */
    MPI_Comm_split(communicators->all, (rankSame < bluegene_np), communicators->all_rank, &communicators->work);
    MPI_Comm_size(communicators->work, &communicators->work_size);
    MPI_Comm_rank(communicators->work, &communicators->work_rank);
    if (rankSame >= bluegene_np) {
      /* not working task */
      communicators->work_size = communicators->work_rank = -1;
      communicators->local_size = communicators->local_rank = -1;
    }

    /*  If only one file will be used => dont split further */
    /*  if numfile > 1 sion will generate correct local communicator */
    if (numfiles >= 1) {
      communicators->local = communicators->work;
    }
    else if(numfiles<0) {
      if(numfiles==-1) {
	/*  Split the common communicator for each IO node to get a local comm with only the writing tasks for this IO Node */
	MPI_Comm_split(commSame, (rankSame < bluegene_np), communicators->all_rank, &communicators->local);
      } else {
	/*  local communicator contains only one task per IO-node */
	/* bluegene_np has to be 512 */
	communicators->local=commDiff;
      }
    } 
    MPI_Comm_size(communicators->local, &communicators->local_size);
    MPI_Comm_rank(communicators->local, &communicators->local_rank);

    /* determine filenumber */
    if (numfiles < 1) {
      /* one file per I/O-node */
      if(numfiles==-1)  communicators->file_number = rankDiff;
      else              communicators->file_number = rankSame;
    }
    else {
      communicators->file_number = -1;
    }

    /* print log message about location, ...  */
    sprintf(cbuffer, "");
    if (rankSame < bluegene_np) {
      if (verbose) {
        sprintf(cbuffer, "BGP[%05d] diff_comm: %4d of %4d  same_comm: %5d of %5d file_comm: %5d of %5d %s phys_xyzt(%d,%d,%d,%d)\n",
                communicators->all_rank, rankDiff + 1, sizeDiff, rankSame + 1, sizeSame, communicators->local_rank + 1, communicators->local_size,
                location, x, y, z, t);
      }
    }
    collective_print_gather(cbuffer, communicators->work);

  }
#endif

#ifdef _SION_BGL
  if (bluegene) {               /* order MPI-tasks by I/O-node */
    _BGP_Personality_t personality;
    MPI_Comm  commSame, commDiff, commTemp;
    int       sizeSame, sizeDiff;
    int       rankSame, rankDiff;
    char      location[BGLPERSONALITY_MAX_LOCATION];
    unsigned  procid, x, y, z, t;
    BGLPersonality personality;
    char      cbuffer[MAXCHARLEN];

    /* get location information */
    rts_get_personality(&personality, sizeof(personality));
    BGLPersonality_getLocationString(&personality, location);
    x = BGLPersonality_xCoord(&personality),
      y = BGLPersonality_yCoord(&personality), z = BGLPersonality_zCoord(&personality), t = rts_get_processor_id()

      /* task of communicator working with different I/O-nodes */
      PMI_Pset_diff_comm_create(&commDiff);
    MPI_Comm_size(commDiff, &sizeDiff);
    MPI_Comm_rank(commDiff, &rankDiff);
    *ionode_number = rankDiff;

    /* communicator consists of all task working with the same I/O-node */
    PMI_Pset_same_comm_create(&commSame);
    MPI_Comm_size(commSame, &sizeSame);
    MPI_Comm_rank(commSame, &rankSame);

    /*  if -p not specified all proc will write! */
    if (bluegene_np == 0) {
      bluegene_np = sizeSame;
    }

    /*  Get a communicator with all writing tasks => new global communicator */
    MPI_Comm_split(communicators->all, (rankSame < bluegene_np), communicators->all_rank, &communicators->work);
    MPI_Comm_size(communicators->work, &communicators->work_size);
    MPI_Comm_rank(communicators->work, &communicators->work_rank);


    /*  If only one file will be used => dont split further */
    if (numfiles == 1) {
      communicators->local = communicators->work;
    }
    else {
      /*  Split the common communicator for each IO node to get a local comm with only the writing tasks for this IO Node */
      MPI_Comm_split(commSame, (rankSame < bluegene_np), communicators->all_rank, &communicators->local);
    }
    MPI_Comm_size(communicators->local, &communicators->local_size);
    MPI_Comm_rank(communicators->local, &communicators->local_rank);

    /* determine filenumber */
    if (numfiles < 1) {
      /* one file per I/O-node */
      communicators->file_number = rankDiff;
    }
    else {
      communicators->file_number = -1;
    }

    /* print log message about location, ...  */
    sprintf(cbuffer, "");
    if (rankSame < bluegene_np) {
      if (verbose) {
        sprintf(cbuffer, "BGL[%05d] diff_comm: %4d of %4d  same_comm: %5d of %5d file_comm: %5d of %5d %s phys_xyzt(%d,%d,%d,%d)\n",
                communicators->all_rank, rankDiff + 1, sizeDiff, rankSame + 1, sizeSame,
                communicators->local_rank + 1, communicators->local_size, location, x, y, z, t);
      }
    }
    collective_print_gather(cbuffer, communicators->work);

  }
#endif

#ifdef _SION_AIX
  /* no communicator adjustment */
#endif

  /* initial set of communicators */
  if (communicators->work_size == -2) {
    /* all task do work */
    communicators->work = communicators->all;
    MPI_Comm_size(communicators->work, &communicators->work_size);
    MPI_Comm_rank(communicators->work, &communicators->work_rank);
  }
  /* local communicators  */
  if (communicators->local_size == -2) {
    if (numfiles == 1) {
      communicators->local = communicators->work;
    }
    /* set a default distribution on files, will be computed again by sion_open */
    if (numfiles < 1) {
      numfiles = communicators->work_size / 2;
      if (numfiles == 0)
        numfiles = 1;
    }
    proc_per_file = communicators->work_size / numfiles;

    /* remaining tasks are write/read to/from the last file */
    if (communicators->work_rank >= (numfiles * proc_per_file)) {
      communicators->file_number = numfiles - 1;
    }
    else {
      communicators->file_number = communicators->work_rank / proc_per_file;
    }

    MPI_Comm_split(communicators->work, communicators->file_number, communicators->all_rank, &communicators->local);

    MPI_Comm_size(communicators->local, &communicators->local_size);
    MPI_Comm_rank(communicators->local, &communicators->local_rank);

    communicators->ionode_number = communicators->file_number;

  }

#ifdef _SION_LINUX
  if (verbose) {
    char      location[256];
    gethostname(location, 256);
    char      cbuffer[MAXCHARLEN];
    sprintf(cbuffer, "LINUX[%03d] diff_comm: %4d of %4d  same_comm: %4d of %4d file_comm: %4d of %4d %s\n",
            communicators->all_rank, communicators->all_rank, communicators->all_size,
            communicators->work_rank, communicators->work_size, communicators->local_rank, communicators->local_size, location);
    collective_print_gather(cbuffer, communicators->all);
  }

#endif

  /* _XT TODO */
#ifdef _SION_AIX
  if (verbose) {
    char      location[256];
    gethostname(location, 256);
    int       sizeSame = 0, sizeDiff = 0;
    int       rankSame, rankDiff;
    char      cbuffer[MAXCHARLEN];
    sprintf(cbuffer, "AIX[%03d] diff_comm: %4d of %4d  same_comm: %4d of %4d file_comm: %4d of %4d %s\n",
            communicators->all_rank, communicators->all_rank, communicators->all_size,
            communicators->work_rank, communicators->work_size, communicators->local_rank, communicators->local_size, location);
    collective_print_gather(cbuffer, communicators->all);
  }

#endif

#ifdef _SION_XT
  if (verbose) {
    char      location[256];
    gethostname(location, 256);
    int       sizeSame = 0, sizeDiff = 0;
    int       rankSame, rankDiff;
    char      cbuffer[MAXCHARLEN];
    sprintf(cbuffer, "XT[%03d] diff_comm: %4d of %4d  same_comm: %4d of %4d file_comm: %4d of %4d %s\n",
            communicators->all_rank, communicators->all_rank, communicators->all_size,
            communicators->work_rank, communicators->work_size, communicators->local_rank, communicators->local_size, location);
    collective_print_gather(cbuffer, communicators->all);
  }
#endif

  /* shift working tasks  */
  {
    int newtasknr;
    newtasknr=(communicators->work_rank+read_task_offset)%communicators->work_size;
    MPI_Comm_split(communicators->work, 0, newtasknr, &communicators->workread);

    MPI_Comm_size(communicators->workread, &communicators->workread_size);
    MPI_Comm_rank(communicators->workread, &communicators->workread_rank);
/*     printf("WF: %d %d %% %d-> %d (%d %d)\n",communicators->work_rank,read_task_offset,communicators->work_size,newtasknr,communicators->workread_rank,communicators->workread_size); */
   
  }

  return(1);
}
