/****************************************************************************
**  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 <math.h>
#include <mpi.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <time.h>
#include <unistd.h>

#include "sion.h"
#include "sion_debug.h"

#define ONRANK if (rank == 0)

char *_lsion_buddy_role_to_str(unsigned int flag);

/*

allnodes/testB.out

siondump: number of files:           5
siondump: file   1:                    testB.out
siondump: file   2:                    testB.out.000001
siondump: file   3:                    testB.out.000002
siondump: file   4:                    testB.out.000003
siondump: file   5:                    testB.out.000004
------------------------------------------------------------
  file[000]:  0:00000  1:00001
  file[001]:  0:00002  1:00003  2:00004
  file[002]:  0:00005  1:00006  2:00007  3:00008
  file[003]:  0:00009  1:00010  2:00011  3:00012  4:00013
  file[004]:  0:00014  1:00015
------------------------------------------------------------

------------------------------------------------------------
siondump: number of files:           5
siondump: file   1:                    testB.out
siondump: file   2:                    testB.out.000001
siondump: file   3:                    testB.out.000002
siondump: file   4:                    testB.out.000003
siondump: file   5:                    testB.out.000004
------------------------------------------------------------
  file[000]:  2:00000  3:00001
  file[001]:  2:00002  3:00003  4:00004
  file[002]:  3:00005  4:00006  5:00007  6:00008
  file[003]:  4:00009  5:00010  6:00011  7:00012  8:00013
  file[004]:  5:00014  6:00015
------------------------------------------------------------

 */

int print_io_info(int sid, int rank)
{
  sion_io_stat_t *p;
  int c;

  p = sion_get_io_info(sid);
  if (p == NULL) {
    return 0;
  }

  printf("on rank %2d: io_info [nfiles=%d]\n", rank, p->nfiles);
  for (c = 0; c < p->nfiles; c++) {
    printf("on rank %2d: io_info [#%02d: fn=%-30s size=%6d role=%-20s]\n", rank, c, p->names[c], (int)p->sizes[c],
      _lsion_buddy_role_to_str(p->roles[c]));
  }
  return 1;
}

int main(int argc, char **argv)
{
  int rank, size;
  char debugname[80];
  MPI_Comm lcomm1;
  MPI_Comm lcomm2;

  /* -------------------------- */
  /* PROLOG */
  /* -------------------------- */

  MPI_Init(&argc, &argv);
  MPI_Comm_size(MPI_COMM_WORLD, &size);
  MPI_Comm_rank(MPI_COMM_WORLD, &rank);

  /* -------------------------- */
  /* TEST A: write a empty file */
  /* -------------------------- */
  if (0) {
    int sid;
    sion_mpi_options options = sion_mpi_options_new();

    sion_mpi_options_set_chunksize(&options, 100);
    sion_mpi_options_set_fsblksize(&options, 10);
    sion_mpi_options_set_buddy(&options);

    sid = sion_paropen_mpi_with_options("testA.out", SION_OPEN_WRITE, MPI_COMM_WORLD, &options);

    if (sid >= 0) {
      sion_parclose_mpi(sid);
    }
  }

  fflush(stderr);
  fflush(stdout);
  MPI_Barrier(MPI_COMM_WORLD);
  ONRANK printf("on rank %2d: END of TEST A\n", rank);
  MPI_Barrier(MPI_COMM_WORLD);

  /* ------------------------------------------------------------- */
  /* TEST A1: write a small file with some different pattern (Coll) */
  /* ------------------------------------------------------------- */
  if (1) {
#define BUFSIZE 1000
#define CHUNKSIZE 1200
    char dirname[80];
    int sid;
    char buffer[BUFSIZE];
    int64_t bytes_written = -1;
    long sum, gsum;
    size_t bwrote;
    int i, filenum, lrank, lsize;
    sion_mpi_options options = sion_mpi_options_new();

    for (i = 0; i < BUFSIZE; i++) {
      buffer[i] = 'A' + rank;
    }

    /* filenum=rank/4; */
    if (rank < 2) {
      filenum = 0;
    } else if (rank < 5) {
      filenum = 1;
    } else if (rank < 9) {
      filenum = 2;
    } else if (rank < 14) {
      filenum = 3;
    } else {
      filenum = 4;
    }
    MPI_Comm_split(MPI_COMM_WORLD, filenum, rank, &lcomm1);
    MPI_Comm_size(lcomm1, &lsize);
    MPI_Comm_rank(lcomm1, &lrank);

    sprintf(dirname, "node%03d", filenum);
    if (lrank == 0) {
      mkdir(dirname, 0751);
    }
    MPI_Barrier(MPI_COMM_WORLD);
    chdir(dirname);
    MPI_Barrier(MPI_COMM_WORLD);

    sprintf(debugname, "%s.%05d", "deb_buddy_write", rank);
    sion_debug_on(1023, debugname);

    sion_mpi_options_set_chunksize(&options, CHUNKSIZE);
    sion_mpi_options_set_fsblksize(&options, 100);
    sion_mpi_options_set_buddy(&options);
    sion_mpi_options_set_multifile_communicator(&options, lcomm1);

    sid = sion_paropen_mpi_with_options("testB.out", SION_OPEN_WRITE, MPI_COMM_WORLD, &options);

    if (sid >= 0) {
      sum = 0;
      bwrote = sion_coll_fwrite(buffer, 1, rank, sid);
      for (i = 0; i < bwrote; i++) {
        sum = sum + buffer[i];
      }
      ONRANK printf("on rank %2d: bwrote=%5d overall blocksum=%8d \n", rank, (int)bwrote, (int)sum);
      bwrote = sion_coll_fwrite(buffer, 1, BUFSIZE / 2, sid);
      for (i = 0; i < bwrote; i++) {
        sum = sum + buffer[i];
      }
      ONRANK printf("on rank %2d: bwrote=%5d overall blocksum=%8d \n", rank, (int)bwrote, (int)sum);
      bwrote = sion_coll_fwrite(buffer, 1, BUFSIZE, sid);
      for (i = 0; i < bwrote; i++) {
        sum = sum + buffer[i];
      }
      ONRANK printf("on rank %2d: bwrote=%5d overall blocksum=%8d \n", rank, (int)bwrote, (int)sum);

      bytes_written = sion_get_bytes_written(sid);
      ONRANK printf("on rank %2d: bytes_written=%" PRId64 "\n", rank, bytes_written);

      print_io_info(sid, rank);

      sion_parclose_mpi(sid);
    } else {
      ONRANK fprintf(stderr, "on rank %2d: error sid = %d\n", rank, sid);
    }
    MPI_Reduce(&sum, &gsum, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);
    ONRANK printf("on rank %2d: global blocksum = %d\n", rank, (int)gsum);

    chdir("..");
  }

#undef BUFSIZE
#undef CHUNKSIZE

  fflush(stderr);
  fflush(stdout);
  MPI_Barrier(MPI_COMM_WORLD);
  ONRANK printf("on rank %2d: END of TEST B write coll\n", rank);
  MPI_Barrier(MPI_COMM_WORLD);

  /* ------------------------------------------------------ */
  /* TEST C: read small file from original files       */
  /* ------------------------------------------------------ */
  if (1) {
#define BUFSIZE 1200
#define CHUNKSIZE 1200
    char dirname[80];
    int sid;
    char buffer[BUFSIZE];
    int64_t bytes_avail = -1;
    size_t bytestoread, bread;
    long sum, gsum;
    int i, filenum, b, lrank, lsize;
    sion_mpi_options options = sion_mpi_options_new();

    for (i = 0; i < BUFSIZE; i++) {
      buffer[i] = 'A' + rank;
    }

    if (rank == 0) {
      /* rename("./node002/testB.out.000002", "./node002/testB.out.000002_deleted"); */
    }

    /* filenum=rank/4; */
    if (rank < 2) {
      filenum = 0;
    } else if (rank < 5) {
      filenum = 1;
    } else if (rank < 9) {
      filenum = 2;
    } else if (rank < 14) {
      filenum = 3;
    } else {
      filenum = 4;
    }
    MPI_Comm_split(MPI_COMM_WORLD, filenum, rank, &lcomm2);
    MPI_Comm_size(lcomm2, &lsize);
    MPI_Comm_rank(lcomm2, &lrank);

    sprintf(dirname, "node%03d", filenum);
    MPI_Barrier(MPI_COMM_WORLD);
    chdir(dirname);
    MPI_Barrier(MPI_COMM_WORLD);

    sprintf(debugname, "%s.%05d", "deb_buddy_read_C", rank);
    sion_debug_on(1023, debugname);

    sion_mpi_options_set_chunksize(&options, CHUNKSIZE);
    sion_mpi_options_set_fsblksize(&options, 10);
    sion_mpi_options_set_buddy(&options);
    sion_mpi_options_set_multifile_communicator(&options, lcomm2);

    sid = sion_paropen_mpi_with_options("testB.out", SION_OPEN_READ, MPI_COMM_WORLD, &options);

    if (sid >= 0) {
      printf("on rank %2d: sid           =%d\n", rank, sid);
      ONRANK printf("on rank %2d: numfiles      =%d\n", rank, sion_get_number_of_files(sid));
      ONRANK printf("on rank %2d: chunksize     =%d\n", rank, (int)sion_get_chunksize(sid));
      ONRANK printf("on rank %2d: globalrank    =%d\n", rank, sion_get_globalrank(sid));
      ONRANK printf("on rank %2d: newfname      =%s\n", rank, sion_get_fname(sid));

      b = 0;
      sum = 0;
      while ((!sion_feof(sid))) {
        bytes_avail = sion_bytes_avail_in_block(sid);
        printf("on rank %2d: block=%d           bytes_avail=%d\n", rank, b, (int)bytes_avail);
        bytestoread = bytes_avail;
        bread = sion_coll_fread(buffer, 1, bytestoread, sid);
        for (i = 0; i < bread; i++) {
          sum = sum + buffer[i];
        }
        printf("on rank %2d: block=%d           bread=%3d overall blocksum=%8d \n", rank, b, (int)bread, (int)sum);
        b++;
      }

      sion_parclose_mpi(sid);
    } else {
      fprintf(stderr, "on rank %2d: error sid = %d\n", rank, sid);
    }

    chdir("..");

    MPI_Reduce(&sum, &gsum, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);
    ONRANK printf("on rank %2d: global blocksum = %d\n", rank, (int)gsum);
  }

  fflush(stderr);
  fflush(stdout);
  MPI_Barrier(MPI_COMM_WORLD);
  ONRANK printf("on rank %2d: END of TEST C read collective\n", rank);
  MPI_Barrier(MPI_COMM_WORLD);

  /* ------------------------------------------------------ */
  /* TEST D: read small file partly from buddy files        */
  /* ------------------------------------------------------ */
  if (1) {
#define BUFSIZE 1200
#define CHUNKSIZE 1200
    char dirname[80];
    int sid;
    char buffer[BUFSIZE];
    int64_t bytes_avail = -1;
    size_t bytestoread, bread;
    long sum, gsum;
    int i, filenum, b, lrank, lsize;
    sion_mpi_options options = sion_mpi_options_new();

    for (i = 0; i < BUFSIZE; i++) {
      buffer[i] = 'A' + rank;
    }

    if (rank == 0) {
      rename("./node002/testB.out.000002", "./node002/testB.out.000002_deleted");
    }

    /* filenum=rank/4; */
    if (rank < 2) {
      filenum = 0;
    } else if (rank < 5) {
      filenum = 1;
    } else if (rank < 9) {
      filenum = 2;
    } else if (rank < 14) {
      filenum = 3;
    } else {
      filenum = 4;
    }
    MPI_Comm_split(MPI_COMM_WORLD, filenum, rank, &lcomm2);
    MPI_Comm_size(lcomm2, &lsize);
    MPI_Comm_rank(lcomm2, &lrank);

    sprintf(dirname, "node%03d", filenum);
    MPI_Barrier(MPI_COMM_WORLD);
    chdir(dirname);
    MPI_Barrier(MPI_COMM_WORLD);

    sprintf(debugname, "%s.%05d", "deb_buddy_read_D", rank);
    sion_debug_on(1023, debugname);

    sion_mpi_options_set_chunksize(&options, CHUNKSIZE);
    sion_mpi_options_set_fsblksize(&options, 10);
    sion_mpi_options_set_buddy(&options);
    sion_mpi_options_set_multifile_communicator(&options, lcomm2);

    sid = sion_paropen_mpi_with_options("testB.out", SION_OPEN_READ, MPI_COMM_WORLD, &options);

    if (sid >= 0) {
      printf("on rank %2d: sid           =%d\n", rank, sid);
      ONRANK printf("on rank %2d: numfiles      =%d\n", rank, sion_get_number_of_files(sid));
      ONRANK printf("on rank %2d: chunksize     =%d\n", rank, (int)sion_get_chunksize(sid));
      ONRANK printf("on rank %2d: globalrank    =%d\n", rank, sion_get_globalrank(sid));
      ONRANK printf("on rank %2d: newfname      =%s\n", rank, sion_get_fname(sid));

      b = 0;
      sum = 0;
      while ((!sion_feof(sid))) {
        bytes_avail = sion_bytes_avail_in_block(sid);
        printf("on rank %2d: block=%d           bytes_avail=%d\n", rank, b, (int)bytes_avail);
        bytestoread = bytes_avail;
        bread = sion_coll_fread(buffer, 1, bytestoread, sid);
        for (i = 0; i < bread; i++) {
          sum = sum + buffer[i];
        }
        printf("on rank %2d: block=%d           bread=%3d overall blocksum=%8d \n", rank, b, (int)bread, (int)sum);
        b++;
      }

      sion_parclose_mpi(sid);
    } else {
      fprintf(stderr, "on rank %2d: error sid = %d\n", rank, sid);
    }

    chdir("..");

    MPI_Reduce(&sum, &gsum, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);
    ONRANK printf("on rank %2d: global blocksum = %d\n", rank, (int)gsum);

    if (rank == 0) {
      rename("./node002/testB.out.000002_deleted", "./node002/testB.out.000002");
    }
  }

  fflush(stderr);
  fflush(stdout);
  MPI_Barrier(MPI_COMM_WORLD);
  ONRANK printf("on rank %2d: END of TEST D read collective\n", rank);
  MPI_Barrier(MPI_COMM_WORLD);

  /* ------------------------------------------------------ */
  /* TEST E: read small file partly from buddy files        */
  /* ------------------------------------------------------ */
  if (1) {
#define BUFSIZE 1200
#define CHUNKSIZE 1200
    char dirname[80];
    int sid;
    char buffer[BUFSIZE];
    int64_t bytes_avail = -1;
    size_t bytestoread, bread;
    long sum, gsum;
    int i, filenum, b, lrank, lsize;
    sion_mpi_options options = sion_mpi_options_new();

    for (i = 0; i < BUFSIZE; i++) {
      buffer[i] = 'A' + rank;
    }

    if (rank == 0) {
      rename("./node002/testB.out.000002", "./node002/testB.out.000002_deleted");
      rename("./node001/testB.out_BUDDY_00.000002", "./node002/testB.out_BUDDY_00.000002");
    }

    /* filenum=rank/4; */
    if (rank < 2) {
      filenum = 0;
    } else if (rank < 5) {
      filenum = 1;
    } else if (rank < 9) {
      filenum = 2;
    } else if (rank < 14) {
      filenum = 3;
    } else {
      filenum = 4;
    }
    MPI_Comm_split(MPI_COMM_WORLD, filenum, rank, &lcomm2);
    MPI_Comm_size(lcomm2, &lsize);
    MPI_Comm_rank(lcomm2, &lrank);

    sprintf(dirname, "node%03d", filenum);
    MPI_Barrier(MPI_COMM_WORLD);
    chdir(dirname);
    MPI_Barrier(MPI_COMM_WORLD);

    sprintf(debugname, "%s.%05d", "deb_buddy_read_E", rank);
    sion_debug_on(1023, debugname);

    sion_mpi_options_set_chunksize(&options, CHUNKSIZE);
    sion_mpi_options_set_fsblksize(&options, 10);
    sion_mpi_options_set_buddy(&options);
    sion_mpi_options_set_multifile_communicator(&options, lcomm2);

    sid = sion_paropen_mpi_with_options("testB.out", SION_OPEN_READ, MPI_COMM_WORLD, &options);

    if (sid >= 0) {
      printf("on rank %2d: sid           =%d\n", rank, sid);
      ONRANK printf("on rank %2d: numfiles      =%d\n", rank, sion_get_number_of_files(sid));
      ONRANK printf("on rank %2d: chunksize     =%d\n", rank, (int)sion_get_chunksize(sid));
      ONRANK printf("on rank %2d: globalrank    =%d\n", rank, sion_get_globalrank(sid));
      ONRANK printf("on rank %2d: newfname      =%s\n", rank, sion_get_fname(sid));

      b = 0;
      sum = 0;
      while ((!sion_feof(sid))) {
        bytes_avail = sion_bytes_avail_in_block(sid);
        printf("on rank %2d: block=%d           bytes_avail=%d\n", rank, b, (int)bytes_avail);
        bytestoread = bytes_avail;
        bread = sion_coll_fread(buffer, 1, bytestoread, sid);
        for (i = 0; i < bread; i++) {
          sum = sum + buffer[i];
        }
        printf("on rank %2d: block=%d           bread=%3d overall blocksum=%8d \n", rank, b, (int)bread, (int)sum);
        b++;
      }

      sion_parclose_mpi(sid);
    } else {
      fprintf(stderr, "on rank %2d: error sid = %d\n", rank, sid);
    }

    chdir("..");

    MPI_Reduce(&sum, &gsum, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);
    ONRANK printf("on rank %2d: global blocksum = %d\n", rank, (int)gsum);

    if (rank == 0) {
      rename("./node002/testB.out.000002_deleted", "./node002/testB.out.000002");
      rename("./node002/testB.out_BUDDY_00.000002", "./node001/testB.out_BUDDY_00.000002");
    }
  }

  fflush(stderr);
  fflush(stdout);
  MPI_Barrier(MPI_COMM_WORLD);
  ONRANK printf("on rank %2d: END of TEST E read collective\n", rank);
  MPI_Barrier(MPI_COMM_WORLD);

  /* ------------------------------------------------------ */
  /* TEST F: restart completely from buddy files            */
  /* ------------------------------------------------------ */
  if (1) {
#define BUFSIZE 1200
#define CHUNKSIZE 1200
    char dirname[80];
    int sid;
    char buffer[BUFSIZE];
    int64_t bytes_avail = -1;
    size_t bytestoread, bread;
    long sum, gsum;
    int i, filenum, b, lrank, lsize;
    sion_mpi_options options = sion_mpi_options_new();

    for (i = 0; i < BUFSIZE; i++) {
      buffer[i] = 'A' + rank;
    }

    if (rank == 0) {
      rename("./node000/testB.out", "./node000/testB.out_deleted");
      rename("./node001/testB.out.000001", "./node001/testB.out.000001_deleted");
      rename("./node002/testB.out.000002", "./node002/testB.out.000002_deleted");
      rename("./node003/testB.out.000003", "./node003/testB.out.000003_deleted");
      rename("./node004/testB.out.000004", "./node003/testB.out.000004_deleted");
    }

    /* filenum=rank/4; */
    if (rank < 2) {
      filenum = 0;
    } else if (rank < 5) {
      filenum = 1;
    } else if (rank < 9) {
      filenum = 2;
    } else if (rank < 14) {
      filenum = 3;
    } else {
      filenum = 4;
    }
    MPI_Comm_split(MPI_COMM_WORLD, filenum, rank, &lcomm2);
    MPI_Comm_size(lcomm2, &lsize);
    MPI_Comm_rank(lcomm2, &lrank);

    sprintf(dirname, "node%03d", filenum);
    MPI_Barrier(MPI_COMM_WORLD);
    chdir(dirname);
    MPI_Barrier(MPI_COMM_WORLD);

    sprintf(debugname, "%s.%05d", "deb_buddy_read_F", rank);
    sion_debug_on(1023, debugname);

    sion_mpi_options_set_chunksize(&options, CHUNKSIZE);
    sion_mpi_options_set_fsblksize(&options, 10);
    sion_mpi_options_set_buddy(&options);
    sion_mpi_options_set_multifile_communicator(&options, lcomm2);

    sid = sion_paropen_mpi_with_options("testB.out", SION_OPEN_READ, MPI_COMM_WORLD, &options);

    if (sid >= 0) {
      printf("on rank %2d: sid           =%d\n", rank, sid);
      ONRANK printf("on rank %2d: numfiles      =%d\n", rank, sion_get_number_of_files(sid));
      ONRANK printf("on rank %2d: chunksize     =%d\n", rank, (int)sion_get_chunksize(sid));
      ONRANK printf("on rank %2d: globalrank    =%d\n", rank, sion_get_globalrank(sid));
      ONRANK printf("on rank %2d: newfname      =%s\n", rank, sion_get_fname(sid));

      b = 0;
      sum = 0;
      while ((!sion_feof(sid))) {
        bytes_avail = sion_bytes_avail_in_block(sid);
        printf("on rank %2d: block=%d           bytes_avail=%d\n", rank, b, (int)bytes_avail);
        bytestoread = bytes_avail;
        bread = sion_coll_fread(buffer, 1, bytestoread, sid);
        for (i = 0; i < bread; i++) {
          sum = sum + buffer[i];
        }
        printf("on rank %2d: block=%d           bread=%3d overall blocksum=%8d \n", rank, b, (int)bread, (int)sum);
        b++;
      }

      sion_parclose_mpi(sid);
    } else {
      fprintf(stderr, "on rank %2d: error sid = %d\n", rank, sid);
    }

    chdir("..");

    MPI_Reduce(&sum, &gsum, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);
    ONRANK printf("on rank %2d: global blocksum = %d\n", rank, (int)gsum);

    if (rank == 0) {
      rename("./node000/testB.out_deleted", "./node000/testB.out");
      rename("./node001/testB.out.000001_deleted", "./node001/testB.out.000001");
      rename("./node002/testB.out.000002_deleted", "./node002/testB.out.000002");
      rename("./node003/testB.out.000003_deleted", "./node003/testB.out.000003");
      rename("./node003/testB.out.000004_deleted", "./node004/testB.out.000004");
    }
  }

  fflush(stderr);
  fflush(stdout);
  MPI_Barrier(MPI_COMM_WORLD);
  ONRANK printf("on rank %2d: END of TEST F read collective\n", rank);
  MPI_Barrier(MPI_COMM_WORLD);

  /* ------------------------------------------------------ */
  /* TEST G: all original files are on one node             */
  /* ------------------------------------------------------ */
  if (1) {
#define BUFSIZE 1200
#define CHUNKSIZE 1200
    char dirname[80];
    int sid;
    char buffer[BUFSIZE];
    int64_t bytes_avail = -1;
    size_t bytestoread, bread;
    long sum, gsum;
    int i, filenum, b, lrank, lsize;
    sion_mpi_options options = sion_mpi_options_new();

    for (i = 0; i < BUFSIZE; i++) {
      buffer[i] = 'A' + rank;
    }

    if (rank == 0) {
      rename("./node000/testB.out", "./node004/testB.out");
      rename("./node001/testB.out.000001", "./node004/testB.out.000001");
      rename("./node002/testB.out.000002", "./node004/testB.out.000002");
      rename("./node003/testB.out.000003", "./node004/testB.out.000003");
    }

    /* filenum=rank/4; */
    if (rank < 2) {
      filenum = 0;
    } else if (rank < 5) {
      filenum = 1;
    } else if (rank < 9) {
      filenum = 2;
    } else if (rank < 14) {
      filenum = 3;
    } else {
      filenum = 4;
    }
    MPI_Comm_split(MPI_COMM_WORLD, filenum, rank, &lcomm2);
    MPI_Comm_size(lcomm2, &lsize);
    MPI_Comm_rank(lcomm2, &lrank);

    sprintf(dirname, "node%03d", filenum);
    MPI_Barrier(MPI_COMM_WORLD);
    chdir(dirname);
    MPI_Barrier(MPI_COMM_WORLD);

    sprintf(debugname, "%s.%05d", "deb_buddy_read_G", rank);
    sion_debug_on(1023, debugname);

    sion_mpi_options_set_chunksize(&options, CHUNKSIZE);
    sion_mpi_options_set_fsblksize(&options, 10);
    sion_mpi_options_set_buddy(&options);
    sion_mpi_options_set_multifile_communicator(&options, lcomm2);

    sid = sion_paropen_mpi_with_options("testB.out", SION_OPEN_READ, MPI_COMM_WORLD, &options);

    if (sid >= 0) {
      printf("on rank %2d: sid           =%d\n", rank, sid);
      ONRANK printf("on rank %2d: numfiles      =%d\n", rank, sion_get_number_of_files(sid));
      ONRANK printf("on rank %2d: chunksize     =%d\n", rank, (int)sion_get_chunksize(sid));
      ONRANK printf("on rank %2d: globalrank    =%d\n", rank, sion_get_globalrank(sid));
      ONRANK printf("on rank %2d: newfname      =%s\n", rank, sion_get_fname(sid));

      b = 0;
      sum = 0;
      while ((!sion_feof(sid))) {
        bytes_avail = sion_bytes_avail_in_block(sid);
        printf("on rank %2d: block=%d           bytes_avail=%d\n", rank, b, (int)bytes_avail);
        bytestoread = bytes_avail;
        bread = sion_coll_fread(buffer, 1, bytestoread, sid);
        for (i = 0; i < bread; i++) {
          sum = sum + buffer[i];
        }
        printf("on rank %2d: block=%d           bread=%3d overall blocksum=%8d \n", rank, b, (int)bread, (int)sum);
        b++;
      }

      sion_parclose_mpi(sid);
    } else {
      fprintf(stderr, "on rank %2d: error sid = %d\n", rank, sid);
    }

    chdir("..");

    MPI_Reduce(&sum, &gsum, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);
    ONRANK printf("on rank %2d: global blocksum = %d\n", rank, (int)gsum);

    if (rank == 0) {
      rename("./node004/testB.out", "./node000/testB.out");
      rename("./node004/testB.out.000001", "./node001/testB.out.000001");
      rename("./node004/testB.out.000002", "./node002/testB.out.000002");
      rename("./node004/testB.out.000003", "./node003/testB.out.000003");
    }
  }

  fflush(stderr);
  fflush(stdout);
  MPI_Barrier(MPI_COMM_WORLD);
  ONRANK printf("on rank %2d: END of TEST G read collective\n", rank);
  MPI_Barrier(MPI_COMM_WORLD);

  /* ------------------------------------------------------ */
  /* TEST H: all buddy files are on one node                */
  /* ------------------------------------------------------ */
  if (1) {
#define BUFSIZE 1200
#define CHUNKSIZE 1200
    char dirname[80];
    int sid;
    char buffer[BUFSIZE];
    int64_t bytes_avail = -1;
    size_t bytestoread, bread;
    long sum, gsum;
    int i, filenum, b, lrank, lsize;
    sion_mpi_options options = sion_mpi_options_new();

    for (i = 0; i < BUFSIZE; i++) {
      buffer[i] = 'A' + rank;
    }

    if (rank == 0) {
      rename("./node000/testB.out", "./node000/testB.out_deleted");
      rename("./node001/testB.out.000001", "./node001/testB.out.000001_deleted");
      rename("./node002/testB.out.000002", "./node002/testB.out.000002_deleted");
      rename("./node003/testB.out.000003", "./node003/testB.out.000003_deleted");
      rename("./node004/testB.out.000004", "./node003/testB.out.000004_deleted");
      rename("node000/testB.out_BUDDY_00.000001", "node003/testB.out_BUDDY_00.000001");
      rename("node001/testB.out_BUDDY_00.000002", "node003/testB.out_BUDDY_00.000002");
      rename("node002/testB.out_BUDDY_00.000003", "node003/testB.out_BUDDY_00.000003");
      rename("node004/testB.out_BUDDY_00", "node003/testB.out_BUDDY_00");
    }

    /* filenum=rank/4; */
    if (rank < 2) {
      filenum = 0;
    } else if (rank < 5) {
      filenum = 1;
    } else if (rank < 9) {
      filenum = 2;
    } else if (rank < 14) {
      filenum = 3;
    } else {
      filenum = 4;
    }
    MPI_Comm_split(MPI_COMM_WORLD, filenum, rank, &lcomm2);
    MPI_Comm_size(lcomm2, &lsize);
    MPI_Comm_rank(lcomm2, &lrank);

    sprintf(dirname, "node%03d", filenum);
    MPI_Barrier(MPI_COMM_WORLD);
    chdir(dirname);
    MPI_Barrier(MPI_COMM_WORLD);

    sprintf(debugname, "%s.%05d", "deb_buddy_read_G", rank);
    sion_debug_on(1023, debugname);

    sion_mpi_options_set_chunksize(&options, CHUNKSIZE);
    sion_mpi_options_set_fsblksize(&options, 10);
    sion_mpi_options_set_buddy(&options);
    sion_mpi_options_set_multifile_communicator(&options, lcomm2);

    sid = sion_paropen_mpi_with_options("testB.out", SION_OPEN_READ, MPI_COMM_WORLD, &options);

    if (sid >= 0) {
      printf("on rank %2d: sid           =%d\n", rank, sid);
      ONRANK printf("on rank %2d: numfiles      =%d\n", rank, sion_get_number_of_files(sid));
      ONRANK printf("on rank %2d: chunksize     =%d\n", rank, (int)sion_get_chunksize(sid));
      ONRANK printf("on rank %2d: globalrank    =%d\n", rank, sion_get_globalrank(sid));
      ONRANK printf("on rank %2d: newfname      =%s\n", rank, sion_get_fname(sid));

      b = 0;
      sum = 0;
      while ((!sion_feof(sid))) {
        bytes_avail = sion_bytes_avail_in_block(sid);
        printf("on rank %2d: block=%d           bytes_avail=%d\n", rank, b, (int)bytes_avail);
        bytestoread = bytes_avail;
        bread = sion_coll_fread(buffer, 1, bytestoread, sid);
        for (i = 0; i < bread; i++) {
          sum = sum + buffer[i];
        }
        printf("on rank %2d: block=%d           bread=%3d overall blocksum=%8d \n", rank, b, (int)bread, (int)sum);
        b++;
      }

      sion_parclose_mpi(sid);
    } else {
      fprintf(stderr, "on rank %2d: error sid = %d\n", rank, sid);
    }

    chdir("..");

    MPI_Reduce(&sum, &gsum, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);
    ONRANK printf("on rank %2d: global blocksum = %d\n", rank, (int)gsum);

    if (rank == 0) {
      rename("./node000/testB.out_deleted", "./node000/testB.out");
      rename("./node001/testB.out.000001_deleted", "./node001/testB.out.000001");
      rename("./node002/testB.out.000002_deleted", "./node002/testB.out.000002");
      rename("./node003/testB.out.000003_deleted", "./node003/testB.out.000003");
      rename("./node003/testB.out.000004_deleted", "./node004/testB.out.000004");
      rename("node003/testB.out_BUDDY_00.000001", "node000/testB.out_BUDDY_00.000001");
      rename("node003/testB.out_BUDDY_00.000002", "node001/testB.out_BUDDY_00.000002");
      rename("node003/testB.out_BUDDY_00.000003", "node002/testB.out_BUDDY_00.000003");
      rename("node003/testB.out_BUDDY_00", "node004/testB.out_BUDDY_00");
    }
  }

  fflush(stderr);
  fflush(stdout);
  MPI_Barrier(MPI_COMM_WORLD);
  ONRANK printf("on rank %2d: END of TEST H read collective\n", rank);
  MPI_Barrier(MPI_COMM_WORLD);

  /* -------------------------- */
  /* EPILOG */
  /* -------------------------- */
  MPI_Finalize();

  return 0;
}
