ShiftTilePattern.h

#ifndef DASH__SHIFT_TILE_PATTERN_H_
#define DASH__SHIFT_TILE_PATTERN_H_

#include <cassert>
#include <functional>
#include <cstring>
#include <array>
#include <type_traits>
#include <utility>

#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/PatternProperties.h>
#include <dash/pattern/internal/PatternArguments.h>

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

namespace dash {

template<
  dim_t NumDimensions,
  MemArrange Arrangement = ROW_MAJOR,
  typename IndexType     = dash::default_index_t>
class ShiftTilePattern
{
public:
  static constexpr char const * PatternName = "ShiftTilePattern";

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
          > partitioning_properties;
  typedef pattern_mapping_properties<
              // Same number of blocks assigned to every unit.
              pattern_mapping_tag::balanced,
              // Number of blocks assigned to a unit may differ.
              pattern_mapping_tag::unbalanced,
              // Every unit mapped in any single slice in every dimension.
              pattern_mapping_tag::diagonal
          > mapping_properties;
  typedef pattern_layout_properties<
              // Elements are contiguous in local memory within single block.
              pattern_layout_tag::blocked,
              // Local element order corresponds to a logical linearization
              // within single blocks.
              pattern_layout_tag::linear
          > layout_properties;

private:
  typedef typename std::make_unsigned<IndexType>::type
    SizeType;
  typedef ShiftTilePattern<NumDimensions, Arrangement, IndexType>
    self_t;
  typedef CartesianIndexSpace<NumDimensions, Arrangement, IndexType>
    MemoryLayout_t;
  typedef CartesianIndexSpace<NumDimensions, Arrangement, IndexType>
    LocalMemoryLayout_t;
  typedef CartesianIndexSpace<NumDimensions, Arrangement, SizeType>
    BlockSpec_t;
  typedef CartesianIndexSpace<NumDimensions, Arrangement, SizeType>
    BlockSizeSpec_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:
  DistributionSpec_t          _distspec;
  dash::Team                * _team            = nullptr;
  TeamSpec_t                  _teamspec;
  MemoryLayout_t              _memory_layout;
  SizeType                    _nunits          = dash::Team::All().size();
  dim_t                       _major_tiled_dim{};
  dim_t                       _minor_tiled_dim{};
  BlockSizeSpec_t             _blocksize_spec;
  BlockSpec_t                 _blockspec;
  BlockSpec_t                 _local_blockspec;
  LocalMemoryLayout_t         _local_memory_layout;
  SizeType                    _local_capacity;
  IndexType                   _lbegin;
  IndexType                   _lend;

public:
  template<typename ... Args>
  ShiftTilePattern(
    SizeType arg,
    Args && ... args)
  : ShiftTilePattern(PatternArguments_t(arg, args...)) {
    DASH_LOG_TRACE("ShiftTilePattern()", "Constructor with Argument list");
    initialize_local_range();
  }

  ShiftTilePattern(
      const SizeSpec_t &sizespec,
      DistributionSpec_t dist,
      TeamSpec_t teamspec,
      dash::Team &team = dash::Team::All())
    : _distspec(std::move(dist))
    , _team(&team)
    ,
    // Degrading to 1-dimensional team spec for now:
    //  _teamspec(_distspec, *_team),
    //  _teamspec(
    //    teamspec,
    //    _distspec,
    //    *_team),
    _teamspec(std::move(teamspec))
    , _memory_layout(sizespec.extents())
    , _nunits(_teamspec.size())
    , _major_tiled_dim(initialize_major_tiled_dim(_distspec))
    , _minor_tiled_dim((_major_tiled_dim + 1) % NumDimensions)
    , _blocksize_spec(
          initialize_blocksizespec(sizespec, _distspec, _teamspec))
    , _blockspec(initialize_blockspec(sizespec, _blocksize_spec, _teamspec))
    , _local_blockspec(
          initialize_local_blockspec(_blockspec, _major_tiled_dim, _nunits))
    , _local_memory_layout(initialize_local_extents(_team->myid()))
    , _local_capacity(initialize_local_capacity())
  {
    DASH_LOG_TRACE("ShiftTilePattern()", "(sizespec, dist, teamspec, team)");
    initialize_local_range();
  }

  ShiftTilePattern(
      const SizeSpec_t &sizespec,
      DistributionSpec_t dist = DistributionSpec_t(),
      Team &team = dash::Team::All())
    : _distspec(std::move(dist))
    , _team(&team)
    , _teamspec(_distspec, *_team)
    , _memory_layout(sizespec.extents())
    , _nunits(_teamspec.size())
    , _major_tiled_dim(initialize_major_tiled_dim(_distspec))
    , _minor_tiled_dim((_major_tiled_dim + 1) % NumDimensions)
    , _blocksize_spec(
          initialize_blocksizespec(sizespec, _distspec, _teamspec))
    , _blockspec(initialize_blockspec(sizespec, _blocksize_spec, _teamspec))
    , _local_blockspec(
          initialize_local_blockspec(_blockspec, _major_tiled_dim, _nunits))
    , _local_memory_layout(initialize_local_extents(_team->myid()))
    , _local_capacity(initialize_local_capacity())
  {
    DASH_LOG_TRACE("ShiftTilePattern()", "(sizespec, dist, team)");
    initialize_local_range();
  }

  ShiftTilePattern(const self_t & other)
  : _distspec(other._distspec),
    _team(other._team),
    _teamspec(other._teamspec),
    _memory_layout(other._memory_layout),
    _nunits(other._nunits),
    _major_tiled_dim(other._major_tiled_dim),
    _minor_tiled_dim(other._minor_tiled_dim),
    _blocksize_spec(other._blocksize_spec),
    _blockspec(other._blockspec),
    _local_blockspec(other._local_blockspec),
    _local_memory_layout(other._local_memory_layout),
    _local_capacity(other._local_capacity),
    _lbegin(other._lbegin),
    _lend(other._lend) {
  }

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

  bool operator==(
    const self_t & other
  ) const {
    if (this == &other) {
      return true;
    }
    // no need to compare all members as most are derived from
    // constructor arguments.
    return(
      _distspec       == other._distspec &&
      _teamspec       == other._teamspec &&
      _memory_layout  == other._memory_layout &&
      _blockspec      == other._blockspec &&
      _blocksize_spec == other._blocksize_spec &&
      _nunits         == other._nunits
    );
  }

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

  ShiftTilePattern & operator=(const ShiftTilePattern & other) {
    if (this != &other) {
      _distspec            = other._distspec;
      _team                = other._team;
      _teamspec            = other._teamspec;
      _memory_layout       = other._memory_layout;
      _local_memory_layout = other._local_memory_layout;
      _blocksize_spec      = other._blocksize_spec;
      _blockspec           = other._blockspec;
      _local_blockspec     = other._local_blockspec;
      _local_capacity      = other._local_capacity;
      _nunits              = other._nunits;
      _minor_tiled_dim     = other._minor_tiled_dim;
      _major_tiled_dim     = other._major_tiled_dim;
      _lbegin              = other._lbegin;
      _lend                = other._lend;
    }
    return *this;
  }

  IndexType lbegin() const {
    return _lbegin;
  }

  IndexType lend() const {
    return _lend;
  }


  team_unit_t unit_at(
    const std::array<IndexType, NumDimensions> & coords,
    const ViewSpec_t & viewspec) const
  {
    DASH_LOG_TRACE("ShiftTilePattern.unit_at()",
                   "coords:",   coords,
                   "viewspec:", viewspec);
    // Unit id from diagonals in cartesian index space,
    // e.g (x + y + z) % nunits
    team_unit_t unit_id{0};
    for (auto d = 0; d < NumDimensions; ++d) {
      auto vs_coord     = coords[d] + viewspec.offset(d);
      // Global block coordinate:
      auto block_coord  = vs_coord / _blocksize_spec.extent(d);
      unit_id          += block_coord;
    }
    unit_id %= _nunits;
    DASH_LOG_TRACE_VAR("ShiftTilePattern.unit_at >", unit_id);
    return unit_id;
  }

  team_unit_t unit_at(
    const std::array<IndexType, NumDimensions> & coords) const
  {
    DASH_LOG_TRACE("ShiftTilePattern.unit_at()",
                   "coords:",    coords,
                   "blocksize:", _blocksize_spec.extents());
    // Unit id from diagonals in cartesian index space,
    // e.g (x + y + z) % nunits
    team_unit_t unit_id{0};
    for (auto d = 0; d < NumDimensions; ++d) {
      // Global block coordinate:
      auto block_coord  = coords[d] / _blocksize_spec.extent(d);
      unit_id          += block_coord;
    }
    unit_id %= _nunits;
    DASH_LOG_TRACE_VAR("ShiftTilePattern.unit_at >", unit_id);
    return unit_id;
  }

  team_unit_t unit_at(
    IndexType global_pos,
    const ViewSpec_t & viewspec) const
  {
    auto global_coords = _memory_layout.coords(global_pos);
    return unit_at(global_coords, viewspec);
  }

  team_unit_t unit_at(
    IndexType global_pos) const
  {
    auto global_coords = _memory_layout.coords(global_pos);
    return unit_at(global_coords);
  }


  SizeType extent(dim_t dim) const {
    if (dim >= NumDimensions || dim < 0) {
      DASH_THROW(
        dash::exception::OutOfRange,
        "Wrong dimension for ShiftTilePattern::local_extent. "
        << "Expected dimension between 0 and " << NumDimensions-1 << ", "
        << "got " << dim);
    }
    return _memory_layout.extent(dim);
  }

  SizeType local_extent(dim_t dim) const {
    if (dim >= NumDimensions || dim < 0) {
      DASH_THROW(
        dash::exception::OutOfRange,
        "Wrong dimension for ShiftTilePattern::local_extent. "
        << "Expected dimension between 0 and " << NumDimensions-1 << ", "
        << "got " << dim);
    }
    return _local_memory_layout.extent(dim);
  }

  std::array<SizeType, NumDimensions> local_extents() const {
    // Same local memory layout for all units:
    return _local_memory_layout.extents();
  }


  IndexType local_at(
    const std::array<IndexType, NumDimensions> & local_coords,
    const ViewSpec_t & viewspec) const
  {
    DASH_LOG_TRACE("ShiftTilePattern.local_at()", local_coords,
                   "view:",         viewspec,
                   "local blocks:", _local_blockspec.extents());
    // Phase coordinates of element:
    std::array<IndexType, NumDimensions> phase_coords{};
    // Coordinates of the local block containing the element:
    std::array<IndexType, NumDimensions> block_coords_l{};
    for (auto d = 0; d < NumDimensions; ++d) {
      auto vs_offset_d  = viewspec.offset(d);
      auto vs_coord_d   = local_coords[d] + vs_offset_d;
      auto block_size_d = _blocksize_spec.extent(d);
      phase_coords[d]   = vs_coord_d % block_size_d;
      block_coords_l[d] = vs_coord_d / block_size_d;
    }
    DASH_LOG_TRACE("ShiftTilePattern.local_at",
                   "local block coords:", block_coords_l,
                   "phase coords:",       phase_coords);
    // Number of blocks preceeding the coordinates' block:
    auto block_offset_l = _local_blockspec.at(block_coords_l);
    auto local_index    =
           block_offset_l * _blocksize_spec.size() + // preceeding blocks
           _blocksize_spec.at(phase_coords);         // element phase
    DASH_LOG_TRACE_VAR("ShiftTilePattern.local_at >", local_index);
    return local_index;
  }

  IndexType local_at(
    const std::array<IndexType, NumDimensions> & local_coords) const
  {
    DASH_LOG_TRACE("ShiftTilePattern.local_at()", local_coords,
                   "local blocks:", _local_blockspec.extents());
    // Phase coordinates of element:
    std::array<IndexType, NumDimensions> phase_coords{};
    // Coordinates of the local block containing the element:
    std::array<IndexType, NumDimensions> block_coords_l{};
    for (auto d = 0; d < NumDimensions; ++d) {
      auto vs_coord_d   = local_coords[d];
      auto block_size_d = _blocksize_spec.extent(d);
      phase_coords[d]   = vs_coord_d % block_size_d;
      block_coords_l[d] = vs_coord_d / block_size_d;
    }
    DASH_LOG_TRACE("ShiftTilePattern.local_at",
                   "local block coords:", block_coords_l,
                   "phase coords:",       phase_coords);
    // Number of blocks preceeding the coordinates' block:
    auto block_offset_l = _local_blockspec.at(block_coords_l);
    auto local_index    =
           block_offset_l * _blocksize_spec.size() + // preceeding blocks
           _blocksize_spec.at(phase_coords);         // element phase
    DASH_LOG_TRACE_VAR("ShiftTilePattern.local_at >", local_index);
    return local_index;
  }

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

  local_index_t local(
    IndexType g_index) const
  {
    DASH_LOG_TRACE_VAR("ShiftTilePattern.local()", g_index);
    // TODO: Implement dedicated method for this, conversion to/from
    //       global coordinates is expensive.
    auto g_coords = coords(g_index);
    return local_index(g_coords);
  }

  std::array<IndexType, NumDimensions> local_coords(
    const std::array<IndexType, NumDimensions> & global_coords) const
  {
    std::array<IndexType, NumDimensions> local_coords = global_coords;
    auto blocksize_d = _blocksize_spec.extent(_major_tiled_dim);
    auto coord_d     = global_coords[_major_tiled_dim];
    local_coords[_major_tiled_dim] =
      // Local block offset
      (coord_d / (blocksize_d * _nunits)) * blocksize_d +
      // Phase
      (coord_d % blocksize_d);
    return local_coords;
  }

  local_index_t local_index(
    const std::array<IndexType, NumDimensions> & global_coords) const
  {
    DASH_LOG_TRACE_VAR("Pattern.local_index()", global_coords);
    // Local offset of the element within all of the unit's local
    // elements:
    auto unit = unit_at(global_coords);
#ifdef __OLD__
    // Global coords to local coords:
    std::array<IndexType, NumDimensions> l_coords =
      local_coords(global_coords);
    // Local coords to local offset:
    auto l_index = local_at(l_coords);
#endif
    auto l_index = at(global_coords);
    DASH_LOG_TRACE_VAR("Pattern.local_index >", l_index);

    return local_index_t { team_unit_t(unit), l_index };
  }


  std::array<IndexType, NumDimensions> global(
      team_unit_t unit,
      const std::array<IndexType, NumDimensions> & local_coords) const {
    DASH_LOG_DEBUG("ShiftTilePattern.global()",
                   "unit:",    unit,
                   "lcoords:", local_coords);
    // Global coordinate of local element:
    std::array<IndexType, NumDimensions> global_coords = local_coords;
    // Local block coordinate of local element:
    auto blocksize_maj     = _blocksize_spec.extent(_major_tiled_dim);
    auto blocksize_min     = _blocksize_spec.extent(_minor_tiled_dim);
    auto l_block_coord_maj = local_coords[_major_tiled_dim] /
                               blocksize_maj;
    auto l_block_coord_min = (NumDimensions > 1)
                             ? local_coords[_minor_tiled_dim] /
                               blocksize_min
                             : 0;
    DASH_LOG_TRACE("ShiftTilePattern.global",
                   "minor tiled dim:",   _minor_tiled_dim,
                   "major tiled dim:",   _major_tiled_dim,
                   "l_block_coord_min:", l_block_coord_min,
                   "l_block_coord_maj:", l_block_coord_maj);
    // Apply diagonal shift in major tiled dimension:
    auto num_shift_blocks = (_nunits + unit -
                              (l_block_coord_min % _nunits))
                            % _nunits;
    num_shift_blocks     += _nunits * l_block_coord_maj;
    DASH_LOG_TRACE("ShiftTilePattern.global",
                   "num_shift_blocks:", num_shift_blocks,
                   "blocksize_maj:",    blocksize_maj);
    global_coords[_major_tiled_dim] =
      (num_shift_blocks * blocksize_maj) +
      local_coords[_major_tiled_dim] % blocksize_maj;
    DASH_LOG_DEBUG_VAR("ShiftTilePattern.global >", global_coords);
    return global_coords;
  }

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

  IndexType global(
    IndexType local_index) const
  {
    DASH_LOG_TRACE("ShiftTilePattern.global()",
                   "local_index:", local_index,
                   "unit:",        team().myid());
    auto block_size    = _blocksize_spec.size();
    auto phase         = local_index % block_size;
    auto l_block_index = local_index / block_size;
    // Block coordinate in local memory:
    auto l_block_coord = _local_blockspec.coords(l_block_index);
    // Coordinate of element in block:
    auto phase_coord   = _blocksize_spec.coords(phase);
    DASH_LOG_TRACE("ShiftTilePattern.global",
                   "local block index:",  l_block_index,
                   "local block coords:", l_block_coord,
                   "phase coords:",       phase_coord);
    // Coordinate of element in local memory:
    std::array<IndexType, NumDimensions> l_coords{};
    for (auto d = 0; d < NumDimensions; ++d) {
      l_coords[d] = l_block_coord[d] * _blocksize_spec.extent(d) +
                    phase_coord[d];
    }
    std::array<IndexType, NumDimensions> g_coords =
      global(team().myid(), l_coords);
    auto offset = global_at(g_coords);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.global >", offset);
    return offset;
  }

  IndexType global_index(
    team_unit_t unit,
    const std::array<IndexType, NumDimensions> & local_coords) const
  {
    DASH_LOG_TRACE("ShiftTilePattern.global_index()",
                   "unit:",         unit,
                   "local_coords:", local_coords);
    std::array<IndexType, NumDimensions> global_coords =
      global(unit, local_coords);
    auto g_index = global_at(global_coords);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.global_index >", g_index);
    return g_index;
  }

  IndexType global_at(
    const std::array<IndexType, NumDimensions> & global_coords,
    const ViewSpec_t                           & viewspec) const
  {
    DASH_LOG_TRACE("ShiftTilePattern.global_at()",
                   "gcoords:",  global_coords,
                   "viewspec:", viewspec);
    // Phase coordinates of element:
    std::array<IndexType, NumDimensions> phase_coords{};
    // Coordinates of the block containing the element:
    std::array<IndexType, NumDimensions> block_coords{};
    for (auto d = 0; d < NumDimensions; ++d) {
      auto vs_coord     = global_coords[d] + viewspec.offset(d);
      phase_coords[d]   = vs_coord % _blocksize_spec.extent(d);
      block_coords[d]   = vs_coord / _blocksize_spec.extent(d);
    }
    DASH_LOG_TRACE("ShiftTilePattern.global_at",
                   "block coords:", block_coords,
                   "phase coords:", phase_coords);
    // Number of blocks preceeding the coordinates' block, equivalent
    // to linear global block offset:
    auto block_index = _blockspec.at(block_coords);
    DASH_LOG_TRACE("ShiftTilePattern.global_at",
                   "block index:",   block_index);
    auto offset = block_index * _blocksize_spec.size() + // preceed. blocks
                  _blocksize_spec.at(phase_coords);      // element phase
    DASH_LOG_TRACE_VAR("ShiftTilePattern.global_at >", offset);
    return offset;
  }

  IndexType global_at(
    const std::array<IndexType, NumDimensions> & global_coords) const
  {
    DASH_LOG_TRACE("ShiftTilePattern.global_at()",
                   "gcoords:",  global_coords);
    // Phase coordinates of element:
    std::array<IndexType, NumDimensions> phase_coords;
    // Coordinates of the block containing the element:
    std::array<IndexType, NumDimensions> block_coords;
    for (auto d = 0; d < NumDimensions; ++d) {
      auto vs_coord     = global_coords[d];
      phase_coords[d]   = vs_coord % _blocksize_spec.extent(d);
      block_coords[d]   = vs_coord / _blocksize_spec.extent(d);
    }
    DASH_LOG_TRACE("ShiftTilePattern.global_at",
                   "block coords:", block_coords,
                   "phase coords:", phase_coords);
    // Number of blocks preceeding the coordinates' block, equivalent
    // to linear global block offset:
    auto block_index = _blockspec.at(block_coords);
    DASH_LOG_TRACE("ShiftTilePattern.global_at",
                   "block index:",   block_index);
    auto offset = block_index * _blocksize_spec.size() + // preceed. blocks
                  _blocksize_spec.at(phase_coords);      // element phase
    DASH_LOG_TRACE_VAR("ShiftTilePattern.global_at >", offset);
    return offset;
  }


  IndexType at(
    const std::array<IndexType, NumDimensions> & global_coords,
    const ViewSpec_t                           & viewspec) const
  {
    DASH_LOG_TRACE("ShiftTilePattern.at()",
                   "gcoords:",  global_coords,
                   "viewspec:", viewspec);
    // Phase coordinates of element:
    std::array<IndexType, NumDimensions> phase_coords;
    // Coordinates of the block containing the element:
    std::array<IndexType, NumDimensions> block_coords;
    for (auto d = 0; d < NumDimensions; ++d) {
      auto vs_coord     = global_coords[d] + viewspec.offset(d);
      phase_coords[d]   = vs_coord % _blocksize_spec.extent(d);
      block_coords[d]   = vs_coord / _blocksize_spec.extent(d);
    }
    DASH_LOG_TRACE("ShiftTilePattern.at",
                   "block_coords:", block_coords,
                   "phase_coords:", phase_coords);
    // Number of blocks preceeding the coordinates' block, equivalent
    // to linear global block offset divided by team size:
    DASH_LOG_TRACE_VAR("ShiftTilePattern.at", _blockspec.extents());
    auto block_index   = _blockspec.at(block_coords);
    auto block_index_l = block_index / _nunits;
    DASH_LOG_TRACE("ShiftTilePattern.at",
                   "global block index:",block_index,
                   "nunits:",            _nunits,
                   "local block index:", block_index_l);
    auto offset = block_index_l * _blocksize_spec.size() + // preceed. blocks
                  _blocksize_spec.at(phase_coords);        // element phase
    DASH_LOG_TRACE_VAR("ShiftTilePattern.at >", offset);
    return offset;
  }

  IndexType at(
    std::array<IndexType, NumDimensions> global_coords) const
  {
    // Note:
    // Expects extent[d] to be a multiple of blocksize[d] * nunits[d]
    // to ensure the balanced property.
    DASH_LOG_TRACE_VAR("ShiftTilePattern.at()", global_coords);
    // Phase coordinates of element:
    std::array<IndexType, NumDimensions> phase_coords{};
    // Coordinates of the block containing the element:
    std::array<IndexType, NumDimensions> block_coords{};
    for (auto d = 0; d < NumDimensions; ++d) {
      auto coord      = global_coords[d];
      phase_coords[d] = coord % _blocksize_spec.extent(d);
      block_coords[d] = coord / _blocksize_spec.extent(d);
    }
    DASH_LOG_TRACE_VAR("ShiftTilePattern.at", block_coords);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.at", phase_coords);
    // Number of blocks preceeding the coordinates' block, equivalent
    // to linear global block offset divided by team size:
    DASH_LOG_TRACE_VAR("ShiftTilePattern.at", _blockspec.extents());
    auto block_offset   = _blockspec.at(block_coords);
    auto block_offset_l = block_offset / _nunits;
    DASH_LOG_TRACE_VAR("ShiftTilePattern.at", block_offset);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.at", _nunits);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.at", block_offset_l);
    return block_offset_l * _blocksize_spec.size() + // preceeding blocks
           _blocksize_spec.at(phase_coords);         // element phase
  }

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


  bool has_local_elements(
    dim_t dim,
    IndexType dim_offset,
    team_unit_t unit,
    const ViewSpec_t & viewspec) const
  {
    DASH_LOG_TRACE_VAR("ShiftTilePattern.has_local_elements()", dim);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.has_local_elements()", dim_offset);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.has_local_elements()", unit);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.has_local_elements()", viewspec);
    // Apply viewspec offset in dimension to given position
    dim_offset += viewspec.offset(dim);
    // Offset to block offset
    IndexType block_coord_d    = dim_offset / _blocksize_spec.extent(dim);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.has_local_elements", block_coord_d);
    // Coordinate of unit in team spec in given dimension
    IndexType teamspec_coord_d = block_coord_d % _teamspec.extent(dim);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.has_local_elements",
                       teamspec_coord_d);
    // Check if unit id lies in cartesian sub-space of team spec
    return _teamspec.includes_index(
              teamspec_coord_d,
              dim,
              dim_offset);
  }

  bool is_local(
    IndexType    index,
    team_unit_t unit) const
  {
    auto glob_coords = coords(index);
    auto coords_unit = unit_at(glob_coords);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.is_local >", (coords_unit == unit));
    return coords_unit == unit;
  }

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


  index_type block_at(
    const std::array<index_type, NumDimensions> & g_coords) const
  {
    std::array<index_type, NumDimensions> block_coords;
    // Coord to block coord to unit coord:
    for (auto d = 0; d < NumDimensions; ++d) {
      block_coords[d] = g_coords[d] / _blocksize_spec.extent(d);
    }
    // Block coord to block index:
    auto block_idx = _blockspec.at(block_coords);
    DASH_LOG_TRACE("ShiftTilePattern.block_at",
                   "coords", g_coords,
                   "> block index", block_idx);
    return block_idx;
  }

  local_index_t local_block_at(
    const std::array<index_type, NumDimensions> & g_coords) const
  {
    DASH_THROW(
      dash::exception::NotImplemented,
      "ShiftTilePattern.local_block_at is not implemented");
  }

  ViewSpec_t block(
    index_type global_block_index) const
  {
    DASH_LOG_TRACE_VAR("ShiftTilePattern.block()", global_block_index);
    // block index -> block coords -> offset
    auto block_coords = _blockspec.coords(global_block_index);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.block", block_coords);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.block", _blocksize_spec.extents());
    std::array<index_type, NumDimensions> offsets{};
    std::array<size_type, NumDimensions>  extents{};
    for (auto d = 0; d < NumDimensions; ++d) {
      extents[d] = _blocksize_spec.extent(d);
      offsets[d] = block_coords[d] * extents[d];
    }
    DASH_LOG_TRACE("ShiftTilePattern.block",
                   "offsets:", offsets,
                   "extents:", extents);
    auto block_vs = ViewSpec_t(offsets, extents);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.block >", block_vs);
    return block_vs;
  }

  ViewSpec_t local_block(
    index_type local_block_index) const {
    DASH_LOG_TRACE_VAR("ShiftTilePattern.local_block()", local_block_index);
    // Local block index to local block coords:
    auto l_block_coords = _local_blockspec.coords(local_block_index);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.local_block()", l_block_coords);
    std::array<index_type, NumDimensions> l_elem_coords{};
    // TODO: This is convenient but less efficient:
    // Translate local coordinates of first element in local block to global
    // coordinates:
    for (auto d = 0; d < NumDimensions; ++d) {
      auto blocksize_d = _blocksize_spec.extent(d);
      l_elem_coords[d] = static_cast<index_type>(
                           l_block_coords[d] * blocksize_d);
    }
    // Global coordinates of first element in block:
    auto g_elem_coords = global(l_elem_coords);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.local_block()", g_elem_coords);
    std::array<index_type, NumDimensions> offsets{};
    std::array<size_type, NumDimensions>  extents{};
    for (auto d = 0; d < NumDimensions; ++d) {
      offsets[d] = g_elem_coords[d];
      extents[d] = _blocksize_spec.extent(d);
    }
    ViewSpec_t block_vs(offsets, extents);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.local_block >", block_vs);
#ifdef __TODO__
    // Coordinates of the unit within the team spec:
    std::array<IndexType, NumDimensions> unit_ts_coord =
      _teamspec.coords(unit);
    for (auto d = 0; d < NumDimensions; ++d) {
      const Distribution & dist = _distspec[d];
      auto blocksize_d          = block_vs[d].extent;
      auto num_units_d          = _teamspec.extent(d);
      auto num_blocks_d         = _blockspec.extent(d);
      // Local to global block coords:
      auto g_block_coord_d      = (l_block_coords[d] + _myid) *
                                  _teamspec.extent(d);
      block_vs[d].offset        = g_block_coord_d * blocksize_d;
    }
#endif
    return block_vs;
  }

  ViewSpec_t local_block_local(
    index_type local_block_index) const {
    DASH_LOG_TRACE_VAR("ShiftTilePattern.local_block_local()", local_block_index);
    // Initialize viewspec result with block extents:
    std::array<index_type, NumDimensions> offsets{};
    std::array<size_type, NumDimensions>  extents =
      _blocksize_spec.extents();
    // Local block index to local block coords:
    auto l_block_coords = _local_blockspec.coords(local_block_index);
    // Local block coords to local element offset:
    for (auto d = 0; d < NumDimensions; ++d) {
      auto blocksize_d  = extents[d];
      offsets[d]        = l_block_coords[d] * blocksize_d;
    }
    ViewSpec_t block_vs(offsets, extents);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.local_block >", block_vs);
    return block_vs;
  }

  const BlockSpec_t & blockspec() const
  {
    return _blockspec;
  }

  const BlockSpec_t & local_blockspec() const
  {
    return _local_blockspec;
  }

  SizeType blocksize(
    dim_t dimension) const {
    return _blocksize_spec.extent(dimension);
  }

  inline SizeType max_blocksize() const {
    return _blocksize_spec.size();
  }

  inline SizeType local_capacity() const {
    // Balanced pattern, local capacity identical for every unit and
    // same as local size.
    return local_size();
  }

  inline SizeType local_size(
    team_unit_t unit = UNDEFINED_TEAM_UNIT_ID) const
  {
    return _local_memory_layout.size();
  }

  inline IndexType num_units() const {
    return _teamspec.size();
  }

  inline IndexType capacity() const {
    return _memory_layout.size();
  }

  inline IndexType size() const {
    return _memory_layout.size();
  }

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

  inline const DistributionSpec_t & distspec() const {
    return _distspec;
  }

  SizeSpec_t sizespec() const {
    return SizeSpec_t(_memory_layout.extents());
  }

  const std::array<SizeType, NumDimensions> & extents() const {
    return _memory_layout.extents();
  }

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

  std::array<IndexType, NumDimensions> coords(
    IndexType index) const {

    ::std::array<IndexType, NumDimensions> pos{};
    auto block_coords = _blockspec.coords(index / _blocksize_spec.size());
    auto phase_coords = _blocksize_spec.coords(index % _blocksize_spec.size());
    for (auto d = 0; d < NumDimensions; ++d) {
      pos[d] = block_coords[d]*_blocksize_spec.extent(d) + phase_coords[d];
    }
    return pos;
  }

  std::array<IndexType, NumDimensions> coords(
    IndexType          index,
    const ViewSpec_t & viewspec) const {

    ::std::array<IndexType, NumDimensions> pos;
    auto block_coords = _blockspec.coords(index / _blocksize_spec.size(),
                                          viewspec);
    auto phase_coords = _blocksize_spec.coords(index % _blocksize_spec.size(),
                                               viewspec);
    for (auto d = 0; d < NumDimensions; ++d) {
      pos[d] = block_coords[d]*_blocksize_spec.extent(d) + phase_coords[d];
    }
    return pos;
  }

  constexpr static MemArrange memory_order() {
    return Arrangement;
  }

  constexpr static dim_t ndim() {
    return NumDimensions;
  }

private:

  ShiftTilePattern(const PatternArguments_t & arguments)
  : _distspec(arguments.distspec()),
    _team(&arguments.team()),
    _teamspec(arguments.teamspec()),
    _memory_layout(arguments.sizespec().extents()),
    _nunits(_teamspec.size()),
    _major_tiled_dim(initialize_major_tiled_dim(_distspec)),
    _minor_tiled_dim((_major_tiled_dim + 1) % NumDimensions),
    _blocksize_spec(initialize_blocksizespec(
        arguments.sizespec(),
        _distspec,
        _teamspec)),
    _blockspec(initialize_blockspec(
        arguments.sizespec(),
        _blocksize_spec,
        _teamspec)),
    _local_blockspec(initialize_local_blockspec(
        _blockspec,
        _major_tiled_dim,
        _nunits)),
    _local_memory_layout(
        initialize_local_extents(_team->myid())),
    _local_capacity(initialize_local_capacity())
  {}
  BlockSizeSpec_t initialize_blocksizespec(
    const SizeSpec_t         & sizespec,
    const DistributionSpec_t & distspec,
    const TeamSpec_t         & teamspec) const {
    DASH_LOG_TRACE("ShiftTilePattern.init_blocksizespec()");
    // Extents of a single block:
    std::array<SizeType, NumDimensions> s_blocks{};
    for (auto d = 0; d < NumDimensions; ++d) {
      const Distribution & dist = distspec[d];
      DASH_LOG_TRACE("ShiftTilePattern.init_blocksizespec d",
                     "sizespec extent[d]:", sizespec.extent(d),
                     "teamspec extent[d]:", teamspec.extent(d));
      SizeType max_blocksize_d  = dist.max_blocksize_in_range(
        sizespec.extent(d),  // size of range (extent)
        teamspec.extent(d)); // number of blocks (units)
      s_blocks[d] = max_blocksize_d;
    }
    DASH_LOG_TRACE_VAR("ShiftTilePattern.init_blocksizespec >", s_blocks);
    return BlockSizeSpec_t(s_blocks);
  }

  BlockSpec_t initialize_blockspec(
    const SizeSpec_t         & sizespec,
    const BlockSizeSpec_t    & blocksizespec,
    const TeamSpec_t         & teamspec) const
  {
    DASH_LOG_TRACE("ShiftTilePattern.init_blockspec()",
                   "pattern size:", sizespec.extents(),
                   "block size:",   blocksizespec.extents(),
                   "team size:",    teamspec.extents());
    // Number of blocks in all dimensions:
    std::array<SizeType, NumDimensions> n_blocks{};
    for (auto d = 0; d < NumDimensions; ++d) {
      SizeType max_blocksize_d = blocksizespec.extent(d);
      SizeType max_blocks_d    = dash::math::div_ceil(
                                   sizespec.extent(d),
                                   max_blocksize_d);
      n_blocks[d] = max_blocks_d;
    }
    BlockSpec_t blockspec(n_blocks);
    DASH_LOG_TRACE_VAR("ShiftTilePattern.init_blockspec >", n_blocks);
    return blockspec;
  }

  BlockSpec_t initialize_local_blockspec(
    const BlockSpec_t        & blockspec,
    dim_t                      major_tiled_dim,
    size_t                     nunits) const
  {
    DASH_LOG_TRACE_VAR("ShiftTilePattern.init_local_blockspec()",
                       blockspec.extents());
    DASH_LOG_TRACE_VAR("ShiftTilePattern.init_local_blockspec()",
                       nunits);
    // Number of local blocks in all dimensions:
    auto l_blocks = blockspec.extents();
    l_blocks[major_tiled_dim] /= nunits;
    DASH_ASSERT_GT(l_blocks[major_tiled_dim], 0,
                   "ShiftTilePattern: Size must be divisible by team size");
    DASH_LOG_TRACE_VAR("ShiftTilePattern.init_local_blockspec >", l_blocks);
    return BlockSpec_t(l_blocks);
  }

  SizeType initialize_local_capacity() const
  {
    // Assumes balanced distribution property, i.e.
    // range = k * blocksz * nunits
    auto l_capacity = size() / _nunits;
    DASH_LOG_TRACE_VAR("ShiftTilePattern.init_local_capacity >", l_capacity);
    return l_capacity;
  }

  void initialize_local_range() {
    auto local_size = _local_memory_layout.size();
    DASH_LOG_DEBUG_VAR("ShiftTilePattern.init_local_range()", local_size);
    if (local_size == 0) {
      _lbegin = 0;
      _lend   = 0;
    } else {
      // First local index transformed to global index
      _lbegin = global(0);
      // Index past last local index transformed to global index
      _lend   = global(local_size - 1) + 1;
    }
    DASH_LOG_DEBUG_VAR("ShiftTilePattern.init_local_range >",
                       _local_memory_layout.extents());
    DASH_LOG_DEBUG_VAR("ShiftTilePattern.init_local_range >", _lbegin);
    DASH_LOG_DEBUG_VAR("ShiftTilePattern.init_local_range >", _lend);
  }

  dim_t initialize_major_tiled_dim(const DistributionSpec_t & ds)
  {
    DASH_LOG_TRACE("ShiftTilePattern.init_major_tiled_dim()");
    if (Arrangement == dash::COL_MAJOR) {
      DASH_LOG_TRACE("ShiftTilePattern.init_major_tiled_dim", "column major");
      for (auto d = 0; d < NumDimensions; ++d) {
        if (ds[d].type == dash::internal::DIST_TILE) {
          DASH_LOG_TRACE("ShiftTilePattern.init_major_tiled_dim >", d);
          return d;
        }
      }
    } else {
      DASH_LOG_TRACE("ShiftTilePattern.init_major_tiled_dim", "row major");
      for (auto d = NumDimensions-1; d >= 0; --d) {
        if (ds[d].type == dash::internal::DIST_TILE) {
          DASH_LOG_TRACE("ShiftTilePattern.init_major_tiled_dim >", d);
          return d;
        }
      }
    }
    DASH_THROW(dash::exception::InvalidArgument,
              "Distribution is not tiled in any dimension");
  }

  std::array<SizeType, NumDimensions> initialize_local_extents(
      team_unit_t unit) const {
    // Coordinates of local unit id in team spec:
#ifdef DASH_ENABLE_LOGGING
    auto unit_ts_coords = _teamspec.coords(unit);
    DASH_LOG_DEBUG_VAR("ShiftTilePattern._local_extents()", unit);
    DASH_LOG_TRACE_VAR("ShiftTilePattern._local_extents", unit_ts_coords);
#endif
    ::std::array<SizeType, NumDimensions> l_extents{};
    for (auto d = 0; d < NumDimensions; ++d) {
      // Number of units in dimension:
      auto num_units_d        = _teamspec.extent(d);
      // Number of blocks in dimension:
      auto num_blocks_d       = _blockspec.extent(d);
      // Maximum extent of single block in dimension:
      auto blocksize_d        = _blocksize_spec.extent(d);
      // Minimum number of blocks local to every unit in dimension:
      auto min_local_blocks_d = num_blocks_d / num_units_d;
      // Coordinate of this unit id in teamspec in dimension:
//    auto unit_ts_coord      = unit_ts_coords[d];
      // Possibly there are more blocks than units in dimension and no
      // block left for this unit. Local extent in d then becomes 0.
      l_extents[d] = min_local_blocks_d * blocksize_d;
    }
    DASH_LOG_DEBUG_VAR("ShiftTilePattern._local_extents >", l_extents);
    return l_extents;
  }
};

} // namespace dash

#include <dash/pattern/ShiftTilePattern1D.h>

#endif // DASH__SHIFT_TILE_PATTERN_H_