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

#define _XOPEN_SOURCE 700

#include <stdint.h>
#include <string.h>

#include "sion_common.h"
#include "sion_const.h"
#include "sion_debug.h"
#include "sion_error_handler.h"
#include "sion_filedesc.h"
#include "sion_internal_collstat.h"
#include "sion_internal_startptr.h"
#include "sion_metadata.h"

/* INTERNAL */

#define DFUNCTION "_lsion_calculate_set_alignment"
int64_t _lsion_calculate_set_alignment(_sion_filedesc *sion_filedesc, int t)
{
  int64_t lsize = SION_SIZE_NOT_VALID;

  /* alignment: fsv 3, WRITE: -> align lsize  */
  /*          : fsv 3, READ:  -> align lsize  */
  /*          : fsv 4, WRITE: -> align lsize+chunksize  */
  /*          : fsv 4, READ:  -> lsize=chunksize  */

  if (sion_filedesc->fileversion <= 3) {
    lsize = (sion_filedesc->all_chunksizes[t] % sion_filedesc->fsblksize == 0)
      ? sion_filedesc->all_chunksizes[t]
      : ((sion_filedesc->all_chunksizes[t] / sion_filedesc->fsblksize) + 1) * sion_filedesc->fsblksize;
    DPRINTFP((2048, DFUNCTION, _SION_DEFAULT_RANK, "old file version (<=3) task=%d lsize=%ld\n", t, (long)lsize));
  } else {
    if (sion_filedesc->mode == SION_FILEMODE_WRITE) {
      lsize = (sion_filedesc->all_chunksizes[t] % sion_filedesc->fsblksize == 0)
        ? sion_filedesc->all_chunksizes[t]
        : ((sion_filedesc->all_chunksizes[t] / sion_filedesc->fsblksize) + 1) * sion_filedesc->fsblksize;
      /* expand chunksize for alignment */
      sion_filedesc->all_chunksizes[t] = lsize;
      DPRINTFP(
        (2048, DFUNCTION, _SION_DEFAULT_RANK, "new file version (>3) task=%d lsize=%ld, WRITE chunksize set\n", t, (long)lsize));
    } else {
      lsize = sion_filedesc->all_chunksizes[t];
      DPRINTFP((2048, DFUNCTION, _SION_DEFAULT_RANK, "new file version (>3) task=%d lsize=%ld, READ no align\n", t, (long)lsize));
    }
  }

  return lsize;
}
#undef DFUNCTION

#define DFUNCTION "_lsion_get_size_metadatablock1"
int _lsion_get_size_metadatablock1(_sion_filedesc *sion_filedesc, int32_t mapping_size)
{
  int rc = 0;
  int firstsize;

  // calculate size of first block for meta-information
  // length of SION_ID + endianness + fsblksize + ntasks + nfiles + filenumber + 3 * versioninfo
  firstsize = strlen(SION_ID) + 8 * sizeof(int32_t) + 2 * sion_filedesc->ntasks * sizeof(int64_t) // globalranks + chunksizes
    + 2 * sizeof(int64_t); // 2 x 64 bit flags
  if (sion_filedesc->fileversion <= 5) {
    firstsize += SION_FILENAME_LENGTH; // prefix
  } else {
    firstsize += sion_filedesc->ntasks * sizeof(int64_t) // sizes
      + (mapping_size * 2 + 1) * sizeof(int32_t); // mapping information
  }
  rc = firstsize;

  return rc;
}
#undef DFUNCTION

#define DFUNCTION "_lsion_calculate_startpointers"
int _lsion_calculate_startpointers(_sion_filedesc *sion_filedesc, int32_t mapping_size)
{
  int rc = SION_SUCCESS;
  int i, firstsize;
  int64_t lsize;

  DPRINTFP((2, DFUNCTION, _SION_DEFAULT_RANK, "enter ntasks=%d fsblksize=%d chunksizes[0]=%lld\n", sion_filedesc->ntasks,
    sion_filedesc->fsblksize, sion_filedesc->all_chunksizes[0]));

  firstsize = _lsion_get_size_metadatablock1(sion_filedesc, mapping_size);

#ifdef WFLARGEMETABLOCK
  if (firstsize < 4 * 1024 * 1024) {
    firstsize = 4 * 1024 * 1024;
  }
#endif

  DPRINTFP((2, DFUNCTION, _SION_DEFAULT_RANK, "firstsize=%d\n", firstsize));

  sion_filedesc->all_startpointers[0] = (firstsize % sion_filedesc->fsblksize == 0)
    ? firstsize
    : ((firstsize / sion_filedesc->fsblksize) + 1) * sion_filedesc->fsblksize;

  i = 0;
  DPRINTFP((2048, DFUNCTION, _SION_DEFAULT_RANK, "  startpointers[%2d]=%10lld (%10.4fMB) chunksizes[%2d]=%8lld\n", i,
    sion_filedesc->all_startpointers[i], sion_filedesc->all_startpointers[i] / 1024.0 / 1024.0, i,
    sion_filedesc->all_chunksizes[i]));
  sion_filedesc->globalskip = 0;
  for (i = 1; i < sion_filedesc->ntasks; i++) {
    lsize = _lsion_calculate_set_alignment(sion_filedesc, i - 1);

    sion_filedesc->globalskip += lsize;
    sion_filedesc->all_startpointers[i] = sion_filedesc->all_startpointers[i - 1] + lsize;

    DPRINTFP((2048, DFUNCTION, _SION_DEFAULT_RANK,
      "  startpointers[%2d]=%10lld (%10.4fMB) chunksizes[%2d]=%8lld chunksizes[%2d]=%8lld\n", i,
      sion_filedesc->all_startpointers[i], sion_filedesc->all_startpointers[i] / 1024.0 / 1024.0, i,
      sion_filedesc->all_chunksizes[i], i - 1, sion_filedesc->all_chunksizes[i - 1]));
  }

  lsize = _lsion_calculate_set_alignment(sion_filedesc, sion_filedesc->ntasks - 1);
  sion_filedesc->globalskip += lsize;

  DPRINTFP((2, DFUNCTION, _SION_DEFAULT_RANK, "leave globalskip is %lld\n", sion_filedesc->globalskip));
  return rc;
}
#undef DFUNCTION

#define DFUNCTION "_lsion_calculate_num_collector"
int _lsion_calculate_num_collector(_sion_filedesc *sion_filedesc, _sion_collstat *collstat, int *num_collectors)
{
  int rc = SION_SUCCESS;
  int max_num_collectors;

  /* max. number: one fsblksize per collector */
  max_num_collectors = (int)(collstat->gsize / sion_filedesc->fsblksize);
  if (collstat->gsize % sion_filedesc->fsblksize > 0) {
    max_num_collectors++;
  }
  DPRINTFP((2, DFUNCTION, _SION_DEFAULT_RANK, "max_num_collectors=%d\n", max_num_collectors));

  if (sion_filedesc->collsize > 0) {
    /* user defined collsize (number of tasks per collector) */
    *num_collectors = (int)(sion_filedesc->ntasks / sion_filedesc->collsize);
    if (sion_filedesc->ntasks % sion_filedesc->collsize > 0) {
      (*num_collectors)++;
    }

    /* limit the user specification */
    if (*num_collectors > max_num_collectors) {
      *num_collectors = max_num_collectors;
    }

    DPRINTFP((32, DFUNCTION, _SION_DEFAULT_RANK, "user given collsize %d -> set num_collectors to %d\n", sion_filedesc->collsize,
      *num_collectors));
  } else {
    /* determine number of collectors by heuristic */
    *num_collectors = max_num_collectors;

    /* limit the user specification */
    if (*num_collectors > sion_filedesc->ntasks) {
      *num_collectors = sion_filedesc->ntasks;
    }

    /* some heuristics */
    if ((sion_filedesc->ntasks >= 512) && (*num_collectors > 32)) {
      *num_collectors = 32;
    } else if ((sion_filedesc->ntasks >= 256) && (*num_collectors > 16)) {
      *num_collectors = 16;
    } else if ((sion_filedesc->ntasks >= 128) && (*num_collectors > 8)) {
      *num_collectors = 8;
    } else if ((sion_filedesc->ntasks >= 64) && (*num_collectors > 8)) {
      *num_collectors = 8;
    } else if ((sion_filedesc->ntasks >= 32) && (*num_collectors > 8)) {
      *num_collectors = 8;
    } else if ((sion_filedesc->ntasks >= 16) && (*num_collectors > 4)) {
      *num_collectors = 4;
    }

    /* TODO: check I/O infrastructure characteristics for better values */

    DPRINTFP((2, DFUNCTION, _SION_DEFAULT_RANK, "SIONlib heuristic collsize=%d num_collectors=%d\n", sion_filedesc->collsize,
      *num_collectors));
  }

  /* data per collector */
  collstat->avg_data_per_collector =
    ((int64_t)(collstat->gsize / *num_collectors) / sion_filedesc->fsblksize) * sion_filedesc->fsblksize;

  if ((collstat->gsize / *num_collectors) % sion_filedesc->fsblksize > 0) {
    collstat->avg_data_per_collector += sion_filedesc->fsblksize;
  }
  DPRINTFP((32, DFUNCTION, _SION_DEFAULT_RANK, "avg_data_per_collectors=%ld\n", (long)collstat->avg_data_per_collector));

  return rc;
}
#undef DFUNCTION

#define DFUNCTION "_lsion_calculate_startpointers_collective"
int _lsion_calculate_startpointers_collective(_sion_filedesc *sion_filedesc, int32_t mapping_size)
{
  int rc = SION_SUCCESS;
  int i, firstsize, num_collectors, numsender, lastcoll, s;

  int64_t currentsize, aligned_size, startpointer;
  _sion_collstat *collstat;

  DPRINTFP((2, DFUNCTION, _SION_DEFAULT_RANK, "enter ntasks=%d fsblksize=%d chunksizes[0]=%lld\n", sion_filedesc->ntasks,
    sion_filedesc->fsblksize, sion_filedesc->all_chunksizes[0]));

  if (sion_filedesc->fileversion <= 3) {
    return _lsion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN,
      "_lsion_calculate_startpointers_collective: files with old sionlib file format (<3) can not be read by collective calls, "
      "please use standard read calls, aborting ...\n");
  }

  /* statistics */
  collstat = _lsion_create_and_init_collstat(sion_filedesc);

  /* calculate size of first block for meta-information */
  collstat->firstsize = firstsize = _lsion_get_size_metadatablock1(sion_filedesc, mapping_size);
  _lsion_calculate_num_collector(sion_filedesc, collstat, &num_collectors);
  collstat->req_num_collectors = num_collectors;

  DPRINTFP((2, DFUNCTION, _SION_DEFAULT_RANK, "firstsize=%d collsize=%d num_collectors=%d\n", firstsize, sion_filedesc->collsize,
    num_collectors));

  numsender = 1;
  lastcoll = 0;
  currentsize = sion_filedesc->all_chunksizes[0];

  /* 1ST STEP: calculate mapping to collectors */
  for (i = 1; i < sion_filedesc->ntasks; i++) {
    if ((currentsize + sion_filedesc->all_chunksizes[i] <= collstat->avg_data_per_collector)
      || (sion_filedesc->all_coll_capability[i] == SION_CAPABILITY_ONLY_SENDER)) {
      /* task will add to last collector */
      currentsize += sion_filedesc->all_chunksizes[i];
      numsender++;
    } else { /* new collector needed */

      /* adjust last senders */
      for (s = lastcoll; s < i; s++) {
        sion_filedesc->all_coll_collector[s] = lastcoll;
        sion_filedesc->all_coll_collsize[s] = numsender;
      }

      /* expand last one of group to align next one */
      if (sion_filedesc->mode == SION_FILEMODE_WRITE) {
        aligned_size = ((int64_t)currentsize / sion_filedesc->fsblksize) * sion_filedesc->fsblksize;
        if (currentsize % sion_filedesc->fsblksize > 0) {
          aligned_size += sion_filedesc->fsblksize;
        }

        DPRINTFP((128, DFUNCTION, _SION_DEFAULT_RANK, "  align chunksizes[%2d]=%8lld + %lld\n", i - 1,
          sion_filedesc->all_chunksizes[i - 1], aligned_size - currentsize));

        sion_filedesc->all_chunksizes[i - 1] += aligned_size - currentsize;
      }

      /* init data for next collector */
      numsender = 1;
      lastcoll = i;
      currentsize = sion_filedesc->all_chunksizes[i];
    }
  }

  /* align last sender */
  if (sion_filedesc->mode == SION_FILEMODE_WRITE) {
    aligned_size = ((int64_t)currentsize / sion_filedesc->fsblksize) * sion_filedesc->fsblksize;
    if (currentsize % sion_filedesc->fsblksize > 0) {
      aligned_size += sion_filedesc->fsblksize;
    }
    DPRINTFP((128, DFUNCTION, _SION_DEFAULT_RANK, "  align chunksizes[%2d]=%8lld + %lld\n", sion_filedesc->ntasks - 1,
      sion_filedesc->all_chunksizes[sion_filedesc->ntasks - 1], aligned_size - currentsize));
    sion_filedesc->all_chunksizes[sion_filedesc->ntasks - 1] += aligned_size - currentsize;
  }

  /* adjust sender of last collector */
  for (s = lastcoll; s < sion_filedesc->ntasks; s++) {
    sion_filedesc->all_coll_collector[s] = lastcoll;
    sion_filedesc->all_coll_collsize[s] = numsender;
  }

  /* 2ND STEP: calculate startpointers */

  /* align first, not necessary, only for debugging  */
  /* startpointer=firstsize; */
  startpointer = (firstsize % sion_filedesc->fsblksize == 0)
    ? firstsize
    : ((firstsize / sion_filedesc->fsblksize) + 1) * sion_filedesc->fsblksize;
  sion_filedesc->globalskip = 0;
  /* calculate mapping to collectors */
  for (i = 0; i < sion_filedesc->ntasks; i++) {
    sion_filedesc->all_startpointers[i] = startpointer;
    startpointer += sion_filedesc->all_chunksizes[i];
    sion_filedesc->globalskip += sion_filedesc->all_chunksizes[i];
  }

  /* statistics */
  if (sion_filedesc->colldebug >= 1) {
    _lsion_update_collstat(collstat, sion_filedesc);
    _lsion_print_collstat(collstat, sion_filedesc);
  }

  _lsion_debugprint_collstat(collstat, sion_filedesc);

  _lsion_destroy_collstat(collstat);

  DPRINTFP((2, DFUNCTION, _SION_DEFAULT_RANK, "leave globalskip is %lld\n", sion_filedesc->globalskip));
  return rc;
}
#undef DFUNCTION

#define DFUNCTION "_lsion_calculate_startpointers_collective_merge"
int _lsion_calculate_startpointers_collective_merge(_sion_filedesc *sion_filedesc, int32_t mapping_size)
{
  int rc = SION_SUCCESS;
  int i, firstsize, num_collectors, numsender, lastcoll, s;

  int64_t currentsize, newsize, aligned_size, startpointer;
  _sion_collstat *collstat;

  DPRINTFP((2, DFUNCTION, _SION_DEFAULT_RANK, "enter ntasks=%d fsblksize=%d chunksizes[0]=%lld\n", sion_filedesc->ntasks,
    sion_filedesc->fsblksize, sion_filedesc->all_chunksizes[0]));

  if (sion_filedesc->fileversion <= 3) {
    return _lsion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN,
      "_lsion_calculate_startpointers_collective: files with old sionlib file format (<3) can not be read by collective calls, "
      "please use standard read calls, aborting ...\n");
  }

  /* statistics */
  collstat = _lsion_create_and_init_collstat(sion_filedesc);

  /* calculate size of first block for meta-information */
  collstat->firstsize = firstsize = _lsion_get_size_metadatablock1(sion_filedesc, mapping_size);
  _lsion_calculate_num_collector(sion_filedesc, collstat, &num_collectors);
  collstat->req_num_collectors = num_collectors;

  DPRINTFP((2, DFUNCTION, _SION_DEFAULT_RANK, "firstsize=%d collsize=%d num_collectors=%d\n", firstsize, sion_filedesc->collsize,
    num_collectors));

  numsender = 1;
  lastcoll = 0;
  currentsize = sion_filedesc->all_chunksizes[0];

  DPRINTFP((128, DFUNCTION, _SION_DEFAULT_RANK, "  currentsize=%lld chunksizes[%2d]=%8lld\n", currentsize, 0,
    sion_filedesc->all_chunksizes[0]));

  /* 1ST STEP: calculate mapping to collectors */
  for (i = 1; i < sion_filedesc->ntasks; i++) {
    if ((currentsize + sion_filedesc->all_chunksizes[i] <= collstat->avg_data_per_collector)
      || (sion_filedesc->all_coll_capability[i] == SION_CAPABILITY_ONLY_SENDER)) {
      /* task will add to last collector */
      currentsize += sion_filedesc->all_chunksizes[i];
      numsender++;
      DPRINTFP((128, DFUNCTION, _SION_DEFAULT_RANK, "  currentsize=%lld chunksizes[%2d]=%8lld\n", currentsize, i,
        sion_filedesc->all_chunksizes[i]));
    } else { /* new collector needed */

      /* increase chunksize of last collector */
      if (sion_filedesc->mode == SION_FILEMODE_WRITE) {
        newsize = currentsize;

        /* expand chunksize to align next one */
        aligned_size = ((int64_t)newsize / sion_filedesc->fsblksize) * sion_filedesc->fsblksize;
        if (newsize % sion_filedesc->fsblksize > 0) {
          aligned_size += sion_filedesc->fsblksize;
        }

        DPRINTFP((128, DFUNCTION, _SION_DEFAULT_RANK, "  resize chunksizes[%2d]=%8lld + %lld\n", lastcoll,
          sion_filedesc->all_chunksizes[lastcoll], aligned_size - sion_filedesc->all_chunksizes[lastcoll]));
        sion_filedesc->all_chunksizes[lastcoll] = aligned_size;

        for (s = lastcoll + 1; s < i; s++) {
          /* leave chunksize of sender on aligned original size to
             allow storing data in intermeadiate non-collective
             flushes */
          _lsion_calculate_set_alignment(sion_filedesc, s);

          /* adjust chunksize of the sender to one file system block as spare space */
          /* sion_filedesc->all_chunksizes[s]=sion_filedesc->fsblksize; */
        }
      }

      /* adjust last senders */
      for (s = lastcoll; s < i; s++) {
        sion_filedesc->all_coll_collector[s] = lastcoll;
        sion_filedesc->all_coll_collsize[s] = numsender;
      }

      /* init data for next collector */
      numsender = 1;
      lastcoll = i;
      currentsize = sion_filedesc->all_chunksizes[i];
    }
  }

  /* align last sender */
  if (sion_filedesc->mode == SION_FILEMODE_WRITE) {
    newsize = currentsize;

    /* expand chunksize to align next one */
    aligned_size = ((int64_t)newsize / sion_filedesc->fsblksize) * sion_filedesc->fsblksize;
    if (newsize % sion_filedesc->fsblksize > 0) {
      aligned_size += sion_filedesc->fsblksize;
    }

    DPRINTFP((128, DFUNCTION, _SION_DEFAULT_RANK, "  resize chunksizes[%2d]=%8lld + %lld\n", lastcoll,
      sion_filedesc->all_chunksizes[lastcoll], aligned_size - sion_filedesc->all_chunksizes[lastcoll]));
    sion_filedesc->all_chunksizes[lastcoll] = aligned_size;

    /* adjust chunksize of the sender to one file system block as spare space */
    for (s = lastcoll + 1; s < i; s++) {
      /* leave chunksize of sender on aligned original size to
         allow storing data in intermeadiate non-collective
         flushes */
      _lsion_calculate_set_alignment(sion_filedesc, s);

      /* adjust chunksize of the sender to one file system block as spare space */
      /* sion_filedesc->all_chunksizes[s]=sion_filedesc->fsblksize; */
    }
  }

  /* adjust sender of last collector */
  for (s = lastcoll; s < sion_filedesc->ntasks; s++) {
    sion_filedesc->all_coll_collector[s] = lastcoll;
    sion_filedesc->all_coll_collsize[s] = numsender;
  }

  /* 2ND STEP: calculate startpointers */

  /* align first, not necessary, only for debugging  */
  /* startpointer=firstsize; */
  startpointer = (firstsize % sion_filedesc->fsblksize == 0)
    ? firstsize
    : ((firstsize / sion_filedesc->fsblksize) + 1) * sion_filedesc->fsblksize;
  sion_filedesc->globalskip = 0;
  /* calculate mapping to collectors */
  for (i = 0; i < sion_filedesc->ntasks; i++) {
    sion_filedesc->all_startpointers[i] = startpointer;
    startpointer += sion_filedesc->all_chunksizes[i];
    sion_filedesc->globalskip += sion_filedesc->all_chunksizes[i];
  }

  /* statistics */
  if (sion_filedesc->colldebug >= 1) {
    _lsion_update_collstat(collstat, sion_filedesc);
    _lsion_print_collstat(collstat, sion_filedesc);
  }

  _lsion_debugprint_collstat(collstat, sion_filedesc);

  _lsion_destroy_collstat(collstat);

  DPRINTFP((2, DFUNCTION, _SION_DEFAULT_RANK, "leave globalskip is %lld\n", sion_filedesc->globalskip));
  return rc;
}
#undef DFUNCTION
