/****************************************************************************
**  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 <inttypes.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#if defined(_SION_BGQ)
#include <stdio_ext.h>
#endif

#if defined(_SION_BGP)
#include <stdio_ext.h>
#endif

#include "sion_buffer.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_hints.h"
#include "sion_internal.h"
#include "sion_internal_positioning.h"
#include "sion_internal_startptr.h"
#include "sion_keyvalue.h"
#include "sion_metadata.h"
#include "sion_serial.h"

/* INTERNAL */

int _sion_open_write(const char *fname, int n, const sion_options *options)
{
  if (n < 0) {
    return _sion_errorprint(
      SION_ID_UNDEF, _SION_ERROR_RETURN, "sion_open: wrong number of logical files specific: n=%d (<0), returning ...\n", n);
  }

  // allocate and initialise internal data structure with default values (NULL and -1)
  _sion_filedesc *filedesc = _sion_alloc_filedesc();
  if (filedesc == NULL) {
    return _sion_errorprint(SION_ID_UNDEF, _SION_ERROR_RETURN,
      "sion_open: cannot allocate filedescriptor structure of size %lu (filedesc), aborting ...\n",
      (unsigned long)sizeof(filedesc));
  }
  _sion_init_filedesc(filedesc);
  filedesc->fname = strdup(fname);

  // New sion file handle
  int sid = _sion_newvcd(filedesc, SION_FILEDESCRIPTOR);
  filedesc->sid = sid;

  // Allocate memory for storing MAXCHUNKS chunksize infos in internal structure
  _sion_realloc_filedesc_blocklist(filedesc, MAXCHUNKS);

  // determine endianness
  int32_t endianness = _sion_get_endianness_with_override(options->endianness);

  DPRINTFP((1, "sion_open", 0, " it is a write operation #tasks=%d\n", n));

  filedesc->state = SION_FILESTATE_SEROPEN; // Serial state
  filedesc->mode = SION_FILEMODE_WRITE; // Write mode
  filedesc->endianness = endianness; // Endianness
  filedesc->swapbytes = 0; // Endianness, swapping bytes
  filedesc->ntasks = n; // Number of logical files
  filedesc->nfiles = 1;
  filedesc->filenumber = 1;

  int64_t apiflag;
  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;
  }

  // memory allocation for internal fields
  _sion_alloc_filedesc_arrays(filedesc);

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

  // set file system block size
  if (options->fsblksize <= 0) {
    int64_t fsblocksize = (int64_t)_sion_file_get_opt_blksize(fileptr);
    if ((fsblocksize < 0) || (fsblocksize > SION_MAX_FSBLOCKSIZE)) {
      fsblocksize = SION_DEFAULT_FSBLOCKSIZE;
    }
    filedesc->fsblksize = fsblocksize;
    DPRINTFP((32, __func__, 0, "setting fsblksize to %lld\n", fsblocksize));
  } else {
    filedesc->fsblksize = options->fsblksize;
  }

  // set chunk sizes
  if (options->chunksizes) {
    if (options->n_chunksizes != n) {
      return _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
        "%s: size of chunksizes array in options (%d) does not match number of logical files (%d)\n", __func__,
        options->n_chunksizes, n);
    }
    for (int i = 0; i < n; i++) {
      filedesc->all_chunksizes[i] = options->chunksizes[i];
    }
  } else {
    for (int i = 0; i < n; i++) {
      filedesc->all_chunksizes[i] = filedesc->fsblksize;
    }
  }

  // set globalranks (0, 1, ...)
  for (int i = 0; i < n; i++) {
    filedesc->all_globalranks[i] = i;
  }

  // check for keyval parameter, needed at this point to set flag before writing header (flag1)
  _sion_keyval_check_env_with_options(filedesc, SION_OPEN_WRITE, options->keyval_mode);
  if (filedesc->keyvalmode != SION_KEYVAL_NONE) {
    _sion_alloc_filedesc_all_keyvalptr(filedesc);
  }

  _sion_calculate_startpointers(filedesc, 0);

  // write header
  _sion_write_header(filedesc);

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

  // write mapping info to meta data block
  _sion_write_header_var_part_mapping(filedesc, 0, NULL);

  // initalize current positions
  _sion_alloc_filedesc_block_arrays(filedesc);
  for (int i = 0; i < n; i++) {
    filedesc->all_blockcount[i] = 1;
    filedesc->all_currentpos[i] = filedesc->all_startpointers[i];
    filedesc->all_currentblocknr[i] = 0;
    filedesc->all_blocksizes[0 * n + i] = 0;
  }

  // set position to first block of rank 0
  filedesc->rank = 0;
  filedesc->chunksize = filedesc->all_chunksizes[0];
  filedesc->startpos = filedesc->all_startpointers[0];
  filedesc->currentpos = filedesc->startpos;
  filedesc->lastchunknr = 0;
  filedesc->currentblocknr = 0;
  _sion_file_set_position(fileptr, filedesc->currentpos);

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

  return sid;
}

int _sion_open_read(const char *fname, int read_all, const sion_options *options)
{
  /* allocate and initialise internal data structure with default values (NULL and -1) */
  _sion_filedesc *filedesc = _sion_alloc_filedesc();
  if (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(filedesc));
  }
  _sion_init_filedesc(filedesc);
  filedesc->fname = strdup(fname); /* Set the filename */

  /* New sion file handle */
  int sid = _sion_newvcd(filedesc, SION_FILEDESCRIPTOR);
  filedesc->sid = sid;

  int64_t apiflag;
  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;
  }

  DPRINTFP((1, "sion_open", 0, " it is a read operation\n"));

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

  /* read part of header which does not depend on ntasks */
  int rc = _sion_read_header_fix_part(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);
  }
  filedesc->rank = 0;
  filedesc->state = SION_FILESTATE_SEROPEN;
  filedesc->mode = SION_FILEMODE_READ;

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

  /* read part of header which depends on ntasks */
  rc = _sion_read_header_var_part(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);
  }

  /* read mapping file, will be moved later to master file */
  _sion_read_header_var_part_mapping(filedesc);

  /* set up internal data */
  _sion_calculate_startpointers(filedesc, filedesc->mapping_size);

  /* allocate memory and read all blocksizes of all tasks from meta data 2 */
  _sion_alloc_filedesc_block_arrays(filedesc);
  rc = _sion_read_header_var_part_blocksizes(filedesc);
  if (rc != SION_SUCCESS) {
    return _sion_errorprint(SION_ID_UNDEF, _SION_ERROR_RETURN,
      "sion_open: cannot read header (var part block sizes) from file %s, aborting ...\n", fname);
  }

  /* check for keyval parameter */
  _sion_keyval_check_env_with_options(filedesc, SION_OPEN_READ, options->keyval_mode);
  if (filedesc->keyvalmode != SION_KEYVAL_NONE) {
    _sion_alloc_filedesc_all_keyvalptr(filedesc);
  }

  /* read complete mapping */
  if ((filedesc->nfiles > 1) && (filedesc->filenumber == 0)) {
    if (read_all == _SION_READ_ALL_OF_MULTI_FILES) {
      sid = _sion_open_read_master(fname, options, filedesc);
    } else {
      sid = _sion_open_read_single(fname, options, filedesc);
    }
  } else { /* not opened on master file */
    sid = _sion_open_read_single(fname, options, filedesc);
  } /* end single file */

  return sid;
}

int _sion_open_read_single(const char *fname, const sion_options *options, _sion_filedesc *filedesc)
{
  int i, sid, blknum;

  /* set info for current rank and position */
  sid = filedesc->sid;
  filedesc->chunksize = filedesc->all_chunksizes[filedesc->rank];
  filedesc->size = filedesc->all_sizes[filedesc->rank];
  filedesc->startpos = filedesc->all_startpointers[filedesc->rank];
  filedesc->currentpos = filedesc->startpos;
  filedesc->currentblocknr = 0;
  filedesc->lastchunknr = filedesc->all_blockcount[filedesc->rank] - 1;
  for (blknum = 0; blknum <= filedesc->lastchunknr; blknum++) {
    filedesc->blocksizes[blknum] = filedesc->all_blocksizes[filedesc->ntasks * blknum + filedesc->rank];
  }
  _sion_file_set_position(filedesc->fileptr, filedesc->currentpos);

  /* initalize current positions */
  for (i = 0; i < filedesc->ntasks; i++) {
    filedesc->all_currentpos[i] = filedesc->all_startpointers[i];
    filedesc->all_currentblocknr[i] = 0;
  }

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

  return sid;
}

int _sion_open_read_master(const char *fname, const sion_options *options, _sion_filedesc *filedesc)
{
  int i, sid, blknum;
  int sid_master, subsid, lfile, lrank, filenr;
  _sion_filedesc *filedesc_master, *filedesc_sub;

  /* set info for current rank and position */
  filedesc->chunksize = filedesc->all_chunksizes[filedesc->rank];
  filedesc->size = filedesc->all_sizes[filedesc->rank];
  filedesc->startpos = filedesc->all_startpointers[filedesc->rank];
  filedesc->currentpos = filedesc->startpos;
  filedesc->currentblocknr = 0;
  filedesc->lastchunknr = filedesc->all_blockcount[filedesc->rank] - 1;
  for (blknum = 0; blknum <= filedesc->lastchunknr; blknum++) {
    filedesc->blocksizes[blknum] = filedesc->all_blocksizes[filedesc->ntasks * blknum + filedesc->rank];
  }
  _sion_file_set_position(filedesc->fileptr, filedesc->currentpos);

  /* initalize current positions */
  for (i = 0; i < filedesc->ntasks; i++) {
    filedesc->all_currentpos[i] = filedesc->all_startpointers[i];
    filedesc->all_currentblocknr[i] = 0;
  }

  /* create a master datastructure */
  filedesc_master = _sion_alloc_filedesc();
  if (filedesc_master == NULL) {
    return _sion_errorprint(SION_ID_UNDEF, _SION_ERROR_RETURN,
      "sion_open: cannot allocate filedescriptor structure of size %lu (filedesc_master), aborting ...\n",
      (unsigned long)sizeof(filedesc_master));
  }
  _sion_init_filedesc(filedesc_master);
  filedesc_master->fname = strdup(fname); /* Set the filename */
  filedesc_master->state = SION_FILESTATE_SEROPENMASTER;
  filedesc_master->mode = SION_FILEMODE_READ;

  /* New sion file handle */
  sid_master = _sion_newvcd(filedesc_master, SION_FILEDESCRIPTOR);
  filedesc_master->sid = sid_master;

  /* allocate vector for all datastructures */
  filedesc_master->multifiles = malloc(filedesc->nfiles * sizeof(_sion_filedesc *));
  if (filedesc_master->multifiles == NULL) {
    return _sion_errorprint(SION_ID_UNDEF, _SION_ERROR_RETURN,
      "cannot allocate filedescriptor structure vector of size %lu (filedesc), aborting ...\n",
      (unsigned long)filedesc->nfiles * sizeof(_sion_filedesc *));
  }

  /* first file is already opened */
  filedesc_master->multifiles[0] = filedesc;

  for (filenr = 1; filenr < filedesc->nfiles; filenr++) {
    char *multi_fname;
    multi_fname = _sion_get_multi_filename(fname, filenr);
    DPRINTFP((32, "sion_open", 0, "open now sub file = %d %s\n", filenr, multi_fname));
    subsid = _sion_open_read(multi_fname, _SION_READ_ALL_OF_MULTI_FILES, options);
    filedesc_sub = _sion_get_filedesc(subsid);
    filedesc_master->multifiles[filenr] = filedesc_sub;
    DPRINTFP((32, "sion_open", 0, "sub file = %d %s opened\n", filenr, multi_fname));
    free(multi_fname);
  }

  /* move mapping to master */
  filedesc_master->mapping_size = filedesc->mapping_size;
  filedesc->mapping_size = -1;
  filedesc_master->mapping = filedesc->mapping;
  filedesc->mapping = NULL;

  /* set rest of master data */
  filedesc_master->rank = 0;

  /* lookup file which contains current rank */
  lfile = filedesc_master->mapping[filedesc_master->rank * 2 + 0];
  lrank = filedesc_master->mapping[filedesc_master->rank * 2 + 1];
  filedesc_sub = filedesc_master->multifiles[lfile];

  filedesc_master->ntasks = filedesc_master->mapping_size;
  filedesc_master->endianness = filedesc_sub->endianness;
  filedesc_master->swapbytes = filedesc_sub->swapbytes;
  filedesc_master->fileversion = filedesc_sub->fileversion;
  filedesc_master->filesionversion = filedesc_sub->filesionversion;
  filedesc_master->filesionpatchlevel = filedesc_sub->filesionpatchlevel;
  filedesc_master->fsblksize = filedesc_sub->fsblksize;
  filedesc_master->swapbytes = filedesc_sub->swapbytes;
  filedesc_master->nfiles = filedesc_sub->nfiles;
  filedesc_master->flag1 = filedesc_sub->flag1;
  filedesc_master->flag2 = filedesc_sub->flag2;
  filedesc_master->keyvalmode = filedesc_sub->keyvalmode;
  filedesc_master->filenumber = 0;

  /* set info for current rank and position */
  filedesc_master->chunksize = filedesc_sub->all_chunksizes[lrank];
  filedesc_master->size = filedesc_sub->all_sizes[lrank];
  filedesc_master->startpos = filedesc_sub->all_startpointers[lrank];
  filedesc_master->currentpos = filedesc_master->startpos;
  filedesc_master->globalskip = filedesc_sub->globalskip;

  filedesc_master->currentblocknr = 0;
  filedesc_master->lastchunknr = filedesc_sub->all_blockcount[lrank] - 1;
  filedesc_master->start_of_varheader = filedesc_sub->start_of_varheader;

  /* set maxusedchunks */
  filedesc_master->maxusedchunks = filedesc->maxusedchunks;
  for (filenr = 1; filenr < filedesc->nfiles; filenr++)
    if (filedesc_master->maxusedchunks < filedesc_master->multifiles[filenr]->maxusedchunks) {
      filedesc_master->maxusedchunks = filedesc_master->multifiles[filenr]->maxusedchunks;
    }

  _sion_realloc_filedesc_blocklist(filedesc_master, filedesc_master->maxusedchunks);
  for (blknum = 0; blknum < filedesc_sub->all_blockcount[lrank]; blknum++) {
    filedesc_master->blocksizes[blknum] = filedesc_sub->all_blocksizes[filedesc_sub->ntasks * blknum + lrank];
  }

  /* set file pointer */
  filedesc_master->fileptr = filedesc_master->multifiles[lfile]->fileptr;

  /* switch to master */
  filedesc = filedesc_master;
  sid = sid_master;

  /* set position */
  _sion_file_set_position(filedesc->fileptr, filedesc->currentpos);

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

  return sid;
}

int _sion_open_rank(const char *fname, sion_open_mode mode, int rank, const sion_rank_options *options)
{
  int rc = SION_NOT_SUCCESS, sid = SION_ID_UNDEF;
  int mapping_filenr, mapping_lrank;

  DPRINTFTS(rank, "before open rank");
  // FIXME re-enable debug output of file_mode
  // DPRINTFP((1, "_sion_open_rank", *rank, "enter open of file %s in %s mode\n", fname, file_mode));

  /* allocate and initialise internal data structure with default values (NULL and -1) */
  _sion_filedesc *filedesc = _sion_alloc_filedesc();
  if (filedesc == NULL) {
    return _sion_errorprint_on_rank(SION_ID_UNDEF, _SION_ERROR_RETURN, rank,
      "_sion_open_rank: cannot allocate filedescriptor structure of size %lu (sion_filedesc), aborting ...",
      (unsigned long)sizeof(filedesc));
  }
  _sion_init_filedesc(filedesc);
  filedesc->fname = strdup(fname); /* Set the filename */

  /* New sion file handle */
  sid = _sion_newvcd(filedesc, SION_FILEDESCRIPTOR);
  filedesc->sid = sid;

  int64_t apiflag;
  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;
  }

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

    DPRINTFP((1, "_sion_open_rank", rank, " it is a write operation\n"));

    return _sion_errorprint_on_rank(
      SION_ID_UNDEF, _SION_ERROR_RETURN, rank, "_sion_open_rank: %s for writing currently not supported, aborting ...", fname);
  }

  /* **************** READ mode **************** */
  DPRINTFTS(rank, "start open read");
  DPRINTFP((1, "_sion_open_rank", rank, " it is a read operation rank=%d\n", rank));

  /* open file */
  _sion_fileptr *fileptr = _sion_file_open(fname, apiflag | SION_FILE_FLAG_READ, 0);
  if (!fileptr) {
    return _sion_errorprint_on_rank(
      SION_ID_UNDEF, _SION_ERROR_RETURN, rank, "_sion_open_rank: cannot open %s for reading, aborting ...", fname);
  }
  filedesc->fileptr = fileptr;
  DPRINTFTS(rank, "start after open file 1");
  rc = _sion_read_header_fix_part(filedesc);
  DPRINTFTS(rank, "start after read header fix part 1");
  if (rc == SION_NOT_SUCCESS) {
    return _sion_errorprint_on_rank(
      SION_ID_UNDEF, _SION_ERROR_RETURN, rank, "_sion_open_rank: cannot read header from file %s, aborting ...", fname);
  }
  filedesc->rank = rank;
  filedesc->state = SION_FILESTATE_SEROPENRANK;
  filedesc->mode = SION_FILEMODE_READ;

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

  /* read part of header which depends on ntasks */
  rc = _sion_read_header_var_part(filedesc);
  DPRINTFTS(rank, "start after read header var part 1");
  if (rc == SION_NOT_SUCCESS) {
    return _sion_errorprint_on_rank(
      SION_ID_UNDEF, _SION_ERROR_RETURN, rank, "_sion_open_rank: cannot read var part header from file %s, aborting ...", fname);
  }

  DPRINTFP((1, "_sion_open_rank", rank, "found number of files %d in %s rank=%d\n", filedesc->nfiles, filedesc->fname, rank));
  if (filedesc->nfiles > 1) {
    /* read mapping for rank */
    rc = _sion_read_header_var_part_mapping_rank(filedesc);
    DPRINTFTS(rank, "start after read mapping rank 1");
    if (rc == SION_NOT_SUCCESS) {
      return _sion_errorprint_on_rank(
        SION_ID_UNDEF, _SION_ERROR_RETURN, rank, "cannot read var part mapping header from file %s, aborting ...", fname);
    }
    mapping_filenr = filedesc->mapping[0];
    mapping_lrank = filedesc->mapping[1];
    DPRINTFP((1, "_sion_open_rank", rank, "data for rank %d is in file %d (lrank=%d)\n", rank, mapping_filenr, mapping_lrank));
    DPRINTFTS(rank, "start after read mapping 1");

    /* reopen one of other multi files if necessary */
    if (mapping_filenr > 0) {
      /* close current file */
      _sion_free_filedesc_arrays(filedesc);
      _sion_file_close(filedesc->fileptr);
      filedesc->fileptr = NULL;
      DPRINTFTS(rank, "start after close file 1");

      /*  and open again the correct one */
      filedesc->fname = _sion_get_multi_filename(fname, mapping_filenr);

      DPRINTFP((1, "_sion_open_rank", rank, "open file filedesc->fname=%s  fname=%s rank=%d\n", filedesc->fname, fname, rank));

      fileptr = _sion_file_open(filedesc->fname, apiflag | SION_FILE_FLAG_READ, 0);
      if (!fileptr) {
        return _sion_errorprint_on_rank(
          SION_ID_UNDEF, _SION_ERROR_RETURN, rank, "_sion_open_rank: cannot open %s for reading, aborting ...", fname);
      }
      filedesc->fileptr = fileptr;
      DPRINTFTS(rank, "start after open file i");

      rc = _sion_read_header_fix_part(filedesc);
      if (rc == SION_NOT_SUCCESS) {
        return _sion_errorprint_on_rank(
          SION_ID_UNDEF, _SION_ERROR_RETURN, rank, "sion_open: cannot read header from file %s, aborting ...", fname);
      }
      DPRINTFTS(rank, "start after read header fix part i");
      _sion_alloc_filedesc_arrays(filedesc);
      rc = _sion_read_header_var_part(filedesc);
      if (rc == SION_NOT_SUCCESS) {
        return _sion_errorprint_on_rank(
          SION_ID_UNDEF, _SION_ERROR_RETURN, rank, "cannot var mapping data from file %s, aborting ...", fname);
      }
      DPRINTFTS(rank, "start after read header var part i");

    } else {
      filedesc->fname = strdup(fname);
    }
  } else {
    mapping_lrank = rank;
  }

  DPRINTFP((1, "_sion_open_rank", rank, "max blocks=%d\n", filedesc->maxusedchunks));
  _sion_realloc_filedesc_blocklist(filedesc, filedesc->maxusedchunks);

  rc = _sion_read_header_var_part_mapping_size(filedesc);
  if (rc == SION_NOT_SUCCESS) {
    return _sion_errorprint_on_rank(
      SION_ID_UNDEF, _SION_ERROR_RETURN, rank, "cannot read mapping size from file %s, aborting ...", filedesc->fname);
  }

  _sion_calculate_startpointers(filedesc, filedesc->mapping_size);

  filedesc->rank = mapping_lrank;

  rc = _sion_read_header_var_part_blocksizes_rank(filedesc);
  if (rc == SION_NOT_SUCCESS) {
    return _sion_errorprint_on_rank(
      SION_ID_UNDEF, _SION_ERROR_RETURN, filedesc->rank, "cannot read header from file %s, aborting ...", fname);
  }
  DPRINTFTS(rank, "start after read header blocksizes i");

  /* set info for current rank and position */
  filedesc->chunksize = filedesc->all_chunksizes[filedesc->rank];
  filedesc->size = filedesc->all_sizes[filedesc->rank];
  filedesc->startpos = filedesc->all_startpointers[filedesc->rank];
  filedesc->currentpos = filedesc->startpos;
  filedesc->currentblocknr = 0;

  /* set file pointer */
  _sion_file_set_position(filedesc->fileptr, filedesc->currentpos);

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

  // FIXME re-enable debug output of file_mode
  // DPRINTFP((1, "_sion_open_rank", 0, "leave open of file %s in %s mode\n", fname, file_mode));
  DPRINTFTS(rank, "after open rank");

  return sid;
}

/* sion_filedesc has to be freed by calling function */
int _sion_close(_sion_filedesc *sion_filedesc)
{
  int rc = SION_SUCCESS;
  int blknum, rank, currentrank;
#ifdef SION_SERIAL_MASTER
  int filenr;
#endif

  DPRINTFP((1, "_sion_close", -1, "enter close   sid=%d currentpos=%15lld\n", sion_filedesc->sid, sion_filedesc->currentpos));
  _sion_print_filedesc(sion_filedesc, 512, "_sion_close", 1);

  if ((sion_filedesc->state != SION_FILESTATE_SEROPEN) && (sion_filedesc->state != SION_FILESTATE_SEROPENRANK)
    && (sion_filedesc->state != SION_FILESTATE_SEROPENMASTER)) {
    return _sion_errorprint_on_rank(SION_NOT_SUCCESS, _SION_ERROR_RETURN, sion_filedesc->rank,
      "_sion_close: invalid file open state (!SEROPEN), aborting %d ...", sion_filedesc->sid);
  }

  if (sion_filedesc->mode == SION_FILEMODE_WRITE) {
    /* write rest of metadata in headers */

    DPRINTFP((1, "_sion_close", -1, " file was opened in write mode sid=%d setpos to %lld\n", sion_filedesc->sid,
      sion_filedesc->end_of_header));

    if (sion_filedesc->usebuffer) {
      _sion_buffer_flush(sion_filedesc);
    }

    /* store data of current rank */
    currentrank = sion_filedesc->rank;
    sion_filedesc->all_currentpos[currentrank] = sion_filedesc->startpos + sion_filedesc->blocksizes[sion_filedesc->lastchunknr];
    sion_filedesc->all_currentblocknr[currentrank] = sion_filedesc->lastchunknr;
    sion_filedesc->all_blockcount[currentrank] = sion_filedesc->lastchunknr + 1;
    sion_filedesc->all_sizes[currentrank] = sion_filedesc->size;
    for (blknum = 0; blknum <= sion_filedesc->lastchunknr; blknum++) {
      sion_filedesc->all_blocksizes[sion_filedesc->ntasks * blknum + currentrank] = sion_filedesc->blocksizes[blknum];
    }

    /* search and set maxusedchunks */
    sion_filedesc->maxusedchunks = -1;
    for (blknum = 0; blknum < sion_filedesc->ntasks; blknum++)
      if (sion_filedesc->all_blockcount[blknum] > sion_filedesc->maxusedchunks) {
        sion_filedesc->maxusedchunks = (int)sion_filedesc->all_blockcount[blknum];
      }

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

    /* write sizes to first meta data block */
    _sion_write_header_sizes(sion_filedesc);

    /* set blocksizes of all not used chunks to zero */
    for (blknum = 0; blknum < sion_filedesc->maxusedchunks; blknum++) {
      for (rank = 0; rank < sion_filedesc->ntasks; rank++) {
        if (blknum >= sion_filedesc->all_blockcount[rank]) {
          sion_filedesc->all_blocksizes[sion_filedesc->ntasks * blknum + rank] = 0;
        } else {
          DPRINTFP((1, "_sion_close", -1, " blocksize rank %2d blk %2d -> %lld\n", rank, blknum,
            sion_filedesc->all_blocksizes[sion_filedesc->ntasks * blknum + rank]));
        }
      }
    }

    _sion_print_filedesc(sion_filedesc, 512, "_sion_close", 1);

    /* close file and free data structure */
    _sion_file_close(sion_filedesc->fileptr);
    sion_filedesc->fileptr = NULL;

  } else {
    /* read */

#ifdef SION_SERIAL_MASTER
    /* close all files if multi file */
    ;
    if (sion_filedesc->state == SION_FILESTATE_SEROPENMASTER) {
      for (filenr = 1; filenr < sion_filedesc->nfiles; filenr++) {
        _sion_close(sion_filedesc->multifiles[filenr]);
        _sion_freevcd(sion_filedesc->multifiles[filenr]->sid);
        _sion_free_filedesc(sion_filedesc->multifiles[filenr]);
        sion_filedesc->multifiles[filenr] = NULL;
      }
    } else {
#endif

      /* close file and free data structure */
      _sion_file_close(sion_filedesc->fileptr);
      sion_filedesc->fileptr = NULL;

#ifdef SION_SERIAL_MASTER
    }
#endif
  }

  DPRINTFP((1, "_sion_close", -1, "leave close   sid=%d\n", sion_filedesc->sid));

  return rc;
}

int _sion_close_sid(int sid)
{
  int rc = SION_SUCCESS;
  _sion_filedesc *sion_filedesc;

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

  if (rc == SION_SUCCESS) {
    _sion_freevcd(sid);
    _sion_free_filedesc(sion_filedesc);
  }

  return rc;
}

char *_sion_get_multi_filename(const char *fname, int filenumber)
{
  char *newfname;

  newfname = malloc(SION_FILENAME_LENGTH);
  if (newfname == NULL) {
    _sion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN, "_sion_get_multi_filename: Cannot allocate string newfname\n");
    return NULL;
  }
  if (filenumber > 0) {
    if (strlen(fname) < SION_FILENAME_LENGTH - 7) {
      sprintf(newfname, "%s.%06d", fname, filenumber);
    } else {
      _sion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN, "_sion_get_multi_filename: filename too long\n");
      free(newfname);
      return NULL;
    }
  } else {
    strcpy(newfname, fname);
  }

  return newfname;
}

int _sion_flush_file(int sid)
{
  int rc = SION_SUCCESS;
  _sion_filedesc *sion_filedesc;

  if ((_sion_vcdtype(sid) != SION_FILEDESCRIPTOR) || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    return _sion_errorprint(
      SION_NOT_SUCCESS, _SION_ERROR_RETURN, "_sion_flush_file: invalid sion_filedesc, aborting %d ...\n", sid);
  }
  DPRINTFP((32, "_sion_flush_file", sion_filedesc->rank, "flush sid=%d\n", sid));

  _sion_file_flush(sion_filedesc->fileptr);
  return rc;
}

int _sion_check_on_collective_mode(_sion_filedesc *sion_filedesc)
{
  int rc = SION_SUCCESS;

  /* don't check if collective merge  */
  /* branch to merge mode if enabled */
  if (sion_filedesc->collmergemode) {
    return rc;
  }

  /* this functions checks in non-collective call after a collective one was called
     this function ensures that buffer are flushed accordantly */
  /* if(sion_filedesc->fileptr_exported) { */
  if (sion_filedesc->usecoll) {
    sion_filedesc->usecoll = 0;
    if ((!sion_filedesc->collector) && (sion_filedesc->collcmdused)) {
      _sion_file_set_position(sion_filedesc->fileptr, sion_filedesc->currentpos);
    }
    return rc;
    /* return _sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_WARN,"_sion_check_on_collective_mode: individual SION read/write function call in collective mode, switching back to no-collective mode"); */
  }
  /* } */
  return rc;
}

int32_t _sion_get_endianness_with_flags(int64_t flags)
{
  int32_t endianness = 0;

  DPRINTFP((2, "_sion_get_endianness_flags", -1, "enter with flags 0x%x\n", flags));

  /* determine endianness */
  endianness = sion_get_endianness();
  /* make value symmetric (first byte = last byte) -> endianness independent */
  endianness |= endianness << 24;
  if (flags & _SION_FMODE_ENDIANNESS_SET) {
    /* manually set endianness on middle bytes */
    if (flags & _SION_FMODE_ENDIANNESS_BIG) {
      endianness |= 0x0101 << 8;
    }
  } else if (endianness & 1) {
    /* no manual setting of endianness: set middle bytes matching the out ones */
    endianness |= 0x0101 << 8;
  }

  DPRINTFP((2, "_sion_get_endianness_flags", -1, "leave with endianness 0x%x\n", endianness));

  return endianness;
}

int32_t _sion_get_endianness_with_override(sion_endianness override)
{
  // determine host endianness
  int32_t endianness = sion_get_endianness();
  // make value symmetric (first byte = last byte) -> endianness independent
  endianness |= endianness << 24;
  // if host endianness is big or override is big endian, set middle bytes
  if (override == SION_ENDIANNESS_BIG || (override == SION_ENDIANNESS_HOST && endianness & 1)) {
    endianness |= 0x0101 << 8;
  }

  DPRINTFP((2, __func__, -1, "leave with endianness 0x%x\n", endianness));
  return endianness;
}

#define DFUNCTION "_sion_getenv"
char *_sion_getenv(const char *name)
{
  char *name_with_prefix = NULL;
  char *getenv_result = NULL;
  const int full_len = strlen(name) + strlen(SION_LVERSION_PREFIX) + 1;

  DPRINTFP((2, DFUNCTION, -1, "enter: name = %s\n", name));

  name_with_prefix = malloc(full_len);
  sprintf(name_with_prefix, "%s%s", SION_LVERSION_PREFIX, name);

  getenv_result = getenv(name_with_prefix);
  free(name_with_prefix);

  DPRINTFP((2, DFUNCTION, -1, "leave: getenv_result = %s\n", getenv_result));

  return getenv_result;
}
#undef DFUNCTION

/* write data block to current and following chunks, data could be distributed over several chunks */
size_t _sion_write_multi_chunk(_sion_filedesc *filedesc, const void *data, int64_t bytes_to_write)
{
  DPRINTFP((2, __func__, -1, "enter\n"));

  _sion_range_iterator ranges = _sion_range_iterator_from_filedesc(filedesc, bytes_to_write);
  _sion_range_kind kind;
  _sion_range range;
  while ((kind = _sion_range_iterator_next(&ranges, &range))) {
    if (kind == _SION_NEXT) {
      _sion_apply_hints(filedesc, SION_HINTS_FREE_TYPE_CHUNK);
      filedesc->currentblocknr++;
      _sion_apply_hints(filedesc, SION_HINTS_ACCESS_TYPE_CHUNK);

      _sion_file_set_position(filedesc->fileptr, range.start);
    }

    DPRINTFP(
      (2, __func__, -1, " writing %" PRId64 " bytes at position %" PRId64 " (chunk # %" PRId64 "), chunksize %" PRId64 "\n",
        range.length, filedesc->currentblocknr, range.start, filedesc->chunksize));

    int64_t bytes_written = _sion_file_write(data, range.length, filedesc->fileptr);
    if (bytes_written != range.length) {
      return _sion_errorprint_on_rank(SION_SIZE_NOT_VALID, _SION_ERROR_RETURN, filedesc->rank,
        "could not write data (%" PRId64 " bytes) to file (bytes written=%" PRId64 " sid=%d) ...", range.length, bytes_written,
        filedesc->sid);
    }

    data = ((char *)data) + range.length;
  }

  filedesc->currentpos = _sion_file_get_position(filedesc->fileptr);
  int64_t new_size = _sion_get_logical_position(filedesc);
  if (new_size > filedesc->size) {
    filedesc->size = new_size;
    filedesc->lastchunknr = new_size / filedesc->chunksize
      + !!(new_size % filedesc->chunksize)
      - 1;
  }

  DPRINTFP((2, __func__, -1, "leave rc=%" PRId64 "\n", bytes_to_write));
  return bytes_to_write;
}

/* read data block from current and following chunks, data could be distributed over several chunks */
size_t _sion_read_multi_chunk(_sion_filedesc *filedesc, void *data, int64_t bytes_to_read)
{
  DPRINTFP((2, __func__, -1, "enter\n"));

  _sion_limited_range_iterator ranges = _sion_limited_range_iterator_from_filedesc(filedesc, bytes_to_read);
  _sion_range_kind kind;
  _sion_range range;
  int64_t total_bytes_read = 0;
  while ((kind = _sion_limited_range_iterator_next(&ranges, &range))) {
    if (kind == _SION_NEXT) {
      filedesc->currentblocknr++;
      _sion_file_set_position(filedesc->fileptr, range.start);
    }

    DPRINTFP(
      (2, __func__, -1, " reading %" PRId64 " bytes at position %" PRId64 " (chunk # %" PRId64 "), chunksize %" PRId64 "\n",
        range.length, filedesc->currentblocknr, range.start, filedesc->chunksize));

    int64_t bytes_read = _sion_file_read((char *)data + total_bytes_read, range.length, filedesc->fileptr);
    if (bytes_read != range.length) {
      return _sion_errorprint_on_rank(SION_SIZE_NOT_VALID, _SION_ERROR_RETURN, filedesc->rank,
        "could not read data (%" PRId64 " bytes) to file (bytes read=%" PRId64 " sid=%d) ...", range.length, bytes_read,
        filedesc->sid);
    }

    total_bytes_read += bytes_read;
  }

  filedesc->currentpos = _sion_file_get_position(filedesc->fileptr);
  DPRINTFP((2, __func__, -1, "leave rc=%" PRId64 "\n", total_bytes_read));
  return total_bytes_read;
}

#define DFUNCTION "_sion_move_to_next_chunk"
int _sion_move_to_next_chunk(_sion_filedesc *sion_filedesc)
{
  int rc = SION_SUCCESS;

  DPRINTFP((2, DFUNCTION, -1, "enter\n"));

  /* is another chink available */
  if (sion_filedesc->currentblocknr < sion_filedesc->lastchunknr) {
    /* move to next chunk */
    sion_filedesc->currentblocknr++;
    sion_filedesc->currentpos = sion_filedesc->startpos + sion_filedesc->currentblocknr * sion_filedesc->globalskip;
    _sion_file_set_position(sion_filedesc->fileptr, sion_filedesc->currentpos);
  } else {
    rc = SION_NOT_SUCCESS;
  }

  DPRINTFP((2, DFUNCTION, -1, "leave, rc = %d\n", rc));
  return rc;
}
#undef DFUNCTION

sion_io_stat_t *_sion_alloc_io_info(int p_nf)
{
  sion_io_stat_t *p = NULL;
  int i;

  /* allocate data structure */
  p = malloc(sizeof(sion_io_stat_t));
  if (p == NULL) {
    _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "_sion_alloc_io_info: cannot allocate memory of size %lu (p_fn), aborting ...\n", (unsigned long)sizeof(sion_io_stat_t));
    return NULL;
  }

  p->nfiles = p_nf;

  /* allocate vectors */
  p->names = malloc(p_nf * sizeof(char *));
  if (p->names == NULL) {
    _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "_sion_alloc_io_info: cannot allocate memory of size %lu (names), aborting ...\n", (unsigned long)p_nf * sizeof(char *));
    free(p);
    return NULL;
  }

  p->sizes = malloc(p_nf * sizeof(size_t));
  if (p->sizes == NULL) {
    _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "_sion_alloc_io_info: cannot allocate memory of size %lu (sizes), aborting ...\n", (unsigned long)p_nf * sizeof(size_t));
    free(p->names);
    free(p);
    return NULL;
  }

  p->roles = malloc(p_nf * sizeof(int));
  if (p->roles == NULL) {
    _sion_errorprint(SION_ID_NOT_VALID, _SION_ERROR_RETURN,
      "_sion_alloc_io_info: cannot allocate memory of size %lu (roles), aborting ...\n", (unsigned long)p_nf * sizeof(int));
    free(p->names);
    free(p->sizes);
    free(p);
    return NULL;
  }
  /* init fields */
  for (i = 0; i < p_nf; i++) {
    p->names[i] = NULL;
    p->sizes[i] = 0;
    p->roles[i] = SION_ROLE_NONE;
  }

  return p;
}

int _sion_free_io_info(sion_io_stat_t *p)
{
  int rc = SION_SUCCESS;

  free(p->names);
  free(p->sizes);
  free(p->roles);
  free(p);

  return rc;
}

void _sion_write_dry_run(_sion_filedesc *filedesc, int64_t amount)
{
  assert(filedesc);
  assert(amount >= 0);

  _sion_chunk_sequence sequence = _sion_chunk_sequence_from_filedesc(filedesc);
  filedesc->currentpos = _sion_chunk_sequence_translate_l2p(&sequence, _sion_get_logical_position(filedesc) + amount);
  filedesc->currentblocknr = (filedesc->currentpos - filedesc->startpos) / filedesc->globalskip;
  if (filedesc->fileptr) {
    _sion_file_set_position(filedesc->fileptr, filedesc->currentpos);
  }

  int64_t new_size = _sion_get_logical_position(filedesc);
  if (new_size > filedesc->size) {
    filedesc->size = new_size;
    filedesc->lastchunknr = new_size / filedesc->chunksize
      + !!(new_size % filedesc->chunksize)
      - 1;
  }
}

void _sion_read_dry_run(_sion_filedesc *filedesc, int64_t amount)
{
  assert(filedesc);
  assert(amount >= 0);

  _sion_limited_range_iterator ranges = _sion_limited_range_iterator_from_filedesc(filedesc, amount);
  _sion_range range;
  while (_sion_limited_range_iterator_next(&ranges, &range)) {
  }
  filedesc->currentpos = range.start + range.length;
  filedesc->currentblocknr = (filedesc->currentpos - filedesc->startpos) / filedesc->globalskip;
  if (filedesc->fileptr) {
    _sion_file_set_position(filedesc->fileptr, filedesc->currentpos);
  }
}
