/****************************************************************************
**  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       **
****************************************************************************/
/*
 * \file sion_internal_par.c
 * \date Sep 2, 2008
 * \author gstu0808
 *
 * \brief Internal Functions(parallel)
 */

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>

#include "sion.h"
#include "sion_fd.h"
#include "printts.h"

/*!\brief Generic parallel open of one direct access file.
 *
 * @param *fname                filename to use
 * @param *file_mode            like the type parameter of fopen: "rb", "wb"
 * @param *prefix               prefix to be used in case of multiple files
 * @param *numFiles             Number of files to open
 * @param *chunksize            chunksize on this task
 * @param *fsblksize            blocksize of filesystem (must be equal on all processors)
 * @param rank                  rank of the current process
 * @param ntasks                number of processes
 * @param *globalrank           any global unique id for this task, will be stored in sion file, usefull if comm is not MPI_COMM_WORLD
 *                              typical: globalrank= rank in MPI_COMM_WORLD
 * @param flag                  sion options flag
 * @param **fileptr             filepointer for this task
 * @param gather_cb             gather function to be used
 * @param scatter_cb            scatter function to be used
 * @param bcast_cb              broadcast function
 * @param barrier_cb            barrier function
 * @param *cb_parm1             dummy param 1 (parallel implementation dependent)
 * @param *cb_parm2             dummy param 2 (parallel implementation dependent)
 *
 * @return sid                  sion file handle
 *                              -1 if error occured
 */
int _sion_paropen_multi_generic(char *fname,
                                char *file_mode,
                                char *prefix,
                                int *numFiles,
                                int *filenumber,
                                sion_int64 *chunksize,
                                sion_int32 *fsblksize,
                                int rank,
                                int ntasks,
                                int *globalrank,
                                int flag,
                                FILE **fileptr,
                                int gather_cb(void *, void *, void *, int),
                                int scatter_cb(void *, void *, void *, int),
                                int bcast_cb(void *, void *, int), int barrier_cb(void *, void *), void *cb_parm1, void *cb_parm2)
{

  int       i, j, sid;
  int       size_f, rc;

  int       nfiles;
  sion_int32 lnfiles;
  int       filenum;
  sion_int32 lfilenum;
  sion_int64 *sion_chunksizes = NULL, lchunksize;
  sion_int64 *sion_startpointers = NULL, lstartpointer;
  sion_int64 *sion_globalranks = NULL, lglobalrank, lglobalskip;
  _sion_filedesc *sion_filedesc;
  sion_int32 lendianness, lswapbytes, maxblocks;
  sion_int64 lblocksize, lntasks, lsion_fileformat_version;
  sion_int64 start_of_varheader;
  int       nread;
  int       endianness, swapbytes;
  sion_int64  new_fsblocksize;
  sion_int64 helpint64;
  sion_int64 *sion_tmpintfield = NULL;
  sion_int32 sion_version, sion_version_patchlevel, sion_fileformat_version;
  sion_int64 flag1 = _SION_FLAG1_NONE, flag2 = _SION_FLAG1_NONE;

  DPRINTFP((2, "_sion_paropen_multi_generic", rank, "enter parallel open of file %s in %s mode #tasks=%d\n", fname, file_mode, ntasks));
  DPRINTFP((32, "_sion_paropen_multi_generic", rank, "sizeof: int=%d long=%d longlong=%d sion_int32=%d sion_int64=%d\n", sizeof(int), sizeof(long),
            sizeof(long long), sizeof(sion_int32), sizeof(sion_int64)));

  size_f = ntasks;
  nfiles = *numFiles;
  filenum = *filenumber;
  sid = -1;


  _sion_alloc_fd_list();

  sion_filedesc = (_sion_filedesc *) malloc(sizeof(_sion_filedesc));
  if (sion_filedesc == NULL) {
    return(_sion_errorprint(-1,_SION_ERROR_RETURN,"_sion_paropen_multi_generic: cannot allocate filedescriptor structure of size %lu (sion_filedesc), aborting ...\n", 
			    (unsigned long) sizeof(sion_filedesc)));
  }
  _sion_init_filedesc(sion_filedesc);

  sid = _sion_newvcd(sion_filedesc, SION_FILEDESCRIPTOR);
  sion_filedesc->maxblocks = -1;
  sion_filedesc->chunksizes = NULL;
  sion_filedesc->fsblksize = *fsblksize;
  _sion_realloc_blocklist(sion_filedesc, MAXBLOCKS);
  sion_filedesc->numblocks = 0;
  sion_filedesc->fname = strdup(fname);

#ifdef SION_MPI
  sion_filedesc->lComm = *((MPI_Comm *) cb_parm1);
  sion_filedesc->gComm = *((MPI_Comm *) cb_parm2);
#endif


  if (!strcmp(file_mode, "bw")) {
    /* **************** WRITE mode **************** */

    /*                                                                      */ DPRINTFTS(rank, "before alloc");
    if (rank == 0) {
      /* memory allocation */
      _sion_allocate_sion_arrays(ntasks, &sion_chunksizes, &sion_startpointers, &sion_globalranks);
    }
    /*                                                                      */ DPRINTFTS(rank, "after alloc");


    DPRINTFP((32, "_sion_paropen_multi_generic", rank, " starting open for write #tasks=%d\n", ntasks));

    /* open file */
    if (rank == 0) {
      *fileptr = _sion_create_and_open_file_for_write(fname);
      if(*fsblksize<=-1) {
	/* check with fstat fsblksize */
	new_fsblocksize=(sion_int64) _sion_get_opt_blksize(*fileptr);
	if((new_fsblocksize<0) || (new_fsblocksize>SION_MAX_FSBLOCKSIZE)) new_fsblocksize=SION_DEFAULT_FSBLOCKSIZE;
      }
      fclose(*fileptr);
    }
    barrier_cb(cb_parm1, cb_parm2);

    /* distribute new fsblksize */
    if(*fsblksize==-1) {
      bcast_cb(&new_fsblocksize, cb_parm1, _SION_INT64);
      *fsblksize=new_fsblocksize;
      sion_filedesc->fsblksize = *fsblksize;
      DPRINTFP((32, "_sion_paropen_multi_generic", rank, "setting fsblksize to %lld\n", new_fsblocksize));
    }

    if (flag & SION_DIFFSIZE) {
      /* collect data and init startpointers on PE 0 */
      lchunksize = (sion_int64) *chunksize;
      lglobalrank = *globalrank;

      /*                                                                      */ DPRINTFTS2(rank, "before gather");
      gather_cb(&lchunksize, sion_chunksizes, cb_parm1, _SION_INT64);

      gather_cb(&lglobalrank, sion_globalranks, cb_parm1, _SION_INT64);

      /*                                                                      */ DPRINTFTS2(rank, "after gather");
      if (rank == 0) {
        /*                                                                      */ DPRINTFTS(rank, "before calculate");
        _sion_calculate_startpointers(ntasks, sion_chunksizes, *fsblksize, sion_startpointers, &lglobalskip);
        /*                                                                      */ DPRINTFTS(rank, "after calculate");
      }
    }
    else {
      if (rank == 0) {
        /*                                                                      */ DPRINTFTS(rank, "before im. calculate");
        /* size and globalranks are implicit given on rank 0 */
        for (i = 0; i < ntasks; i++) {
          sion_chunksizes[i] = (sion_int64) *chunksize;
          sion_globalranks[i] = (sion_int64) i;
        }
        _sion_calculate_startpointers(ntasks, sion_chunksizes, *fsblksize, sion_startpointers, &lglobalskip);
        /*                                                                      */ DPRINTFTS(rank, "after im. calculate");
      }
    }

    /*                                                                      */ DPRINTFTS(rank, "before open");
    *fileptr = _sion_open_existing_file_for_write(fname);
    barrier_cb(cb_parm1, cb_parm2);
    /*                                                                      */ DPRINTFTS(rank, "after open");

    if (!*fileptr) {
      return(_sion_errorprint(-1,_SION_ERROR_RETURN,"_sion_paropen_multi_generic: cannot open %s for writing on task %d, aborting ...\n", fname, rank));
    }

    /* store data in static data structure (sid)  */
    sion_filedesc->fd = *fileptr;
    sion_filedesc->rank = rank;
    sion_filedesc->ntasks = ntasks;
    sion_filedesc->endianness = sion_get_endianness();  /* Endianness */
    sion_filedesc->nfiles = nfiles;
    sion_filedesc->filenumber = filenum;
    sion_filedesc->end_of_header = -1;
    sion_filedesc->start_of_data = -1;
    sion_filedesc->start_of_varheader = -1;
    sion_filedesc->state = SION_FILESTATE_PAROPEN;
    sion_filedesc->mode = SION_FILEMODE_WRITE;

    sion_filedesc->prefix = strdup(prefix);

    /* write header */
    if (rank == 0) {
      /*                                                                      */ DPRINTFTS(rank, "before writeh");
      _sion_write_header(*fileptr, *fsblksize, ntasks, sion_globalranks, sion_chunksizes, nfiles, filenum, flag1, flag2,
                         prefix, sion_filedesc->endianness);
      /*                                                                      */ DPRINTFTS(rank, "after writeh");

      /* needed for writing pointer to var part of metadata at the end of the file */
      sion_filedesc->end_of_header = _sion_get_position(*fileptr);
      sion_filedesc->start_of_data = sion_startpointers[0];
      /*set max. file size */
      lstartpointer = sion_startpointers[ntasks - 1];
      /*                                                                      */ DPRINTFTS(rank, "before setp(0)");
      _sion_flush_file_fp(*fileptr);
      _sion_set_position(*fileptr, lstartpointer);
      /*                                                                      */ DPRINTFTS(rank, "after setp(0)");

    }

    /* distribute start_pos */
    if (flag & SION_DIFFSIZE) {
      /*                                                                      */ DPRINTFTS(rank, "before scatter");
      scatter_cb(sion_startpointers, &lstartpointer, cb_parm1, _SION_INT64);
      /*                                                                      */ DPRINTFTS(rank, "after scatter");
    }

    /* distribute globalskip */
    bcast_cb(&lglobalskip, cb_parm1, _SION_INT64);
    sion_filedesc->globalskip = lglobalskip;

    DPRINTFP((32, "_sion_paropen_multi_generic", rank, " start position is %10lld %10.4f MB\n", lstartpointer, lstartpointer / 1024.0 / 1024.0));

    /*                                                                      */ DPRINTFTS(rank, "before setp");
    barrier_cb(cb_parm1, cb_parm2);
    _sion_flush_file_fp(*fileptr);
    _sion_set_position(*fileptr, lstartpointer);
    sion_filedesc->startpos = lstartpointer;
    sion_filedesc->chunksize = (sion_int64) *chunksize;

    barrier_cb(cb_parm1, cb_parm2);
    /*                                                                      */ DPRINTFTS(rank, "after setp");
    DPRINTFP((32, "_sion_paropen_multi_generic", rank, " ending open for write #tasks=%d filepos=%lld\n", ntasks, _sion_get_position(SIONFP)));

  }
  else {
    /* **************** READ mode **************** */
    if ((rank == 0) && (SION_DEBUG == 1))
      DPRINTFP((32, "_sion_paropen_multi_generic", rank, " starting open for read #tasks=%d\n", ntasks));

    /*                                                                      */ DPRINTFTS(rank, "before openR");
    *fileptr = _sion_open_file_for_read(fname);
    barrier_cb(cb_parm1, cb_parm2);
    /*                                                                      */ DPRINTFTS(rank, "after openR");
    if (!*fileptr) {
      return(_sion_errorprint(-1,_SION_ERROR_RETURN,"_sion_paropen_multi_generic: cannot open %s for reading, aborting ...\n", fname));
    }

    /* store data in static data structure (sid)  */
    sion_filedesc->fd = *fileptr;
    sion_filedesc->rank = rank;
    sion_filedesc->ntasks = ntasks;
    sion_filedesc->state = SION_FILESTATE_PAROPEN;
    sion_filedesc->mode = SION_FILEMODE_READ;
    sion_filedesc->nfiles = nfiles;


    if (rank == 0) {
      rc = _sion_read_header_fixed_part(*fileptr, &endianness, &swapbytes, fsblksize, &size_f, &nfiles, &filenum, &flag1, &flag2,
                                        prefix, &sion_version, &sion_version_patchlevel, &sion_fileformat_version);
      if (!rc) {
	return(_sion_errorprint(-1,_SION_ERROR_RETURN,"_sion_paropen_multi_generic: cannot read header from file %s, aborting ...\n", fname));
      }
      DPRINTFP((32, "_sion_paropen_multi_generic", rank,
                " read, after read of fix header part endianness=%d blksize=%d ntasks=%d\n", endianness, *fsblksize, size_f));
      lendianness = endianness;
      lswapbytes = swapbytes;
      lblocksize = *fsblksize;
      lntasks = size_f;
      lsion_fileformat_version = sion_fileformat_version;

      sion_filedesc->prefix = prefix;

      /*                                                                      */ DPRINTFTS(rank, "before alloc");
      /* memory allocation */
      _sion_allocate_sion_arrays(size_f, &sion_chunksizes, &sion_startpointers, &sion_globalranks);
      /*                                                                      */ DPRINTFTS(rank, "after alloc");

      rc = _sion_read_header_var_part(*fileptr, size_f, sion_globalranks, sion_chunksizes, &maxblocks, &start_of_varheader, swapbytes);

      if (!rc) {
	return(_sion_errorprint(-1,_SION_ERROR_RETURN,"_sion_paropen_multi_generic: cannot read header from file %s, aborting ...\n", fname));
      }
      _sion_calculate_startpointers(size_f, sion_chunksizes, *fsblksize, sion_startpointers, &lglobalskip);
      /*                                                                      */ DPRINTFTS(rank, "after calculate");
    }

    /* distribute globalskip */
    bcast_cb(&lglobalskip, cb_parm1, _SION_INT64);
    sion_filedesc->globalskip = lglobalskip;

    /* broadcast information read from file */
    bcast_cb(&lendianness, cb_parm1, _SION_INT32);
    bcast_cb(&lswapbytes, cb_parm1, _SION_INT32);
    bcast_cb(&lblocksize, cb_parm1, _SION_INT64);
    bcast_cb(&lntasks, cb_parm1, _SION_INT64);
    bcast_cb(&lsion_fileformat_version, cb_parm1, _SION_INT64);
    *fsblksize = lblocksize;

    if (rank == 0)
      lnfiles = nfiles;
    bcast_cb(&lnfiles, cb_parm1, _SION_INT32);
    sion_filedesc->nfiles = lnfiles;
    *numFiles = nfiles = lnfiles;

    if (rank == 0)
      lfilenum = filenum;
    bcast_cb(&lfilenum, cb_parm1, _SION_INT32);
    sion_filedesc->filenumber = lfilenum;
    *filenumber = filenum = lfilenum;

    bcast_cb(&flag1, cb_parm1, _SION_INT64);
    sion_filedesc->flag1 = flag1;
    bcast_cb(&flag2, cb_parm1, _SION_INT64);
    sion_filedesc->flag2 = flag2;


    if (rank == 0)
      helpint64 = maxblocks;
    bcast_cb(&helpint64, cb_parm1, _SION_INT64);
    maxblocks = helpint64;

    if (maxblocks > MAXBLOCKS)
      _sion_realloc_blocklist(sion_filedesc, MAXBLOCKS);

    /*                                                                      */ DPRINTFTS(rank, "after bcast");

    /*                                                                      */ DPRINTFTS(rank, "before scatter");
    scatter_cb(sion_chunksizes, &lchunksize, cb_parm1, _SION_INT64);
    scatter_cb(sion_globalranks, &lglobalrank, cb_parm1, _SION_INT64);
    scatter_cb(sion_startpointers, &lstartpointer, cb_parm1, _SION_INT64);
    /*                                                                      */ DPRINTFTS(rank, "after scatter");

    if (rank == 0) {
      sion_tmpintfield = (sion_int64 *) malloc(ntasks * sizeof(sion_int64));
      if (sion_tmpintfield == NULL) {
	return(_sion_errorprint(-1,_SION_ERROR_RETURN,"_sion_paropen_multi_generic: cannot allocate temporary memory of size %lu (sion_tmpintfield), aborting ...\n",
				(unsigned long) ntasks * sizeof(sion_int64)));
      }
    }

    /* read number of blocks for each task */
    if (rank == 0) {

      _sion_purge_file_fp(SIONFP);
      _sion_set_position(SIONFP, start_of_varheader);

      nread = fread(sion_tmpintfield, sizeof(sion_int64), ntasks, SIONFP);
      if (nread != ntasks) {
	return(_sion_errorprint(-1,_SION_ERROR_RETURN,"_sion_paropen_multi_generic: cannot read numblocks from meta block 2 of the file, aborting ... (%d,%d)\n", ntasks, nread));
      }
      sion_swap(sion_tmpintfield, sion_tmpintfield, sizeof(sion_int64), ntasks, swapbytes);

      for (j = 0; j < ntasks; j++)
        DPRINTFP((32, "_sion_paropen_multi_generic", rank, " read, numblocks on task %02d is %10lld\n", j, sion_tmpintfield[j]));
    }

    /* and distribute them */
    /*                                                                      */ DPRINTFTS(rank, "before scatter");
    scatter_cb(sion_tmpintfield, &helpint64, cb_parm1, _SION_INT64);
    /*                                                                      */ DPRINTFTS(rank, "after scatter");
    sion_filedesc->numblocks = helpint64;

    for (i = 0; i < maxblocks; i++) {
      if (rank == 0) {
        nread = fread(sion_tmpintfield, sizeof(sion_int64), ntasks, SIONFP);
        for (j = 0; j < ntasks; j++)
          DPRINTFP((32, "_sion_paropen_multi_generic", rank, " read, total chunksize on task %02d for block %2d: %10lld\n", j, i,
                    sion_tmpintfield[j]));

        if (nread != ntasks) {
	  return(_sion_errorprint(-1,_SION_ERROR_RETURN,"_sion_paropen_multi_generic: cannot read chunksizes[%d] from meta block 2 of file, aborting ... (%d)\n", i, ntasks));
        }
        sion_swap(sion_tmpintfield, sion_tmpintfield, sizeof(sion_int64), ntasks, swapbytes);

      }

      /*                                                                      */ DPRINTFTS(rank, "before scatter");
      scatter_cb(sion_tmpintfield, &helpint64, cb_parm1, _SION_INT64);
      /*                                                                      */ DPRINTFTS(rank, "after scatter");
      sion_filedesc->chunksizes[i] = helpint64;
    }

#define BGFLUSH
#ifdef BGFLUSH
    _sion_flush_file_fp(*fileptr);
#endif

    /*                                                                      */ DPRINTFTS(rank, "before setp");
    barrier_cb(cb_parm1, cb_parm2);
    /*     fprintf(stderr,"[SION]               setp on rank %d to %lld\n",rank,lstartpointer);  */
    _sion_purge_file_fp(*fileptr);
    _sion_set_position(*fileptr, lstartpointer);
    sion_filedesc->startpos = lstartpointer;
    sion_filedesc->chunksize = lchunksize;
    sion_filedesc->globalskip = lglobalskip;
    sion_filedesc->currentpos = lstartpointer;
    sion_filedesc->currentblocknr = 0;
    sion_filedesc->endianness = lendianness;
    sion_filedesc->swapbytes = lswapbytes;
    sion_filedesc->fileversion = lsion_fileformat_version;
    *chunksize=lchunksize;
    *globalrank=lglobalrank;

    barrier_cb(cb_parm1, cb_parm2);
    /*                                                                      */ DPRINTFTS(rank, "after setp");

  } /* read */

  _sion_deallocate_sion_arrays(&sion_chunksizes, &sion_startpointers, &sion_globalranks);

  _sion_print_filedesc(sion_filedesc, 32, "_sion_paropen_multi_generic", 1);

  DPRINTFP((32, "_sion_paropen_multi_generic", rank, " start position on task %02d is at end of sion_paropen_generic %10lld\n", rank,
            _sion_get_position(SIONFP)));

  DPRINTFP((2, "_sion_paropen_multi_generic", rank, "leave parallel open of file %s in %s mode #tasks=%d\n", fname, file_mode, ntasks));

  return (sid);

}                               /*  END OF _sion_paropen_multi_generic */


/*!\brief Internal function for closing the sion file in parallel on all tasks
 *
 * @param sid                   sion file handle
 * @param rank                  rank of the current process
 * @param ntasks                number of processes
 * @param mapping               mapping of tasks to filenumber if opened for write and nfiles>1
 * @param gather_cb             gather function to be used
 * @param scatter_cb    scatter function to be used
 * @param bcast_cb              broadcast function
 * @param barrier_cb    barrier function
 * @param cb_parm1              ??
 * @param cb_parm2              ??
 *
 * @return rc                   1 if close is ok
 */
int _sion_parclose_generic(int sid,
                           int rank,
                           int ntasks,
                           int mapping_size,
                           sion_int32 *mapping,
                           int gather_cb(void *, void *, void *, int),
                           int scatter_cb(void *, void *, void *, int),
                           int bcast_cb(void *, void *, int), int barrier_cb(void *, void *), void *cb_parm1, void *cb_parm2)
{

  int       rc = 1;
  int       i, nwrite, lrank;
  sion_int32 maxblocks;
  /*   char marker[4]="sion"; */
  sion_int64 helpint64;
  sion_int64 *sion_tmpintfield = NULL;
  sion_int64 start_of_varheader;
  _sion_filedesc *sion_filedesc;

  if ((_sion_vcdtype(sid) != SION_FILEDESCRIPTOR) || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    return(_sion_errorprint(0,_SION_ERROR_RETURN,"_sion_parclose_generic: invalid sion_filedesc, aborting %d ...\n", sid));
  }

  if (sion_filedesc->state != SION_FILESTATE_PAROPEN) {
    return(_sion_errorprint(0,_SION_ERROR_RETURN,"_sion_parclose_generic: sion file with sid=%d was not opened by a sion_paropen\n", sid));
  }

  DPRINTFP((2, "_sion_parclose_generic", rank, "enter parallel close  sid=%d\n", sid));

  DPRINTFP((1, "_sion_parclose_generic", rank, " Communicator: %x\n", *(int *) cb_parm1));

  /* READ MODE: close file on all tasks  */
  if (sion_filedesc->mode == SION_FILEMODE_READ) {
    if (sion_filedesc->state != SION_FILESTATE_CLOSE) {

      _sion_print_filedesc(sion_filedesc, 32, "_sion_parclose_generic", 1);
      DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close (read mode)  sid=%d, call fclose on file\n", sid));

      fclose(sion_filedesc->fd);
      sion_filedesc->fd = NULL;
      sion_filedesc->state = SION_FILESTATE_CLOSE;
    }
    return (rc);
  }
  else {

    /* WRITE MODE: collect data from all tasks, write metadata, and close file on all tasks  */

    _sion_flush_block(sid);

    _sion_print_filedesc(sion_filedesc, 32, "_sion_parclose_generic", 1);

    /* close file on all other task, except 0  */
    if (rank != 0) {
      if (sion_filedesc->state != SION_FILESTATE_CLOSE) {
        DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close (write mode)  sid=%d, call fclose on file\n", sid));
        fclose(sion_filedesc->fd);
        sion_filedesc->fd = NULL;
        sion_filedesc->state = SION_FILESTATE_CLOSE;
      }
    }

    barrier_cb(cb_parm1, cb_parm2);

    DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close sid=%d: numblocks=%d globalskip=%lld\n", sid, sion_filedesc->numblocks,
              sion_filedesc->globalskip));

    for (i = 0; i < sion_filedesc->numblocks; i++) {
      DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close sid=%d: local block %02d -> %10lld bytes\n", sid, i,
                sion_filedesc->chunksizes[i]));
    }

    if (rank == 0) {
      sion_tmpintfield = (sion_int64 *) malloc(ntasks * sizeof(sion_int64));
      if (sion_tmpintfield == NULL) {
	return(_sion_errorprint(0,_SION_ERROR_RETURN,"_sion_parclose_generic: cannot allocate temporary memory of size %lu (sion_tmpintfield), aborting ...\n",
				(unsigned long) ntasks * sizeof(sion_int64)));
      }
    }
    /* gather number of blocks of each tasks */
    helpint64 = sion_filedesc->numblocks + 1;
    /*                                                                      */ DPRINTFTS2(rank, "before gather");
    gather_cb(&helpint64, sion_tmpintfield, cb_parm1, _SION_INT64);
    /*                                                                      */ DPRINTFTS2(rank, "after gather");

    /* reposition file on rank 0 */
    if (rank == 0) {
      DPRINTFP((32, "_sion_paropen_generic", rank, " parallel close sid=%d: setpos to %lld\n", sid, sion_filedesc->end_of_header));
      _sion_flush_file_fp(sion_filedesc->fd);
      _sion_set_position(sion_filedesc->fd, sion_filedesc->end_of_header);
      maxblocks = -1;
      for (i = 0; i < ntasks; i++)
        if (sion_tmpintfield[i] > maxblocks)
          maxblocks = (sion_int32) sion_tmpintfield[i];

      DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close sid=%d: max. number of blocks is %d\n", sid, maxblocks));
      DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close sid=%d: filepos before write maxblocks %lld\n", sid,
                _sion_get_position(sion_filedesc->fd)));

      nwrite = fwrite(&maxblocks, 1, sizeof(maxblocks), sion_filedesc->fd);
      if (nwrite != sizeof(maxblocks)) {
	return(_sion_errorprint(0,_SION_ERROR_RETURN,"_sion_parclose_generic: cannot write header (maxblocks) to file, aborting ... (%d)\n", nwrite));
      }
      DPRINTFP((32, "_sion_paropen_generic", rank, " parallel close sid=%d: filepos after write maxblocks %lld nwrite=%d\n", sid,
                _sion_get_position(sion_filedesc->fd), nwrite));

      start_of_varheader = sion_filedesc->start_of_data + maxblocks * sion_filedesc->globalskip;

      nwrite = fwrite(&start_of_varheader, 1, sizeof(sion_int64), sion_filedesc->fd);
      if (nwrite != sizeof(sion_int64)) {
	return(_sion_errorprint(0,_SION_ERROR_RETURN,"_sion_parclose_generic: cannot write header (start_of_varheader) to file, aborting ... (%d)\n", nwrite));
      }
      DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close sid=%d: filepos after write start_of_varheader %lld nwrite=%d\n", sid,
                _sion_get_position(sion_filedesc->fd), nwrite));

      /*     nwrite=fwrite(marker,sizeof(char),4,sion_filedesc->fd); */

      _sion_flush_file_fp(sion_filedesc->fd);
      _sion_set_position(sion_filedesc->fd, start_of_varheader);

      DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close sid=%d: filepos after move to start_of_varheader %lld\n", sid,
                _sion_get_position(sion_filedesc->fd)));

      nwrite = fwrite(sion_tmpintfield, sizeof(sion_int64), ntasks, sion_filedesc->fd);
      if (nwrite != ntasks) {
	return(_sion_errorprint(0,_SION_ERROR_RETURN,"_sion_parclose_generic: cannot write numblocks[] field to meta block 2 of the file, aborting ... (%d)\n", ntasks));
      }
      DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close sid=%d: filepos after write numblocks[] %lld nwrite=%d\n", sid,
                _sion_get_position(sion_filedesc->fd), nwrite));

      /* for distribute data */
      helpint64 = maxblocks;

    }

    /* distribute maxblocks to all tasks */
    bcast_cb(&helpint64, cb_parm1, _SION_INT64);
    maxblocks = helpint64;

    /* collect chunksizes of each block from each task and write it to file */
    for (i = 0; i < maxblocks; i++) {
      if (i <= sion_filedesc->numblocks) {
        helpint64 = sion_filedesc->chunksizes[i];
      }
      else {
        helpint64 = 0;
      }

      /*                                                                      */ DPRINTFTS2(rank, "before gather");
      gather_cb(&helpint64, sion_tmpintfield, cb_parm1, _SION_INT64);
      /*                                                                      */ DPRINTFTS2(rank, "after gather");

      if (rank == 0) {
        for (lrank = 0; lrank < ntasks; lrank++)
          DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close sid=%d: write total chunksize for block %d: %2lld rank=%d\n", sid, i,
                    sion_tmpintfield[lrank], lrank));

        nwrite = fwrite(sion_tmpintfield, sizeof(sion_int64), ntasks, sion_filedesc->fd);
        if (nwrite != ntasks) {
	  return(_sion_errorprint(0,_SION_ERROR_RETURN,"_sion_parclose_generic: cannot write blocksizes[%d] to varheader of file, aborting ... (%d)\n", i, ntasks));
        }
        DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close sid=%d: filepos after write blocksizes[%d] %lld nwrite=%d\n", sid, i,
                  _sion_get_position(sion_filedesc->fd), nwrite));
      }
    }

    /* write mapping to file if more than one physical file is used, mapping_size is the number of global tasks */
    if (rank == 0) {
      /* mapping_size will be written in each case, is 0 if (number of physical file is one) */
      nwrite = fwrite(&mapping_size, sizeof(sion_int32), 1, sion_filedesc->fd);
      if (nwrite != 1) {
	return(_sion_errorprint(0,_SION_ERROR_RETURN,"_sion_parclose_generic: cannot write mapping_size to varheader of file, aborting ... (%d)\n", 1));
      }
      if (mapping_size > 0) {
        nwrite = fwrite(mapping, sizeof(sion_int32), 2 * mapping_size, sion_filedesc->fd);
        if (nwrite != 2*mapping_size) {
 	  return(_sion_errorprint(0,_SION_ERROR_RETURN,"_sion_parclose_generic: cannot write mapping to varheader of file, aborting ... (%d)\n", 2*mapping_size));
        }
        for (lrank = 0; lrank < mapping_size; lrank++) {
          DPRINTFP((32, "_sion_parclose_generic", rank, " mapping[%4d]=(%10ld,%10ld)\n", lrank,mapping[lrank*2],mapping[lrank*2+1]));
	}

        DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close sid=%d: filepos after write mapping %lld nwrite=%d\n", sid,
                  _sion_get_position(sion_filedesc->fd), nwrite));
      }
    }

    /* close file on task 0  */
    if (rank == 0) {
      DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close (write mode)  sid=%d, call fclose on file\n", sid));
      fclose(sion_filedesc->fd);
      sion_filedesc->fd = NULL;
      sion_filedesc->state = SION_FILESTATE_CLOSE;
    }

  }                             /* write */

  _sion_freevcd(sid);
  _sion_free_filedesc(sion_filedesc);
  sion_filedesc = NULL;


  DPRINTFP((2, "_sion_parclose_generic", rank, "leave parallel close  sid=%d\n", sid));

  return (rc);
}                               /*  END OF _sion_parclose_generic */
