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

#define _XOPEN_SOURCE 700

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

#include "sion_buffer.h"
#include "sion_common.h"
#include "sion_const.h"
#include "sion_debug.h"
#include "sion_error_handler.h"
#include "sion_fd.h"
#include "sion_file.h"
#include "sion_filedesc.h"
#include "sion_generic_buddy.h"
#include "sion_generic_collective.h"
#include "sion_generic_internal.h"
#include "sion_hints.h"
#include "sion_internal.h"
#include "sion_internal_startptr.h"
#include "sion_keyvalue.h"
#include "sion_metadata.h"

int _sion_paropen_generic_one_file(int sid, const char *fname, sion_open_mode mode, int numfiles, int filenumber, int rank,
  int ntasks, int globalrank, int flag, _sion_generic_gendata *gendata, _sion_generic_buddy *buddy,
  const sion_generic_options *options)
{
  int64_t lchunksize, lstartpointer, lglobalrank, new_fsblocksize, helpint64, apiflag;
  int64_t *sion_tmpintfield = NULL;
  int32_t *sion_tmpintfield_map = NULL, helpint32;
  int32_t *sion_tmpintfield_buddy32 = NULL;
  int64_t *sion_tmpintfield_buddy64 = NULL;

  switch (options->lowlevel_api) {
  case SION_LOWLEVEL_API_POSIX:
    apiflag = SION_FILE_FLAG_POSIX;
    break;
  case SION_LOWLEVEL_API_C: // FALLTHROUGH
  default:
    apiflag = SION_FILE_FLAG_ANSI;
  }

  // FIXME re-enable debug output of file_mode
  // DPRINTFP((2, "_sion_paropen_generic_one_file", rank, "enter parallel open of file %s in mode %d #tasks=%d\n", fname,
  //  (int)flags_store->mask, ntasks));
  DPRINTFP((2, "_sion_paropen_generic_one_file", rank, "enter parallel open of file %s #tasks=%d\n", fname, ntasks));
  DPRINTFP((32, "_sion_paropen_generic_one_file", rank, "sizeof: int=%d long=%d longlong=%d int32_t=%d int64_t=%d\n", sizeof(int),
    sizeof(long), sizeof(long long), sizeof(int32_t), sizeof(int64_t)));

  /* select local communicator */
  void *comm_group = NULL;
  if (flag & _SION_INTERNAL_FLAG_NORMAL) {
    comm_group = gendata->comm_data_local;
  }
  if (flag & _SION_INTERNAL_FLAG_BUDDY_NORMAL) {
    comm_group = gendata->comm_data_local;
  }
  if (flag & _SION_INTERNAL_FLAG_BUDDY_SEND) {
    comm_group = buddy->buddy_send.commgroup;
  }
  if (flag & _SION_INTERNAL_FLAG_BUDDY_COLL) {
    comm_group = buddy->buddy_coll.commgroup;
  }
  if (flag & _SION_INTERNAL_FLAG_BUDDY_READ) {
    comm_group = buddy->groups[buddy->currentgroup]->commgroup;
  }

  /* decide if file has to be opened physically  */
  int do_open_file = 1;
  if (flag & _SION_INTERNAL_FLAG_BUDDY_SEND) {
    do_open_file = 0;
  }
  if ((flag & _SION_INTERNAL_FLAG_BUDDY_COLL) && (rank > 0)) {
    do_open_file = 0;
  }
  if ((flag & _SION_INTERNAL_FLAG_BUDDY_READ) && (rank > 0)) {
    do_open_file = 0;
  }
  /* WF: todo decide if data is local, these tasks can read data directly? */
  DPRINTFP((2, "_sion_paropen_generic_one_file", rank, "do_open_file=%d\n", do_open_file));

  _sion_filedesc *filedesc = _sion_alloc_filedesc();
  if (!filedesc) {
    _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_ABORT,
      "_sion_paropen_omp: cannot allocate filedescriptor structure of size %lu (_filedesc), aborting ...\n",
      (unsigned long)sizeof(filedesc));
  }
  _sion_init_filedesc(filedesc);
  filedesc->fname = strdup(fname); /* Set the filename */

  _sion_reassignvcd(sid, filedesc, SION_FILEDESCRIPTOR);
  filedesc->sid = sid;

  /* store generic communication data */
  filedesc->dataptr = gendata;

  /* Allocate memory for storing MAXCHUNKS chunksize infos in internal structure */
  _sion_realloc_filedesc_blocklist(filedesc, MAXCHUNKS);
  filedesc->lastchunknr = 0; /* Set the current number of chunks */
  filedesc->currentblocknr = 0; /* Set the current block number */

  if (mode == SION_OPEN_WRITE) {
    /* **************** WRITE mode **************** */

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

    /* check parameter */
    if (ntasks < 0) {
      return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
        "sion_paropen: wrong number of tasks specific: ntasks=%d (<0), returning ...\n", (int)ntasks);
    }

    /* check parameter */
    if ((flag & _SION_INTERNAL_FLAG_NORMAL) && (globalrank < 0)) {
      return _sion_errorprint(
        SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen: ((globalrank != NULL) && (*globalrank<0)), returning ...\n");
    }

    filedesc->state = SION_FILESTATE_PAROPEN;
    filedesc->mode = SION_FILEMODE_WRITE;
    filedesc->endianness = _sion_get_endianness_with_override(options->endianness); /* Endianness */
    filedesc->swapbytes = 0; /* Endianness, swapping bytes */
    filedesc->fsblksize = options->fsblksize;
    filedesc->rank = rank;
    filedesc->globalrank = globalrank;
    filedesc->ntasks = ntasks;
    filedesc->nfiles = numfiles;
    filedesc->filenumber = filenumber;
    filedesc->usecoll = !!options->collsize;
    filedesc->collmergemode = options->collective_merge;
    filedesc->usebuddy = options->buddylevel > 0;
    if (filedesc->usebuddy) {
      filedesc->buddylevel = options->buddylevel;
    }

    /* open file on rank 0, first time to create file and get fsblksize if necessary */
    if (rank == 0) {
      _sion_fileptr *fileptr = _sion_file_open(fname, apiflag | SION_FILE_FLAG_WRITE | SION_FILE_FLAG_CREATE, 0);
      if (!fileptr) {
        return _sion_errorprint(
          SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_generic: cannot open %s for writing, aborting ...\n", fname);
      }
      if (options->fsblksize < 0) {
        /* check with fstat fsblksize */
        new_fsblocksize = (int64_t)_sion_file_get_opt_blksize(fileptr);
        if ((new_fsblocksize < 0) || (new_fsblocksize > SION_MAX_FSBLOCKSIZE)) {
          new_fsblocksize = SION_DEFAULT_FSBLOCKSIZE;
        }
      }
      _sion_file_close(fileptr);
    }
    gendata->apidesc->barrier_cb(comm_group);
    /* printf("WF: rank=%2d after first barrier fsblksize=%d\n",rank,(int) *fsblksize); */

    /* distribute new fsblksize */
    if (options->fsblksize == -1) {
      gendata->apidesc->bcastr_cb(&new_fsblocksize, comm_group, _SION_INT64, 1, 0);
      filedesc->fsblksize = new_fsblocksize;
      DPRINTFP((32, "_sion_paropen_generic_one_file", rank, "setting fsblksize to %lld\n", new_fsblocksize));
    }

    /* check for buffer, needed at this point to set flag before writing header */
    _sion_buffer_check_env(filedesc);

    /* check for keyval parameter, needed at this point to set flag before writing header (flag1) */
    if (rank == 0) {
      _sion_keyval_check_env_with_options(filedesc, mode, options->keyval_mode);
    }
    gendata->apidesc->bcastr_cb(&filedesc->keyvalmode, comm_group, _SION_INT32, 1, 0);

    /* check for collective options */
    if (rank == 0) {
      if (options->collsize > 0) {
        filedesc->collsize = options->collsize;
      }
      _sion_coll_check_env(filedesc);
    }

    if ((flag & _SION_INTERNAL_FLAG_BUDDY_NORMAL) || (flag & _SION_INTERNAL_FLAG_BUDDY_SEND)
      || (flag & _SION_INTERNAL_FLAG_BUDDY_COLL)) {
      /* overwrite collective info if not set*/
      if (!filedesc->usecoll) {
        filedesc->usecoll = 1;
        filedesc->collsize = filedesc->ntasks;
      };
    }

    gendata->apidesc->bcastr_cb(&filedesc->usecoll, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->collsize, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->collmergemode, comm_group, _SION_INT32, 1, 0);

    /* check if API support coalescing I/O */
    if (filedesc->usecoll) {
      if (gendata->apidesc->level != SION_GENERIC_API_LEVEL_FULL) {
        _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_WARN,
          "sion_paropen_generic: requested coalescing I/O but API does not support this mode, falling back to individual mode "
          "...\n");
        filedesc->usecoll = 0;
      }
    }

    /* check for hints options */
    if (rank == 0) {
      _sion_hints_check_env(filedesc);
    }
    gendata->apidesc->bcastr_cb(&filedesc->usehints, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->hinttype, comm_group, _SION_INT32, 1, 0);

    DPRINTFTS(rank, "before alloc");
    if (rank == 0) {
      /* memory allocation for internal fields */
      _sion_alloc_filedesc_arrays(filedesc);
      if (filedesc->usecoll) {
        _sion_alloc_filedesc_coll_arrays(filedesc);
      }
    }
    DPRINTFTS(rank, "after alloc");

    /* collect data and init startpointers on PE 0 */
    lchunksize = (options->chunksize >= 0) ? options->chunksize : filedesc->fsblksize;
    lglobalrank = globalrank;
    filedesc->chunksize_req = lchunksize;
    DPRINTFP((32, "_sion_paropen_generic_one_file", rank, "lchunksize=%lld lglobalrank=%lld\n", lchunksize, lglobalrank));

    DPRINTFTS2(rank, "before gather");
    gendata->apidesc->gatherr_cb(&lchunksize, filedesc->all_chunksizes, comm_group, _SION_INT64, 1, 0);
    gendata->apidesc->gatherr_cb(&lglobalrank, filedesc->all_globalranks, comm_group, _SION_INT64, 1, 0);

    /* check capability of tasks */
    if (filedesc->usecoll) {
      filedesc->coll_capability = gendata->apidesc->get_capability_cb(comm_group);
      gendata->apidesc->gatherr_cb(&filedesc->coll_capability, filedesc->all_coll_capability, comm_group, _SION_INT32, 1, 0);
    }

    /* collect mapping information on rank 0 of first file, mapping=NULL for all others */
    int mapping_size = 0;
    int32_t *mapping = NULL;
    if (flag & _SION_INTERNAL_FLAG_NORMAL || flag & _SION_INTERNAL_FLAG_BUDDY_NORMAL) {
      _sion_generic_collect_mapping(filedesc, &mapping_size, &mapping);
    } else {
      _sion_generic_collect_mapping_buddy(buddy, gendata, &mapping_size, &mapping);
    }

    DPRINTFTS2(rank, "after gather");
    if (rank == 0) {
      DPRINTFTS(rank, "before calculate");
      DPRINTFP((32, "_sion_paropen_generic_one_file", rank, "chunksizes[%d - 1]=%ld\n", ntasks,
        (long)filedesc->all_chunksizes[ntasks - 1]));
      if (!filedesc->usecoll) {
        _sion_calculate_startpointers(filedesc, mapping_size);
      } else {
        if (!filedesc->collmergemode) {
          _sion_calculate_startpointers_collective(filedesc, mapping_size);
        } else {
          _sion_calculate_startpointers_collective_merge(filedesc, mapping_size);
        }
      }
      DPRINTFTS(rank, "after calculate");
    }

    DPRINTFTS(rank, "before open");
    /* open not file on non-collector task if buddy-checkpointing */
    if (do_open_file) {
      /* open file on all ranks */
      _sion_fileptr *fileptr = _sion_file_open(fname, apiflag | SION_FILE_FLAG_WRITE, 0);
      if (!fileptr) {
        return _sion_errorprint(
          SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_generic: cannot open %s for writing, aborting ...\n", fname);
      }
      /* store data in static data structure (sid)  */
      filedesc->fileptr = fileptr;
    }
    gendata->apidesc->barrier_cb(comm_group);
    DPRINTFTS(rank, "after open");

    /* write header */
    if (rank == 0) {
      /* apply hint for first meta data block */
      _sion_apply_hints(filedesc, SION_HINTS_ACCESS_TYPE_METADATABLOCK1);

      DPRINTFTS(rank, "before writeh");
      _sion_write_header(filedesc);
      DPRINTFTS(rank, "after writeh");

      /* needed for writing pointer to var part of metadata at the end of the file */
      filedesc->end_of_header = _sion_file_get_position(filedesc->fileptr);
      filedesc->start_of_data = filedesc->all_startpointers[0];

      /* write mapping info to meta data block */
      _sion_write_header_var_part_mapping(filedesc, mapping_size, mapping);

      /*set max. file size */
      lstartpointer = filedesc->all_startpointers[ntasks - 1] + filedesc->all_chunksizes[ntasks - 1];
      DPRINTFTS(rank, "before setp(0)");
      _sion_file_set_position(filedesc->fileptr, lstartpointer);
      DPRINTFTS(rank, "after setp(0)");
    }

    free(mapping);

    /* distribute start_pos */
    DPRINTFTS(rank, "before scatter");
    gendata->apidesc->scatterr_cb(filedesc->all_startpointers, &filedesc->startpos, comm_group, _SION_INT64, 1, 0);
    DPRINTFTS(rank, "after scatter");

    /* distribute chunksize */
    DPRINTFTS(rank, "before scatter");
    gendata->apidesc->scatterr_cb(filedesc->all_chunksizes, &filedesc->chunksize, comm_group, _SION_INT64, 1, 0);
    DPRINTFTS(rank, "after scatter");

    /* distribute information for collective operations */
    if (filedesc->usecoll) {
      gendata->apidesc->scatterr_cb(filedesc->all_coll_collsize, &filedesc->collsize, comm_group, _SION_INT32, 1, 0);
      gendata->apidesc->scatterr_cb(filedesc->all_coll_collector, &filedesc->collector, comm_group, _SION_INT32, 1, 0);

      _sion_free_filedesc_coll_arrays(filedesc);
    }

    /* distribute globalskip */
    gendata->apidesc->bcastr_cb(&filedesc->globalskip, comm_group, _SION_INT64, 1, 0);

    DPRINTFP((32, "_sion_paropen_generic_one_file", rank, " start position is %10lld %10.4f MB chunksize=%10lld %10.4f MB\n",
      filedesc->startpos, filedesc->startpos / 1024.0 / 1024.0, filedesc->chunksize, filedesc->chunksize / 1024.0 / 1024.0));

    /* set filepointer on each task */
    DPRINTFTS(rank, "before setp");
    gendata->apidesc->barrier_cb(comm_group);
    if (do_open_file) {
      _sion_file_set_position(filedesc->fileptr, filedesc->startpos);
    }
    filedesc->currentpos = filedesc->startpos;
    /* given by calculate startpointers ...
       filedesc->chunksize = (int64_t) *chunksize;
    */
    gendata->apidesc->barrier_cb(comm_group);

    /* apply hint for first chunk */
    _sion_apply_hints(filedesc, SION_HINTS_ACCESS_TYPE_CHUNK);

    DPRINTFTS(rank, "after setp");
    DPRINTFP((32, "_sion_paropen_generic_one_file", rank, " ending open for write #tasks=%d filepos=%lld\n", ntasks,
      _sion_file_get_position(filedesc->fileptr)));

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

    /* open not file on non-collector task if buddy-checkpointing */
    _sion_fileptr *fileptr;
    if (do_open_file) {
      DPRINTFTS(rank, "before openR");
      fileptr = _sion_file_open(fname, apiflag | SION_FILE_FLAG_READ, 0);
      DPRINTFTS(rank, "after openR");
      if (!fileptr) {
        DPRINTFP((32, "_sion_paropen_generic_one_file", rank, " cannot open %s for reading, aborting ...\n", fname));
        return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
          "_sion_paropen_generic_one_file: cannot open %s for reading, aborting ...\n", fname);
      }
    } else {
      fileptr = NULL;
    }
    gendata->apidesc->barrier_cb(comm_group);

    /* store data in static data structure (sid)  */
    filedesc->fileptr = fileptr;
    filedesc->rank = rank;
    filedesc->ntasks = ntasks;
    filedesc->state = SION_FILESTATE_PAROPEN;
    filedesc->mode = SION_FILEMODE_READ;
    filedesc->nfiles = numfiles;
    filedesc->usebuddy = options->buddylevel > 0;
    if (filedesc->usebuddy) {
      filedesc->buddylevel = options->buddylevel;
    }

    /* creating of mapping from file ranks to rank used in buddy read */
    if (flag & _SION_INTERNAL_FLAG_BUDDY_READ) {
      /* overwrite collective info if not set*/
      if (!filedesc->usecoll) {
        filedesc->usecoll = 1;
        filedesc->collsize = filedesc->ntasks;
      }

      DPRINTFP((
        32, "_sion_paropen_generic_one_file", rank, " create buddy mapping ntasks=%d filentasks=%d\n", ntasks, filedesc->ntasks));

      if (rank == 0) {
        sion_tmpintfield_buddy32 = malloc(ntasks * sizeof(int32_t));
        if (sion_tmpintfield_buddy32 == NULL) {
          return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
            "_sion_paropen_generic_one_file: cannot allocate temporary memory of size %lu (sion_tmpintfield_buddy), aborting "
            "...\n",
            (unsigned long)ntasks * sizeof(int32_t));
        }
        for (int j = 0; j < ntasks; j++) {
          sion_tmpintfield_buddy32[j] = -1;
        }
        sion_tmpintfield_buddy64 = malloc(ntasks * sizeof(int64_t));
        if (sion_tmpintfield_buddy64 == NULL) {
          free(sion_tmpintfield_buddy32);
          return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
            "_sion_paropen_generic_one_file: cannot allocate temporary memory of size %lu (sion_tmpintfield_buddy), aborting "
            "...\n",
            (unsigned long)ntasks * sizeof(int64_t));
        }
        for (int j = 0; j < ntasks; j++) {
          sion_tmpintfield_buddy64[j] = -1;
        }
        sion_tmpintfield_map = malloc(ntasks * sizeof(int32_t));
        if (sion_tmpintfield_map == NULL) {
          free(sion_tmpintfield_buddy32);
          free(sion_tmpintfield_buddy64);
          return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
            "_sion_paropen_generic_one_file: cannot allocate temporary memory of size %lu (sion_tmpintfield_map), aborting ...\n",
            (unsigned long)ntasks * sizeof(int32_t));
        }
        for (int j = 0; j < ntasks; j++) {
          sion_tmpintfield_map[j] = -1;
        }
      }
      helpint32 = buddy->groups[buddy->currentgroup]->filelrank;
      gendata->apidesc->gatherr_cb(&helpint32, sion_tmpintfield_map, comm_group, _SION_INT32, 1, 0);

      if (rank == 0) {
        for (int j = 0; j < ntasks; j++) {
          DPRINTFP((64, "_sion_paropen_generic_one_file", rank, " buddy map[%d]=%d\n", j, (int)sion_tmpintfield_map[j]));
        }
      }
    }

    if (rank == 0) {
      /* overwrites filedesc->ntasks */
      if (SION_SUCCESS != _sion_read_header_fix_part(filedesc)) {
        free(sion_tmpintfield_buddy32);
        free(sion_tmpintfield_buddy64);
        return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
          "_sion_paropen_generic_one_file: cannot read header from file %s, aborting ...\n", fname);
      }
      DPRINTFP((32, "_sion_paropen_generic_one_file", rank,
        " read, after read of fix header part endianness=0x%x blksize=%d ntasks=%d\n", filedesc->endianness, filedesc->fsblksize,
        filedesc->ntasks));

      DPRINTFTS(rank, "before alloc");
      /* memory allocation */
      _sion_alloc_filedesc_arrays(filedesc);
      DPRINTFTS(rank, "after alloc");

      if (SION_SUCCESS != _sion_read_header_var_part(filedesc)) {
        free(sion_tmpintfield_buddy32);
        free(sion_tmpintfield_buddy64);
        return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
          "_sion_paropen_generic_one_file: cannot read header from file %s, aborting ...\n", fname);
      }

      _sion_read_header_var_part_mapping_size(filedesc);

      if (options->collsize > 0) {
        filedesc->collsize = options->collsize;
      }
      _sion_coll_check_env(filedesc);
      if (filedesc->usecoll) {
        _sion_alloc_filedesc_coll_arrays(filedesc);
      }

      /* collective */
      if (!filedesc->usecoll) {
        _sion_calculate_startpointers(filedesc, filedesc->mapping_size);
      } else {
        if (!filedesc->collmergemode) {
          _sion_calculate_startpointers_collective(filedesc, filedesc->mapping_size);
        } else {
          _sion_calculate_startpointers_collective_merge(filedesc, filedesc->mapping_size);
        }
      }
      DPRINTFTS(rank, "after calculate");

      /* check for keyval parameter, needed at this point to set flag before writing header (flag1) */
      _sion_keyval_check_env_with_options(filedesc, mode, options->keyval_mode);
    } /* rank==0 */

    /* distribute keyvalmode */
    gendata->apidesc->bcastr_cb(&filedesc->keyvalmode, comm_group, _SION_INT32, 1, 0);

    /* distribute collective options */
    gendata->apidesc->bcastr_cb(&filedesc->usecoll, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->collsize, comm_group, _SION_INT32, 1, 0);

    DPRINTFP((32, "_sion_paropen_generic_one_file", rank, " usecoll=%d\n", filedesc->usecoll));

    if (filedesc->usecoll) {
      if (!(flag & _SION_INTERNAL_FLAG_BUDDY_READ)) {
        gendata->apidesc->scatterr_cb(filedesc->all_coll_collsize, &filedesc->collsize, comm_group, _SION_INT32, 1, 0);
        gendata->apidesc->scatterr_cb(filedesc->all_coll_collector, &filedesc->collector, comm_group, _SION_INT32, 1, 0);
      } else {
        /* remap data */
        if (rank == 0)
          for (int j = 0; j < ntasks; j++)
            if (sion_tmpintfield_map[j] >= 0) {
              sion_tmpintfield_buddy32[j] = filedesc->all_coll_collsize[sion_tmpintfield_map[j]];
            }
        gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy32, &filedesc->collsize, comm_group, _SION_INT32, 1, 0);
        /* remap data */
        if (rank == 0)
          for (int j = 0; j < ntasks; j++)
            if (sion_tmpintfield_map[j] >= 0) {
              sion_tmpintfield_buddy32[j] = filedesc->all_coll_collector[sion_tmpintfield_map[j]];
            }
        gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy32, &filedesc->collector, comm_group, _SION_INT32, 1, 0);
      }

      _sion_free_filedesc_coll_arrays(filedesc);
    }

    /* distribute globalskip */
    gendata->apidesc->bcastr_cb(&filedesc->globalskip, comm_group, _SION_INT64, 1, 0);

    /* broadcast information read from file */
    gendata->apidesc->bcastr_cb(&filedesc->endianness, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->swapbytes, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->fsblksize, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->ntasks, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->fileversion, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->nfiles, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->filenumber, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->flag1, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->flag2, comm_group, _SION_INT32, 1, 0);
    gendata->apidesc->bcastr_cb(&filedesc->maxusedchunks, comm_group, _SION_INT32, 1, 0);

    DPRINTFP((32, "_sion_paropen_generic_one_file", rank, " read, after read of maxusedchunks=%d maxchunks=%d (%d)\n",
      filedesc->maxusedchunks, filedesc->maxchunks, MAXCHUNKS));
    if (filedesc->maxusedchunks > MAXCHUNKS) {
      _sion_realloc_filedesc_blocklist(filedesc, filedesc->maxusedchunks);
    }
    DPRINTFTS(rank, "after bcast");

    /* scatter per task information read from file */
    DPRINTFTS(rank, "before scatter");
    if (!(flag & _SION_INTERNAL_FLAG_BUDDY_READ)) {
      gendata->apidesc->scatterr_cb(filedesc->all_chunksizes, &filedesc->chunksize, comm_group, _SION_INT64, 1, 0);
      gendata->apidesc->scatterr_cb(filedesc->all_sizes, &filedesc->size, comm_group, _SION_INT64, 1, 0);
      gendata->apidesc->scatterr_cb(filedesc->all_startpointers, &filedesc->startpos, comm_group, _SION_INT64, 1, 0);
      gendata->apidesc->scatterr_cb(filedesc->all_globalranks, &helpint64, comm_group, _SION_INT64, 1, 0);
      filedesc->globalrank = (int32_t)helpint64;
    } else {
      /* remap data */
      if (rank == 0)
        for (int j = 0; j < ntasks; j++)
          if (sion_tmpintfield_map[j] >= 0) {
            sion_tmpintfield_buddy64[j] = filedesc->all_chunksizes[sion_tmpintfield_map[j]];
          }
      gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy64, &filedesc->chunksize, comm_group, _SION_INT64, 1, 0);
      /* remap data */
      if (rank == 0)
        for (int j = 0; j < ntasks; j++)
          if (sion_tmpintfield_map[j] >= 0) {
            sion_tmpintfield_buddy64[j] = filedesc->all_sizes[sion_tmpintfield_map[j]];
          }
      gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy64, &filedesc->size, comm_group, _SION_INT64, 1, 0);
      /* remap data */
      if (rank == 0)
        for (int j = 0; j < ntasks; j++)
          if (sion_tmpintfield_map[j] >= 0) {
            sion_tmpintfield_buddy64[j] = filedesc->all_startpointers[sion_tmpintfield_map[j]];
          }
      gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy64, &filedesc->startpos, comm_group, _SION_INT64, 1, 0);
      /* remap data */
      if (rank == 0)
        for (int j = 0; j < ntasks; j++)
          if (sion_tmpintfield_map[j] >= 0) {
            sion_tmpintfield_buddy64[j] = filedesc->all_globalranks[sion_tmpintfield_map[j]];
          }
      gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy64, &helpint64, comm_group, _SION_INT64, 1, 0);
      filedesc->globalrank = (int32_t)helpint64;
    }

    DPRINTFTS(rank, "after scatter");

    if (filedesc->fileversion <= 5) {
      /* file is version 5 or older, read number of blocks for each task */
      if (rank == 0) {
        sion_tmpintfield = malloc(filedesc->ntasks * sizeof(int64_t));
        if (sion_tmpintfield == NULL) {
          free(sion_tmpintfield_buddy32);
          return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
            "_sion_paropen_generic_one_file: cannot allocate temporary memory of size %lu (sion_tmpintfield), aborting ...\n",
            (unsigned long)ntasks * sizeof(int64_t));
        }
        _sion_read_header_var_part_blockcount_to_field(filedesc, filedesc->ntasks, sion_tmpintfield);

        for (int j = 0; j < filedesc->ntasks; j++) {
          DPRINTFP((2048, "_sion_paropen_generic_one_file", rank, " read, blockcount on task %02d is %10ld\n", j,
            (long)sion_tmpintfield[j]));
        }
      }

      /* and distribute them */
      DPRINTFTS(rank, "before scatter");
      if (!(flag & _SION_INTERNAL_FLAG_BUDDY_READ)) {
        gendata->apidesc->scatterr_cb(sion_tmpintfield, &helpint64, comm_group, _SION_INT64, 1, 0);
      } else {
        /* remap data */
        if (rank == 0)
          for (int j = 0; j < ntasks; j++)
            if (sion_tmpintfield_map[j] >= 0) {
              sion_tmpintfield_buddy64[j] = sion_tmpintfield[sion_tmpintfield_map[j]];
            }
        gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy64, &helpint64, comm_group, _SION_INT64, 1, 0);
      }
      DPRINTFTS(rank, "after scatter");
      filedesc->lastchunknr = helpint64 - 1;
      DPRINTFP(
        (32, "_sion_paropen_generic_one_file", rank, "  lastchunknr on task %02d is %10ld\n", rank, (long)filedesc->lastchunknr));

      for (int32_t i = 0; i < filedesc->maxusedchunks; i++) {
        if (rank == 0) {
          _sion_read_header_var_part_nextblocksizes_to_field(filedesc, filedesc->ntasks, sion_tmpintfield);
        }
        DPRINTFTS(rank, "before scatter");
        if (!(flag & _SION_INTERNAL_FLAG_BUDDY_READ)) {
          gendata->apidesc->scatterr_cb(sion_tmpintfield, &helpint64, comm_group, _SION_INT64, 1, 0);
        } else {
          /* remap data */
          if (rank == 0)
            for (int j = 0; j < ntasks; j++)
              if (sion_tmpintfield_map[j] >= 0) {
                sion_tmpintfield_buddy64[j] = sion_tmpintfield[sion_tmpintfield_map[j]];
              }
          gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy64, &helpint64, comm_group, _SION_INT64, 1, 0);
        }
        DPRINTFTS(rank, "after scatter");
        filedesc->blocksizes[i] = helpint64;
      }
    } else {
      /* file is version 6 or newer, synthesize block count and size */
      int64_t size = filedesc->size, chunksize = filedesc->chunksize;

      int64_t blockcount = (chunksize > 0) ? size / chunksize + ((size % chunksize > 0) ? 1 : 0) : 0;
      filedesc->lastchunknr = (blockcount > 0) ? blockcount - 1 : 0;

      for (int64_t ichunk = 0; ichunk < filedesc->maxusedchunks; ichunk++) {
        int64_t blocksize;
        if (ichunk < blockcount - 1) {
          blocksize = chunksize;
        } else if (ichunk == blockcount - 1) {
          blocksize = size - ((blockcount - 1) * chunksize);
        } else {
          blocksize = 0;
        }
        filedesc->blocksizes[ichunk] = blocksize;
      }
    }

#define BGFLUSH
#ifdef BGFLUSH
    if (do_open_file) {
      _sion_file_flush(filedesc->fileptr);
    }
#endif

    DPRINTFTS(rank, "before setp");
    gendata->apidesc->barrier_cb(comm_group);
    if (do_open_file) {
      _sion_file_set_position(filedesc->fileptr, filedesc->startpos);
    }
    filedesc->currentpos = filedesc->startpos;
    filedesc->currentblocknr = 0;

    /* free tmp field */
    free(sion_tmpintfield);
    free(sion_tmpintfield_map);
    free(sion_tmpintfield_buddy32);
    free(sion_tmpintfield_buddy64);

    gendata->apidesc->barrier_cb(comm_group);
    DPRINTFTS(rank, "after setp");
    /* end of read */
  } else {
    return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_paropen_multi_mpi: unknown file mode");
  }

  if (rank == 0) {
    /* not needed for rest of sionlib function calls  */
    _sion_free_filedesc_arrays(filedesc);
  }

  _sion_print_filedesc(filedesc, 512, "_sion_paropen_generic_one_file", 1);

  DPRINTFP((32, "_sion_paropen_generic_one_file", rank, " start position on task %02d is at end of sion_paropen_generic %10lld\n",
    rank, _sion_file_get_position(filedesc->fileptr)));

  // FIXME re-enable debug output of file_mode
  // DPRINTFP((2, "_sion_paropen_generic_one_file", rank, "leave parallel open of file %s in mode 0x%lx #tasks=%d\n", fname,
  //   (long)flags_store->mask, ntasks));
  DPRINTFP((2, "_sion_paropen_generic_one_file", rank, "leave parallel open of file %s #tasks=%d\n", fname, ntasks));

  return sid;
}

int _sion_parclose_generic(
  int sid, int rank, int ntasks, int flag, _sion_generic_gendata *sion_gendata, _sion_generic_buddy *buddy_data)
{
  int rc = SION_SUCCESS;
  int blknum;
  int64_t helpint64;
  int64_t *sion_tmpintfield = NULL;
  _sion_filedesc *sion_filedesc;
  void *comm_group = NULL;
  int do_close_file;

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

  if (sion_filedesc->state != SION_FILESTATE_PAROPEN) {
    return _sion_errorprint(SION_NOT_SUCCESS, _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));

  /* select local communicator */
  if (flag & _SION_INTERNAL_FLAG_NORMAL) {
    comm_group = sion_gendata->comm_data_local;
  }
  if (flag & _SION_INTERNAL_FLAG_BUDDY_SEND) {
    comm_group = buddy_data->buddy_send.commgroup;
  }
  if (flag & _SION_INTERNAL_FLAG_BUDDY_COLL) {
    comm_group = buddy_data->buddy_coll.commgroup;
  }
  if (flag & _SION_INTERNAL_FLAG_BUDDY_READ) {
    comm_group = buddy_data->groups[buddy_data->currentgroup]->commgroup;
  }

  /* decide if file has to be opened physically  */
  do_close_file = 1;
  if (flag & _SION_INTERNAL_FLAG_BUDDY_SEND) {
    do_close_file = 0;
  }
  if ((flag & _SION_INTERNAL_FLAG_BUDDY_COLL) && (rank > 0)) {
    do_close_file = 0;
  }
  if ((flag & _SION_INTERNAL_FLAG_BUDDY_READ) && (rank > 0)) {
    do_close_file = 0;
  }

  /* 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, 512, "_sion_parclose_generic", 1);
      DPRINTFP((32, "_sion_parclose_generic", rank, " parallel close (read mode)  sid=%d, call fclose on file\n", sid));

      if (do_close_file) {
        _sion_file_close(sion_filedesc->fileptr);
      }
      sion_filedesc->fileptr = NULL;
      sion_filedesc->state = SION_FILESTATE_CLOSE;
    }
  } else {
    /* WRITE MODE: collect data from all tasks, write metadata, and close file on all tasks  */

    /* clear internal buffer */
    if (sion_filedesc->usebuffer) {
      _sion_buffer_flush(sion_filedesc);
    }

    _sion_print_filedesc(sion_filedesc, 512, "_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));
        if (do_close_file) {
          _sion_file_close(sion_filedesc->fileptr);
        }
        sion_filedesc->fileptr = NULL;
        sion_filedesc->state = SION_FILESTATE_CLOSE;
      }
    }

    sion_gendata->apidesc->barrier_cb(comm_group);

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

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

    /* gather sizes and write to file header */
    DPRINTFTS2(rank, "before gather");
    sion_gendata->apidesc->gatherr_cb(&sion_filedesc->size, sion_tmpintfield, comm_group, _SION_INT64, 1, 0);
    DPRINTFTS2(rank, "after gather");
    if (rank == 0) {
      _sion_write_header_sizes_from_field(sion_filedesc, sion_tmpintfield);
    }

    /* gather number of blocks of each tasks, and search maxusedchunks */
    DPRINTFTS2(rank, "before gather");
    helpint64 = sion_filedesc->lastchunknr + 1;
    sion_gendata->apidesc->gatherr_cb(&helpint64, sion_tmpintfield, comm_group, _SION_INT64, 1, 0);

    if (rank == 0) {
      sion_filedesc->maxusedchunks = -1;
      for (blknum = 0; blknum < sion_filedesc->ntasks; blknum++)
        if (sion_tmpintfield[blknum] > sion_filedesc->maxusedchunks) {
          sion_filedesc->maxusedchunks = (int)sion_tmpintfield[blknum];
        }
    }
    sion_gendata->apidesc->bcastr_cb(&sion_filedesc->maxusedchunks, comm_group, _SION_INT32, 1, 0);
    DPRINTFTS2(rank, "after gather");

    /* calculate and set start_of_varheader */
    sion_filedesc->start_of_varheader = sion_filedesc->start_of_data + sion_filedesc->maxusedchunks * sion_filedesc->globalskip;

    /* 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));
      if (do_close_file) {
        _sion_file_close(sion_filedesc->fileptr);
      }
      sion_filedesc->fileptr = NULL;
      sion_filedesc->state = SION_FILESTATE_CLOSE;

      /* free tmp field */
      free(sion_tmpintfield);
    }

  } /* write */

  _sion_free_filedesc(sion_filedesc);
  sion_filedesc = NULL;

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

  return rc;
}

int _sion_parreinit_generic(int sid, int64_t chunksize, int rank, int ntasks, _sion_generic_gendata *sion_gendata)
{
  int rc = SION_SUCCESS;
  _sion_filedesc *sion_filedesc;
  int64_t lchunksize, lstartpointer, lglobalrank;
  void *comm_group = NULL;

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

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

  DPRINTFP((2, "_sion_parreinit_generic", sion_filedesc->rank, "enter parallel reinit  sid=%d\n", sid));

  comm_group = sion_gendata->comm_data_local;

  if (sion_filedesc->mode == SION_FILEMODE_READ) {
    return _sion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN,
      "_sion_parreinit_generic: sion file with sid=%d only allowed for files openend for write\n", sid);
  }

  DPRINTFTS(sion_filedesc->rank, "before alloc");
  if (sion_filedesc->rank == 0) {
    /* memory allocation for internal fields */
    _sion_alloc_filedesc_arrays(sion_filedesc);
  }
  DPRINTFTS(sion_filedesc->rank, "after alloc");

  /* collect new chunksize data and init startpointers on PE 0 */
  lchunksize = (int64_t)chunksize;
  lglobalrank = (int64_t)sion_filedesc->globalrank;

  DPRINTFTS2(sion_filedesc->rank, "before gather");
  sion_gendata->apidesc->gatherr_cb(&lchunksize, sion_filedesc->all_chunksizes, comm_group, _SION_INT64, 1, 0);
  sion_gendata->apidesc->gatherr_cb(&lglobalrank, sion_filedesc->all_globalranks, comm_group, _SION_INT64, 1, 0);

  DPRINTFTS2(sion_filedesc->rank, "after gather");

  if (sion_filedesc->usecoll) {
    _sion_alloc_filedesc_coll_arrays(sion_filedesc);
  }

  if (sion_filedesc->rank == 0) {
    DPRINTFTS(sion_filedesc->rank, "before calculate");
    if (!sion_filedesc->usecoll) {
      _sion_calculate_startpointers(sion_filedesc, 0);
    } else {
      _sion_calculate_startpointers_collective(sion_filedesc, 0);
    }
    DPRINTFTS(sion_filedesc->rank, "after calculate");
  }

  /* write header again */
  if (sion_filedesc->rank == 0) {
    /* apply hint for first meta data block */
    _sion_apply_hints(sion_filedesc, SION_HINTS_ACCESS_TYPE_METADATABLOCK1);

    lstartpointer = 0;
    _sion_file_set_position(sion_filedesc->fileptr, lstartpointer);

    DPRINTFTS(sion_filedesc->rank, "before writeh");
    _sion_write_header(sion_filedesc);
    DPRINTFTS(sion_filedesc->rank, "after writeh");

    /* needed for writing pointer to var part of metadata at the end of the file */
    sion_filedesc->end_of_header = _sion_file_get_position(sion_filedesc->fileptr);
    sion_filedesc->start_of_data = sion_filedesc->all_startpointers[0];

    /*set max. file size */
    lstartpointer =
      sion_filedesc->all_startpointers[sion_filedesc->ntasks - 1] + sion_filedesc->all_chunksizes[sion_filedesc->ntasks - 1];
    DPRINTFTS(sion_filedesc->rank, "before setp(0)");
    _sion_file_set_position(sion_filedesc->fileptr, lstartpointer);
    DPRINTFTS(sion_filedesc->rank, "after setp(0)");
  }

  /* distribute start_pos */
  DPRINTFTS(sion_filedesc->rank, "before scatter");
  sion_gendata->apidesc->scatterr_cb(sion_filedesc->all_startpointers, &sion_filedesc->startpos, comm_group, _SION_INT64, 1, 0);
  DPRINTFTS(sion_filedesc->rank, "after scatter");

  /* distribute information for collective operations */
  if (sion_filedesc->usecoll) {
    sion_gendata->apidesc->scatterr_cb(sion_filedesc->all_coll_collsize, &sion_filedesc->collsize, comm_group, _SION_INT32, 1, 0);
    sion_gendata->apidesc->scatterr_cb(
      sion_filedesc->all_coll_collector, &sion_filedesc->collector, comm_group, _SION_INT32, 1, 0);

    _sion_free_filedesc_coll_arrays(sion_filedesc);
  }

  /* distribute globalskip */
  sion_gendata->apidesc->bcastr_cb(&sion_filedesc->globalskip, comm_group, _SION_INT64, 1, 0);

  DPRINTFP((32, "_sion_parreinit_generic", sion_filedesc->rank, " start position is %10lld %10.4f MB\n", sion_filedesc->startpos,
    sion_filedesc->startpos / 1024.0 / 1024.0));

  /* set filepointer on each task */
  DPRINTFTS(sion_filedesc->rank, "before setp");
  sion_gendata->apidesc->barrier_cb(comm_group);
  _sion_file_set_position(sion_filedesc->fileptr, sion_filedesc->startpos);
  sion_filedesc->currentpos = sion_filedesc->startpos;
  sion_filedesc->chunksize = (int64_t)chunksize;
  sion_gendata->apidesc->barrier_cb(comm_group);

  /* apply hint for first chunk */
  _sion_apply_hints(sion_filedesc, SION_HINTS_ACCESS_TYPE_CHUNK);

  if (sion_filedesc->rank == 0) {
    /* not needed for rest of sionlib function calls  */
    _sion_free_filedesc_arrays(sion_filedesc);
  }

  DPRINTFTS(sion_filedesc->rank, "after setp");
  DPRINTFP((32, "_sion_parreinit_generic", sion_filedesc->rank, " ending open for write #tasks=%d filepos=%lld\n",
    sion_filedesc->ntasks, _sion_file_get_position(sion_filedesc->fileptr)));

  DPRINTFP((2, "_sion_parreinit_generic", sion_filedesc->rank, "leave parallel reinit of file %s in  #tasks=%d\n",
    sion_filedesc->fname, sion_filedesc->ntasks));

  return rc;
}

#define DFUNCTION "_sion_generic_collect_mapping"
int _sion_generic_collect_mapping(_sion_filedesc *sion_filedesc, int *mapping_size, int32_t **mapping)
{
  int rc = SION_SUCCESS;
  int t;
  _sion_generic_gendata *sion_gendata;
  _sion_generic_apidesc *sion_apidesc;
  int32_t lpos[2], *receivemap = NULL, iamreceiver, receiver = -1;

  sion_gendata = sion_filedesc->dataptr;
  sion_apidesc = sion_gendata->apidesc;

  *mapping = NULL;
  *mapping_size = 0;

  if ((sion_filedesc->mode == SION_FILEMODE_WRITE) && (sion_filedesc->nfiles > 1)) {
    /* collect mapping to files to task 0 */

    /* mapping data will be collected by master of first physical file */
    if ((sion_filedesc->filenumber == 0) && (sion_filedesc->rank == 0)) {
      /* allocate data */
      *mapping_size = sion_gendata->gsize;
      *mapping = malloc(*mapping_size * 2 * sizeof(int32_t));
      if (*mapping == NULL) {
        return _sion_errorprint(
          SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_generic_parclose: Cannot allocate memory for mapping");
      }
    }

    /* gather info about send about global rank of master of first file on grank 0 */
    if (sion_gendata->grank == 0) {
      receivemap = malloc(sion_gendata->gsize * sizeof(int32_t));
      if (receivemap == NULL) {
        return _sion_errorprint(
          SION_ID_NOT_VALID, _SION_ERROR_RETURN, "sion_generic_parclose: Cannot allocate memory for receivemap");
      }
    }

    if ((sion_filedesc->filenumber == 0) && (sion_filedesc->rank == 0)) {
      iamreceiver = sion_gendata->grank;
    } else {
      iamreceiver = -1;
    }
    sion_apidesc->gatherr_cb(&iamreceiver, receivemap, sion_gendata->comm_data_global, _SION_INT32, 1, 0);
    if (sion_gendata->grank == 0) {
      for (t = 0; t < sion_gendata->gsize; t++) {
        if (receivemap[t] >= 0) {
          receiver = receivemap[t];
          break;
        }
      }
      DPRINTFP((1, DFUNCTION, sion_gendata->grank, "receiver of mapping grank=%d\n", receiver));
    }
    sion_apidesc->bcastr_cb(&receiver, sion_gendata->comm_data_global, _SION_INT32, 1, 0);

    /* receive global rank of master of first file on grank 0 */
    lpos[0] = sion_filedesc->filenumber;
    lpos[1] = sion_filedesc->rank;
    sion_apidesc->gatherr_cb(&lpos, *mapping, sion_gendata->comm_data_global, _SION_INT32, 2, receiver);
  }

  free(receivemap);

  return rc;
}
#undef DFUNCTION

/*  END OF _sion_parclose_generic */
