BlockPattern1D.h

#ifndef DASH__BLOCK_PATTERN_1D_H_
#define DASH__BLOCK_PATTERN_1D_H_

#include <dash/Types.h>
#include <dash/Distribution.h>
#include <dash/Exception.h>
#include <dash/Dimensional.h>
#include <dash/Cartesian.h>
#include <dash/Team.h>

#include <dash/pattern/BlockPattern.h>
#include <dash/pattern/PatternProperties.h>
#include <dash/pattern/internal/PatternArguments.h>

#include <dash/internal/Math.h>
#include <dash/internal/Logging.h>

#include <functional>
#include <array>
#include <type_traits>


namespace dash {

template<
  MemArrange Arrangement,
  typename   IndexType
>
class BlockPattern<1, Arrangement, IndexType>
{
private:
  static const dim_t NumDimensions = 1;

public:
  static constexpr const char * PatternName = "BlockPattern1D";

public:
  typedef pattern_partitioning_properties<
              // Block extents are constant for every dimension.
              pattern_partitioning_tag::rectangular,
              // Identical number of elements in every block.
              pattern_partitioning_tag::balanced,
              // Size of blocks may differ.
              pattern_partitioning_tag::unbalanced
          > partitioning_properties;
  typedef pattern_mapping_properties<
              // Number of blocks assigned to a unit may differ.
              pattern_mapping_tag::unbalanced
          > mapping_properties;
  typedef pattern_layout_properties<
              // Local indices iterate over block boundaries.
              pattern_layout_tag::canonical,
              // Local element order corresponds to canonical linearization
              // within entire local memory.
              pattern_layout_tag::linear
          > layout_properties;

private:
  typedef typename std::make_unsigned<IndexType>::type
    SizeType;
  typedef BlockPattern<NumDimensions, Arrangement, IndexType>
    self_t;
  typedef CartesianIndexSpace<NumDimensions, Arrangement, IndexType>
    MemoryLayout_t;
  typedef CartesianIndexSpace<NumDimensions, Arrangement, IndexType>
    LocalMemoryLayout_t;
  typedef CartesianSpace<NumDimensions, SizeType>
    BlockSpec_t;
  typedef DistributionSpec<NumDimensions>
    DistributionSpec_t;
  typedef TeamSpec<NumDimensions, IndexType>
    TeamSpec_t;
  typedef SizeSpec<NumDimensions, SizeType>
    SizeSpec_t;
  typedef ViewSpec<NumDimensions, IndexType>
    ViewSpec_t;
  typedef internal::PatternArguments<NumDimensions, IndexType>
    PatternArguments_t;

public:
  typedef IndexType   index_type;
  typedef SizeType    size_type;
  typedef ViewSpec_t  viewspec_type;
  typedef struct {
    team_unit_t unit;
    IndexType   index;
  } local_index_t;
  typedef struct {
    team_unit_t unit;
    std::array<index_type, NumDimensions> coords;
  } local_coords_t;

private:
  SizeType                    _size            = 0;
  MemoryLayout_t              _memory_layout;
  DistributionSpec_t          _distspec;
  dash::Team *                _team            = nullptr;
  TeamSpec_t                  _teamspec;
  SizeType                    _nunits          = 0;
  SizeType                    _blocksize       = 0;
  SizeType                    _nblocks         = 0;
  SizeType                    _local_size      = 0;
  LocalMemoryLayout_t         _local_memory_layout;
  SizeType                    _nlblocks        = 0;
  SizeType                    _local_capacity  = 0;
  std::array<IndexType, 2>    _lbegin_lend     = { };
  IndexType                   _lend            = -1;

public:
  constexpr BlockPattern() = delete;

  template<typename ... Args>
  BlockPattern(
    SizeType arg,
    Args && ... args)
  : BlockPattern(PatternArguments_t(arg, args...))
  { }

  BlockPattern(
    const SizeSpec_t         sizespec,
    const DistributionSpec_t dist,
    const TeamSpec_t         teamspec,
    dash::Team &             team     = dash::Team::All())
  : _size(sizespec.size()),
    _memory_layout(std::array<SizeType, 1> {{ _size }}),
    _distspec(dist),
    _team(&team),
    _teamspec(
      teamspec,
      _distspec,
      *_team),
    _nunits(_team->size()),
    _blocksize(initialize_blocksize(
        _size,
        _distspec,
        _nunits)),
    _nblocks(initialize_num_blocks(
        _size,
        _blocksize,
        _nunits)),
    _local_size(
        initialize_local_extent(_team->myid())),
    _local_memory_layout(std::array<SizeType, 1> {{ _local_size }}),
    _nlblocks(initialize_num_local_blocks(
        _blocksize,
        _local_size)),
    _local_capacity(initialize_local_capacity()),
    _lbegin_lend(initialize_local_range(_local_size))
  { }

  BlockPattern(
    const SizeSpec_t         sizespec,
    const DistributionSpec_t dist = DistributionSpec_t(),
    Team &                   team = dash::Team::All())
  : _size(sizespec.size()),
    _memory_layout(std::array<SizeType, 1> {{ _size }}),
    _distspec(dist),
    _team(&team),
    _teamspec(_distspec, *_team),
    _nunits(_team->size()),
    _blocksize(initialize_blocksize(
        _size,
        _distspec,
        _nunits)),
    _nblocks(initialize_num_blocks(
        _size,
        _blocksize,
        _nunits)),
    _local_size(
        initialize_local_extent(_team->myid())),
    _local_memory_layout(std::array<SizeType, 1> {{ _local_size }}),
    _nlblocks(initialize_num_local_blocks(
        _blocksize,
        _local_size)),
    _local_capacity(initialize_local_capacity()),
    _lbegin_lend(initialize_local_range(_local_size))
  { }

  constexpr BlockPattern(self_t && other)      = default;

  constexpr BlockPattern(const self_t & other) = default;

  constexpr BlockPattern(self_t & other)
  : BlockPattern(static_cast<const self_t &>(other)) {
  }

  self_t & operator=(self_t && other)         = default;

  self_t & operator=(const self_t & other)    = default;

  constexpr bool operator==(
    const self_t & other
  ) const {
    // no need to compare all members as most are derived from
    // constructor arguments.
    return (
      this == &other
      || ( _size        == other._size &&
           _distspec    == other._distspec &&
           _teamspec    == other._teamspec &&
           _nblocks     == other._nblocks &&
           _nlblocks    == other._nlblocks &&
           _local_size  == other._local_size &&
           _blocksize   == other._blocksize &&
           _nunits      == other._nunits &&
           _lbegin_lend == other._lbegin_lend
         )
      );
  }

  constexpr bool operator!=(
    const self_t & other
  ) const {
    return !(*this == other);
  }

  constexpr IndexType lbegin() const {
    return _lbegin_lend[0];
  }

  constexpr IndexType lend() const {
    return _lbegin_lend[1];
  }


  constexpr team_unit_t unit_at(
    const std::array<IndexType, NumDimensions> & coords,
    const ViewSpec_t & viewspec) const {
    return team_unit_t (((coords[0] + viewspec.offset(0)) / _blocksize)
                        % _nunits);
  }

  constexpr team_unit_t unit_at(
    const std::array<IndexType, NumDimensions> & coords) const {
    return team_unit_t((coords[0] / _blocksize) % _nunits);
  }

  constexpr team_unit_t unit_at(
    IndexType global_pos,
    const ViewSpec_t & viewspec
  ) const {
    return team_unit_t(((global_pos + viewspec.offset(0)) / _blocksize)
                       % _nunits);
  }

  constexpr team_unit_t unit_at(
    IndexType global_pos
  ) const {
    return team_unit_t((global_pos / _blocksize) % _nunits);
  }


  IndexType extent(dim_t dim) const {
    DASH_ASSERT_EQ(
      0, dim,
      "Wrong dimension for Pattern::local_extent. " <<
      "Expected dimension = 0, got " << dim);
    return _size;
  }

  template <dim_t dim = 0>
  constexpr IndexType extent() const {
    static_assert(
      0 == dim,
      "Wrong dimension for Pattern::local_extent. "
      "Expected dimension = 0");
    return _size;
  }

  IndexType local_extent(dim_t dim) const {
    DASH_ASSERT_EQ(
      0, dim,
      "Wrong dimension for Pattern::local_extent. " <<
      "Expected dimension = 0, got " << dim);
    return _local_size;
  }

  template <dim_t dim = 0>
  constexpr IndexType local_extent() const {
    static_assert(
      0 == dim,
      "Wrong dimension for Pattern::local_extent. "
      "Expected dimension = 0");
    return _local_size;
  }

  constexpr std::array<SizeType, NumDimensions> local_extents(
    team_unit_t unit) const {
    return std::array<SizeType, 1> {{
              unit == _team->myid()
              ? _local_size
              : initialize_local_extent(unit)
           }};
  }

  constexpr std::array<SizeType, NumDimensions> local_extents() const {
    return std::array<SizeType, 1> {{  _local_size }};
  }


  constexpr IndexType local_at(
    const std::array<IndexType, NumDimensions> & local_coords,
    const ViewSpec_t & viewspec) const {
    return local_coords[0] + viewspec.offset(0);
  }

  constexpr IndexType local_at(
    const std::array<IndexType, NumDimensions> & local_coords) const {
    return local_coords[0];
  }

  constexpr local_coords_t local(
    const std::array<IndexType, NumDimensions> & global_coords) const {
    return local_coords_t {
             unit_at(global_coords),      // .unit
             local_coords(global_coords)  // .coords
           };
  }

  constexpr local_index_t local(
    IndexType g_index) const {
    return local_index(coords(g_index));
  }

  constexpr std::array<IndexType, NumDimensions> local_coords(
    const std::array<IndexType, NumDimensions> & global_coords
  ) const noexcept {
    return std::array<IndexType, 1> {{
             static_cast<IndexType>(
               (((global_coords[0] / _blocksize) / _nunits) * _blocksize)
               + (global_coords[0] % _blocksize)
             )
           }};
  }

  constexpr local_index_t local_index(
    const std::array<IndexType, NumDimensions> & g_coords) const {
    return local_index_t {
             team_unit_t(unit_at(g_coords[0])),
             at(g_coords)
           };
  }


  constexpr std::array<IndexType, NumDimensions> global(
    team_unit_t unit,
    const std::array<IndexType, NumDimensions> & local_coords) const {
    return (_nunits < 2)
            ? local_coords
            : std::array<IndexType, 1> {{
                static_cast<IndexType>(
                  (( _distspec[0].local_index_to_block_coord(
                       static_cast<IndexType>(unit),
                       local_coords[0],
                       _nunits)
                   ) * _blocksize)
                  + (local_coords[0] % _blocksize)
                )
              }};
  }

  constexpr std::array<IndexType, NumDimensions> global(
    const std::array<IndexType, NumDimensions> & l_coords) const
  {
    return global(_team->myid(), l_coords);
  }

  constexpr IndexType global(
    IndexType l_index) const
  {
    return global(_team->myid(), std::array<IndexType, 1> {{ l_index }})[0];
  }

  constexpr IndexType global_index(
    team_unit_t unit,
    const std::array<IndexType, NumDimensions> & l_coords) const
  {
    return global(unit, l_coords)[0];
  }

  constexpr IndexType global_at(
    const std::array<IndexType, NumDimensions> & global_coords,
    const ViewSpec_t & viewspec) const {
    return global_coords[0] + viewspec.offset(0);
  }

  constexpr IndexType global_at(
    const std::array<IndexType, NumDimensions> & global_coords) const {
    return global_coords[0];
  }



  constexpr IndexType at(
    const std::array<IndexType, NumDimensions> & g_coords) const {
    return local_coords(g_coords)[0];
  }

  constexpr IndexType at(
    const std::array<IndexType, 1> & g_coords,
    const ViewSpec_t & viewspec) const {
    return local_coords(
             std::array<IndexType, 1> {{
               g_coords[0] + viewspec.offset(0)
             }}
           )[0];
  }

  template<typename ... Values>
  constexpr IndexType at(IndexType value, Values ... values) const {
    static_assert(
      sizeof...(values) == NumDimensions-1,
      "Wrong parameter number");
    return at(std::array<IndexType, NumDimensions> {
                value, (IndexType)values...
              });
  }


  constexpr bool has_local_elements(
    dim_t dim,
    IndexType dim_offset,
    team_unit_t unit) const {
    // Check if unit id lies in cartesian sub-space of team spec
    return _teamspec.includes_index(
              unit,
              dim,
              dim_offset);
  }

  constexpr bool is_local(
    IndexType index,
    team_unit_t unit) const {
    return unit_at(index) == unit;
  }

  constexpr bool is_local(
    IndexType index) const {
    return is_local(index, team().myid());
  }


  constexpr BlockSpec_t blockspec() const {
    return BlockSpec_t({ _nblocks });
  }

  constexpr BlockSpec_t local_blockspec() const
  {
    return BlockSpec_t({ _nlblocks });
  }

  constexpr index_type block_at(
    const std::array<index_type, NumDimensions> & g_coords) const {
    return g_coords[0] / _blocksize;
  }

  constexpr local_index_t local_block_at(
    const std::array<index_type, NumDimensions> & g_coords) const {
    return local_index_t {
             // unit id:
             static_cast<team_unit_t>(
                (g_coords[0] / _blocksize) % _teamspec.size()),
             // local block index:
             static_cast<index_type>(
                (g_coords[0] / _blocksize) / _teamspec.size())
           };
  }

  constexpr ViewSpec_t block(
    index_type g_block_index) const
  {
    return ViewSpec_t(
      {{ static_cast<index_type>(g_block_index * _blocksize) }},
      {{ static_cast<size_type>(
          _blocksize - ( g_block_index < _nblocks - 1
                         ? 0
                         : underfilled_blocksize(0) )
         ) }}
    );
  }

  constexpr ViewSpec_t local_block(
    index_type l_block_index) const
  {
    // Local block index to local block coords:
    return ViewSpec_t(
      {{ static_cast<index_type>( global(l_block_index * _blocksize) ) }},
      {{ static_cast<size_type>(
          (l_block_index == (_nlblocks - 1)
          ? (_local_size % _blocksize == 0
             ? _blocksize
             : _local_size % _blocksize )
          : _blocksize )
         ) }}
    );
  }

  ViewSpec_t local_block_local(
    index_type l_block_index) const
  {
    DASH_LOG_DEBUG_VAR("BlockPattern<1>.local_block_local()", l_block_index);
    index_type offset = l_block_index * _blocksize;
    std::array<index_type, NumDimensions> offsets = {{ offset }};
    std::array<size_type, NumDimensions>  extents = {{ _blocksize }};
    if(l_block_index == (_nlblocks - 1))
    {
      size_type remaining = _local_size % extents[0];
      extents[0] = (remaining == 0) ? extents[0] : remaining;
    }
    ViewSpec_t block_vs(offsets, extents);
    DASH_LOG_DEBUG_VAR("BlockPattern<1>.local_block_local >", block_vs);
    return block_vs;
  }

  constexpr SizeType blocksize(
    dim_t dimension) const {
    return _blocksize;
  }

  constexpr SizeType max_blocksize() const {
    return _blocksize;
  }

  constexpr SizeType local_capacity() const
  {
    return _local_capacity;
  }

  constexpr SizeType local_size(team_unit_t unit) const
  {
    return (unit == _team->myid().id)
           ? _local_size
           : initialize_local_extent(unit);
  }

  constexpr SizeType local_size() const noexcept
  {
    return _local_size;
  }

  constexpr IndexType num_units() const noexcept {
    return _nunits;
  }

  constexpr IndexType capacity() const noexcept {
    return _size;
  }

  constexpr IndexType size() const noexcept {
    return _size;
  }

  constexpr dash::Team & team() const noexcept {
    return *_team;
  }

  constexpr const DistributionSpec_t & distspec() const noexcept {
    return _distspec;
  }

  constexpr SizeSpec_t sizespec() const {
    return SizeSpec_t(std::array<SizeType, 1> {{ _size }});
  }

  constexpr const std::array<SizeType, NumDimensions> extents() const {
    return std::array<SizeType, 1> {{ _size }};
  }

  constexpr const TeamSpec_t & teamspec() const {
    return _teamspec;
  }

  constexpr std::array<IndexType, NumDimensions> coords(
    IndexType index) const {
    return std::array<IndexType, 1> {{ index }};
  }

  constexpr std::array<IndexType, NumDimensions> coords(
    IndexType          index,
    const ViewSpec_t & viewspec) const {
    return std::array<IndexType, 1> {{ index + viewspec.offset(0) }};
  }

  constexpr static MemArrange memory_order() {
    return Arrangement;
  }

  constexpr static dim_t ndim() {
    return 1;
  }

  constexpr SizeType underfilled_blocksize(
    dim_t dimension) const {
    // Underflow blocksize = regular blocksize - overflow blocksize:
    return ( _blocksize == 0 || _size % _blocksize == 0
             ? 0
             : _blocksize - (_size % _blocksize)
           );
  }

private:

  BlockPattern(const PatternArguments_t & arguments)
  :  _size(arguments.sizespec().size()),
     _memory_layout(std::array<SizeType, 1> {{ _size }}),
     _distspec(arguments.distspec()),
     _team(&arguments.team()),
     _teamspec(arguments.teamspec()),
     _nunits(_team->size()),
     _blocksize(initialize_blocksize(
         _size,
         _distspec,
         _nunits)),
     _nblocks(initialize_num_blocks(
         _size,
         _blocksize,
         _nunits)),
     _local_size(
         initialize_local_extent(_team->myid())),
     _local_memory_layout(std::array<SizeType, 1> {{ _local_size }}),
     _nlblocks(initialize_num_local_blocks(
         _blocksize,
         _local_size)),
     _local_capacity(initialize_local_capacity()),
     _lbegin_lend(initialize_local_range(_local_size))
  {}

  SizeType initialize_blocksize(
    SizeType                   size,
    const DistributionSpec_t & distspec,
    SizeType                   nunits) const {
    DASH_LOG_TRACE_VAR("BlockPattern<1>.init_blocksize", nunits);
    if (nunits == 0) {
      return 0;
    }
    const Distribution & dist = distspec[0];
    return dist.max_blocksize_in_range(
             size,    // size of range (extent)
             nunits); // number of blocks (units)
  }

  SizeType initialize_num_blocks(
    SizeType                   size,
    SizeType                   blocksize,
    SizeType                   nunits) const {
    if (blocksize == 0) {
      return 0;
    }
    DASH_LOG_TRACE("BlockPattern<1>.init_num_blocks()",
                   "size", size, "blocksize", blocksize, "nunits", nunits);
    SizeType n_blocks = dash::math::div_ceil(
                          size,
                          blocksize);
    DASH_LOG_TRACE_VAR("BlockPattern<1>.init_blockspec", n_blocks);
    return n_blocks;
  }

  SizeType initialize_num_local_blocks(
    SizeType                    blocksize,
    SizeType                    local_size) const {
    auto num_l_blocks = local_size;
    if (blocksize > 0) {
      num_l_blocks = dash::math::div_ceil(
                       num_l_blocks,
                       blocksize);
    } else {
      num_l_blocks = 0;
    }
    DASH_LOG_TRACE_VAR("BlockPattern<1>.init_num_local_blocks", num_l_blocks);
    return num_l_blocks;
  }

  SizeType initialize_local_capacity() const {
    SizeType l_capacity = 1;
    if (_nunits == 0) {
      return 0;
    }
    auto max_l_blocks = dash::math::div_ceil(
                          _nblocks,
                          _nunits);
    DASH_LOG_TRACE_VAR("BlockPattern<1>.init_lcapacity.d", _nunits);
    DASH_LOG_TRACE_VAR("BlockPattern<1>.init_lcapacity.d", max_l_blocks);
    l_capacity = max_l_blocks * _blocksize;
    DASH_LOG_DEBUG_VAR("BlockPattern<1>.init_lcapacity >", l_capacity);
    return l_capacity;
  }

  constexpr std::array<IndexType, 2>
  initialize_local_range(SizeType l_size) const {
    return (l_size == 0)
             ? std::array<IndexType, 2> {{ 0, 0 }}
             : std::array<IndexType, 2> {{
                 global(0),
                 global(l_size - 1) + 1
               }};
  }

  constexpr SizeType initialize_local_extent(
    team_unit_t unit) const
  {
    return (_nunits == 0
           ? 0
           : (_nblocks == 1 && _nunits == 1
             ? _size
               // Possibly there are more blocks than units in dimension
               // and no block left for this unit. Local extent in d then
               // becomes 0.
             : ((_nblocks / _nunits) * _blocksize
               // Unbalanced blocks owned by the unit:
               + (static_cast<SizeType>(unit) < (_nblocks % _nunits)
                 ? _blocksize
                 : 0 )
               // If the last block in the dimension is underfilled and
               // assigned to the local unit, subtract the missing extent:
               - (static_cast<SizeType>(unit) ==
                    // Unit id assigned to the last block:
                    (_nblocks % _nunits == 0
                    ? _nunits - 1
                    : ((_nblocks % _nunits) - 1 ))
                 ? underfilled_blocksize(0)
                 : 0 )
               )
             )
           );

  }
};

} // namespace dash

#endif // DASH__BLOCK_PATTERN_1D_H_