sion_generic_internal.c

Go to the documentation of this file.

/****************************************************************************
**  SIONLIB     http://www.fz-juelich.de/jsc/sionlib                       **
*****************************************************************************
**  Copyright (c) 2008-2016                                                **
**  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 <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"

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


}



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

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 */