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

#define _XOPEN_SOURCE 700

#include <assert.h>

#include "sion_internal_positioning.h"

_sion_range _lsion_range_split_off_prefix(_sion_range *range, int64_t length)
{
  assert(length >= 0);
  _sion_range prefix =
    (_sion_range){.start = range->start, .length = (range->length < length) ? range->length : length};

  range->start += prefix.length;
  range->length -= prefix.length;
  return prefix;
}

void _lsion_range_clamp(_sion_range *range, int64_t length)
{
  assert(length >= 0);
  range->length = (range->length < length) ? range->length : length;
}

void _lsion_range_into_spec(_sion_range range, int64_t *spec)
{
  spec[0] = range.start;
  spec[1] = range.length;
}

_sion_chunk_sequence _lsion_chunk_sequence_from_parts(int64_t start, int64_t stride, int64_t chunksize)
{
  assert(start >= 0);
  assert(chunksize > 0);
  assert(stride >= chunksize);

  return (_sion_chunk_sequence){.start = start, .stride = stride, .chunksize = chunksize};
}

_sion_chunk_sequence _lsion_chunk_sequence_from_filedesc(const _sion_filedesc *fd)
{
  return _lsion_chunk_sequence_from_parts(fd->startpos, fd->globalskip, fd->chunksize);
}

bool _lsion_chunk_sequence_position_oob(const _sion_chunk_sequence *sequence, int64_t physical_position)
{
  assert(physical_position - sequence->start >= 0);
  return false;
}

int64_t _lsion_chunk_sequence_translate_l2p(const _sion_chunk_sequence *sequence, int64_t logical_position)
{
  assert(logical_position >= 0);
  return sequence->start + (logical_position / sequence->chunksize) * sequence->stride + (logical_position % sequence->chunksize);
}

int32_t _lsion_chunk_sequence_translate_l2n(const _sion_chunk_sequence *sequence, int64_t logical_position)
{
  assert(logical_position >= 0);
  return logical_position / sequence->chunksize;
}

int64_t _lsion_chunk_sequence_translate_p2l(const _sion_chunk_sequence *sequence, int64_t physical_position)
{
  assert(physical_position - sequence->start >= 0);
  int64_t posinblock = (physical_position - sequence->start) % sequence->stride;
  int32_t blocknr = (physical_position - sequence->start) / sequence->stride;
  return blocknr * sequence->chunksize + ((posinblock < sequence->chunksize) ? posinblock : sequence->chunksize);
}

int32_t _lsion_chunk_sequence_translate_p2n(const _sion_chunk_sequence *sequence, int64_t physical_position)
{
  assert(physical_position - sequence->start >= 0);
  int32_t blocknr = (physical_position - sequence->start) / sequence->stride;
  int64_t posinblock = (physical_position - sequence->start) % sequence->stride;
  return blocknr + ((posinblock < sequence->chunksize) ? 0 : 1);
}

_sion_range _lsion_chunk_sequence_chunk_from_number(const _sion_chunk_sequence *sequence, int32_t chunk_number)
{
  assert(chunk_number >= 0);
  return (_sion_range){.start = sequence->start + chunk_number * sequence->stride, .length = sequence->chunksize};
}

_sion_range _lsion_chunk_sequence_chunk_from_physical_position(const _sion_chunk_sequence *sequence, int64_t physical_position)
{
  int64_t logical_position = _lsion_chunk_sequence_translate_p2l(sequence, physical_position);
  return _lsion_chunk_sequence_chunk_from_logical_position(sequence, logical_position);
}

_sion_range _lsion_chunk_sequence_chunk_from_logical_position(const _sion_chunk_sequence *sequence, int64_t logical_position)
{
  return (_sion_range){.start = _lsion_chunk_sequence_translate_l2p(sequence, logical_position),
    .length = sequence->chunksize - (logical_position % sequence->chunksize)};
}

_sion_limited_chunk_sequence _lsion_limited_chunk_sequence_from_parts(
  int64_t start, int64_t stride, size_t n_chunks, const int64_t *chunk_sizes)
{
  assert(start >= 0);
  assert(chunk_sizes != NULL);
  for (size_t ichunk = 0; ichunk < n_chunks; ichunk++) {
    assert(chunk_sizes[ichunk] >= 0);
    assert(stride >= chunk_sizes[ichunk]);
  }

  int64_t ubound = start + (n_chunks - 1) * stride + ((n_chunks > 0) ? chunk_sizes[n_chunks - 1] : 0);
  return (_sion_limited_chunk_sequence){
    .start = start, .stride = stride, .n_chunks = n_chunks, .chunk_sizes = chunk_sizes, .ubound = ubound};
}

_sion_limited_chunk_sequence _lsion_limited_chunk_sequence_from_filedesc(const _sion_filedesc *fd)
{
  return _lsion_limited_chunk_sequence_from_parts(fd->startpos, fd->globalskip, fd->lastchunknr + 1, fd->blocksizes);
}

bool _lsion_limited_chunk_sequence_position_oob(const _sion_limited_chunk_sequence *sequence, int64_t physical_position)
{
  assert(physical_position - sequence->start >= 0);
  return physical_position >= sequence->ubound;
}

int32_t _lsion_limited_chunk_sequence_translate_p2n(const _sion_limited_chunk_sequence *sequence, int64_t physical_position)
{
  assert(!_lsion_limited_chunk_sequence_position_oob(sequence, physical_position));
  int32_t chunknr = (physical_position - sequence->start) / sequence->stride;
  int64_t posinblock = (physical_position - sequence->start) % sequence->stride;
  return chunknr + ((posinblock >= sequence->chunk_sizes[chunknr]) ? 1 : 0);
}

_sion_range _lsion_limited_chunk_sequence_chunk_from_number(const _sion_limited_chunk_sequence *sequence, int32_t chunk_number)
{
  assert(chunk_number >= 0 && chunk_number < sequence->n_chunks);

  return (_sion_range){.start = sequence->start + chunk_number * sequence->stride, .length = sequence->chunk_sizes[chunk_number]};
}

_sion_range _lsion_limited_chunk_sequence_chunk_from_physical_position(
  const _sion_limited_chunk_sequence *sequence, int64_t physical_position)
{
  assert(!_lsion_limited_chunk_sequence_position_oob(sequence, physical_position));

  int32_t chunknr = (physical_position - sequence->start) / sequence->stride;
  _sion_range chunk;
  // start of chunk
  if (((physical_position - sequence->start) % sequence->stride) >= sequence->chunk_sizes[chunknr]) {
    // current position lies beyond a chunk that belongs to this sequence, skip ahead to start of next chunk
    chunknr++;
    chunk.start = sequence->start + chunknr * sequence->stride;
  } else {
    chunk.start = physical_position;
  }

  // length of chunk
  chunk.length = sequence->chunk_sizes[chunknr] - ((chunk.start - sequence->start) % sequence->stride);

  return chunk;
}

_sion_range_iterator _lsion_range_iterator_from_parts(_sion_chunk_sequence sequence, int64_t position, int64_t amount)
{
  assert(position >= 0);
  assert(amount >= 0);

  return (_sion_range_iterator){.sequence = sequence, .position = position, .remaining = amount};
}

_sion_range_iterator _lsion_range_iterator_from_filedesc(const _sion_filedesc *fd, int64_t amount)
{
  _sion_chunk_sequence sequence = _lsion_chunk_sequence_from_filedesc(fd);
  return _lsion_range_iterator_from_parts(sequence, fd->currentpos, amount);
}

_sion_range_kind _lsion_range_iterator_next(_sion_range_iterator *self, _sion_range *next)
{
  if (self->remaining <= 0 || _lsion_chunk_sequence_position_oob(&(self->sequence), self->position)) {
    return _SION_EXHAUSTED;
  }

  *next = _lsion_chunk_sequence_chunk_from_physical_position(&(self->sequence), self->position);
  _lsion_range_clamp(next, self->remaining);

  _sion_range_kind kind = (next->start == self->position) ? _SION_REMAINDER : _SION_NEXT;

  self->position = next->start + next->length;
  self->remaining -= next->length;

  return kind;
}

_sion_partitioned_range_iterator _lsion_partitioned_range_iterator_from_parts(
  _sion_chunk_sequence sequence, int64_t position, int64_t amount, int64_t partition)
{
  assert(amount == 0 || partition > 0);
  _sion_range_iterator inner = _lsion_range_iterator_from_parts(sequence, position, amount);
  _sion_range current = (_sion_range){.start = 0, .length = 0};
  _lsion_range_iterator_next(&inner, &current);
  return (_sion_partitioned_range_iterator){.inner = inner, .current = current, .partition = partition};
}

_sion_partitioned_range_iterator _lsion_partitioned_range_iterator_from_filedesc(
  const _sion_filedesc *fd, int64_t amount, int64_t partition)
{
  _sion_chunk_sequence sequence = _lsion_chunk_sequence_from_filedesc(fd);
  return _lsion_partitioned_range_iterator_from_parts(sequence, fd->currentpos, amount, partition);
}

_sion_range_kind _lsion_partitioned_range_iterator_next(_sion_partitioned_range_iterator *self, _sion_range *next)
{
  _sion_range_kind kind;
  if (self->current.length <= 0) {
    // need to get a new range, kind is determined by inner iterator
    if (!(kind = _lsion_range_iterator_next(&(self->inner), &(self->current)))) {
      return _SION_EXHAUSTED;
    }
  } else {
    // next range is a part of the range that is currently being split
    kind = _SION_REMAINDER;
  }

  // TODO: align ranges with partitions (fs blocks)
  *next = _lsion_range_split_off_prefix(&(self->current), self->partition);

  return kind;
}

_sion_limited_range_iterator _lsion_limited_range_iterator_from_parts(
  _sion_limited_chunk_sequence sequence, int64_t position, int64_t amount)
{
  assert(position >= 0);
  assert(amount >= 0);

  return (_sion_limited_range_iterator){.sequence = sequence, .position = position, .remaining = amount};
}

_sion_limited_range_iterator _lsion_limited_range_iterator_from_filedesc(const _sion_filedesc *fd, int64_t amount)
{
  _sion_limited_chunk_sequence sequence = _lsion_limited_chunk_sequence_from_filedesc(fd);
  return _lsion_limited_range_iterator_from_parts(sequence, fd->currentpos, amount);
}

_sion_range_kind _lsion_limited_range_iterator_next(_sion_limited_range_iterator *self, _sion_range *next)
{
  if (self->remaining <= 0 || _lsion_limited_chunk_sequence_position_oob(&(self->sequence), self->position)) {
    return _SION_EXHAUSTED;
  }

  *next = _lsion_limited_chunk_sequence_chunk_from_physical_position(&(self->sequence), self->position);
  _lsion_range_clamp(next, self->remaining);

  _sion_range_kind kind = (next->start == self->position) ? _SION_REMAINDER : _SION_NEXT;

  self->position = next->start + next->length;
  self->remaining -= next->length;

  return kind;
}

_sion_partitioned_limited_range_iterator _lsion_partitioned_limited_range_iterator_from_parts(
  _sion_limited_chunk_sequence sequence, int64_t position, int64_t amount, int64_t partition)
{
  assert(amount == 0 || partition > 0);
  _sion_limited_range_iterator inner = _lsion_limited_range_iterator_from_parts(sequence, position, amount);
  _sion_range current = (_sion_range){.start = 0, .length = 0};
  _lsion_limited_range_iterator_next(&inner, &current);
  return (_sion_partitioned_limited_range_iterator){.inner = inner, .current = current, .partition = partition};
}

_sion_partitioned_limited_range_iterator _lsion_partitioned_limited_range_iterator_from_filedesc(
  const _sion_filedesc *fd, int64_t amount, int64_t partition)
{
  _sion_limited_chunk_sequence sequence = _lsion_limited_chunk_sequence_from_filedesc(fd);
  return _lsion_partitioned_limited_range_iterator_from_parts(sequence, fd->currentpos, amount, partition);
}

_sion_range_kind _lsion_partitioned_limited_range_iterator_next(_sion_partitioned_limited_range_iterator *self, _sion_range *next)
{
  _sion_range_kind kind;
  if (self->current.length <= 0) {
    // need to get a new range, kind is determined by inner iterator
    if (!(kind = _lsion_limited_range_iterator_next(&(self->inner), &(self->current)))) {
      return _SION_EXHAUSTED;
    }
  } else {
    // next range is a part of the range that is currently being split
    kind = _SION_REMAINDER;
  }

  // TODO: align ranges with partitions (fs blocks)
  *next = _lsion_range_split_off_prefix(&(self->current), self->partition);

  return kind;
}

int64_t _lsion_get_logical_position(const _sion_filedesc *filedesc)
{
  assert(filedesc->mode & SION_FILEMODE_WRITE);
  _sion_chunk_sequence sequence = _lsion_chunk_sequence_from_filedesc(filedesc);
  return _lsion_chunk_sequence_translate_p2l(&sequence, filedesc->currentpos);
}

int32_t _lsion_get_current_chunk_number(const _sion_filedesc *filedesc)
{
  if (filedesc->mode & SION_FILEMODE_WRITE) {
    _sion_chunk_sequence sequence = _lsion_chunk_sequence_from_filedesc(filedesc);
    return _lsion_chunk_sequence_translate_p2n(&sequence, filedesc->currentpos);
  } else {
    _sion_limited_chunk_sequence sequence = _lsion_limited_chunk_sequence_from_filedesc(filedesc);
    return _lsion_limited_chunk_sequence_translate_p2n(&sequence, filedesc->currentpos);
  }
}

_sion_range _lsion_get_current_chunk(const _sion_filedesc *filedesc)
{
  if (filedesc->mode & SION_FILEMODE_WRITE) {
    _sion_chunk_sequence sequence = _lsion_chunk_sequence_from_filedesc(filedesc);
    int32_t blocknr = _lsion_chunk_sequence_translate_p2n(&sequence, filedesc->currentpos);
    return _lsion_chunk_sequence_chunk_from_number(&sequence, blocknr);
  } else {
    _sion_limited_chunk_sequence sequence = _lsion_limited_chunk_sequence_from_filedesc(filedesc);
    int32_t blocknr = _lsion_limited_chunk_sequence_translate_p2n(&sequence, filedesc->currentpos);
    return _lsion_limited_chunk_sequence_chunk_from_number(&sequence, blocknr);
  }
}
