/****************************************************************************
**  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 <assert.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.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.h"
#include "sion_generic_buddy.h"
#include "sion_generic_collective.h"
#include "sion_generic_internal.h"
#include "sion_internal.h"
#include "sion_internal_positioning.h"
#include "sion_metadata.h"

bool _sion_generic_collective_next_spec_empty(void *iterator, int64_t *range);

int _sion_paropen_generic_buddy(int sid, const char *fname, sion_open_mode mode, int numfiles, int filenumber, int rank,
  int ntasks, int globalrank, _sion_generic_gendata *gendata, const sion_generic_options *options)
{
  int rc = SION_SUCCESS;
  int capability, pass;
  int32_t file_globalrank;
  char *buddy_fn, *nfname = NULL;
  _sion_filedesc *filedesc;
  _sion_generic_buddy *buddies, *buddyptr;

  int buddylevel = options->buddylevel;

  DPRINTFP((1, __func__, gendata->grank, "allocate memory for %d buddy levels\n", buddylevel));

  buddies = malloc(buddylevel * sizeof(_sion_generic_buddy));
  if (!buddies) {
    return _sion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN,
      "cannot allocate buddies structure of size %lu (_sion_generic_buddy), aborting ...\n",
      (unsigned long)sizeof(_sion_generic_buddy));
  }
  buddies->numgroups = 0;

  buddy_fn = calloc(SION_FILENAME_LENGTH + 10, 1);
  if (!buddy_fn) {
    free(buddies);
    return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "_sion_paropen_generic_buddy: cannot allocate temporary memory of size %lu (buddy_fn), aborting ...\n",
      (unsigned long)SION_FILENAME_LENGTH + 10);
  }

  /* get pointer to internal datastructures */
  _sion_generic_apidesc *apidesc = gendata->apidesc;

  if (mode == SION_OPEN_WRITE) {
    /* STEP1: WRITE open normal file */
    nfname =
      (apidesc->get_multi_filename_cb ? apidesc->get_multi_filename_cb : _sion_get_multi_filename)(fname, gendata->filenumber);
    DPRINTFP((1, __func__, gendata->grank, "call parallel open of %d files (current name %s)\n", gendata->numfiles, nfname));
    rc = _sion_paropen_generic_one_file(sid, nfname, mode, gendata->numfiles, gendata->filenumber, gendata->lrank, gendata->lsize,
      globalrank, _SION_INTERNAL_FLAG_BUDDY_NORMAL, gendata, NULL, options);
    DPRINTFP((1, __func__, gendata->grank, "leave parallel open of %d files in #tasks=%d sid=%d globalrank=%d\n",
      gendata->numfiles, gendata->lsize, sid, gendata->grank));
    free(nfname);

    /* test sid and get internal data structure */
    if ((rc < 0) || (sid < 0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR) || !(filedesc = _sion_vcdtovcon(sid))) {
      free(buddies);
      free(buddy_fn);
      return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN, "_sion_paropen_generic_buddy: invalid filedesc %d", sid);
    }

    /* store additional data */
    filedesc->buddies = buddies;

    capability = apidesc->get_capability_cb(gendata->comm_data_global);

    /* open buddy files */
    for (int b = 0; b < filedesc->buddylevel; b++) {
      buddyptr = &buddies[b];

      /* create communicators */
      buddyptr->buddy_send.commgroup = NULL;
      buddyptr->buddy_coll.commgroup = NULL;
      rc = _sion_buddy_map(gendata, capability, b + 1, &buddyptr->buddy_send, &buddyptr->buddy_coll);

      /* file name */
      sprintf(buddy_fn, "%s_BUDDY_%02d", fname, b);

      /* open files */
      for (pass = 1; pass <= _SION_BW_SCHED_NUM_PASSES; pass++) {
        DPRINTFP((1, __func__, gendata->grank, "BUDDYO pass=%d [grank=%2d] op=%d\n", pass, globalrank,
          _sion_buddy_bwsched(buddyptr->buddy_send.groupnum, gendata->numfiles, pass)));

        switch (_sion_buddy_bwsched(buddyptr->buddy_send.groupnum, gendata->numfiles, pass)) {
        case _SION_BW_SCHED_ACTIONA:
          /* parameter */
          file_globalrank = -1 * (globalrank + 1);
          sion_generic_options file_options = *options;
          sion_generic_options_set_chunksize(&file_options, 0);
          nfname = (apidesc->get_multi_filename_cb ? apidesc->get_multi_filename_cb : _sion_get_multi_filename)(
            buddy_fn, buddyptr->buddy_coll.filenum);
          DPRINTFP((1, __func__, gendata->grank,
            "BUDDYO call open_one_file COLL [gendata: b=%d grank=%2d, group=%d of %d, lrank=%d of %d] fn=%s (%d)\n", b,
            globalrank, buddyptr->buddy_coll.groupnum, gendata->numfiles, buddyptr->buddy_coll.rank, buddyptr->buddy_coll.size,
            nfname, buddyptr->buddy_coll.filenum));
          buddyptr->buddy_coll.sid = _sion_newvcd(NULL, SION_FILEDESCRIPTOR);
          rc = _sion_paropen_generic_one_file(buddyptr->buddy_coll.sid, nfname, mode, gendata->numfiles,
            buddyptr->buddy_coll.filenum, buddyptr->buddy_coll.rank, buddyptr->buddy_coll.size, file_globalrank,
            _SION_INTERNAL_FLAG_BUDDY_COLL, gendata, buddyptr, &file_options);
          free(nfname);
          break;
        case _SION_BW_SCHED_ACTIONB:
          /* parameter */
          file_globalrank = globalrank;
          nfname = (apidesc->get_multi_filename_cb ? apidesc->get_multi_filename_cb : _sion_get_multi_filename)(
            buddy_fn, buddyptr->buddy_send.filenum);
          DPRINTFP((1, __func__, gendata->grank,
            "BUDDYO call open_one_file SEND [gendata: b=%d grank=%2d, group=%d of %d, lrank=%d of %d] fn=%s \n", b, globalrank,
            buddyptr->buddy_send.groupnum, gendata->numfiles, buddyptr->buddy_send.rank, buddyptr->buddy_send.size, nfname));
          buddyptr->buddy_send.sid = _sion_newvcd(NULL, SION_FILEDESCRIPTOR);
          rc = _sion_paropen_generic_one_file(buddyptr->buddy_send.sid, nfname, mode, gendata->numfiles,
            buddyptr->buddy_send.filenum, buddyptr->buddy_send.rank, buddyptr->buddy_send.size, file_globalrank,
            _SION_INTERNAL_FLAG_BUDDY_SEND, gendata, buddyptr, options);
          free(nfname);
          break;
        default:
          DPRINTFP((1, __func__, gendata->grank, "BUDDYO no-op [grank=%2d]\n", globalrank));
          int mapping_size;
          int32_t *mapping;
          _sion_generic_collect_mapping_buddy(buddyptr, gendata, &mapping_size, &mapping);
          free(mapping);
          break;
        } /* switch */
      } /* for pass */
    } /* for buddies */

  } else if (mode == SION_OPEN_READ) {
    int filefound, lgroup = filenumber, root = 0, task, tmpsize, group, masterfile_found;
    int32_t filenumber, lrank, lsize, file, file_lgroup, myrole, steprank;
    int32_t helpint32 = 0, numsteps = 0, numgroups = 0, numreaderinthisfile;
    int32_t *sion_tmpintfield1 = NULL; /* allocated on rank 0 */
    /* int32_t *sion_tmpintfield2 = NULL;  */ /* allocated on lranl-0 that has access to master file  */

    int *stepvectorlist[MAXREADSTEPS]; /* to be changed in an dynamic list */
    int *stepvector, readercount = 0, readyflag = 0, datafound = 0;
    int stepcount_coll, stepcount_collreader, stepcount_reader, stepcount_reader_rank, stepcount_noreader,
      stepcount_noreader_rank, groupcounter;
    int step, fit, newrank;
    _sion_generic_buddy_info *buddy_info;
    int32_t helpint7[7];

    DPRINTFP((1, __func__, gendata->grank, "enter buddy open in read mode\n"));

    if (gendata->grank == 0) {
      tmpsize = 1 * gendata->gsize;
      sion_tmpintfield1 = malloc(tmpsize * sizeof(int32_t));
      if (sion_tmpintfield1 == NULL) {
        free(buddies);
        free(buddy_fn);
        return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
          "_sion_paropen_generic_buddy: cannot allocate temporary memory of size %lu (sion_tmpintfield), aborting ...\n",
          (unsigned long)tmpsize * sizeof(int32_t));
      }
    }

    /* Loop1: over all file sets (normal, buddy1, buddy2) until all have found their data chunk */
    groupcounter = 0;
    for (int b = -1; ((b < buddylevel) && (!readyflag)); b++) {
      /* determine filename and check for existence of master file on each lrank=0  */
      if (b == -1) {
        sprintf(buddy_fn, "%s", fname);
      } else {
        sprintf(buddy_fn, "%s_BUDDY_%02d", fname, b);
      }
      nfname = (apidesc->get_multi_filename_cb ? apidesc->get_multi_filename_cb : _sion_get_multi_filename)(buddy_fn, 0);
      DPRINTFP((1, __func__, gendata->grank, "BUDDY[%2d] start checking buddy level (file=%s)\n", b, nfname));

      filefound = 0;
      if (gendata->lrank == 0) {
        if (_sion_file_stat_file(nfname)) {
          filefound = 1;
          DPRINTFP((1, __func__, gendata->grank, "check for existence of file %s --> file found\n", nfname));
        }
      }

      /* distribute info about master file to all others (root=<tasknum> or -1 if file not found) */
      helpint32 = filefound;
      gendata->apidesc->gatherr_cb(&helpint32, sion_tmpintfield1, gendata->comm_data_global, _SION_INT32, 1, 0);
      if (gendata->grank == 0) {
        root = -1;
        if (sion_tmpintfield1 == NULL) {
          free(buddies);
          free(buddy_fn);
          return _sion_errorprint(
            SION_NOT_SUCCESS, _SION_ERROR_ABORT, "%s: gatherr_cb returned sion_tmpintfield1 == NULL", __func__);
        }
        for (task = 0; task < gendata->gsize; task++) {
          if (sion_tmpintfield1[task] == 1) {
            root = task;
          }
        }
        helpint32 = root;
        DPRINTFP((1, __func__, gendata->grank, "after scan master: root for this file is %d #tasks=%d\n", root, gendata->gsize));
      }
      gendata->apidesc->bcastr_cb(&helpint32, gendata->comm_data_global, _SION_INT32, 1, 0);
      root = helpint32;
      DPRINTFP((1, __func__, gendata->grank, "root for this file is %d\n", root));

      /* get mapping from file and distribute it */
      if (root != -1) {
        _sion_generic_buddy_get_and_distribute_info_from_file(gendata, nfname, root, &filenumber, &numfiles, &lrank, &lsize);
        DPRINTFP((1, __func__, gendata->grank, "BUDDY[%d] after read mapping file=%d of %d, lrank=%d of %d\n", b, filenumber,
          numfiles, lrank, lsize));
        masterfile_found = 1;
      } else {
        filenumber = lrank = lsize = -1;
        masterfile_found = 0;
      }
      free(nfname);

      /* numfiles will be defined from from input parameter, other info will be defined on-the-fly */
      DPRINTFP((32, __func__, gendata->grank, "Info avail: b=%d numfiles=%d root=%d lgroup=%d\n", b, numfiles, root, lgroup));

      /* Loop2: for each physical file of this file set */
      for (file = 0; ((file < numfiles) && (!readyflag)); file++) {
        /* check physical file */
        nfname = (apidesc->get_multi_filename_cb ? apidesc->get_multi_filename_cb : _sion_get_multi_filename)(buddy_fn, file);
        filefound = -1;
        if (gendata->lrank == 0) {
          if (_sion_file_stat_file(nfname)) {
            filefound = lgroup;
            DPRINTFP((1, __func__, gendata->grank, "check for existence of file %s --> file found\n", nfname));
          }
        }

        /* gather and distribute info (root and lgroup) about file to all others (root=<tasknum> or -1 if file not found) */
        helpint32 = filefound;
        gendata->apidesc->gatherr_cb(&helpint32, sion_tmpintfield1, gendata->comm_data_global, _SION_INT32, 1, 0);
        if (gendata->grank == 0) {
          root = -1;
          for (task = 0; task < gendata->gsize; task++) {
            if (sion_tmpintfield1[task] > -1) {
              root = task;
              file_lgroup = sion_tmpintfield1[task];
            }
          }
          DPRINTFP(
            (1, __func__, gendata->grank, "after scan one file: root for this file is %d file_lgroup=%d\n", root, file_lgroup));
        }
        if (gendata->grank == 0) {
          helpint32 = root;
        }
        gendata->apidesc->bcastr_cb(&helpint32, gendata->comm_data_global, _SION_INT32, 1, 0);
        root = helpint32;
        DPRINTFP((1, __func__, gendata->grank, "root for this file is %d\n", root));

        if (root >= 0) {
          /* file found */

          if (!masterfile_found) {
            /* get avail info from file: lrank/lsize for local tasks */
            DPRINTFP((1, __func__, gendata->grank, "BUDDY[%d] befre read globalranks file=%d of %d, lrank=%d of %d\n", b,
              filenumber, numfiles, lrank, lsize));
            _sion_generic_buddy_get_and_distribute_info_from_one_file(
              gendata, nfname, root, &filenumber, &numfiles, &lrank, &lsize);
            DPRINTFP((1, __func__, gendata->grank, "BUDDY[%d] after read globalranks file=%d of %d, lrank=%d of %d\n", b,
              filenumber, numfiles, lrank, lsize));
          }

          /* bcast lgroup */
          if (gendata->grank == 0) {
            helpint32 = file_lgroup;
          }
          gendata->apidesc->bcastr_cb(&helpint32, gendata->comm_data_global, _SION_INT32, 1, 0);
          file_lgroup = helpint32;
          DPRINTFP((1, __func__, gendata->grank, "file_lgroup for this file is %d\n", file_lgroup));

          /* determine my role for this file */
          myrole = SION_ROLE_NONE; /* not acting on that file */
          if ((filenumber == file) && (!datafound)) {
            datafound = 1; /* fantastic ... */
            if (gendata->grank == root) {
              myrole = SION_ROLE_COLLECTOR_READER; /* collector and reader */
            } else {
              myrole = SION_ROLE_READER; /* reader */
            }
          } else {
            if (gendata->grank == root) {
              myrole = SION_ROLE_COLLECTOR; /* collector */
            } else if (file_lgroup == lgroup) {
              myrole = SION_ROLE_NOREADER; /* task on the same lgroup but not collector */
            }
          }

          /* collect on rank 0 info about each task acting on this file  */
          helpint32 = myrole;
          gendata->apidesc->gatherr_cb(&helpint32, sion_tmpintfield1, gendata->comm_data_global, _SION_INT32, 1, 0);

          /* check if at least one tasks needs data  */
          if (gendata->grank == 0) {
            numreaderinthisfile = 0;
            if (sion_tmpintfield1 == NULL) {
              return _sion_errorprint(
                SION_NOT_SUCCESS, _SION_ERROR_ABORT, "%s: gatherr_cb returned sion_tmpintfield1 == NULL", __func__);
            }
            for (task = 0; ((task < gendata->gsize)); task++)
              if ((sion_tmpintfield1[task] == SION_ROLE_COLLECTOR_READER) || (sion_tmpintfield1[task] == SION_ROLE_READER)) {
                numreaderinthisfile++;
              }
          }
          gendata->apidesc->bcastr_cb(&numreaderinthisfile, gendata->comm_data_global, _SION_INT32, 1, 0);

          if (numreaderinthisfile > 0) {
            DPRINTFP((1, __func__, gendata->grank, "myrole for file %s is %s\n", nfname, _sion_buddy_role_to_str(myrole)));

            /* check for step in which the group can be created */
            if (gendata->grank == 0) {
              /* check in existing steps if group can created there */
              fit = 0;
              for (step = 0; ((step < numsteps) && (fit == 0)); step++) {
                stepvector = stepvectorlist[step];
                fit = 1;
                for (task = 0; ((task < gendata->gsize) && (fit == 1)); task++) {
                  if (sion_tmpintfield1[task] != SION_ROLE_NONE) {
                    if (stepvector[task] != SION_ROLE_NONE) {
                      fit = 0;
                    }
                  }
                }
              }
              DPRINTFP((1, __func__, gendata->grank, "BUDDY STEP assign: after scanning %d steps fit=%d\n", numsteps, fit));
              if (fit == 0) {
                /* no space in available steps, create a new one */
                stepvector = malloc(gendata->gsize * sizeof(int));
                if (stepvector == NULL) {
                  return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
                    "_sion_paropen_generic_buddy: cannot allocate temporary memory of size %lu (stepvector), aborting ...\n",
                    (unsigned long)gendata->gsize * sizeof(int));
                }
                for (task = 0; (task < gendata->gsize); task++) {
                  stepvector[task] = SION_ROLE_NONE;
                }
                numsteps++;
                step++;
                stepvectorlist[numsteps - 1] = stepvector;
                DPRINTFP((1, __func__, gendata->grank, "BUDDY STEP assign: allocating new step %d\n", numsteps - 1));
              }
              /* store roles in stepvector and analyse group */
              stepcount_coll = stepcount_collreader = stepcount_reader = stepcount_noreader = 0;
              for (task = 0; (task < gendata->gsize); task++) {
                if (sion_tmpintfield1[task] != SION_ROLE_NONE) {
                  stepvector[task] = sion_tmpintfield1[task];
                  if (sion_tmpintfield1[task] == SION_ROLE_COLLECTOR) {
                    stepcount_coll++;
                  }
                  if (sion_tmpintfield1[task] == SION_ROLE_READER) {
                    readercount++;
                    stepcount_reader++;
                  }
                  if (sion_tmpintfield1[task] == SION_ROLE_COLLECTOR_READER) {
                    readercount++;
                    stepcount_collreader++;
                  }
                  if (sion_tmpintfield1[task] == SION_ROLE_NOREADER) {
                    stepcount_noreader++;
                  }
                }
              }
              /* DPRINTFP((1, __func__, gendata->grank, "BUDDY STEP assign: stepcount_coll=%d stepcount_collreader=%d stepcount_reader=%d stepcount_noreader=%d\n", */
              /* 	  stepcount_coll,stepcount_collreader, stepcount_reader, stepcount_noreader)); */
              DPRINTFP((1, __func__, gendata->grank, "BUDDY STEP assign: group assigned to step %d (tasks with no data: %d)\n",
                numsteps - 1, gendata->gsize - readercount));
              if (gendata->gsize - readercount == 0) {
                readyflag = 1;
              }

              /* calculate ranks */
              stepcount_noreader_rank = stepcount_coll;
              stepcount_reader_rank = stepcount_noreader_rank + stepcount_noreader;
              for (task = 0; (task < gendata->gsize); task++) {
                /*DPRINTFP((1, __func__, gendata->grank, "BUDDY STEP assign1: task %d stepcount_reader_rank=%d stepcount_noreader_rank=%d %s\n",
                  task, stepcount_reader_rank, stepcount_noreader_rank,_sion_buddy_role_to_str(sion_tmpintfield1[task])));*/
                if (sion_tmpintfield1[task] == SION_ROLE_NONE) {
                  newrank = -1;
                }
                if (sion_tmpintfield1[task] == SION_ROLE_COLLECTOR) {
                  newrank = 0;
                }
                if (sion_tmpintfield1[task] == SION_ROLE_COLLECTOR_READER) {
                  newrank = stepcount_reader_rank++;
                }
                if (sion_tmpintfield1[task] == SION_ROLE_NOREADER) {
                  newrank = stepcount_noreader_rank++;
                }
                if (sion_tmpintfield1[task] == SION_ROLE_READER) {
                  newrank = stepcount_reader_rank++;
                }
                sion_tmpintfield1[task] = newrank;
                /* DPRINTFP((1, __func__, gendata->grank, "BUDDY STEP assign2: rank %d to task %d stepcount_reader_rank=%d stepcount_noreader_rank=%d\n",
                   sion_tmpintfield1[task],task, stepcount_reader_rank, stepcount_noreader_rank));*/
              }
            } /* if(rank==0) --> stepvector handling */

            /* scatter rank info to tasks */
            gendata->apidesc->scatterr_cb(sion_tmpintfield1, &steprank, gendata->comm_data_global, _SION_INT32, 1, 0);

            /* bcast info ), ready flag, size, collsize, to_index  */
            if (gendata->grank == 0) {
              helpint7[0] = step - 1; /* step number */
              helpint7[1] = readyflag; /* readyflag */
              helpint7[2] = stepcount_coll + stepcount_collreader + stepcount_reader + stepcount_noreader; /* size */
              helpint7[3] = stepcount_collreader + stepcount_reader; /* collsize */
              helpint7[4] = stepcount_coll + stepcount_noreader; /* from_index */
              helpint7[5] = stepcount_coll + stepcount_noreader + stepcount_collreader + stepcount_reader - 1; /* to_index */
              helpint7[6] = groupcounter++; /* unique Id of group */
              DPRINTFP(
                (1, __func__, gendata->grank, "define group #%d (step=%d, readyflag=%d, size=%d, collsize=%d, from=%d, to=%d) \n",
                  helpint7[6], helpint7[0], helpint7[1], helpint7[2], helpint7[3], helpint7[4], helpint7[5]));
            }
            gendata->apidesc->bcastr_cb(helpint7, gendata->comm_data_global, _SION_INT32, 7, 0);
            readyflag = helpint7[1];

            /* create new data structure for storing info about group if task is acting in this group */
            if (myrole != SION_ROLE_NONE) {
              buddy_info = malloc(sizeof(_sion_generic_buddy_info));
              if (buddy_info == NULL) {
                free(buddies);
                return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
                  "_sion_paropen_generic_buddy: cannot allocate temporary memory of size %lu (buddy_info), aborting ...\n",
                  (unsigned long)sizeof(_sion_generic_buddy_info));
              }
              buddies->groups[buddies->numgroups] = buddy_info;
              buddies->numgroups++;
              buddy_info->groupid = helpint7[6];
              buddy_info->stepnum = helpint7[0];
              buddy_info->rank = steprank;
              buddy_info->size = helpint7[2];
              buddy_info->commgroup = NULL; /* will be created later */
              buddy_info->collsize = helpint7[3];
              buddy_info->from_index = helpint7[4];
              buddy_info->to_index = helpint7[5];
              buddy_info->myrole = myrole;
              buddy_info->bnum = b;
              if ((myrole == SION_ROLE_READER) || (myrole == SION_ROLE_COLLECTOR_READER)) {
                buddy_info->filelrank = lrank;
              } else {
                buddy_info->filelrank = -1;
              }
              buddy_info->filenum = file;
              DPRINTFP((1, __func__, gendata->grank,
                "create group #%d (step=%d, rank=%d, size=%d, collsize=%d, from=%2d, to=%2d) (bnum=%2d file# %2d) with role %s "
                "\n",
                buddy_info->groupid, buddy_info->stepnum, buddy_info->rank, buddy_info->size, buddy_info->collsize,
                buddy_info->from_index, buddy_info->to_index, buddy_info->bnum, buddy_info->filenum,
                _sion_buddy_role_to_str(buddy_info->myrole)));

            } /* create group */
          } /* (numreaderinthisfile>0) */
        } /* if (file found on one of the lgroups) */
      } /* end of loop over physical files of file set */
    } /* end of loop over file sets (buddies) */

    if (!readyflag) {
      /* some tasks could not find their data */
      free(buddies);
      free(buddy_fn);
      return _sion_errorprint(
        SION_ID_NOT_VALID, _SION_ERROR_RETURN, "_sion_paropen_generic_buddy: cannot open file, not all files available");
    }

    /* bcast max. number of steps */
    gendata->apidesc->bcastr_cb(&numsteps, gendata->comm_data_global, _SION_INT32, 1, 0);
    buddies->numsteps = numsteps;

    /* bcast max. number of groups */
    if (gendata->grank == 0) {
      numgroups = groupcounter;
    }
    gendata->apidesc->bcastr_cb(&numgroups, gendata->comm_data_global, _SION_INT32, 1, 0);

    /* build commgroups */
    for (step = 0; step < numsteps; step++) {
      /* search for group with this step */
      group = -1;
      for (int b = 0; ((b < buddies->numgroups) && (group == -1)); b++) {
        DPRINTFP((1, __func__, gendata->grank, "check group #%d of %d in stepnum=%d  step=%d\n", b, buddies->numgroups,
          buddies->groups[b]->stepnum, step));
        if (buddies->groups[b]->stepnum == step) {
          group = b;
        }
      }

      if (group != -1) {
        /* create group for step */
        buddy_info = buddies->groups[group];
        rc = gendata->apidesc->create_lcg_cb(&buddy_info->commgroup, gendata->comm_data_global, gendata->grank, gendata->gsize,
          buddy_info->rank, buddy_info->size, buddy_info->groupid, numgroups);
        DPRINTFP(
          (256, __func__, _SION_DEFAULT_RANK, "COMM_created active [grank=%2d of %2d, group=%2d of %2d, lrank=%2d of %2d]\n",
            gendata->grank, gendata->gsize, buddy_info->groupid, numgroups, buddy_info->rank, buddy_info->size));
      } else {
        /* nothing to do in this step, create dummy comm */
        void *dummycommgroup = NULL;
        rc = gendata->apidesc->create_lcg_cb(
          &dummycommgroup, gendata->comm_data_global, gendata->grank, gendata->gsize, 0, 1, -1, numgroups);
        DPRINTFP(
          (256, __func__, _SION_DEFAULT_RANK, "COMM_created dummy  [grank=%2d of %2d, group=%2d of %2d, lrank=%2d of %2d]\n",
            gendata->grank, gendata->gsize, -1, numgroups, 0, 1));
      }

      if (group != -1) {
        /* open file */
        buddy_info = buddies->groups[group];
        buddies->currentgroup = group;

        if (buddy_info->bnum == -1) {
          sprintf(buddy_fn, "%s", fname);
        } else {
          sprintf(buddy_fn, "%s_BUDDY_%02d", fname, buddy_info->bnum);
        }
        nfname = (apidesc->get_multi_filename_cb ? apidesc->get_multi_filename_cb : _sion_get_multi_filename)(
          buddy_fn, buddy_info->filenum);

        DPRINTFP((256, __func__, _SION_DEFAULT_RANK,
          "open file for group #%d %s [step=%d grank=%2d of %2d, lrank=%2d of %2d filelrank=%2d]\n", buddy_info->groupid, nfname,
          buddy_info->stepnum, gendata->grank, gendata->gsize, buddy_info->rank, buddy_info->size, buddy_info->filelrank));

        /* test if tsk is a reader, a task is only in one step a reader */
        if ((buddy_info->myrole == SION_ROLE_READER) || (buddy_info->myrole == SION_ROLE_COLLECTOR_READER)) {
          buddy_info->sid = sid; /* use already defined sid, this will be the master data structure */
          rc = _sion_paropen_generic_one_file(buddy_info->sid, nfname, mode, gendata->numfiles, buddy_info->filenum,
            buddy_info->rank, buddy_info->size, gendata->grank, _SION_INTERNAL_FLAG_BUDDY_READ, gendata, buddies, options);

          /* store internal buddy data structure */
          /* test sid and get internal data structure */
          if ((rc < 0) || (sid < 0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR) || !(filedesc = _sion_vcdtovcon(sid))) {
            free(buddy_fn);
            return _sion_errorprint(
              SION_ID_NOT_VALID, _SION_ERROR_RETURN, "_sion_paropen_generic_buddy: invalid filedesc %d", sid);
          }

          /* store additional data */
          filedesc->buddies = buddies;

        } else {
          buddy_info->sid = _sion_newvcd(NULL, SION_FILEDESCRIPTOR); /* create a new sid */
          rc = _sion_paropen_generic_one_file(buddy_info->sid, nfname, mode, gendata->numfiles, buddy_info->filenum,
            buddy_info->rank, buddy_info->size, gendata->grank, _SION_INTERNAL_FLAG_BUDDY_READ, gendata, buddies, options);
        }

        DPRINTFP((1, __func__, gendata->grank, "BUDDYO afteropen [grank=%2d step=%d] rc=%d\n", gendata->grank, step, rc));
        free(nfname);

      } else {
        /* nothing to do in this step */
        DPRINTFP((1, __func__, gendata->grank, "BUDDYO no-op [grank=%2d]\n", gendata->grank));
      }

      /* barrier between steps */
      DPRINTFP((1, __func__, gendata->grank, "BUDDYO beforegbarrier [grank=%2d]\n", gendata->grank));
      apidesc->barrier_cb(gendata->comm_data_global);
      DPRINTFP((1, __func__, gendata->grank, "BUDDYO aftergbarrier [grank=%2d]\n", gendata->grank));

    } /* for steps */

  } else {
    free(buddies);
    free(buddy_fn);
    return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN, "_sion_paropen_generic_buddy: unknown file mode");
  }
  /* 'buddies' is still used later and should not be freed here. */
  /* free(buddies); */
  free(buddy_fn);
  apidesc->barrier_cb(gendata->comm_data_global);

  return rc;
}

#define DFUNCTION "_sion_parclose_generic_buddy"
int _sion_parclose_generic_buddy(int sid, int rank, int ntasks, _sion_generic_gendata *sion_gendata)
{
  int rc = SION_SUCCESS;
  _sion_filedesc *sion_filedesc;
  /* _sion_generic_apidesc *sion_apidesc; */
  _sion_generic_buddy *buddies, *buddyptr;
  int group, step;
  _sion_generic_buddy_info *buddy_info;

  int b, pass;

  /* obtain pointers to internal data structures */
  if ((sid < 0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR) || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN, "_sion_parclose_generic_buddy: invalid sion_filedesc %d", sid);
  }
  /* sion_apidesc=sion_gendata->apidesc; */
  buddies = sion_filedesc->buddies;

  if (sion_filedesc->mode == SION_FILEMODE_WRITE) {
    /* close buddy files */
    for (b = 0; b < sion_filedesc->buddylevel; b++) {
      buddyptr = &buddies[b];

      /* close files */
      for (pass = 1; pass <= _SION_BW_SCHED_NUM_PASSES; pass++) {
        DPRINTFP((1, DFUNCTION, sion_gendata->grank, "BUDDYC pass=%d [grank=%2d] op=%d\n", pass, sion_gendata->grank,
          _sion_buddy_bwsched(buddyptr->buddy_send.groupnum, sion_gendata->numfiles, pass)));

        switch (_sion_buddy_bwsched(buddyptr->buddy_send.groupnum, sion_gendata->numfiles, pass)) {
        case _SION_BW_SCHED_ACTIONA:
          DPRINTFP((1, DFUNCTION, sion_gendata->grank,
            "BUDDYc call parallel close COLL [gendata: grank=%2d, group=%d of %d, lrank=%d of %d]\n", sion_gendata->grank,
            buddyptr->buddy_coll.groupnum, sion_gendata->numfiles, buddyptr->buddy_coll.rank, buddyptr->buddy_coll.size));
          _sion_parclose_generic(buddyptr->buddy_coll.sid, buddyptr->buddy_coll.rank, buddyptr->buddy_coll.size,
            _SION_INTERNAL_FLAG_BUDDY_COLL, sion_gendata, buddyptr);
          break;
        case _SION_BW_SCHED_ACTIONB:
          DPRINTFP((1, DFUNCTION, sion_gendata->grank,
            "BUDDYc call parallel close SEND [gendata: grank=%2d, group=%d of %d, lrank=%d of %d]\n", sion_gendata->grank,
            buddyptr->buddy_send.groupnum, sion_gendata->numfiles, buddyptr->buddy_send.rank, buddyptr->buddy_send.size));
          _sion_parclose_generic(buddyptr->buddy_send.sid, buddyptr->buddy_send.rank, buddyptr->buddy_send.size,
            _SION_INTERNAL_FLAG_BUDDY_SEND, sion_gendata, buddyptr);
          break;
        default:
          DPRINTFP((1, DFUNCTION, sion_gendata->grank, "BUDDYC no-op [grank=%2d]\n", sion_gendata->grank));
          break;
        } /* switch */
      } /* for pass */
    } /* for buddies */

    free(buddies);

    /* STEP: close normal files */
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "call parallel close of %d files, sid=%d\n", sion_gendata->numfiles, sid));
    rc = _sion_parclose_generic(sid, sion_filedesc->rank, sion_filedesc->ntasks, _SION_INTERNAL_FLAG_NORMAL, sion_gendata, NULL);
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "leave parallel close of %d files in #tasks=%d globalrank=%d\n",
      sion_gendata->numfiles, sion_gendata->lsize, sion_gendata->grank));

  } else if (sion_filedesc->mode == SION_FILEMODE_READ) {
    /* loop over all groups */
    for (step = 0; step < buddies->numsteps; step++) {
      /* search for group with this step */
      group = -1;
      for (b = 0; ((b < buddies->numgroups) && (group == -1)); b++) {
        DPRINTFP((1, DFUNCTION, sion_gendata->grank, "check group #%d of %d %d==%d  %x\n", b, buddies->numgroups,
          buddies->groups[b]->stepnum, step, buddies->groups[b]));
        if (buddies->groups[b]->stepnum == step) {
          group = b;
        }
      }
      buddy_info = buddies->groups[group];

      if (group != -1) {
        /* call parclose for this group */
        DPRINTFP((1, DFUNCTION, sion_gendata->grank, "call parallel close of %d files, sid=%d\n", sion_gendata->numfiles, sid));
        buddies->currentgroup = group;
        rc = _sion_parclose_generic(
          buddy_info->sid, buddy_info->rank, buddy_info->size, _SION_INTERNAL_FLAG_BUDDY_READ, sion_gendata, buddies);
        DPRINTFP((1, DFUNCTION, sion_gendata->grank, "leave parallel close of %d files in #tasks=%d globalrank=%d\n",
          sion_gendata->numfiles, sion_gendata->lsize, sion_gendata->grank));
      }
    } /* step */

  } else {
    return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN, "_sion_parclose_generic_buddy: unknown file mode");
  }

  return rc;
}
#undef DFUNCTION

#define DFUNCTION "_sion_coll_fwrite_buddy"
int64_t _sion_coll_fwrite_buddy(const void *data, size_t size, size_t nitems, int sid, _sion_generic_gendata *sion_gendata)
{
  _sion_filedesc *sion_filedesc, *sion_filedesc_coll, *sion_filedesc_send;
  _sion_generic_apidesc *sion_apidesc;
  _sion_generic_buddy *buddies, *buddyptr;
  int collector = 0;

  DPRINTFP((1, DFUNCTION, sion_gendata->grank, "enter parallel write buddy\n"));

  /* obtain pointers to internal data structures */
  if ((sid < 0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR) || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    _sion_errorprint(SION_SIZE_NOT_VALID, _SION_ERROR_ABORT, "_sion_coll_fwrite_buddy: invalid sion_filedesc %d", sid);
    assert(0);
  }
  sion_apidesc = sion_gendata->apidesc;
  buddies = sion_filedesc->buddies;

  /* buddy files */
  int64_t bytes_written = 0;
  for (int b = 0; b < sion_filedesc->buddylevel; b++) {
    buddyptr = &buddies[b];

    /* get pointer to data structures */
    if ((buddyptr->buddy_coll.sid < 0) || (_sion_vcdtype(buddyptr->buddy_coll.sid) != SION_FILEDESCRIPTOR)
      || !(sion_filedesc_coll = _sion_vcdtovcon(buddyptr->buddy_coll.sid))) {
      _sion_errorprint(
        SION_SIZE_NOT_VALID, _SION_ERROR_ABORT, "_sion_coll_fwrite_buddy: invalid sion_filedesc %d", buddyptr->buddy_coll.sid);
      assert(0);
    }
    if ((buddyptr->buddy_send.sid < 0) || (_sion_vcdtype(buddyptr->buddy_send.sid) != SION_FILEDESCRIPTOR)
      || !(sion_filedesc_send = _sion_vcdtovcon(buddyptr->buddy_send.sid))) {
      _sion_errorprint(
        SION_SIZE_NOT_VALID, _SION_ERROR_ABORT, "_sion_coll_fwrite_buddy: invalid sion_filedesc %d", buddyptr->buddy_send.sid);
      assert(0);
    }

    /* open files */
    for (int pass = 1; pass <= _SION_BW_SCHED_NUM_PASSES; pass++) {
      DPRINTFP((1, DFUNCTION, sion_gendata->grank, "BUDDYw pass=%d [grank=%2d] op=%d\n", pass, sion_gendata->grank,
        _sion_buddy_bwsched(buddyptr->buddy_send.groupnum, sion_gendata->numfiles, pass)));

      switch (_sion_buddy_bwsched(buddyptr->buddy_send.groupnum, sion_gendata->numfiles, pass)) {
      case _SION_BW_SCHED_ACTIONA:
        DPRINTFP((1, DFUNCTION, sion_gendata->grank,
          "BUDDYw call parallel write COLL [gendata: grank=%2d, group=%d of %d, lrank=%d of %d] %d..%d\n", sion_gendata->grank,
          buddyptr->buddy_coll.groupnum, sion_gendata->numfiles, buddyptr->buddy_coll.rank, buddyptr->buddy_coll.size,
          buddyptr->buddy_coll.from_index, buddyptr->buddy_coll.to_index));
        int64_t ownnewposition = 0;
        if (sion_filedesc_coll->rank == collector) {
          ownnewposition = sion_filedesc_coll->currentpos;
          DPRINTFP((1, DFUNCTION, sion_gendata->grank, "BUDDYw call parallel write COLL currentpos=%d\n",
            (int)sion_filedesc_coll->currentpos));
        }

        bytes_written = sion_apidesc->gather_execute_cb(data, sion_filedesc_coll->fsblksize, buddyptr->buddy_coll.commgroup,
          collector, buddyptr->buddy_coll.from_index, buddyptr->buddy_coll.to_index, buddyptr->buddy_coll.sid,
          _sion_generic_collective_process_write, NULL, _sion_generic_collective_next_spec_empty);

        if (sion_filedesc_coll->rank == collector) {
          _sion_file_set_position(sion_filedesc_coll->fileptr, ownnewposition);
          sion_filedesc_coll->currentpos = ownnewposition;
          DPRINTFP((1, DFUNCTION, sion_gendata->grank, "BUDDYw call parallel write COLL currentpos=%d\n",
            (int)sion_filedesc_coll->currentpos));
        }
        break;
      case _SION_BW_SCHED_ACTIONB:
        DPRINTFP((1, DFUNCTION, sion_gendata->grank,
          "BUDDYw call parallel write SEND [gendata: grank=%2d, group=%d of %d, lrank=%d of %d] %d..%d\n", sion_gendata->grank,
          buddyptr->buddy_send.groupnum, sion_gendata->numfiles, buddyptr->buddy_send.rank, buddyptr->buddy_send.size,
          buddyptr->buddy_send.from_index, buddyptr->buddy_send.to_index));

        _sion_partitioned_range_iterator ranges =
          _sion_partitioned_range_iterator_from_filedesc(sion_filedesc_send, size * nitems, sion_filedesc_send->fsblksize);
        bytes_written = sion_apidesc->gather_execute_cb(data, sion_filedesc_send->fsblksize, buddyptr->buddy_send.commgroup,
          collector, buddyptr->buddy_send.from_index, buddyptr->buddy_send.to_index, buddyptr->buddy_send.sid,
          _sion_generic_collective_process_write, &ranges, _sion_generic_collective_next_write_spec);

        _sion_write_dry_run(sion_filedesc_send, nitems * size);
        break;
      default:
        DPRINTFP((1, DFUNCTION, sion_gendata->grank, "BUDDYw no-op [grank=%2d]\n", sion_gendata->grank));
        break;
      } /* switch */
    } /* for pass */
  } /* for buddies */

  DPRINTFP((1, DFUNCTION, sion_gendata->grank, "leave parallel write buddy\n"));

  int64_t items_written = size ? bytes_written / size : 0;

  return items_written;
}
#undef DFUNCTION

bool _sion_generic_collective_next_spec_empty(void *iterator, int64_t *range)
{
  (void)iterator;
  (void)range;

  return false;
}

#define DFUNCTION "_sion_coll_fread_buddy"
int64_t _sion_coll_fread_buddy(void *data, size_t size, size_t nitems, int sid)
{
  _sion_filedesc *sion_filedesc, *b_sion_filedesc;
  _sion_generic_gendata *sion_gendata;
  _sion_generic_apidesc *sion_apidesc;
  int64_t items_read = 0;

  /* obtain pointers to internal data structures */
  if ((sid < 0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR) || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    _sion_errorprint(SION_SIZE_NOT_VALID, _SION_ERROR_ABORT, "_sion_coll_fread_buddy: invalid sion_filedesc %d", sid);
    assert(0);
  }
  DPRINTFP((4, DFUNCTION, _SION_DEFAULT_RANK, "enter usecoll=%d collector=%d collsize=%d (%d tasks, %d files)\n",
    sion_filedesc->usecoll, sion_filedesc->collector, sion_filedesc->collsize, sion_filedesc->ntasks, sion_filedesc->nfiles));

  sion_gendata = sion_filedesc->dataptr;
  sion_apidesc = sion_gendata->apidesc;
  _sion_generic_buddy *buddies = sion_filedesc->buddies;

  /* needed for avoiding subsequent non-collective calls */
  sion_filedesc->collcmdused = 1;

  /* check collsize */
  if (sion_filedesc->collsize <= 0) {
    _sion_errorprint(SION_SIZE_NOT_VALID, _SION_ERROR_ABORT, "_sion_coll_fread_buddy: collsize=%d <= 0, returning ...\n",
      (int)sion_filedesc->collsize);
    assert(0);
  }

  /* read parts and scatter these to sender tasks via callback function */
  if (!sion_apidesc->execute_scatter_cb) {
    _sion_errorprint(SION_SIZE_NOT_VALID, _SION_ERROR_ABORT,
      "_sion_coll_fread_buddy: API %s not correctly initalized, collective I/O calls missing, aborting", sion_apidesc->name);
    assert(0);
  }

  /* loop over all groups */
  int64_t bytes_read = 0;
  for (int step = 0; step < buddies->numsteps; step++) {
    /* search for group with this step */
    int group = -1;
    for (int b = 0; ((b < buddies->numgroups) && (group == -1)); b++) {
      DPRINTFP((1, DFUNCTION, sion_gendata->grank, "check group #%d of %d %d==%d  %x\n", b, buddies->numgroups,
        buddies->groups[b]->stepnum, step, buddies->groups[b]));
      if (buddies->groups[b]->stepnum == step) {
        group = b;
      }
    }

    if (group != -1) {
      /* activity needed for this group */
      _sion_generic_buddy_info *buddy_info = buddies->groups[group];

      if ((buddy_info->sid < 0) || (_sion_vcdtype(buddy_info->sid) != SION_FILEDESCRIPTOR)
        || !(b_sion_filedesc = _sion_vcdtovcon(buddy_info->sid))) {
        _sion_errorprint(
          SION_SIZE_NOT_VALID, _SION_ERROR_ABORT, "_sion_coll_fread_buddy: invalid sion_filedesc %d", buddy_info->sid);
        assert(0);
      }

      /* parameter of callback function */
      int collector = 0, firstsender = buddy_info->from_index, lastsender = buddy_info->to_index;

      /* read own part */
      if (buddy_info->myrole == SION_ROLE_COLLECTOR_READER) {
        bytes_read = _sion_read_multi_chunk(b_sion_filedesc, data, nitems * size);
        firstsender++; /* must not read again */
      }

      if ((buddy_info->myrole == SION_ROLE_COLLECTOR) || (buddy_info->myrole == SION_ROLE_COLLECTOR_READER)) {
        /* save own position before reading at other positions */
        int64_t ownnewposition = sion_filedesc->currentpos;

        DPRINTFP((1, DFUNCTION, sion_gendata->grank,
          "call execute_gather #%d (step=%d, rank=%d, size=%d, collsize=%d, from=%2d, to=%2d) (bnum=%2d file# %2d) with role %s "
          "\n",
          buddy_info->groupid, buddy_info->stepnum, buddy_info->rank, buddy_info->size, buddy_info->collsize,
          buddy_info->from_index, buddy_info->to_index, buddy_info->bnum, buddy_info->filenum,
          _sion_buddy_role_to_str(buddy_info->myrole)));

        // TODO: handle return code
        sion_apidesc->execute_scatter_cb(data, b_sion_filedesc->fsblksize, buddy_info->commgroup, collector, firstsender,
          lastsender, buddy_info->sid, _sion_generic_collective_process_read, NULL, _sion_generic_collective_next_read_spec);

        /* set own position to end of own block read in this call */
        _sion_file_set_position(b_sion_filedesc->fileptr, ownnewposition);
        b_sion_filedesc->currentpos = ownnewposition;
      } else if (buddy_info->myrole == SION_ROLE_READER) {
        _sion_partitioned_limited_range_iterator ranges =
          _sion_partitioned_limited_range_iterator_from_filedesc(b_sion_filedesc, size * nitems, b_sion_filedesc->fsblksize);

        bytes_read =
          sion_apidesc->execute_scatter_cb(data, b_sion_filedesc->fsblksize, buddy_info->commgroup, collector, firstsender,
            lastsender, buddy_info->sid, _sion_generic_collective_process_read, &ranges, _sion_generic_collective_next_read_spec);

        _sion_read_dry_run(b_sion_filedesc, nitems * size);
      }

      if ((buddy_info->myrole == SION_ROLE_READER) || (buddy_info->myrole == SION_ROLE_COLLECTOR_READER)) {
        items_read = size ? bytes_read / size : 0;
      }
    } /* froup != -1 */

    /* barrier between steps */
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "BUDDYO beforegbarrier [grank=%2d]\n", sion_gendata->grank));
    sion_apidesc->barrier_cb(sion_gendata->comm_data_global);
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "BUDDYO aftergbarrier [grank=%2d]\n", sion_gendata->grank));
  } /* for step */

  DPRINTFP((4, DFUNCTION, _SION_DEFAULT_RANK, "leave usecoll=%d collector=%d collsize=%d (%d tasks, %d files) rc=%d\n",
    sion_filedesc->usecoll, sion_filedesc->collector, sion_filedesc->collsize, sion_filedesc->ntasks, sion_filedesc->nfiles,
    items_read));

  return items_read;
}
#undef DFUNCTION

#define DFUNCTION "_sion_buddy_map"
int _sion_buddy_map(_sion_generic_gendata *sion_gendata, int capability, int buddy_idx, _sion_generic_buddy_info *buddy_send,
  _sion_generic_buddy_info *buddy_coll)
{
  int rc = SION_SUCCESS;
  int tmpsize, t, g, g_map, p, p_map, n, n_map, orig, distance, distance_1step, pass;
  int *size_per_group = NULL, *group_map_orig_to_new, *group_map_new_to_orig, mynewgroupnr;
  int32_t helpint2[2];
  int32_t *tasktogroup = NULL;
  int32_t *tmpintfield1 = NULL;
  int comm_send_rank, comm_send_size, comm_send_grpnum;
  int comm_coll_rank, comm_coll_size, comm_coll_grpnum;
  void *dummycommgroup = NULL;

  DPRINTFP(
    (256, DFUNCTION, _SION_DEFAULT_RANK, "buddy_idx=%d capability=%d [gendata: grank=%d of %d, group=%d of %d, lrank=%d of %d]\n",
      buddy_idx, capability, sion_gendata->grank, sion_gendata->gsize, sion_gendata->filenumber, sion_gendata->numfiles,
      sion_gendata->lrank, sion_gendata->lsize));

  distance = buddy_idx; /* higher level buddy CP: groups will be remapped to d=1 buddy CP  */
  distance_1step = 1;

  /* allocate some fields */
  if (sion_gendata->grank == 0) {
    tmpsize = 1 * sion_gendata->gsize;
    tasktogroup = malloc(tmpsize * sizeof(int32_t));
    if (tasktogroup == NULL) {
      return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
        "_sion_buddy_map: cannot allocate temporary memory of size %lu (tasktogroup), aborting ...\n",
        (unsigned long)tmpsize * sizeof(int32_t));
    }

    tmpsize = 2 * sion_gendata->gsize;
    tmpintfield1 = malloc(tmpsize * sizeof(int32_t));
    if (tmpintfield1 == NULL) {
      free(tasktogroup);
      return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
        "_sion_buddy_map: cannot allocate temporary memory of size %lu (tmpintfield1), aborting ...\n",
        (unsigned long)tmpsize * sizeof(int32_t));
    }

    size_per_group = malloc(sion_gendata->numfiles * sizeof(int));
    if (size_per_group == NULL) {
      free(tasktogroup);
      free(tmpintfield1);
      return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
        "_sion_buddy_map: cannot allocate temporary memory of size %lu (size_per_group), aborting ...\n",
        (unsigned long)sion_gendata->numfiles * sizeof(int));
    }
  }

  /* on all tasks */
  group_map_orig_to_new = malloc(sion_gendata->numfiles * sizeof(int));
  if (group_map_orig_to_new == NULL) {
    free(tasktogroup);
    free(size_per_group);
    free(tmpintfield1);
    return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "_sion_buddy_map: cannot allocate temporary memory of size %lu (group_map_orig_to_new), aborting ...\n",
      (unsigned long)sion_gendata->numfiles * sizeof(int));
  }

  group_map_new_to_orig = malloc(sion_gendata->numfiles * sizeof(int));
  if (group_map_new_to_orig == NULL) {
    free(size_per_group);
    free(group_map_orig_to_new);
    free(tasktogroup);
    free(tmpintfield1);
    return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "_sion_buddy_map: cannot allocate temporary memory of size %lu (group_map_new_to_orig), aborting ...\n",
      (unsigned long)sion_gendata->numfiles * sizeof(int));
  }

  /* Step1: gather group number */
  helpint2[0] = sion_gendata->filenumber;
  sion_gendata->apidesc->gatherr_cb(helpint2, tasktogroup, sion_gendata->comm_data_global, _SION_INT32, 1, 0);

  /* Step2: gather local size */
  helpint2[0] = sion_gendata->lsize;
  helpint2[1] = capability;
  sion_gendata->apidesc->gatherr_cb(helpint2, tmpintfield1, sion_gendata->comm_data_global, _SION_INT32, 2, 0);
  if (sion_gendata->grank == 0) {
    for (t = 0; t < sion_gendata->gsize; t++) {
      DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK, "task-map --> t=%2d grpnr=%2d lsize=%2d capability=%d\n", t, tasktogroup[t],
        tmpintfield1[2 * t + 0], tmpintfield1[2 * t + 1]));
    }
    /* --> tasktogroup[t] = groupnr,  tmpintfield1[t][0] = lsize tmpintfield1[t][1] = capability  */
  }

  /* Step2: build vector with size info per group and group map */
  if (sion_gendata->grank == 0) {
    for (t = 0; t < sion_gendata->gsize; t++) {
      size_per_group[tasktogroup[t]] = tmpintfield1[2 * t + 0];
    }
  }

  for (g = 0; g < sion_gendata->numfiles; g++) {
    orig = (g * distance) % sion_gendata->numfiles;
    group_map_orig_to_new[orig] = g;
    group_map_new_to_orig[g] = orig;
  }

  for (g = 0; g < sion_gendata->numfiles; g++) {
    DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK, "grp-map --> g=%2d->%2d\n", g, group_map_orig_to_new[g]));
  }

  /* Step3: compute send group */
  if (sion_gendata->grank == 0) {
    for (t = 0; t < sion_gendata->gsize; t++) {
      g = tasktogroup[t];
      g_map = group_map_orig_to_new[g];
      p_map = (g_map - distance_1step + sion_gendata->numfiles) % sion_gendata->numfiles;
      p = group_map_new_to_orig[p_map];
      tmpintfield1[2 * t + 0] = g_map;
      tmpintfield1[2 * t + 1] = size_per_group[p];
      DPRINTFP(
        (256, DFUNCTION, _SION_DEFAULT_RANK, "comm-send --> t=%2d grpnr=%2d grpmap=%2d prevgr=%2d prevgrmap=%2d prev_lsize=%2d\n",
          t, g, tmpintfield1[2 * t + 0], p, p_map, tmpintfield1[2 * t + 1]));
    }
    /* --> tmpintfield1[t][0] = own_groupnr,  tmpintfield1[t][1] = lsize of prev_group  */
  }

  /* Step4: scatter info about send group */
  sion_gendata->apidesc->scatterr_cb(tmpintfield1, helpint2, sion_gendata->comm_data_global, _SION_INT32, 2, 0);

  /* Step5: compute ranks for send group */
  mynewgroupnr = comm_send_grpnum = helpint2[0];
  buddy_send->rank = comm_send_rank = helpint2[1] + sion_gendata->lrank;
  buddy_send->size = comm_send_size = helpint2[1] + sion_gendata->lsize;
  buddy_send->collsize = sion_gendata->lsize;
  buddy_send->groupnum = mynewgroupnr;
  buddy_send->filenum = group_map_new_to_orig[mynewgroupnr];
  buddy_send->from_index = helpint2[1];
  buddy_send->to_index = helpint2[1] + sion_gendata->lsize - 1;
  DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
    "COMM_SEND d=%d grpnum=%2d lrank=%2d lsize=%2d mynewgroupnr=%2d collsize=%2d %d..%d fnum=%d\n", distance, comm_send_grpnum,
    comm_send_rank, comm_send_size, mynewgroupnr, buddy_send->collsize, buddy_send->from_index, buddy_send->to_index,
    buddy_send->filenum));

  /* Step6: compute coll group */
  if (sion_gendata->grank == 0) {
    for (t = 0; t < sion_gendata->gsize; t++) {
      g = tasktogroup[t];
      g_map = group_map_orig_to_new[g];
      n_map = (g_map + distance_1step) % sion_gendata->numfiles;
      n = group_map_new_to_orig[n_map];
      tmpintfield1[2 * t + 0] = n_map;
      tmpintfield1[2 * t + 1] = size_per_group[n];
      DPRINTFP(
        (256, DFUNCTION, _SION_DEFAULT_RANK, "comm-coll --> t=%2d grpnr=%2d grpmap=%2d nextgr=%2d nextgrmap=%2d next_lsize=%2d\n",
          t, g, g_map, n, tmpintfield1[2 * t + 0], tmpintfield1[2 * t + 1]));
    }
    /* --> tmpintfield1[t][0] = next_groupnr_map, tmpintfield1[t][0] = lsize of next_group  */
  }

  /* Step7: scatter info about coll group */
  sion_gendata->apidesc->scatterr_cb(tmpintfield1, helpint2, sion_gendata->comm_data_global, _SION_INT32, 2, 0);

  /* Step8: compute ranks for coll group */
  comm_coll_grpnum = helpint2[0];
  buddy_coll->rank = comm_coll_rank = sion_gendata->lrank;
  buddy_coll->size = comm_coll_size = sion_gendata->lsize + helpint2[1];
  buddy_coll->collsize = helpint2[1];
  buddy_coll->groupnum = comm_coll_grpnum;
  buddy_coll->filenum = group_map_new_to_orig[comm_coll_grpnum];
  buddy_coll->from_index = sion_gendata->lsize;
  buddy_coll->to_index = sion_gendata->lsize + helpint2[1] - 1;
  DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK, "COMM_COLL d=%d grpnum=%2d lrank=%2d lsize=%2d collsize=%2d %d..%d fnum=%d\n",
    distance, comm_coll_grpnum, comm_coll_rank, comm_coll_size, buddy_coll->collsize, buddy_coll->from_index,
    buddy_coll->to_index, buddy_coll->filenum));

  /* Step9: create local communicators for send/coll */

  for (pass = 1; pass <= _SION_BW_SCHED_NUM_PASSES; pass++) {
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "pass=%d [grank=%2d] op=%d\n", pass, sion_gendata->grank,
      _sion_buddy_bwsched(mynewgroupnr, sion_gendata->numfiles, pass)));

    switch (_sion_buddy_bwsched(mynewgroupnr, sion_gendata->numfiles, pass)) {
    case _SION_BW_SCHED_ACTIONA:
      rc = sion_gendata->apidesc->create_lcg_cb(&buddy_coll->commgroup, sion_gendata->comm_data_global, sion_gendata->grank,
        sion_gendata->gsize, comm_coll_rank, comm_coll_size, comm_coll_grpnum, sion_gendata->numfiles);
      DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
        "COMM_created comm_coll (d=%1d, p=%1d) [gendata: mygrp=%2d grank=%2d of %2d, group=%2d of %2d, lrank=%2d of %2d]\n",
        distance, pass, mynewgroupnr, sion_gendata->grank, sion_gendata->gsize, comm_coll_grpnum, sion_gendata->numfiles,
        comm_coll_rank, comm_coll_size));
      break;
    case _SION_BW_SCHED_ACTIONB:
      rc = sion_gendata->apidesc->create_lcg_cb(&buddy_send->commgroup, sion_gendata->comm_data_global, sion_gendata->grank,
        sion_gendata->gsize, comm_send_rank, comm_send_size, comm_send_grpnum, sion_gendata->numfiles);
      DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
        "COMM_created comm_send (d=%1d, p=%1d) [gendata: mygrp=%2d grank=%2d of %2d, group=%2d of %2d, lrank=%2d of %2d]\n",
        distance, pass, mynewgroupnr, sion_gendata->grank, sion_gendata->gsize, comm_send_grpnum, sion_gendata->numfiles,
        comm_send_rank, comm_send_size));
      break;
    case _SION_BW_SCHED_NOACTION:
      rc = sion_gendata->apidesc->create_lcg_cb(&dummycommgroup, sion_gendata->comm_data_global, sion_gendata->grank,
        sion_gendata->gsize, 0, 1, -1, sion_gendata->numfiles);
      DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
        "COMM_created dummy (d=%1d, p=%1d) [gendata: mygrp=%2d grank=%2d of %2d, group=%2d of %2d, lrank=%2d of %2d]\n", distance,
        pass, mynewgroupnr, sion_gendata->grank, sion_gendata->gsize, -1, sion_gendata->numfiles, 0, 1));
      break;

    default:
      break;
    }
  }

  /* free fields */
  if (sion_gendata->grank == 0) {
    free(tasktogroup);
    free(tmpintfield1);
    free(size_per_group);
  }
  free(group_map_new_to_orig);
  free(group_map_orig_to_new);

  return rc;
}
#undef DFUNCTION

/* pass=1..3 */
#define DFUNCTION "_sion_buddy_bwsched"
int _sion_buddy_bwsched(int groupnr, int numgroups, int pass)
{
  int res = _SION_BW_SCHED_NOACTION;
  if (numgroups % 2 == 0) { /* #groups even */
    if (groupnr % 2 == 0) { /* groupnr even */
      if (pass == 1) {
        res = _SION_BW_SCHED_ACTIONA;
      }
      if (pass == 2) {
        res = _SION_BW_SCHED_ACTIONB;
      }
    } else { /* groupnr even */
      if (pass == 1) {
        res = _SION_BW_SCHED_ACTIONB;
      }
      if (pass == 2) {
        res = _SION_BW_SCHED_ACTIONA;
      }
    }
  } else {
    if (groupnr % 2 == 0) { /* groupnr even */
      if ((pass == 1) && (groupnr != (numgroups - 1))) {
        res = _SION_BW_SCHED_ACTIONA;
      }
      if ((pass == 2) && (groupnr != 0)) {
        res = _SION_BW_SCHED_ACTIONB;
      }
      if ((pass == 3) && (groupnr == 0)) {
        res = _SION_BW_SCHED_ACTIONB;
      }
      if ((pass == 3) && (groupnr == (numgroups - 1))) {
        res = _SION_BW_SCHED_ACTIONA;
      }
    } else { /* groupnr even */
      if (pass == 1) {
        res = _SION_BW_SCHED_ACTIONB;
      }
      if (pass == 2) {
        res = _SION_BW_SCHED_ACTIONA;
      }
    }
  }
  return res;
}
#undef DFUNCTION

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

  sion_apidesc = sion_gendata->apidesc;

  *mapping = NULL;
  *mapping_size = 0;

  /* mapping data will be collected by master of first physical file */
  if ((buddyptr->buddy_coll.groupnum == 0) && (buddyptr->buddy_coll.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_collect_mapping_buddy: 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_collect_mapping_buddy: Cannot allocate memory for receivemap");
    }
  }

  if ((buddyptr->buddy_coll.filenum == 0) && (buddyptr->buddy_coll.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] = buddyptr->buddy_send.filenum;
  lpos[1] = buddyptr->buddy_send.rank;
  sion_apidesc->gatherr_cb(&lpos, *mapping, sion_gendata->comm_data_global, _SION_INT32, 2, receiver);

  free(receivemap);

  return rc;
}
#undef DFUNCTION

#define DFUNCTION "_sion_generic_buddy_get_and_distribute_info_from_file"
int _sion_generic_buddy_get_and_distribute_info_from_file(_sion_generic_gendata *sion_gendata, char *fname, int root,
  int32_t *filenumber, int32_t *numfiles, int32_t *lrank, int32_t *lsize)
{
  int sid = -1, t;
  int rc = SION_SUCCESS;
  int *tasksinfile = NULL;
  int mapping_size = -1;
  int32_t *mapping = NULL;
  int32_t lpos[2];
  _sion_generic_apidesc *sion_apidesc;

  sion_apidesc = sion_gendata->apidesc;

  DPRINTFP((1, DFUNCTION, sion_gendata->grank, "entering function sion_gendata->grank=%d\n", sion_gendata->grank));

  if (sion_gendata->grank == root) {
    /* open and get mapping of sion file */
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "before open\n"));
    sid = _sion_open_read(fname, _SION_READ_MASTER_ONLY_OF_MULTI_FILES, &SION_OPTIONS_INIT);
    if (sid >= 0) {
      DPRINTFP((1, DFUNCTION, sion_gendata->grank, "after open\n"));
      rc = sion_get_mapping(sid, &mapping_size, &mapping, numfiles);
      DPRINTFP((1, DFUNCTION, sion_gendata->grank, "sion file %d files\n", *numfiles));
    } else {
      *numfiles = -1;
    }
  }

  /* each task has to know if more than one file was used in sion file */
  sion_apidesc->bcastr_cb(numfiles, sion_gendata->comm_data_global, _SION_INT32, 1, root);
  DPRINTFP((1, DFUNCTION, sion_gendata->grank, "%s: numfiles=%d\n", fname, *numfiles));

  if ((sion_gendata->grank == root) && (*numfiles > 1)) {
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "%s: mapping_size=%d sion_gendata->gsize=%d\n", fname, *numfiles, mapping_size,
      sion_gendata->gsize));
    if (mapping_size != sion_gendata->gsize) {
      return _sion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN,
        "_sion_generic_buddy_get_and_distribute_info_from_file: Incorrect sum of ntasks of files %d <> %d\n", mapping_size,
        sion_gendata->gsize);
    }
  }

  if (*numfiles < 0) {
    return _sion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN,
      "_sion_generic_buddy_get_and_distribute_info_from_file: could not get numfiles from sion file\n");
  }

  if (*numfiles > 1) {
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "before scatter\n"));
    if (sion_gendata->grank == root) {
      for (t = 0; t < mapping_size; t++) {
        DPRINTFP((1, DFUNCTION, sion_gendata->grank, "  %d -> (%d,%d)\n", t, mapping[t * 2], mapping[t * 2 + 1]));
      }
    }

    /* scatter mapping to all tasks */
    sion_apidesc->scatterr_cb(mapping, lpos, sion_gendata->comm_data_global, _SION_INT32, 2, root);
    *filenumber = lpos[0];
    *lrank = lpos[1];
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "after scatter filenum+lrank (%d,%d)\n", *filenumber, *lrank));

    /* compute and scatter number of tasks in each file  */
    if (sion_gendata->grank == root) {
      tasksinfile = malloc(*numfiles * sizeof(int));
      if (tasksinfile == NULL) {
        return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
          "_sion_generic_get_and_distribute_info_from_file: Cannot allocate memory for tasksinfile counter vector");
      }
      for (t = 0; t < *numfiles; t++) {
        tasksinfile[t] = 0; /* init counter */
      }
      for (t = 0; t < mapping_size; t++) {
        tasksinfile[mapping[t * 2]]++; /* count tasks in file  */
      }
      for (t = 0; t < mapping_size; t++) {
        mapping[t * 2 + 1] = tasksinfile[mapping[t * 2]]; /* set 2nd value of mapping to lsize  */
      }
    }
    sion_apidesc->scatterr_cb(mapping, lpos, sion_gendata->comm_data_global, _SION_INT32, 2, root);
    *lsize = lpos[1];
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "after scatter lsize (%d, %d of %d)\n", *filenumber, *lrank, *lsize));

    if (sion_gendata->grank == root) {
      free(tasksinfile);
    }
    /* WARNING: mapping file of sion file is now destroyed and should not be used until close  */

  } else {
    *filenumber = 0;
    *lrank = sion_gendata->grank;
    *lsize = sion_gendata->gsize;
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "only one file -> filenumber=%d lRank=%d\n", *filenumber, *lrank));
  }

  if (sion_gendata->grank == root) {
    /* frees also mapping vector */
    if (sid >= 0) {
      _sion_close_sid(sid);
    }
  }

  return rc;
}
#undef DFUNCTION

#define DFUNCTION "_sion_generic_buddy_get_and_distribute_info_from_one_file"
int _sion_generic_buddy_get_and_distribute_info_from_one_file(_sion_generic_gendata *sion_gendata, char *fname, int root,
  int32_t *filenumber, int32_t *numfiles, int32_t *lrank, int32_t *lsize)
{
  int rc = SION_SUCCESS;
  int t, mapping_size = 0, grank, *mapping = NULL;
  int32_t file_filenumber, file_numfiles, file_lrank, file_lsize;
  _sion_filedesc *sion_filedesc = NULL;
  _sion_fileptr *sion_fileptr;
  _sion_generic_apidesc *sion_apidesc;

  sion_apidesc = sion_gendata->apidesc;

  DPRINTFP((1, DFUNCTION, sion_gendata->grank, "entering function sion_gendata->grank=%d\n", sion_gendata->grank));

  if (sion_gendata->grank == root) {
    /* open and get globalranks of sion file */
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "before open\n"));
    /* allocate and initialise internal data structure with default values (NULL and -1) */
    sion_filedesc = _sion_alloc_filedesc();
    if (sion_filedesc == NULL) {
      return _sion_errorprint(SION_ID_UNDEF, SION_ID_UNDEF,
        "sion_open: cannot allocate filedescriptor structure of size %lu (sion_filedesc), aborting ...\n",
        (unsigned long)sizeof(sion_filedesc));
    }
    _sion_init_filedesc(sion_filedesc);

    /* open file */
    sion_fileptr = _sion_file_open(fname, SION_FILE_FLAG_POSIX | SION_FILE_FLAG_READ, 0);
    if (!sion_fileptr) {
      return _sion_errorprint(SION_ID_UNDEF, _SION_ERROR_RETURN, "sion_open: cannot open %s for reading, aborting ...\n", fname);
    }
    sion_filedesc->fileptr = sion_fileptr;

    /* read part of header which does not depend on ntasks */
    rc = _sion_read_header_fix_part(sion_filedesc);
    if (rc != SION_SUCCESS) {
      return _sion_errorprint(
        SION_ID_UNDEF, _SION_ERROR_RETURN, "sion_open: cannot read header from file %s, aborting ...\n", fname);
    }
    sion_filedesc->rank = 0;
    sion_filedesc->state = SION_FILESTATE_SEROPEN;
    sion_filedesc->mode = SION_FILEMODE_READ;

    /* memory allocation for internal fields */
    _sion_alloc_filedesc_arrays(sion_filedesc);

    /* read part of header which depends on ntasks */
    rc = _sion_read_header_var_part(sion_filedesc);
    if (rc != SION_SUCCESS) {
      return _sion_errorprint(
        SION_ID_UNDEF, _SION_ERROR_RETURN, "sion_open: cannot read header (var part) from file %s, aborting ...\n", fname);
    }

    file_numfiles = sion_filedesc->nfiles;
    file_filenumber = sion_filedesc->filenumber;
    file_lsize = sion_filedesc->ntasks;

    /* allocate mapping vector */
    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_buddy_get_and_distribute_info_from_one_file: cannot allocate temporary memory of size %lu (mapping), "
        "aborting ...\n",
        (unsigned long)mapping_size * 2 * sizeof(int32_t));
    }
  }

  /* each task has to know some number about local file */
  sion_apidesc->bcastr_cb(&file_numfiles, sion_gendata->comm_data_global, _SION_INT32, 1, root);
  DPRINTFP((1, DFUNCTION, sion_gendata->grank, "%s: numfiles=%d\n", fname, file_numfiles));
  sion_apidesc->bcastr_cb(&file_filenumber, sion_gendata->comm_data_global, _SION_INT32, 1, root);
  DPRINTFP((1, DFUNCTION, sion_gendata->grank, "%s: filenumber=%d\n", fname, file_filenumber));
  sion_apidesc->bcastr_cb(&file_lsize, sion_gendata->comm_data_global, _SION_INT32, 1, root);
  DPRINTFP((1, DFUNCTION, sion_gendata->grank, "%s: file_lsize=%d\n", fname, file_lsize));

  if (file_numfiles != *numfiles) {
    free(mapping);
    return _sion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN,
      "_sion_generic_buddy_get_and_distribute_info_from_one_file: Incorrect number of files %d <> %d\n", file_numfiles,
      *numfiles);
  }

  /* init mapping vector lrank */
  DPRINTFP((1, DFUNCTION, sion_gendata->grank, "before scatter\n"));
  if (sion_gendata->grank == root) {
    for (t = 0; t < mapping_size; t++) {
      mapping[t] = -1;
    }
    for (t = 0; t < file_lsize; t++) {
      grank = sion_filedesc->all_globalranks[t];
      if (grank >= 0) {
        mapping[grank] = t;
        DPRINTFP((1, DFUNCTION, sion_gendata->grank, "  file=%s mapping[%d] -> (%d)\n", fname, grank, mapping[grank]));
      }
    }
  }

  /* scatter mapping to all tasks */
  sion_apidesc->scatterr_cb(mapping, &file_lrank, sion_gendata->comm_data_global, _SION_INT32, 1, root);
  if (file_lrank != -1) {
    *filenumber = file_filenumber;
    *lrank = file_lrank;
    *lsize = file_lsize;
    DPRINTFP((1, DFUNCTION, sion_gendata->grank, "after scatter filenum=%d lrank=%d lsize=%d\n", *filenumber, *lrank, *lsize));
  }

  if (sion_gendata->grank == root) {
    /* close current file */
    _sion_free_filedesc_arrays(sion_filedesc);
    _sion_file_close(sion_filedesc->fileptr);
    sion_filedesc->fileptr = NULL;
    free(mapping);
  }

  return rc;
}
#undef DFUNCTION
