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

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
#include <assert.h>

#if defined(_BGL)
#include <rts.h>
#ifndef __USE_FILE_OFFSET64
#define __USE_FILE_OFFSET64
#endif
#endif

#include <sys/time.h>

#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

#include <unistd.h>

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


FILE     *sion_get_fp(int sid)
{
  _sion_filedesc *sion_filedesc;

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

  return (SIONFP);
}

/*!
 * @brief Function to open a sion file in serial mode.
 *
 * This function opens all tasks of a sion file in serial mode. The
 * meta-data of all tasks will be loaded in memory, which allows to
 * access different parts of the file by using sion_seek_fp(), for
 * example to switch between ranks. Default position fo the
 * filepointer is before the first byte of the first task.
 *
 * In the case of multi-file sion-files sion_open() open internally
 * each physical file with sion_open recursively. sion_open manage for
 * each physical file an own sion file descriptor and data structure.
 *
 * Warning: multi-file-support is not implemented in sion_open of the
 * current version (1.1p5).
 *
 * @param[in]           fname           name of file, should equal on all tasks
 * @param[in,out]       file_mode       like the type parameter of fopen (currently recognized options: "rb", "wb")
 * @param[in]           ntasks          number of tasks used to write this file
 * @param[in]           nfiles          number of physical files
 * @param[in,out]       chunksizes      chunksize for each task
 * @param[in,out]       fsblksize       blocksize of filesystem, must be equal on all processors
 * @param[in]           globalranks     rank numbers for which the file should be open;
 *                                      will be stored in sion file, useful if comm is not MPI_COMM_WORLD
 *                                      typical: globalrank= rank in MPI_COMM_WORLD
 * @param[out]          fileptr         filepointer for this task
 *
 * @retval              sid             sion file handle or -1 if error occured
 */

int sion_open(char *fname, char *file_mode, int *ntasks, int *nfiles, sion_int64 **chunksizes, sion_int32 *fsblksize, int **globalranks, FILE **fileptr)
{

  int       rc, i, sid, filenumber;
  char     *prefix;

  sion_int32 maxblocks;
  int       endianness, swapbytes, mapping_size;
  sion_int64 *sion_chunksizes, *sion_startpointers, *sion_globalranks, *sion_blocksizes, *sion_blockcount;
  sion_int64 start_of_varheader, *sion_currentpos, *sion_currentblocknr, lglobalskip;
  _sion_filedesc *sion_filedesc;
  sion_int32 sion_version, sion_version_patchlevel, sion_fileformat_version;
  sion_int64 flag1 = _SION_FLAG1_NONE, flag2 = _SION_FLAG2_NONE;
  sion_int32 *mapping;
  sion_int64  new_fsblocksize;
  /*   sion_int64 lstartpointer,lpartnum,lglobalskip; */

  DPRINTFP((1, "sion_open", 0, "enter open of file %s in %s mode\n", fname, file_mode));

  prefix = malloc(SION_FILENAME_LENGTH);
  if (prefix == NULL) {
    return(_sion_errorprint(-1,_SION_ERROR_RETURN,"cannot allocate prefix string of size %lu , aborting ...\n", (unsigned long) SION_FILENAME_LENGTH));
  }

  _sion_alloc_fd_list();

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

  /* Initialise with default values (NULL and -1) */
  _sion_init_filedesc(sion_filedesc);

  /* New Sion file handle */
  sid = _sion_newvcd(sion_filedesc, SION_FILEDESCRIPTOR);

  //  sion_filedesc->maxblocks   = -1;
  //  sion_filedesc->chunksizes  = NULL;

  /* Allocate memory for an internal sion structure with MAXBLOCKS chunks */
  _sion_realloc_blocklist(sion_filedesc, MAXBLOCKS);

  sion_filedesc->numblocks = 0; /* Set the current number of chunks */
  sion_filedesc->fname = strdup(fname); /* Set the filename */
  //  sion_filedesc->end_of_header=-1;
  sion_filedesc->debug = 0;     /* disable the debug flag in the file */

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

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

    sion_filedesc->state = SION_FILESTATE_SEROPEN;      /* Serial state */
    sion_filedesc->mode = SION_FILEMODE_WRITE;  /* Write mode */
    sion_filedesc->endianness = sion_get_endianness();  /* Endianness */
    sion_filedesc->swapbytes = 0;       /* Endianness, swapping bytes */
    sion_filedesc->fsblksize = *fsblksize;      /* Filesystem block size */
    sion_filedesc->ntasks = *ntasks;    /* Number of tasks using this file */
    sion_filedesc->nfiles = 1;
    sion_filedesc->filenumber = 1;


    /* memory allocation and default init */
    _sion_allocate_sion_arrays(*ntasks, &sion_chunksizes, &sion_startpointers, &sion_globalranks);

    sion_filedesc->all_chunksizes  = sion_chunksizes;
    sion_filedesc->all_globalranks = sion_globalranks;
    sion_filedesc->all_startpointers = sion_startpointers;

    /* open file */
    *fileptr = _sion_create_and_open_file_for_write(fname);
    if(*fsblksize==-1) {
      /* check with fstat fsblksize */
      new_fsblocksize=(sion_int64) _sion_get_opt_blksize(*fileptr);
      if((new_fsblocksize<0) || (new_fsblocksize>SION_MAX_FSBLOCKSIZE)) new_fsblocksize=SION_DEFAULT_FSBLOCKSIZE;
    }
    if(*fsblksize==-1) {
      *fsblksize=new_fsblocksize;
      sion_filedesc->fsblksize = *fsblksize;
      DPRINTFP((32, "_sion_paropen_multi_generic", 0, "setting fsblksize to %lld\n", new_fsblocksize));
    }

    /* initialize chunksizes and partnums */
    for (i = 0; i < *ntasks; i++) {
      sion_chunksizes[i] = (sion_int64) (*chunksizes)[i];
      sion_globalranks[i] = (sion_int64) (*globalranks)[i];
    }
    _sion_calculate_startpointers(*ntasks, sion_chunksizes, *fsblksize, sion_startpointers, &lglobalskip);
    sion_filedesc->globalskip = lglobalskip;

    /* open file */
    if (!*fileptr) {
      return(_sion_errorprint(-1,_SION_ERROR_RETURN,"sion_open: cannot open %s for writing, aborting ...\n", fname));
    }
    sion_filedesc->fd = *fileptr;
    if(*fsblksize==-1) {
      /* check with fstat fsblksize */
      new_fsblocksize=(sion_int64) _sion_get_opt_blksize(*fileptr);
      if((new_fsblocksize<0) || (new_fsblocksize>SION_MAX_FSBLOCKSIZE)) new_fsblocksize=SION_DEFAULT_FSBLOCKSIZE;
      *fsblksize=new_fsblocksize;
      sion_filedesc->fsblksize = *fsblksize;
      DPRINTFP((32, "_sion_paropen_multi_generic", 0, "setting fsblksize to %lld\n", new_fsblocksize));
    }

    /* write header */
    _sion_write_header(*fileptr, *fsblksize, *ntasks, sion_globalranks, sion_chunksizes, 1, 1, flag1, flag2, fname, sion_filedesc->endianness);

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

    /* initalize current positions */

    _sion_allocate_sion_arrays_locations(*ntasks, sion_filedesc->maxblocks,
                                         &sion_blockcount, &sion_currentpos, &sion_currentblocknr, &sion_blocksizes);
    sion_filedesc->all_blockcount = sion_blockcount;
    sion_filedesc->all_currentpos = sion_currentpos;
    sion_filedesc->all_currentblocknr = sion_currentblocknr;
    sion_filedesc->all_blocksizes = sion_blocksizes;

    for (i = 0; i < *ntasks; i++) {
      sion_blockcount[i] = 0;
      sion_currentpos[i] = sion_startpointers[i];
      sion_currentblocknr[i] = 0;
      sion_blocksizes[0 * i * *ntasks] = 0;
    }

    /* set position to first block of rank 0 */
    sion_filedesc->rank = 0;
    sion_filedesc->chunksize = sion_filedesc->all_chunksizes[0];
    sion_filedesc->numblocks = sion_filedesc->all_blockcount[0];
    sion_filedesc->startpos = sion_filedesc->all_startpointers[0];

    _sion_flush_file_fp(*fileptr);
    _sion_set_position(sion_filedesc->fd, sion_filedesc->startpos);
    sion_seek(sid, 0, SION_CURRENT_BLK, SION_CURRENT_POS);

  }
  else {
    /* **************** READ mode **************** */
    DPRINTFP((1, "sion_open", 0, " it is a read operation\n"));

    /* open file */
    *fileptr = _sion_open_file_for_read(fname);
    if (!*fileptr) {
      return(_sion_errorprint(-1,_SION_ERROR_RETURN,"sion_open: cannot open %s for writing, aborting ...\n", fname));
    }
    sion_filedesc->fd = *fileptr;

    rc =
      _sion_read_header_fixed_part(*fileptr, &endianness, &swapbytes,
                                   fsblksize, ntasks, nfiles, &filenumber,
                                   &flag1, &flag2, prefix, &sion_version, &sion_version_patchlevel, &sion_fileformat_version);
    if (!rc) {
      return(_sion_errorprint(-1,_SION_ERROR_RETURN,"sion_open: cannot read header from file %s, aborting ...\n", fname));
    }
    sion_filedesc->endianness = endianness;
    sion_filedesc->swapbytes = swapbytes;
    sion_filedesc->fsblksize = *fsblksize;
    sion_filedesc->ntasks = *ntasks;
    sion_filedesc->nfiles = *nfiles;
    sion_filedesc->filenumber = filenumber;
    sion_filedesc->rank = 0;
    sion_filedesc->state = SION_FILESTATE_SEROPEN;
    sion_filedesc->mode = SION_FILEMODE_READ;
    sion_filedesc->fileversion = sion_fileformat_version;

    _sion_allocate_sion_arrays(*ntasks, &sion_chunksizes, &sion_startpointers, &sion_globalranks);
    sion_filedesc->all_chunksizes = sion_chunksizes;
    sion_filedesc->all_globalranks = sion_globalranks;
    sion_filedesc->all_startpointers = sion_startpointers;

    rc = _sion_read_header_var_part(*fileptr, *ntasks, sion_globalranks, sion_chunksizes, &maxblocks, &start_of_varheader, swapbytes);
    sion_filedesc->maxblocks = maxblocks;
    sion_filedesc->start_of_varheader = start_of_varheader;
    sion_filedesc->prefix = strdup(prefix);

    /* only for outgoing parameters chunksizes and globalranks */
    if (chunksizes != NULL) {

      sion_int64 *helpptr = NULL;

      if ((*chunksizes) == NULL) {
        helpptr = (sion_int64 *) malloc(*ntasks * sizeof(sion_int64));
        if (helpptr == NULL) {
	  return(_sion_errorprint(-1,_SION_ERROR_RETURN,"cannot allocate filedescriptor structure of size %lu (chunksizes), aborting ...\n", (unsigned long) sizeof(sion_int64)));
        }
      }
      *chunksizes = helpptr;
      for (i = 0; i < *ntasks; i++) {
        helpptr[i] = sion_chunksizes[i];
      }
    }
    if (globalranks != NULL) {
      int      *helpptr = NULL;
      if ((*globalranks) == NULL) {
        helpptr = (int *) malloc(*ntasks * sizeof(int));
        if (helpptr == NULL) {
	  return(_sion_errorprint(-1,_SION_ERROR_RETURN,"sion_open: cannot allocate memory of size %lu (globalranks), aborting ...\n", (unsigned long) (*ntasks) * sizeof(int)));
        }
      }
      *globalranks = helpptr;
      for (i = 0; i < (*ntasks); i++) {
        helpptr[i] = (int) sion_globalranks[i];
      }
    }


    _sion_calculate_startpointers(*ntasks, sion_chunksizes, *fsblksize, sion_startpointers, &lglobalskip);
    sion_filedesc->globalskip = lglobalskip;

    _sion_allocate_sion_arrays_locations(*ntasks, maxblocks, &sion_blockcount, &sion_currentpos, &sion_currentblocknr, &sion_blocksizes);
    sion_filedesc->all_blockcount = sion_blockcount;
    sion_filedesc->all_currentpos = sion_currentpos;
    sion_filedesc->all_currentblocknr = sion_currentblocknr;
    sion_filedesc->all_blocksizes = sion_blocksizes;

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

    rc = _sion_read_header_var_part_locations(*fileptr, *ntasks, maxblocks, start_of_varheader, sion_blockcount, sion_blocksizes, swapbytes);

    if(sion_filedesc->nfiles>1) {
      /* read complete mapping */
      rc = _sion_read_header_var_part_mapping(*fileptr, *ntasks, maxblocks, start_of_varheader, &mapping_size, &mapping, swapbytes);
      sion_filedesc->mapping_size=mapping_size;
      sion_filedesc->mapping=mapping;

      { int rank;
	for (rank = 0; rank < mapping_size; rank++) {
	  DPRINTFP((32, "sion_open", 0, " mapping[%d] = %d , %d \n", rank,mapping[rank*2+0],mapping[rank*2+1]));
	}
      }

    }

    sion_seek(sid, SION_CURRENT_RANK, 0, 0);

    if (!rc) {
      return(_sion_errorprint(-1,_SION_ERROR_RETURN,"sion_open: cannot read header from file %s, aborting ...\n", fname));
    }


  }                             /* READ mode */
  free(prefix);

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

  DPRINTFP((1, "sion_open", 0, "leave open of file %s in %s mode\n", fname, file_mode));

  return (sid);

}

/*!
 * @brief Function to open a sion file for a specific rank
 *
 * This function opens a sion file for a specific rank.  It can be
 * used to open the sion file independently from each task.
 * (e.g. if no MPI is available or only a subset of tasks chunks are needed)
 *
 * Using this function the meta data at beginning of the sion file are read
 * from each task instead of read once and distribute (sion_open).
 * sion_open_rank reads only the tasks specific meta data  from the meta data
 * block at the end of the sion file.
 *
 * All metadata will be read and stored in internal data structures.
 *
 * Warning: Only read operations are currently supported.
 *
 * @param[in]           fname           name of file, should equal on all tasks
 * @param[in,out]       file_mode       like the type parameter of fopen (currently recognized options: "rb", "wb")
 * @param[in,out]       chunksize       chunksize for this task
 * @param[in,out]       fsblksize       blocksize of filesystem, must be equal on all processors
 * @param[in]           rank            rank number for which the file should be open;
 *                                      will be stored in sion file, usefull if comm is not MPI_COMM_WORLD
 *                                      typical: globalrank= rank in MPI_COMM_WORLD
 * @param[out]          fileptr         filepointer for this task
 *
 * @retval              sid             sion file handle or -1 if error occured
 */
int sion_open_rank(char *fname, char *file_mode, sion_int64 *chunksize, sion_int32 *fsblksize, int *rank, FILE **fileptr)
{

  int             rc, sid, ntasks, nfiles, filenumber;
  int             endianness, swapbytes, mapping_filenr, mapping_lrank;
  char           *prefix;
  sion_int64     *sion_chunksizes, *sion_startpointers, *sion_globalranks, flag1, flag2;
  sion_int64      start_of_varheader, lglobalskip, numblocks;
  sion_int32      sion_version, sion_version_patchlevel, sion_fileformat_version, maxblocks;
  _sion_filedesc *sion_filedesc;

  /*   sion_int64 lstartpointer,lpartnum,lglobalskip; */
  /*                                                                      */ DPRINTFTS(*rank, "before open rank");
  DPRINTFP((1, "sion_open_rank", *rank, "enter open of file %s in %s mode\n", fname, file_mode));

  prefix = malloc(SION_FILENAME_LENGTH);
  if (prefix == NULL) {
    return(_sion_errorprint(-1,_SION_ERROR_RETURN,"cannot allocate prefix string of size %lu , aborting ...\n", (unsigned long) SION_FILENAME_LENGTH));
  }

  _sion_alloc_fd_list();

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

  sid = _sion_newvcd(sion_filedesc, SION_FILEDESCRIPTOR);
/*   _sion_realloc_blocklist(sion_filedesc, MAXBLOCKS); */

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

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

    return(_sion_errorprint(-1,_SION_ERROR_RETURN,"sion_open_rank: %s for writing currently not supported, aborting ...\n", fname));

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

    /* open file */
    *fileptr = _sion_open_file_for_read(fname);
    if (!*fileptr) {
      return(_sion_errorprint(-1,_SION_ERROR_RETURN,"cannot open %s for reading, aborting ...\n", fname));
    }
    sion_filedesc->fd = *fileptr;
    sion_filedesc->maxblocks          = -1;
    sion_filedesc->chunksizes         = NULL;
    sion_filedesc->numblocks          = 0;
    sion_filedesc->end_of_header      = -1;
    /*                                                                      */ DPRINTFTS(*rank, "start after open file 1");

    rc =_sion_read_header_fixed_part(*fileptr, &endianness, &swapbytes,
				     fsblksize, &ntasks, &nfiles,
				     &filenumber, &flag1, &flag2, prefix, &sion_version, 
				     &sion_version_patchlevel, &sion_fileformat_version);
    if (!rc) {
      return(_sion_errorprint(-1,_SION_ERROR_RETURN,"cannot read header from file %s, aborting ...\n", fname));
    }

    _sion_allocate_sion_arrays(ntasks, &sion_chunksizes, &sion_startpointers, &sion_globalranks);
    rc = _sion_read_header_var_part(*fileptr, ntasks, sion_globalranks, sion_chunksizes, &maxblocks, &start_of_varheader, swapbytes);
    if (!rc) {
      return(_sion_errorprint(-1,_SION_ERROR_RETURN,"cannot read var part header from file %s, aborting ...\n", fname));
    }
    /*                                                                      */ DPRINTFTS(*rank, "start after read header 1");

    DPRINTFP((1, "sion_open_rank", *rank, "found number of files %d in %s rank=%d\n", nfiles,fname,*rank));
    if(nfiles>1) {
      /* read mapping for rank */
      rc=_sion_read_header_var_part_mapping_rank(*fileptr, *rank, ntasks, maxblocks, start_of_varheader,
						 &mapping_filenr,&mapping_lrank,swapbytes);
      if (!rc) {
	return(_sion_errorprint(-1,_SION_ERROR_RETURN,"cannot read var part mapping header from file %s, aborting ...\n", fname));
      }
      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_sion_arrays(&sion_chunksizes, &sion_startpointers, &sion_globalranks);
	fclose(*fileptr);
	/*                                                                      */ DPRINTFTS(*rank, "start after close file 1");

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

	DPRINTFP((1, "sion_open_rank", *rank, "open file sion_filedesc->fname=%s  fname=%s rank=%d\n", sion_filedesc->fname,fname,*rank));

	*fileptr = _sion_open_file_for_read(sion_filedesc->fname);
	if (!*fileptr) {
	  return(_sion_errorprint(-1,_SION_ERROR_RETURN,"cannot open %s for reading, aborting ...\n", fname));
	}
	/*                                                                      */ DPRINTFTS(*rank, "start after open file i");

	sion_filedesc->fd = *fileptr;
	rc =_sion_read_header_fixed_part(*fileptr, &endianness, &swapbytes,
					 fsblksize, &ntasks, &nfiles,
					 &filenumber, &flag1, &flag2, fname, &sion_version, 
					 &sion_version_patchlevel, &sion_fileformat_version);
	if (!rc) {
	  return(_sion_errorprint(-1,_SION_ERROR_RETURN,"sion_open: cannot read header from file %s, aborting ...\n", fname));
	}
	_sion_allocate_sion_arrays(ntasks, &sion_chunksizes, &sion_startpointers, &sion_globalranks);
	rc = _sion_read_header_var_part(*fileptr, ntasks, sion_globalranks, sion_chunksizes, &maxblocks, &start_of_varheader, swapbytes);
	if (!rc) {
	  return(_sion_errorprint(-1,_SION_ERROR_RETURN,"cannot var mapping data from file %s, aborting ...\n", fname));
	}
	/*                                                                      */ DPRINTFTS(*rank, "start after read header file i");
	
      } else {
	sion_filedesc->fname              = strdup(fname);
      }
    } else {
      mapping_filenr=0;
      mapping_lrank=*rank;
    }

    DPRINTFP((1, "sion_open_rank", *rank, "max blocks=%d\n", maxblocks));
    _sion_realloc_blocklist(sion_filedesc, maxblocks);

    sion_filedesc->maxblocks          = maxblocks;
    sion_filedesc->start_of_varheader = start_of_varheader;
    sion_filedesc->all_chunksizes    = sion_chunksizes;
    sion_filedesc->all_globalranks    = sion_globalranks;
    sion_filedesc->all_startpointers  = sion_startpointers;
    sion_filedesc->endianness         = endianness;
    sion_filedesc->swapbytes          = swapbytes;
    sion_filedesc->fsblksize          = *fsblksize;
    sion_filedesc->ntasks             = ntasks;
    sion_filedesc->nfiles             = nfiles;
    sion_filedesc->filenumber         = filenumber;
    sion_filedesc->flag1              = flag1;
    sion_filedesc->flag2              = flag2;
    sion_filedesc->state              = SION_FILESTATE_SEROPENRANK;
    sion_filedesc->mode               = SION_FILEMODE_READ;
    sion_filedesc->fileversion        = sion_fileformat_version;
    sion_filedesc->prefix             = strdup(prefix);

    _sion_calculate_startpointers(ntasks, sion_chunksizes, *fsblksize, sion_startpointers, &lglobalskip);


    sion_filedesc->globalskip         = lglobalskip;
    sion_filedesc->all_blockcount     = NULL;
    sion_filedesc->all_currentpos     = NULL;
    sion_filedesc->all_currentblocknr = NULL;
    sion_filedesc->all_blocksizes     = NULL;
    sion_filedesc->rank = mapping_lrank;

    rc = _sion_read_header_var_part_locations_rank(*fileptr, sion_filedesc->rank, ntasks, start_of_varheader, 
						   &numblocks, sion_filedesc->chunksizes, sion_filedesc, swapbytes);
    if (!rc) {
      return(_sion_errorprint(-1,_SION_ERROR_RETURN,"cannot read header from file %s, aborting ...\n", fname));
    }

    sion_filedesc->numblocks = numblocks;
/*     sion_filedesc->rank = *rank; */
    sion_filedesc->startpos = sion_startpointers[sion_filedesc->rank];
    sion_filedesc->currentpos = sion_filedesc->startpos;
    sion_filedesc->currentblocknr = 0;

    sion_seek(sid, SION_CURRENT_RANK, 0, 0);

  }                             /* READ mode */
  free(prefix);

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

  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);

}

/*!
 * @brief Function that returns pointers to internal fields.
 *
 * \sa This function returns pointers to internal fields, containing the number of chunks written
 * by each task (sion_chunkcount) and their sizes (sion_chunksizes).
 * This function is only needed if the sion file was opened for reading in serial mode.
 *
 * @param[in]   sid                     sion file handle
 * @param[out]  ntasks                  number of tasks wrote to the sion file
 * @param[out]  maxblocks               maximum number blocks in the file
 * @param[out]  globalskip              distance in bytes between the first bytes of two subsequent chunks of a task
 * @param[out]  start_of_varheader      start position of the meta data block at the end of sion file
 * @param[out]  sion_chunksizes         field containing requested chunk size of each task
 * @param[out]  sion_globalranks        field containing global unique id of each task
 * @param[out]  sion_blockcount         field containing number of blocks used by each task
 * @param[out]  sion_blocksizes         field containing for each task and block the number of bytes used in this chunk
 *                                      e.g. access: sion_blocksizes[rank*blocks+blocknr]
 * @return      1 if ok
 */
int sion_get_locations(int sid,
		       int *ntasks,
		       int *maxblocks,
		       sion_int64 *globalskip,
		       sion_int64 *start_of_varheader,
		       sion_int64 **sion_chunksizes, 
		       sion_int64 **sion_globalranks, 
		       sion_int64 **sion_blockcount, 
		       sion_int64 **sion_blocksizes ) { 
  int       rc = 1;
  _sion_filedesc *sion_filedesc;

  if ((sid<0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR)
      || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    return(_sion_errorprint(0,_SION_ERROR_RETURN,"invalid sion_filedesc, aborting %d ...\n", sid));
  }
  DPRINTFP((1, "sion_get_locations", sion_filedesc->rank, "enter\n"));

  *ntasks = sion_filedesc->ntasks;
  *maxblocks = sion_filedesc->maxblocks;
  *sion_chunksizes = sion_filedesc->all_chunksizes;
  *sion_globalranks = sion_filedesc->all_globalranks;
  *sion_blockcount = sion_filedesc->all_blockcount;

  *sion_blocksizes = sion_filedesc->all_blocksizes;

  *globalskip = sion_filedesc->globalskip;
  *start_of_varheader = sion_filedesc->start_of_varheader;

  DPRINTFP((1, "sion_get_locations", sion_filedesc->rank, "leave\n"));

  return (rc);
}

/*!
 * @brief Function that returns current position in file and pointer fiels containing chunk sizes.
 *
 * \sa This function returns returns current position in file and pointer fiels containing chunk sizes.
 * This function is only needed if the sion file was opened for reading in parallel mode (sion_paropen..., sion_open_rank).
 *
 * @param[in]   sid                     sion file handle
 * @param[out]  currentchunknr          number of current block in the file
 * @param[out]  currentpos              position in current block in the file
 * @param[out]  numchunks               maximum number blocks in the file
 * @param[out]  chunksizes              field containing chunk size of each task
 * @return      1 if ok
 */
int sion_get_current_location(int sid, int *currentchunknr, sion_int64 *currentpos, int *numchunks, sion_int64 **chunksizes)
{
  int       rc = 1;
  _sion_filedesc *sion_filedesc;

  if ((sid<0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR)
      || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    return(_sion_errorprint(0,_SION_ERROR_RETURN,"invalid sion_filedesc, aborting %d ...\n", sid));
  }
  DPRINTFP((1, "sion_get_current_location", sion_filedesc->rank, "enter\n"));

  *currentchunknr = sion_filedesc->currentblocknr;
  *currentpos     = sion_filedesc->currentpos - (sion_filedesc->startpos + sion_filedesc->currentblocknr * sion_filedesc->globalskip);
  *numchunks      = sion_filedesc->numblocks;
  *chunksizes     = sion_filedesc->chunksizes;

  DPRINTFP((1, "sion_get_current_location", sion_filedesc->rank, "leave\n"));
  return (rc);
}

/*!
 * @brief Function that returns pointers to the internal field mapping.
 *
 * \sa This function returns a pointer to the internal field mapping
 * This function is only needed if the sion file was opened for reading in serial mode.
 *
 * @param[in]   sid                     sion file handle
 * @param[out]  *mapping_size           size of mapping vector
 * @param[out]  **mapping               pointer to mapping vector
 * @return      1 if ok
 */
int sion_get_mapping(int          sid,
		     int         *mapping_size,
		     sion_int32 **mapping,
		     int         *numfiles ) { 
  int       rc = 1;
  _sion_filedesc *sion_filedesc;
  
  if ((sid<0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR)
      || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    return(_sion_errorprint(0,_SION_ERROR_RETURN,"invalid sion_filedesc, aborting %d ...\n", sid));
  }
  DPRINTFP((1, "sion_get_mapping", sion_filedesc->rank, "enter\n"));

  if(sion_filedesc->mapping_size>0) {
    *mapping_size=sion_filedesc->mapping_size;
    *mapping=sion_filedesc->mapping;
  } else {
    rc = -1; 
  }
    *numfiles=sion_filedesc->nfiles;

  DPRINTFP((1, "sion_get_mapping", sion_filedesc->rank, "leave (mapping_size=%d)\n",mapping_size));

  return (rc);
}

int sion_get_file_endianness(int sid)
{
  _sion_filedesc *sion_filedesc;

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

  return (sion_filedesc->endianness);
}

int sion_get_current_locations(int sid, int *ntasks, sion_int64 **sion_currentpos, sion_int64 **sion_currentblocknr)
{
  int       rc = 1;
  _sion_filedesc *sion_filedesc;

  if ((sid<0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR)
      || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    return(_sion_errorprint(0,_SION_ERROR_RETURN,"invalid sion_filedesc, aborting %d ...\n", sid));
  }
  DPRINTFP((1, "sion_get_current_locations", sion_filedesc->rank, "enter\n"));

  *ntasks = sion_filedesc->ntasks;
  *sion_currentpos = sion_filedesc->all_currentpos;
  *sion_currentblocknr = sion_filedesc->all_currentblocknr;

  DPRINTFP((1, "sion_get_current_locations", sion_filedesc->rank, "leave\n"));
  return (rc);
}

int sion_get_number_of_files(int sid)
{
  _sion_filedesc *sion_filedesc;

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

  return (sion_filedesc->nfiles);
}

int sion_get_filenumber(int sid)
{
  _sion_filedesc *sion_filedesc;

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

  return (sion_filedesc->filenumber);
}

sion_int64 sion_get_bytes_written(int sid)
{
  _sion_filedesc *sion_filedesc;
  sion_int64 bytes=-1;
  int i;

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

  if (sion_filedesc->mode == SION_FILEMODE_WRITE) {
    _sion_flush_block(sid);
    bytes=0;
    for(i=0;i<=sion_filedesc->numblocks;i++) bytes+=sion_filedesc->chunksizes[i];
  }

  return (bytes);
}

sion_int64 sion_get_bytes_read(int sid)
{
  _sion_filedesc *sion_filedesc;
  sion_int64 bytes=-1;
  int i;

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

  if (sion_filedesc->mode == SION_FILEMODE_READ) {
    _sion_purge_file_fp(SIONFP);
    sion_filedesc->currentpos = _sion_get_position(SIONFP);

    bytes=0;
    for(i=0;i<sion_filedesc->currentblocknr;i++) bytes+=sion_filedesc->chunksizes[i];
    /* current chunk */
    bytes+=sion_filedesc->currentpos - (sion_filedesc->startpos + sion_filedesc->currentblocknr * sion_filedesc->globalskip);
  }

  return (bytes);
}

size_t sion_fwrite(const void *data, size_t size, size_t nitems, int sid)
{
  size_t rc=0;
  sion_ensure_free_space(sid,size*nitems);
  rc = fwrite(data, size, nitems, sion_get_fp(sid));
  return(rc);
}

size_t sion_fread( void *ptr, size_t size, size_t nmemb, int sid) {
  sion_int64 bytes,bread, btoread;
  size_t rc;
  FILE *fptr;

  if ((sid<0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR)
      || !(fptr=sion_get_fp(sid))) {
    return(_sion_errorprint(-1,_SION_ERROR_RETURN,"invalid sion_filedesc, aborting %d ...\n", sid));
  }

  bread=0;
  bytes=size*nmemb;
  while(!sion_feof(sid) && (bytes>0)) {
    btoread=sion_bytes_avail_in_chunk(sid);
    if(btoread>bytes) btoread=bytes;
      rc=fread(ptr,1,btoread,fptr);
      if(rc>=0) return(rc);
      bread+=rc;
  }
  return(bread);
}

/*!
 * @brief Function to set the file pointer to a new position (not usable for sion multi-file opened by sion_open).
 *
 * This function sets the file pointer to a new position according to the specified parameters.
 * It can only be used when the sion file was opened for reading in serial mode.
 * In Write mode currently only the rank can be modified!
 * SION_CURRENT_BLK and SION_CURRENT_POS are required
 *
 * @param[in]   sid             sion file handle
 * @param[in]   rank            rank number of the process (SION_CURRENT_RANK to select the current rank)
 * @param[in]   currentblocknr  block number (SION_CURRENT_BLK to select the current block)
 * @param[in]   posinblk        position in the block (SION_CURRENT_POS to select the current position)
 *
 * @retval      1 if file pointer can be moved to new position
 *              0 otherwise
 */
int sion_seek(int sid, int rank, int currentblocknr, sion_int64 posinblk)
{
  FILE *fileptr;
  int rc;
  /* TBD: check if not sion_open and multi file */
  rc=sion_seek_fp(sid,rank,currentblocknr,posinblk,&fileptr);
  return(rc);

}

/*!
 * @brief Function to set the file pointer to a new position (multi-file version).
 *
 * This function sets the file pointer to a new position according to the specified parameters.
 * It can only be used when the sion file was opened for reading in serial mode.
 * In Write mode currently only the rank can be modified!
 * SION_CURRENT_BLK and SION_CURRENT_POS are required
 *
 * @param[in]   sid             sion file handle
 * @param[in]   rank            rank number of the process (SION_CURRENT_RANK to select the current rank)
 * @param[in]   currentblocknr  block number (SION_CURRENT_BLK to select the current block)
 * @param[in]   posinblk        position in the block (SION_CURRENT_POS to select the current position)
 * @param[OUT]  **fileptr       filepointer to corresponding file of a multi-file set
 *
 * @retval      1 if file pointer can be moved to new position
 *              0 otherwise
 */
int sion_seek_fp(int sid, int rank, int currentblocknr, sion_int64 posinblk, FILE **fileptr)
{
  int       rc = 1;
  int       currentrank, i;
  sion_int64 position;
  sion_int64 bytes=-1;
  _sion_filedesc *sion_filedesc;

  DPRINTFP((1, "sion_seek", -1, " called with sid=%dk\n", sid));
  assert((_sion_vcdtype(sid) == SION_FILEDESCRIPTOR));

  if ((sid<0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR)
      || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_seek: invalid sion_filedesc, aborting %d ...\n", sid));
  }
  DPRINTFP((1, "sion_seek", sion_filedesc->rank, "enter seek\n"));

  if (sion_filedesc->mode == SION_FILEMODE_WRITE) {
    if ((sion_filedesc->state != SION_FILESTATE_SEROPEN)
	&& (sion_filedesc->state != SION_FILESTATE_SEROPENRANK)) {
      return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_seek: invalid file open state (!SEROPEN), aborting %d ...\n", sid));
    }
  }
  if (rank >= sion_filedesc->ntasks) {
    return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_seek: parameter rank %d (max. %d) out of range, aborting %d ...\n", rank, sion_filedesc->ntasks, sid));
  }


  /* --- READ ---  */
  if (sion_filedesc->mode == SION_FILEMODE_READ) {

    /* --- PARALLEL READ */
    if (sion_filedesc->state == SION_FILESTATE_PAROPEN) {

      /* check rank */
      if ( (rank != SION_CURRENT_RANK) && (rank != sion_filedesc->rank) ) {
	return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_seek: parameter rank is different from current rank in parallel openened file, aborting %d ...\n", sid));
      }
      currentrank=sion_filedesc->rank;
      
      /* check requested block number */
      if (currentblocknr == SION_CURRENT_BLK) {
	currentblocknr=sion_filedesc->currentblocknr;
      }
      if ((currentblocknr == SION_ABSOLUTE_POS) && (posinblk>=0)) {
	/* search position */
	bytes=0;currentblocknr=0;
	while ( (currentblocknr<sion_filedesc->numblocks) && (bytes+sion_filedesc->chunksizes[currentblocknr]<posinblk) ) {
	  bytes+=sion_filedesc->chunksizes[currentblocknr];
	  currentblocknr++;
	}
	posinblk-=bytes;
      }


      if ( (currentblocknr < 0) || (currentblocknr >= sion_filedesc->numblocks) ) {
	return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_seek: parameter chunk number (%d) is out of range (0 .. %d), aborting  ...\n", currentblocknr,sion_filedesc->numblocks-1));
        return (0);
      }

      /* check requested position in block */
      if (posinblk == SION_CURRENT_POS) {
	_sion_flush_file_fp(SIONFP);
	sion_filedesc->currentpos = _sion_get_position(SIONFP);
	posinblk=sion_filedesc->currentpos - (sion_filedesc->startpos + sion_filedesc->currentblocknr * sion_filedesc->globalskip);
      }

      if ( (posinblk <0) || (posinblk > sion_filedesc->chunksizes[currentblocknr]) ) {
	return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_seek: parameter posiinblk (%lld) is out of range (0 .. %lld), aborting ...\n", posinblk,
				sion_filedesc->chunksizes[currentblocknr]));
        return (0);
      }
      
      /* set new position */
      sion_filedesc->currentblocknr = currentblocknr;
      position = sion_filedesc->startpos + sion_filedesc->currentblocknr * sion_filedesc->globalskip + posinblk;
      sion_filedesc->currentpos = position;
      _sion_purge_file_fp(SIONFP);
      _sion_set_position(SIONFP, position);
      
      DPRINTFP((1, "sion_seek", sion_filedesc->rank,
		" sid=%d rank=%4d currentblocknr=%4d posinblk=%4ld set fileptr to position %14lld (%14lld)\n",
		sid, rank, currentblocknr, posinblk, _sion_get_position(SIONFP),position));
      

    }      
    /* --- SERIAL READ ---  */
    if (sion_filedesc->state == SION_FILESTATE_SEROPEN) {

      currentrank = sion_filedesc->rank;
      if (rank == SION_CURRENT_RANK) rank = currentrank;

      if ((sion_filedesc->all_blockcount == NULL)
          || (sion_filedesc->all_blocksizes == NULL)) {
	return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_seek: internal error, data structure not initialized, aborting %d ...\n", sid));
      }
      if (currentblocknr >= sion_filedesc->maxblocks) {
	return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_seek: parameter currentblocknr %d (max. %d) out of range, aborting %d ...\n", 
				currentblocknr, sion_filedesc->maxblocks, sid));
      }
      position = sion_filedesc->all_startpointers[rank] + currentblocknr * sion_filedesc->globalskip + posinblk;
    } else {
      /* only moving in own rank (sion_open_rank) */
      position = sion_filedesc->startpos + currentblocknr * sion_filedesc->globalskip + posinblk;
    }

    sion_filedesc->currentpos = position;
    sion_filedesc->currentblocknr = currentblocknr;
    _sion_purge_file_fp(SIONFP);
    _sion_set_position(SIONFP, position);

    DPRINTFP((1, "sion_seek", sion_filedesc->rank,
              " sid=%d rank=%4d currentblocknr=%4d posinblk=%4ld set fileptr to position %14lld (%14lld)\n",
              sid, rank, currentblocknr, posinblk, _sion_get_position(SIONFP),position));

  } else {

    /* --- WRITE ---  */

    /* SERIAL WRITE to new file */
    currentrank = sion_filedesc->rank;
    if (rank == SION_CURRENT_RANK)
      rank = currentrank;

    if (currentblocknr != SION_CURRENT_BLK) {
      return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_seek: serial write to file currently only for SION_CURRENT_BLK implemented, aborting %d ...\n", sid));
    }
    if (posinblk != SION_CURRENT_POS) {
      return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_seek: serial write to file currently only for SION_CURRENT_POS implemented, aborting %d ...\n", sid));
    }

    /* update meta data of current rank */
    _sion_flush_block(sid);

    /* store current information */
    DPRINTFP((1, "sion_seek", sion_filedesc->rank,
              " sid=%d rank=%4d  startpos=%14lld chunksizes=%14lld\n", sid,
              currentrank, sion_filedesc->startpos, sion_filedesc->chunksizes[sion_filedesc->numblocks]));
    sion_filedesc->all_currentpos[currentrank] = sion_filedesc->startpos + sion_filedesc->chunksizes[sion_filedesc->numblocks];
    sion_filedesc->all_currentblocknr[currentrank] = sion_filedesc->numblocks;
    sion_filedesc->all_blockcount[currentrank] = sion_filedesc->numblocks + 1;
    for (i = 0; i <= sion_filedesc->numblocks; i++) {
      sion_filedesc->all_blocksizes[sion_filedesc->ntasks * i + currentrank] = sion_filedesc->chunksizes[i];
    }

    /* switch to new position */
    sion_filedesc->rank = rank;

    sion_filedesc->numblocks = sion_filedesc->all_currentblocknr[rank];
    sion_filedesc->startpos = sion_filedesc->all_startpointers[rank];
    /*
       if(currentblocknr==SION_CURRENT_BLK)   currentblocknr  =sion_filedesc->numblocks;
       if(posinblk==SION_CURRENT_POS) posinblk=sion_filedesc->chunksizes[sion_filedesc->numblocks];
     */
    for (i = 0; i <= sion_filedesc->numblocks; i++) {
      sion_filedesc->chunksizes[i] = sion_filedesc->all_blocksizes[sion_filedesc->ntasks * i + rank];
    }

    /* goto new position in file */
    position = sion_filedesc->all_currentpos[rank];
    DPRINTFP((1, "sion_seek", sion_filedesc->rank, " currentpos on rank %4d is %14lld\n", rank, position));

    _sion_flush_file_fp(SIONFP);
    _sion_set_position(SIONFP, position);

    DPRINTFP((1, "sion_seek", sion_filedesc->rank,
              " sid=%d set to rank=%4d set fileptr to position %14lld\n", sid, rank, _sion_get_position(SIONFP)));

  }

  /* TBD: implement sion_open and sion_seek_fp for multi-files */
  *fileptr=SIONFP;

  DPRINTFP((1, "sion_seek", sion_filedesc->rank, "leave seek\n"));

  return (rc);
}

/*!
 * @brief Function that indicates the end of file for this task.
 *
 * This function indicates if the end of file is reached for this task.
 * This means that the file pointer is behind of the last bytes of last chunk of this task.
 * If all bytes of the current chunk are already read and there are more chunks available
 * for this task, sion_feof will advance the filepointer to the start position of the
 * next chunk in the sion file.
 * The function is a task local function, which can be called independently from other MPI tasks.
 *
 * @param[in]   sid     sion file handle (in)
 *
 * @return      1 if end of last chunk reached
 *              0 otherwise
 */
int sion_feof(int sid)
{
  int       rc = 0;
  sion_int64 maxpos, position;

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

  DPRINTFP((8, "sion_feof", sion_filedesc->rank, "enter feof   sid=%d currentpos=%15lld\n", sid, sion_filedesc->currentpos));

  _sion_flush_file_fp(SIONFP);
  sion_filedesc->currentpos = _sion_get_position(SIONFP);


  maxpos = sion_filedesc->startpos
    + sion_filedesc->currentblocknr * sion_filedesc->globalskip + sion_filedesc->chunksizes[sion_filedesc->currentblocknr];

  DPRINTFP((16, "sion_feof", sion_filedesc->rank,
            " after flush sid=%d currentpos=%15lld maxpos= %lld startpos=%lld curblknr=%d\n",
            sid, sion_filedesc->currentpos, maxpos, sion_filedesc->startpos, sion_filedesc->currentblocknr));

  if (sion_filedesc->currentpos < maxpos) {
    /* end of current block not reached */
    rc = 0;
    DPRINTFP((16, "sion_feof", sion_filedesc->rank, " end of current block %d not reached sid=%d\n", sion_filedesc->currentblocknr, sid));
  }
  else {
    sion_filedesc->currentblocknr++;
    if (sion_filedesc->currentblocknr < sion_filedesc->numblocks) {
      /* end of current block reached, skip to next block */
      position = sion_filedesc->startpos + sion_filedesc->currentblocknr * sion_filedesc->globalskip;
      _sion_purge_file_fp(SIONFP);
      _sion_set_position(SIONFP, position);
      rc = 0;
      DPRINTFP((8, "sion_feof", sion_filedesc->rank,
                " end of block %d reached, skipping to next %lld -> %lld position=%lld gs=%lld sid=%d\n",
                sion_filedesc->currentblocknr - 1,
                sion_filedesc->currentpos, _sion_get_position(SIONFP), position, sion_filedesc->globalskip, sid));
    }
    else {
      /* end of last block reached */
      sion_filedesc->currentblocknr--;
      DPRINTFP((8, "sion_feof", sion_filedesc->rank, " end of last block %d reached sid=%d\n", sion_filedesc->currentblocknr, sid));
      rc = 1;
    }
  }

  DPRINTFP((8, "sion_feof", sion_filedesc->rank, "leave feof   sid=%d currentpos=%15lld\n", sid, sion_filedesc->currentpos));

  return (rc);
}

sion_int64 sion_bytes_avail_in_block(int sid) 
{
  return(sion_bytes_avail_in_chunk(sid));
}



/*!@brief Function that returns the number of bytes available in the current chunk.
 *
 * This function returns the number of bytes currently available in the current chunk (bytes not read).
 * It is a local function, which can be called independently from other MPI tasks.
 *
 * @param[in]   sid             sion file handle
 *
 * @return      rc>0  number of bytes
 *              rc<=0 file position is after end of block
 */
sion_int64 sion_bytes_avail_in_chunk(int sid)
{
  sion_int64 maxpos;
  sion_int64 bytes_avail;
  _sion_filedesc *sion_filedesc;

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


  _sion_purge_file_fp(SIONFP);
  sion_filedesc->currentpos = _sion_get_position(SIONFP);

  maxpos = sion_filedesc->startpos
    + sion_filedesc->currentblocknr * sion_filedesc->globalskip + sion_filedesc->chunksizes[sion_filedesc->currentblocknr];

  bytes_avail = maxpos - sion_filedesc->currentpos;

  DPRINTFP((8, "sion_bytes_avail_in_block", sion_filedesc->rank,
            "leave sid=%d rank=%4d  currentpos=%lld  maxpos= %lld -> bytes to read %ld\n",
            sid, sion_filedesc->rank, sion_filedesc->currentpos, maxpos, bytes_avail));

  return (bytes_avail);
}

/*!
 * @brief Function that returns the current file position
 *
 * \sa This function returns an sion_int64 containing the current file possition.
 *
 * @param[in]   sid     sion file handle (in)
 *
 * @return      current file position
 *
 */
sion_int64 sion_get_position(int sid)
{
  _sion_filedesc *sion_filedesc;

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

  _sion_flush_file_fp(SIONFP);

  return (_sion_get_position(SIONFP));

}

int sion_set_fp_closed(int sid)
{
  int       rc = 1;
  _sion_filedesc *sion_filedesc;

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

int sion_set_second_fp(int sid, FILE *secondfd)
{
  int       rc = 1;
  _sion_filedesc *sion_filedesc;

  if ((sid<0) || (_sion_vcdtype(sid) != SION_FILEDESCRIPTOR)
      || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_set_second_fp: invalid sion_filedesc, aborting %d ...\n", sid));
  }
  sion_filedesc->secondfd = secondfd;
  _sion_print_filedesc(sion_filedesc, 1024, "sion_set_second_fd", 1);
  /*
  if (sion_filedesc->mode == SION_FILEMODE_READ) {
    sion_seek(sid, SION_CURRENT_RANK, 0, 0);
  }
  */
  return (rc);
}

int sion_unset_second_fp(int sid)
{
  int       rc = 1;
  _sion_filedesc *sion_filedesc;

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

int sion_flush(int sid)
{
  int       rc = 1;
  rc = _sion_flush_block(sid);
  return (rc);
}

/*!
 * @brief Funtion to close a sion file.
 *
 * This function closes a sion file which was opened in serial mode with sion_open() or sion_open_rank().
 * In Write mode this function will also save all meta data to the meta data block of the sion file
 * at the beginning and the end of the file.
 * The function is a task local function, which can be called independently from other MPI tasks.
 *
 *  @param[in]  sid     sion file handle
 *
 *  @return     1 if close is ok
 */
int sion_close(int sid)
{

  int       rc = 1;
  int       maxblocks, nwrite, blknum, rank, currentrank, mapping_size;
  _sion_filedesc *sion_filedesc;

#ifdef SION_DEBUG
/*   fprintf(stderr, " sion_close, sid=%d type=%d %x\n",sid,_sion_vcdtype(sid),*(unsigned int *)_sion_vcdtovcon(sid)); */
#endif

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

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


  if ((sion_filedesc->state != SION_FILESTATE_SEROPEN)
      && (sion_filedesc->state != SION_FILESTATE_SEROPENRANK)) {
    return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_close: invalid file open state (!SEROPEN), aborting %d ...\n", sid));
  }

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

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

    /* update meta data of current rank */
    _sion_flush_block(sid);
    DPRINTFP((1, "sion_close", sion_filedesc->rank, " after flush block sid=%d  fileptr is at position %14lld\n", sid, _sion_get_position(SIONFP)));

    currentrank = sion_filedesc->rank;
    sion_filedesc->all_currentpos[currentrank] = sion_filedesc->startpos + sion_filedesc->chunksizes[sion_filedesc->numblocks];
    sion_filedesc->all_currentblocknr[currentrank] = sion_filedesc->numblocks;
    sion_filedesc->all_blockcount[currentrank] = sion_filedesc->numblocks + 1;
    for (blknum = 0; blknum <= sion_filedesc->numblocks; blknum++) {
      sion_filedesc->all_blocksizes[sion_filedesc->ntasks * blknum + currentrank] = sion_filedesc->chunksizes[blknum];
    }


    _sion_flush_file_fp(SIONFP);
    _sion_set_position(SIONFP, sion_filedesc->end_of_header);

    maxblocks = -1;
    for (blknum = 0; blknum < sion_filedesc->ntasks; blknum++)
      if (sion_filedesc->all_blockcount[blknum] > maxblocks)
        maxblocks = (int) sion_filedesc->all_blockcount[blknum];

    nwrite = fwrite(&maxblocks, 1, sizeof(maxblocks), SIONFP);
    if (nwrite != sizeof(maxblocks)) {
      return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_close: cannot write header (maxblocks) to file, aborting ... (%d)\n", nwrite));
    }

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

    nwrite = fwrite(&sion_filedesc->start_of_varheader, 1, sizeof(sion_int64), SIONFP);
    if (nwrite != sizeof(sion_int64)) {
      return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_close: cannot write header (start_of_varheader) to file, aborting ... (%d)\n", nwrite));
    }
    DPRINTFP((1, "sion_close", sion_filedesc->rank, " wrote start_of_varheader=%lld (%d bytes)\n", sion_filedesc->start_of_varheader, nwrite));

    /*     nwrite=fwrite(marker,sizeof(char),4, SIONFP ); */

    _sion_purge_file_fp(SIONFP);
    _sion_set_position(SIONFP, sion_filedesc->start_of_varheader);

    nwrite = fwrite(sion_filedesc->all_blockcount, sizeof(sion_int64), sion_filedesc->ntasks, SIONFP);
    if (nwrite != sion_filedesc->ntasks) {
      return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_close: cannot write header (all_blockcount) to file, aborting ... (%d)\n", nwrite));
    }
    DPRINTFP((1, "sion_close", sion_filedesc->rank, " wrote numblocks field (%lu bytes)\n", (unsigned long) nwrite * sizeof(sion_int64)));

    for (blknum = 0; blknum < maxblocks; 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;
        }
        DPRINTFP((1, "sion_close", sion_filedesc->rank,
                  " blocksize rank %2d blk %2d -> %lld\n", rank, blknum, sion_filedesc->all_blocksizes[sion_filedesc->ntasks * blknum + rank]));
      }
    }

    nwrite = fwrite(sion_filedesc->all_blocksizes, sizeof(sion_int64), sion_filedesc->ntasks * maxblocks, SIONFP);
    if (nwrite != sion_filedesc->ntasks * maxblocks) {
      return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_close: cannot write header (all_blocksizes) to file, aborting ... (%d)\n", nwrite));
    }
    DPRINTFP((1, "sion_close", sion_filedesc->rank, " wrote blocksizes field (%lu bytes)\n", (unsigned long) nwrite * sizeof(sion_int64)));

    /* mapping_size will be written in each case, is 0 if (number of physical file is one) */
    mapping_size = 0;
    nwrite = fwrite(&mapping_size, sizeof(sion_int32), 1, SIONFP);
    if (nwrite != 1) {
      return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_close: cannot write header (mapping_size) to file, aborting ... (%d)\n", nwrite));
    }
    DPRINTFP((32, "sion_close", sion_filedesc->rank, " wrote mapping_size=0 nwrite=%d\n", nwrite));



  }

  fclose(SIONFP);

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

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

  return (rc);
}

/*!
 * @brief Funtion to ensure that enough space is available.
 *
 * This function ensures that there is enough space available for writing numbytes bytes
 * to the sion file. It allocates a new block at the end of the sion file of size chunksize
 * if numbytes cannot be written in the current block.
 * This procedure is only be needed if the caller function from the user program does not count
 * the number of bytes written. It is a task local function, which can be called independently
 * from other MPI tasks, and it moves in some cases the filepointer to a new position and flushes
 * also the local filepointer.
 *
 * @param[in]   sid             sion file handle
 * @param[in]   bytes           number of bytes which will be written
 *
 * @return     1 if ok
 */
int sion_ensure_free_space(int sid, sion_int64 bytes)
{
  _sion_filedesc *sion_filedesc;
  sion_int64 fileposition;
  sion_int64 byteswritten;
  int       rc = 1;

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

  DPRINTFP((8, "_sion_ensure_free_space", sion_filedesc->rank, "enter ensure_free   sid=%d bytes=%lld\n", sid, bytes));

  if (sion_filedesc->mode != SION_FILEMODE_WRITE) {
      return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_ensure_free_space[%2d]: file is opened invalid sion_mode, aborting ...\n", sion_filedesc->rank));
  }
  if (SIONFP == NULL) {
      return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_ensure_free_space[%2d]: file is not open, aborting ...\n", sion_filedesc->rank));
  }


  fileposition = _sion_get_position(SIONFP);
  DPRINTFP((16, "_sion_ensure_free_space", sion_filedesc->rank,
            "  after getpos sid=%d fileposition is %lld bytes=%lld\n", sid, fileposition, bytes));

  _sion_flush_file_fp(SIONFP);
  if (bytes > sion_filedesc->chunksize) {
    return(_sion_errorprint(0,_SION_ERROR_RETURN,"sion_ensure_free_space[%2d]: could not write %lld bytes, chunksize=%lld, aborting ...\n",
			    sion_filedesc->rank, bytes, sion_filedesc->chunksize));
  }

  fileposition = _sion_get_position(SIONFP);
  byteswritten = fileposition - sion_filedesc->startpos;

  DPRINTFP((16, "_sion_ensure_free_space", sion_filedesc->rank,
            " sid=%d byteswritten(%lld) + new bytes (%lld) = %lld\n", sid, byteswritten, bytes, byteswritten + bytes));

  if (((byteswritten + bytes) > sion_filedesc->chunksize)
      && (byteswritten + bytes > sion_filedesc->fsblksize)) {
    /* not enough space for writing next data */
    _sion_create_new_block(sid);
  }

  DPRINTFP((8, "_sion_ensure_free_space", sion_filedesc->rank,
            "leave ensure_free   sid=%d fileposition is %lld bw+bytes=%lld <= chunksize=%lld (fsblksize=%d)\n",
            sid, fileposition, byteswritten + bytes, sion_filedesc->chunksize, sion_filedesc->fsblksize));


  return (rc);
}

/*!\brief SION get endianness
 *
 *  returns (1-> big endian, 0 ->little endian)
 */
int sion_get_endianness()
{

  /* Are we little or big endian?  From Harbison&Steele.  */
  union
  {
    long      l;
    char      c[sizeof(long)];
  } u;

  u.l = 1;
  return (u.c[sizeof(long) - 1] == 1);

}
