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

/*
 * \brief Internal Functions(parallel)
 *
 * \date Sep 2, 2008
 * \author gstu0808
 *
 */

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

#include "sion.h"
#include "sion_debug.h"
#include "sion_file.h"
#include "sion_filedesc.h"
#include "sion_fd.h"
#include "sion_metadata.h"
#include "sion_internal.h"
#include "sion_printts.h"
#include "sion_keyvalue.h"
#include "sion_flags.h"

#include "sion_cache.h" 
#include "sion_buffer.h" 
#include "sion_hints.h" 
#include "sion_generic_internal.h"
#include "sion_generic_collective.h"

/*!\brief Generic parallel open of one direct access file.
 *
 * @param  sid                   sion file handle
 * @param  fname                 filename to use
 * @param  file_mode_flags       flags from already parsed file_mode (see also sion_internal.h)
 * @param  prefix                prefix to be used in case of multiple files
 * @param  numFiles              Number of files to open
 * @param  filenumber            file number
 * @param  chunksize             chunksize on this task
 * @param  fsblksize             blocksize of filesystem (must be equal on all processors)
 * @param  rank                  rank of the current process
 * @param  ntasks                number of processes
 * @param  globalrank            any global unique id for this task, will be stored in sion file, usefull if comm is not MPI_COMM_WORLD
 *                               typical: globalrank= rank in MPI_COMM_WORLD
 * @param  flag                  sion options flag
 * @param  fileptr               filepointer for this task
 * @param  sion_gendata          structure, containing references to commgroups and callbacks
 *
 * @return  sion file handle
 *          -1 if error occured
 */
int _sion_paropen_generic_one_file(
				   int    sid,
				   char  *fname,
				   _sion_flags_store *flags_store,
				   char  *prefix,
				   int   *numFiles,
				   int   *filenumber, 
				   sion_int64  *chunksize,
				   sion_int32  *fsblksize,
				   int    rank,
				   int    ntasks,
				   int   *globalrank,
				   int    flag,
				   FILE **fileptr,
				   _sion_generic_gendata *sion_gendata )
{

	  int       i, j;
	  int       rc;
	  
	  _sion_filedesc *sion_filedesc;
	  _sion_fileptr  *sion_fileptr; 

	  int       nfiles, filenum;
	  sion_int64 lchunksize, lstartpointer, lglobalrank, new_fsblocksize, helpint64, apiflag;
	  sion_int64 *sion_tmpintfield = NULL;

          _sion_flags_entry* flags_entry = NULL;

	  if (flags_store->mask&_SION_FMODE_POSIX) apiflag=SION_FILE_FLAG_POSIX;
	  else                                   apiflag=SION_FILE_FLAG_ANSI;


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

	  /* some shortcuts */
	  nfiles  = *numFiles;
	  filenum = *filenumber;

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

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

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



	  if (flags_store->mask&_SION_FMODE_WRITE) {
	    /* **************** WRITE mode **************** */

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

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

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

	    /* check parameter */
	    if (globalrank<0) {
	      return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,"sion_paropen: globalrank < 0, returning ...\n", (int) *globalrank));
	    }
	    
	    sion_filedesc->state      = SION_FILESTATE_PAROPEN;
	    sion_filedesc->mode       = SION_FILEMODE_WRITE;
	    sion_filedesc->endianness = _sion_get_endianness_with_flags(flags_store->mask); /* Endianness */
	    sion_filedesc->swapbytes  = 0;                       /* Endianness, swapping bytes */
	    sion_filedesc->fsblksize  = *fsblksize;
	    sion_filedesc->rank       = rank;
	    sion_filedesc->globalrank = *globalrank;
	    sion_filedesc->ntasks     = ntasks;
	    sion_filedesc->nfiles     = nfiles;
	    sion_filedesc->filenumber = filenum;
	    sion_filedesc->prefix     = strdup(prefix);
	    sion_filedesc->compress   = flags_store->mask&_SION_FMODE_COMPRESS;
	    sion_filedesc->usecoll       = (flags_store->mask&_SION_FMODE_COLLECTIVE)>0;
	    sion_filedesc->collmergemode = (flags_store->mask&_SION_FMODE_COLLECTIVE_MERGE)>0;

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

	    /* distribute new fsblksize */
	    if(*fsblksize==-1) {
	      sion_gendata->apidesc->bcastr_cb(&new_fsblocksize, sion_gendata->comm_data_local, _SION_INT64, 1, 0);
	      *fsblksize=new_fsblocksize;
	      sion_filedesc->fsblksize = *fsblksize;
	      /* printf("WF: rank=%2d after bcast fsblksize=%d new_fsblocksize=%d\n",rank,(int) *fsblksize,(int) new_fsblocksize); */
	      DPRINTFP((32, "_sion_paropen_generic_one_file", rank, "setting fsblksize to %lld\n", new_fsblocksize));
	    }

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

	    /* check for keyval parameter, needed at this point to set flag before writing header (flag1) */
	    if (rank == 0) {
	      _sion_keyval_check_env(sion_filedesc, flags_store->mask);
	    }
	    sion_gendata->apidesc->bcastr_cb(&sion_filedesc->keyvalmode, sion_gendata->comm_data_local, _SION_INT32, 1, 0);

	    /* check for collective options */
	    if (rank == 0) {
              if ((flags_entry = _sion_flags_get(flags_store, "collsize"))) {
                sion_filedesc->collsize = atoi(flags_entry->val);
              }
	      _sion_coll_check_env(sion_filedesc);
	    }
	    sion_gendata->apidesc->bcastr_cb(&sion_filedesc->usecoll,  sion_gendata->comm_data_local, _SION_INT32, 1, 0);
	    sion_gendata->apidesc->bcastr_cb(&sion_filedesc->collsize, sion_gendata->comm_data_local, _SION_INT32, 1, 0);
	    sion_gendata->apidesc->bcastr_cb(&sion_filedesc->collmergemode, sion_gendata->comm_data_local, _SION_INT32, 1, 0);

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

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

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

	    /* collect data and init startpointers on PE 0 */
	    lchunksize  = (sion_int64) *chunksize;
	    lglobalrank = (sion_int64) *globalrank;
	    sion_filedesc->chunksize_req=lchunksize;
	    DPRINTFP((32, "_sion_paropen_generic_one_file", rank, "lchunksize=%lld lglobalrank=%lld\n", lchunksize,lglobalrank));
	    
	    /*                                                                      */ DPRINTFTS2(rank, "before gather");
	    sion_gendata->apidesc->gatherr_cb(&lchunksize, sion_filedesc->all_chunksizes, sion_gendata->comm_data_local, _SION_INT64, 1, 0);
	    sion_gendata->apidesc->gatherr_cb(&lglobalrank, sion_filedesc->all_globalranks, sion_gendata->comm_data_local, _SION_INT64, 1, 0);

	    /* check capability of tasks */
	    if(sion_filedesc->usecoll) {
	      sion_filedesc->coll_capability=sion_gendata->apidesc->get_capability_cb(sion_gendata->comm_data_global);
	      sion_gendata->apidesc->gatherr_cb(&sion_filedesc->coll_capability, sion_filedesc->all_coll_capability, sion_gendata->comm_data_local, _SION_INT32, 1, 0);
	    }
	    
	    /*                                                                      */ DPRINTFTS2(rank, "after gather");
	    if (rank == 0) {
	      /*                                                                      */ DPRINTFTS(rank, "before calculate");
	      DPRINTFP((32, "_sion_paropen_generic_one_file", rank, "chunksizes[%d - 1]=%ld\n", ntasks,(long) sion_filedesc->all_chunksizes[ntasks - 1]));
	      if (!sion_filedesc->usecoll) _sion_calculate_startpointers(sion_filedesc);
	      else {
		if (!sion_filedesc->collmergemode) _sion_calculate_startpointers_collective(sion_filedesc);
		else                               _sion_calculate_startpointers_collective_merge(sion_filedesc);
	      }
	      /*                                                                      */ DPRINTFTS(rank, "after calculate");
	    }

	    /* open file on all ranks */
	    /*                                                                      */ DPRINTFTS(rank, "before open");
	    sion_fileptr = _sion_file_open(fname,apiflag|SION_FILE_FLAG_WRITE,0);
	    if (!sion_fileptr) {
	      return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,"sion_paropen_generic: cannot open %s for reading, aborting ...\n", fname));
	    }
	    sion_gendata->apidesc->barrier_cb(sion_gendata->comm_data_local);
	    /*                                                                      */ DPRINTFTS(rank, "after open");

	    /* store data in static data structure (sid)  */
	    sion_filedesc->fileptr = sion_fileptr;


	    /* write header */
	    if (rank == 0) {

	      /* apply hint for first meta data block */
	      _sion_apply_hints(sion_filedesc,SION_HINTS_ACCESS_TYPE_METADATABLOCK1);

	      /*                                                                      */ DPRINTFTS(rank, "before writeh");
	      _sion_write_header(sion_filedesc);
	      /*                                                                      */ DPRINTFTS(rank, "after writeh");

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

	    }

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

	    /* distribute chunksize */
	    /*                                                                        */ DPRINTFTS(rank, "before scatter");
	    sion_gendata->apidesc->scatterr_cb(sion_filedesc->all_chunksizes, &sion_filedesc->chunksize, sion_gendata->comm_data_local, _SION_INT64, 1, 0);
	    /*                                                                        */ DPRINTFTS(rank, "after scatter");

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

	      _sion_free_filedesc_coll_arrays(sion_filedesc);
	    }

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

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

	    /* set filepointer on each task */
	    /*                                                                      */ DPRINTFTS(rank, "before setp");
	    sion_gendata->apidesc->barrier_cb(sion_gendata->comm_data_local);
	    _sion_file_flush(sion_filedesc->fileptr);
	    _sion_file_set_position(sion_filedesc->fileptr, sion_filedesc->startpos);
	    sion_filedesc->currentpos = sion_filedesc->startpos;
	    /* given by calculate startpointers ... 
	       sion_filedesc->chunksize = (sion_int64) *chunksize;
	    */
	    sion_gendata->apidesc->barrier_cb(sion_gendata->comm_data_local);

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

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

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

	    /*                                                                      */ DPRINTFTS(rank, "before openR");
	    sion_fileptr = _sion_file_open(fname,apiflag|SION_FILE_FLAG_READ,0);
	    sion_gendata->apidesc->barrier_cb(sion_gendata->comm_data_local);
	    /*                                                                      */ DPRINTFTS(rank, "after openR");
	    if (!sion_fileptr) {
	      return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,"_sion_paropen_generic_one_file: cannot open %s for reading, aborting ...\n", fname));
	    }

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


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

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

	      rc = _sion_read_header_var_part(sion_filedesc);
	      if (rc!=SION_SUCCESS) {
		return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,"_sion_paropen_generic_one_file: cannot read header from file %s, aborting ...\n", fname));
	      }
	      
              if ((flags_entry = _sion_flags_get(flags_store, "collsize"))) {
                sion_filedesc->collsize = atoi(flags_entry->val);
              }
	      _sion_coll_check_env(sion_filedesc);
	      if(sion_filedesc->usecoll)  _sion_alloc_filedesc_coll_arrays(sion_filedesc);
	      
	      /* collective */
	      if (!sion_filedesc->usecoll) _sion_calculate_startpointers(sion_filedesc);
	      else {
		if (!sion_filedesc->collmergemode) _sion_calculate_startpointers_collective(sion_filedesc);
		else                               _sion_calculate_startpointers_collective_merge(sion_filedesc);
	      }
	      /*                                                                      */ DPRINTFTS(rank, "after calculate");

	    /* check for keyval parameter, needed at this point to set flag before writing header (flag1) */
	      _sion_keyval_check_env(sion_filedesc, flags_store->mask);

	    }
	    /* distribute keyvalmode */
	    sion_gendata->apidesc->bcastr_cb(&sion_filedesc->keyvalmode, sion_gendata->comm_data_local, _SION_INT32, 1, 0);

	    /* distribute collective options */
	    sion_gendata->apidesc->bcastr_cb(&sion_filedesc->usecoll, sion_gendata->comm_data_local, _SION_INT32, 1, 0);
	    sion_gendata->apidesc->bcastr_cb(&sion_filedesc->collsize, sion_gendata->comm_data_local, _SION_INT32, 1, 0);

	    if(sion_filedesc->usecoll) {
	      sion_gendata->apidesc->scatterr_cb(sion_filedesc->all_coll_collsize, &sion_filedesc->collsize, sion_gendata->comm_data_local, _SION_INT32, 1, 0);
	      sion_gendata->apidesc->scatterr_cb(sion_filedesc->all_coll_collector, &sion_filedesc->collector, sion_gendata->comm_data_local, _SION_INT32, 1, 0);

	      _sion_free_filedesc_coll_arrays(sion_filedesc);
	    }

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

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

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

	    /* scatter per task information read from file */
	    /*                                                                      */ DPRINTFTS(rank, "before scatter");
	    sion_gendata->apidesc->scatterr_cb(sion_filedesc->all_chunksizes, &sion_filedesc->chunksize, sion_gendata->comm_data_local, _SION_INT64, 1, 0);
	    sion_gendata->apidesc->scatterr_cb(sion_filedesc->all_startpointers, &sion_filedesc->startpos, sion_gendata->comm_data_local, _SION_INT64, 1, 0);
	    sion_gendata->apidesc->scatterr_cb(sion_filedesc->all_globalranks, &helpint64, sion_gendata->comm_data_local, _SION_INT64, 1, 0);sion_filedesc->globalrank=(sion_int32) helpint64;
	    /*                                                                      */ DPRINTFTS(rank, "after scatter");

	    /* read number of blocks for each task */
	    if (rank == 0) {
	      sion_tmpintfield = (sion_int64 *) malloc(ntasks * sizeof(sion_int64));
	      if (sion_tmpintfield == NULL) {
		return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,"_sion_paropen_generic_one_file: cannot allocate temporary memory of size %lu (sion_tmpintfield), aborting ...\n",
					(unsigned long) ntasks * sizeof(sion_int64)));
	      }
	      _sion_read_header_var_part_blockcount_to_field(sion_filedesc, ntasks, sion_tmpintfield);

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

	    /* and distribute them */
	    /*                                                                      */ DPRINTFTS(rank, "before scatter");
	    sion_gendata->apidesc->scatterr_cb(sion_tmpintfield, &helpint64, sion_gendata->comm_data_local, _SION_INT64, 1, 0);
	    /*                                                                      */ DPRINTFTS(rank, "after scatter");
	    sion_filedesc->lastchunknr = helpint64-1;
	    DPRINTFP((32, "_sion_paropen_generic_one_file", rank, "  lastchunknr on task %02d is %10ld\n", rank, (long) sion_filedesc->lastchunknr));

	    for (i = 0; i < sion_filedesc->maxusedchunks; i++) {
	      if (rank == 0) _sion_read_header_var_part_nextblocksizes_to_field(sion_filedesc, ntasks, sion_tmpintfield);
	      /*                                                                      */ DPRINTFTS(rank, "before scatter");
	      sion_gendata->apidesc->scatterr_cb(sion_tmpintfield, &helpint64, sion_gendata->comm_data_local, _SION_INT64, 1, 0);
	      /*                                                                      */ DPRINTFTS(rank, "after scatter");
	      sion_filedesc->blocksizes[i] = helpint64;
	    }

	#define BGFLUSH
	#ifdef BGFLUSH
	    _sion_file_flush(sion_filedesc->fileptr);
	#endif

	    /*                                                                      */ DPRINTFTS(rank, "before setp");
	    sion_gendata->apidesc->barrier_cb(sion_gendata->comm_data_local);
	    _sion_file_purge(sion_filedesc->fileptr);
	    _sion_file_set_position(sion_filedesc->fileptr, sion_filedesc->startpos);
	    sion_filedesc->currentpos     = sion_filedesc->startpos;
	    sion_filedesc->currentblocknr = 0;

	    /* OUTPUT parameters */
	    *fsblksize  = sion_filedesc->fsblksize;
	    *chunksize  = sion_filedesc->chunksize;
	    *globalrank = sion_filedesc->globalrank;

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

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

	  if(fileptr!=NULL) {
	    if(sion_filedesc->fileptr->flags&SION_FILE_FLAG_ANSI) {
	      *fileptr=sion_filedesc->fileptr->fileptr;
	      sion_filedesc->fileptr_exported=1;
	    } else {
	      *fileptr=NULL;
	      sion_filedesc->fileptr_exported=0;
	    }
	  }

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

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

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

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

	  return (sid);


}



/*!\brief Internal function to close parallel opened SION file
 *
 * @param  sid  reference to file description struct (_sion_filedesc)
 * @param  rank  rank
 * @param  ntasks  number of tasks
 * @param  mapping_size  number of global tasks
 * @param  mapping  mapping
 * @param  sion_gendata  generic data struct
 *
 * @return  SION_SUCCESS if successful
 */
int _sion_parclose_generic(int sid,
			   int rank,
			   int ntasks,
			   int mapping_size,
			   sion_int32 *mapping,
			   _sion_generic_gendata *sion_gendata )
{

  int       rc = SION_SUCCESS;
  int       blknum, lrank;
  sion_int64 helpint64;
  sion_int64 *sion_tmpintfield = NULL;
  _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_parclose_generic: invalid sion_filedesc, aborting %d ...\n", sid));
  }

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

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

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

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

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

    /* _sion_buffer_flush(sion_filedesc); */ /* clear internal buffer */
    _sion_flush_block(sion_filedesc);

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

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

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

    sion_gendata->apidesc->barrier_cb(sion_gendata->comm_data_local);

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

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

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

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

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

    /* write rest of first meta data block on rank 0 */
    if (rank == 0) {
      _sion_write_header_var_info(sion_filedesc);

      _sion_write_header_var_part_blockcount_from_field(sion_filedesc,sion_filedesc->ntasks,sion_tmpintfield);

    }

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

      /*                                                                      */ DPRINTFTS2(rank, "before gather");
      sion_gendata->apidesc->gatherr_cb(&helpint64, sion_tmpintfield, sion_gendata->comm_data_local, _SION_INT64, 1, 0);
      /*                                                                      */ DPRINTFTS2(rank, "after gather");

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

	_sion_write_header_var_part_nextblocksizes_from_field(sion_filedesc,sion_filedesc->ntasks,sion_tmpintfield);

      }
    }

    /* write mapping to file if more than one physical file is used, mapping_size is the number of global tasks */
    if (rank == 0) {
      _sion_write_header_var_part_mapping(sion_filedesc, mapping_size, mapping);
    }

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

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

  }                             /* write */

  _sion_free_filedesc(sion_filedesc);
  sion_filedesc = NULL;


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

  return (rc);
}

/*!\brief change chunksize for an already opened SION file (write)
 *
 * @return 1 if successful
 */
int _sion_parreinit_generic(
			    int sid,
			    sion_int64 chunksize,
			    int rank,
			    int ntasks,
			    _sion_generic_gendata *sion_gendata)
{

  int       rc = SION_SUCCESS;
  _sion_filedesc *sion_filedesc;
  sion_int64 lchunksize, lstartpointer, lglobalrank;

  if ((_sion_vcdtype(sid) != SION_FILEDESCRIPTOR) || !(sion_filedesc = _sion_vcdtovcon(sid))) {
    return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_RETURN,"_sion_parreinit_generic: invalid sion_filedesc, aborting %d ...\n", sid));
  }
	  
  if (sion_filedesc->state != SION_FILESTATE_PAROPEN) {
    return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_RETURN,"_sion_parreinit_generic: sion file with sid=%d was not opened by a sion_paropen\n", sid));
  }
	  
  DPRINTFP((2, "_sion_parreinit_generic", sion_filedesc->rank, "enter parallel reinit  sid=%d\n", sid));

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

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

  /* collect new chunksize data and init startpointers on PE 0 */
  lchunksize  = (sion_int64) chunksize;
  lglobalrank = (sion_int64) sion_filedesc->globalrank;
 
  /*                                                                      */ DPRINTFTS2(sion_filedesc->rank, "before gather");
  sion_gendata->apidesc->gatherr_cb(&lchunksize, sion_filedesc->all_chunksizes, sion_gendata->comm_data_local, _SION_INT64, 1, 0);
  sion_gendata->apidesc->gatherr_cb(&lglobalrank, sion_filedesc->all_globalranks, sion_gendata->comm_data_local, _SION_INT64, 1, 0);

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

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

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

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

    _sion_file_flush(sion_filedesc->fileptr);
    lstartpointer=0;
    _sion_file_set_position(sion_filedesc->fileptr, lstartpointer);
    
    /*                                                                      */ DPRINTFTS(sion_filedesc->rank, "before writeh");
    _sion_write_header(sion_filedesc);
    /*                                                                      */ DPRINTFTS(sion_filedesc->rank, "after writeh");
    
    /* needed for writing pointer to var part of metadata at the end of the file */
    sion_filedesc->end_of_header = _sion_file_get_position(sion_filedesc->fileptr);
    sion_filedesc->start_of_data = sion_filedesc->all_startpointers[0];

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

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

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

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

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

  /* set filepointer on each task */
  /*                                                                      */ DPRINTFTS(sion_filedesc->rank, "before setp");
  sion_gendata->apidesc->barrier_cb(sion_gendata->comm_data_local);
  _sion_file_flush(sion_filedesc->fileptr);
  _sion_file_set_position(sion_filedesc->fileptr, sion_filedesc->startpos);
  sion_filedesc->currentpos = sion_filedesc->startpos;
  sion_filedesc->chunksize = (sion_int64) chunksize;
  sion_gendata->apidesc->barrier_cb(sion_gendata->comm_data_local);
  
  /* apply hint for first chunk */
  _sion_apply_hints(sion_filedesc,SION_HINTS_ACCESS_TYPE_CHUNK);

  if (sion_filedesc->rank == 0) {
    /* not needed for rest of sionlib function calls  */
    _sion_free_filedesc_arrays(sion_filedesc);
  }
  
  /*                                                                      */ DPRINTFTS(sion_filedesc->rank, "after setp");
  DPRINTFP((32, "_sion_parreinit_generic", sion_filedesc->rank, " ending open for write #tasks=%d filepos=%lld\n", 
	    sion_filedesc->ntasks, _sion_file_get_position(sion_filedesc->fileptr)));

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

  return (rc);


}
/*  END OF _sion_parclose_generic */
