BlockPattern.h

#ifndef DASH__BLOCK_PATTERN_H_
#define DASH__BLOCK_PATTERN_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/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>
#include <utility>

namespace dash {

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

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 CartesianIndexSpace<NumDimensions, Arrangement, IndexType>
    BlockSpec_t;
  typedef CartesianIndexSpace<NumDimensions, Arrangement, IndexType>
    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;
  SizeType                    _nunits          = 0;
  MemoryLayout_t              _memory_layout;
  BlockSizeSpec_t             _blocksize_spec;
  BlockSpec_t                 _blockspec;
  LocalMemoryLayout_t         _local_memory_layout;
  BlockSpec_t                 _local_blockspec;
  SizeType                    _local_capacity;
  IndexType                   _lbegin;
  IndexType                   _lend;

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

  BlockPattern(
      const SizeSpec_t &sizespec,
      DistributionSpec_t dist,
      const TeamSpec_t &teamspec,
      dash::Team &team = dash::Team::All())
    : _distspec(std::move(dist))
    , _team(&team)
    , _teamspec(teamspec, _distspec, *_team)
    , _nunits(_teamspec.size())
    , _memory_layout(sizespec.extents())
    , _blocksize_spec(
          initialize_blocksizespec(sizespec, _distspec, _teamspec))
    , _blockspec(initialize_blockspec(sizespec, _distspec, _blocksize_spec))
    , _local_memory_layout(initialize_local_extents(_team->myid()))
    , _local_blockspec(
          initialize_local_blockspec(_blocksize_spec, _local_memory_layout))
    , _local_capacity(initialize_local_capacity())
  {
    DASH_LOG_TRACE("BlockPattern()", "(sizespec, dist, teamspec, team)");
    initialize_local_range();
    DASH_LOG_TRACE("BlockPattern()", "BlockPattern initialized");
  }

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

  BlockPattern(const self_t & other)
  : _distspec(other._distspec),
    _team(other._team),
    _teamspec(other._teamspec),
    _nunits(other._nunits),
    _memory_layout(other._memory_layout),
    _blocksize_spec(other._blocksize_spec),
    _blockspec(other._blockspec),
    _local_memory_layout(other._local_memory_layout),
    _local_blockspec(other._local_blockspec),
    _local_capacity(other._local_capacity),
    _lbegin(other._lbegin),
    _lend(other._lend)
  {
    // No need to copy _arguments as it is just used to
    // initialize other members.
    DASH_LOG_TRACE("BlockPattern(other)", "BlockPattern copied");
  }

  BlockPattern(self_t & other)
  : BlockPattern(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);
  }

  BlockPattern & operator=(const BlockPattern & other)
  {
    DASH_LOG_TRACE("BlockPattern.=(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;
      _lbegin              = other._lbegin;
      _lend                = other._lend;
      DASH_LOG_TRACE("BlockPattern.=(other)", "BlockPattern assigned");
    }
    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
  {
    // Apply viewspec offsets to coordinates:
    std::array<IndexType, NumDimensions> vs_coords;
    for (auto d = 0; d < NumDimensions; ++d) {
      vs_coords[d] = coords[d] + viewspec.offset(0);
    }
    return unit_at(vs_coords);
  }

  team_unit_t unit_at(
    const std::array<IndexType, NumDimensions> & coords) const
  {
    std::array<IndexType, NumDimensions> unit_coords{};
    // Coord to block coord to unit coord:
    for (auto d = 0; d < NumDimensions; ++d) {
      unit_coords[d] = (coords[d] / _blocksize_spec.extent(d))
                         % _teamspec.extent(d);
    }
    // Unit coord to unit id:
    team_unit_t unit_id(_teamspec.at(unit_coords));
    DASH_LOG_TRACE("BlockPattern.unit_at",
                   "coords", coords,
                   "> unit id", 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);
  }


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

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

  constexpr std::array<SizeType, NumDimensions> local_extents() const noexcept
  {
    return _local_memory_layout.extents();
  }

  constexpr std::array<SizeType, NumDimensions> local_extents(
      team_unit_t unit) const noexcept
  {
    return ( unit == _team->myid()
             ? _local_memory_layout.extents()
             : initialize_local_extents(unit)
           );
  }


  IndexType local_at(
    const std::array<IndexType, NumDimensions> & local_coords,
    const ViewSpec_t & viewspec) const
  {
    auto coords = local_coords;
    for (auto d = 0; d < NumDimensions; ++d) {
      coords[d] += viewspec.offset(d);
    }
    return _local_memory_layout.at(coords);
  }

  constexpr IndexType local_at(
    const std::array<IndexType, NumDimensions> & local_coords) const
  {
    return _local_memory_layout.at(local_coords);
  }

  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("BlockPattern.local()", g_index);
    // TODO: Implement dedicated method for this, conversion to/from
    //       global coordinates is expensive.
    auto l_coords = coords(g_index);
    return local_index(l_coords);
  }

  std::array<IndexType, NumDimensions> local_coords(
    const std::array<IndexType, NumDimensions> & global_coords) const
  {
    std::array<IndexType, NumDimensions> local_coords{};
    for (auto d = 0; d < NumDimensions; ++d) {
      auto block_size_d     = _blocksize_spec.extent(d);
      auto b_offset_d       = global_coords[d] % block_size_d;
      auto g_block_offset_d = global_coords[d] / block_size_d;
      auto l_block_offset_d = g_block_offset_d / _teamspec.extent(d);
      local_coords[d]       = b_offset_d +
                              (l_block_offset_d * block_size_d);
    }
    return local_coords;
  }

  local_index_t local_index(
    const std::array<IndexType, NumDimensions> & global_coords) const
  {
    DASH_LOG_TRACE_VAR("BlockPattern.local_index()", global_coords);
    auto unit = unit_at(global_coords);
    DASH_LOG_TRACE_VAR("BlockPattern.local_index", unit);
    // Global coords to local coords:
    std::array<IndexType, NumDimensions> l_coords =
      local_coords(global_coords);
    DASH_LOG_TRACE_VAR("BlockPattern.local_index", l_coords);
    if (unit == _team->myid()) {
      // Coords are local to this unit, use pre-generated local memory
      // layout
      return local_index_t { unit, _local_memory_layout.at(l_coords) };
    } else {
      // Cannot use _local_memory_layout as it is only defined for the
      // active unit but does not specify local memory of other units.
      // Generate local memory layout for unit assigned to coords:
      auto l_mem_layout =
        LocalMemoryLayout_t(initialize_local_extents(unit));
      return local_index_t { unit, l_mem_layout.at(l_coords) };
    }
  }


  std::array<IndexType, NumDimensions> global(
    team_unit_t unit,
    const std::array<IndexType, NumDimensions> & local_coords) const
  {
    DASH_LOG_DEBUG_VAR("BlockPattern.global()", local_coords);
    if (_teamspec.size() < 2) {
      return local_coords;
    }
    // Coordinates of the unit within the team spec:
    std::array<IndexType, NumDimensions> unit_ts_coord =
      _teamspec.coords(unit);
    // Index of the element:
    std::array<IndexType, NumDimensions> glob_index{};
    for (auto d = 0; d < NumDimensions; ++d) {
      const Distribution & dist = _distspec[d];
      auto num_units_d          = _teamspec.extent(d);
      auto blocksize_d          = _blocksize_spec.extent(d);
      auto local_index_d        = local_coords[d];
      // TOOD: Use % (blocksize_d - underfill_d)
      auto elem_block_offset_d  = local_index_d % blocksize_d;
      // Global coords of the element's block within all blocks:
      auto block_index_d        = dist.local_index_to_block_coord(
                                    unit_ts_coord[d], // unit ts offset in d
                                    local_index_d,
                                    num_units_d
                                  );
      glob_index[d]  = (block_index_d * blocksize_d) + elem_block_offset_d;
    }
    DASH_LOG_DEBUG_VAR("BlockPattern.global >", glob_index);
    return glob_index;
  }

  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
  {
    std::array<IndexType, NumDimensions> local_coords =
      _local_memory_layout.coords(local_index);
    DASH_LOG_TRACE_VAR("BlockPattern.local_to_global_idx()", local_coords);
    std::array<IndexType, NumDimensions> global_coords =
      global(_team->myid(), local_coords);
    DASH_LOG_TRACE_VAR("BlockPattern.local_to_global_idx >", global_coords);
    return _memory_layout.at(global_coords);
  }

  IndexType global_index(
    team_unit_t unit,
    const std::array<IndexType, NumDimensions> & local_coords) const
  {
    std::array<IndexType, NumDimensions> global_coords =
      global(unit, local_coords);
    DASH_LOG_TRACE_VAR("BlockPattern.local_to_global_idx", global_coords);
    return _memory_layout.at(global_coords);
  }

  IndexType global_at(
    const std::array<IndexType, NumDimensions> & view_coords,
    const ViewSpec_t                           & viewspec) const
  {
    DASH_LOG_TRACE("BlockPattern.global_at()",
                   "view coords:",view_coords,
                   "view:",       viewspec);
    // Global coordinates of the element referenced in the view:
    std::array<IndexType, NumDimensions> global_coords{};
    for (auto d = 0; d < NumDimensions; ++d) {
      global_coords[d] = view_coords[d] + viewspec.offset(d);
    }
    DASH_LOG_TRACE("BlockPattern.global_at",
                   "global coords:", global_coords);
    // Offset in iteration order is identical to offset in canonical order:
    auto offset = _memory_layout.at(global_coords);
    DASH_LOG_TRACE_VAR("BlockPattern.global_at >", offset);
    return offset;
  }

  IndexType global_at(
    const std::array<IndexType, NumDimensions> & global_coords) const
  {
    DASH_LOG_TRACE("BlockPattern.global_at()",
                   "gcoords:",  global_coords);
    // Offset in iteration order is identical to offset in canonical order:
    auto offset = _memory_layout.at(global_coords);
    DASH_LOG_TRACE_VAR("BlockPattern.global_at >", offset);
    return offset;
  }


  IndexType at(
    const std::array<IndexType, NumDimensions> & global_coords) const
  {
    auto unit = unit_at(global_coords);
    // Global coords to local coords:
    std::array<IndexType, NumDimensions> l_coords =
      local_coords(global_coords);
    DASH_LOG_TRACE_VAR("BlockPattern.at", l_coords);
    if (unit == _team->myid()) {
      // Coords are local to this unit, use pre-generated local memory
      // layout
      return _local_memory_layout.at(l_coords);
    } else {
      // Cannot use _local_memory_layout as it is only defined for the
      // active unit but does not specify local memory of other units.
      // Generate local memory layout for unit assigned to coords:
      auto l_mem_layout =
        LocalMemoryLayout_t(initialize_local_extents(unit));
      return l_mem_layout.at(l_coords);
    }
  }

  IndexType at(
    const std::array<IndexType, NumDimensions> & global_coords,
    const ViewSpec_t & viewspec) const
  {
    auto coords = global_coords;
    for (auto d = 0; d < NumDimensions; ++d) {
      coords[d] += viewspec.offset(d);
    }
    DASH_LOG_TRACE_VAR("BlockPattern.at()", coords);
    DASH_LOG_TRACE_VAR("BlockPattern.at()", viewspec);
    return at(coords);
  }

  template<typename ... Values>
  IndexType at(IndexType value, Values ... values) const
  {
    static_assert(
      sizeof...(values) == NumDimensions-1,
      "Wrong parameter number");
    std::array<IndexType, NumDimensions> inputindex = {
      value, (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("BlockPattern.has_local_elements()", dim);
    DASH_LOG_TRACE_VAR("BlockPattern.has_local_elements()", dim_offset);
    DASH_LOG_TRACE_VAR("BlockPattern.has_local_elements()", unit);
    // 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("BlockPattern.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("BlockPattern.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("BlockPattern.is_local >", (coords_unit == unit));
    return coords_unit == unit;
  }

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


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

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

  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("BlockPattern.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
  {
    local_index_t l_pos;

    std::array<IndexType, NumDimensions> l_block_coords{};
    std::array<IndexType, NumDimensions> unit_ts_coords{};
    for (dim_t d = 0; d < NumDimensions; ++d) {
      auto nunits_d      = _teamspec.extent(d);
      auto blocksize_d   = _blocksize_spec.extent(d);
      auto block_coord_d = g_coords[d] / blocksize_d;
      l_block_coords[d]  = block_coord_d / nunits_d;
      unit_ts_coords[d]  = block_coord_d % nunits_d;
    }
    l_pos.unit  = _teamspec.at(unit_ts_coords);
    l_pos.index = _local_blockspec.at(l_block_coords);

    DASH_LOG_TRACE("BlockPattern.local_block_at >",
                   "coords",             g_coords,
                   "unit:",              l_pos.unit,
                   "local block index:", l_pos.index);
    return l_pos;
  }

  ViewSpec_t block(
    index_type global_block_index) const
  {
    // block index -> block coords -> offset
    auto block_coords = _blockspec.coords(global_block_index);
    std::array<index_type, NumDimensions> offsets{};
    std::array<size_type, NumDimensions>  extents{};

    for (auto d = 0; d < NumDimensions; ++d) {
      auto num_blocks_d = _blockspec.extent(d);
      extents[d]        = _blocksize_spec.extent(d);
      if(block_coords[d] == (num_blocks_d - 1)){
        extents[d] -= underfilled_blocksize(d);
      }
      offsets[d] = block_coords[d] * _blocksize_spec.extent(d);
    }
    return ViewSpec_t(offsets, extents);
  }

  ViewSpec_t local_block(
    index_type local_block_index) const
  {
    // Initialize viewspec result with block extents:
    std::array<SizeType, NumDimensions>  block_vs_extents =
      _blocksize_spec.extents();
    std::array<IndexType, NumDimensions> block_vs_offsets {{ }};
    // Local block index to local block coords:
    auto l_block_coords = _local_blockspec.coords(local_block_index);
    auto l_elem_coords  = l_block_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 num_blocks_d =_local_blockspec.extent(d);
      if(l_block_coords[d] == (num_blocks_d - 1)){
          size_type remaining = local_extent(d) % block_vs_extents[d];
          block_vs_extents[d] = (remaining == 0)
                                ? block_vs_extents[d]
                                : remaining;
        // to calculate offset, extent of fully filled blocks are needed
        l_elem_coords[d] *= _blocksize_spec.extent(d);
      } else {
        l_elem_coords[d] *= block_vs_extents[d];
      }
    }
    // Global coordinates of first element in block:
    auto g_elem_coords = global(l_elem_coords);
    for (auto d = 0; d < NumDimensions; ++d) {
      block_vs_offsets[d] = g_elem_coords[d];
    }
#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_extents[d];
      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_offsets[d]       = g_block_coord_d * blocksize_d;
    }
#endif
    ViewSpec_t block_vs(block_vs_offsets, block_vs_extents);
    return block_vs;
  }

  ViewSpec_t local_block_local(
    index_type local_block_index) const
  {
    // Local block index to local block coords:
    auto l_block_coords = _local_blockspec.coords(local_block_index);
    std::array<index_type, NumDimensions> offsets;
    std::array<size_type, NumDimensions>  extents = _blocksize_spec.extents();
    // last block in at least one dimension
    for(dim_t d=0; d<NumDimensions; ++d){
      if(l_block_coords[d] == (_local_blockspec.extent(d)-1)){
          size_type remaining = local_extent(d) % extents[d];
          extents[d] = (remaining == 0) ? extents[d] : remaining;
          // to calculate offset, extent of fully filled blocks are needed
          offsets[d] = l_block_coords[d] * _blocksize_spec.extent(d);
      } else {
        offsets[d] = l_block_coords[d] * extents[d];
      }
    }
    return ViewSpec_t(offsets, extents);
  }

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

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

  constexpr SizeType local_capacity() const noexcept
  {
    return _local_capacity;
  }

  constexpr SizeType local_size(
    team_unit_t unit = UNDEFINED_TEAM_UNIT_ID) const noexcept
  {
    return (unit == UNDEFINED_TEAM_UNIT_ID)
           ? _local_memory_layout.size()
           : initialize_local_extents(unit).size();
  }

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

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

  constexpr IndexType size() const noexcept
  {
    return _memory_layout.size();
  }

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

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

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

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

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

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

  constexpr std::array<IndexType, NumDimensions> coords(
    IndexType          index,
    const ViewSpec_t & viewspec) const noexcept
  {
    return _memory_layout.coords(index, viewspec);
  }

  constexpr static MemArrange memory_order() noexcept
  {
    return Arrangement;
  }

  constexpr static dim_t ndim() noexcept
  {
    return NumDimensions;
  }

  SizeType underfilled_blocksize(
    dim_t dimension) const
  {
    // Underflow blocksize = regular blocksize - overflow blocksize:
    auto ovf_blocksize = _memory_layout.extent(dimension) %
                         blocksize(dimension);
    if (ovf_blocksize == 0) {
      return 0;
    } else {
      auto reg_blocksize = blocksize(dimension);
      return reg_blocksize - ovf_blocksize;
    }
  }

private:

  BlockPattern(const PatternArguments_t & arguments)
  :  _distspec(arguments.distspec()),
     _team(&arguments.team()),
     _teamspec(arguments.teamspec()),
     _nunits(_teamspec.size()),
     _memory_layout(arguments.sizespec().extents()),
     _blocksize_spec(initialize_blocksizespec(
         arguments.sizespec(),
         _distspec,
         _teamspec)),
     _blockspec(initialize_blockspec(
         arguments.sizespec(),
         _distspec,
         _blocksize_spec)),
     _local_memory_layout(
         initialize_local_extents(_team->myid())),
     _local_blockspec(initialize_local_blockspec(
         _blocksize_spec,
         _local_memory_layout)),
     _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_VAR("BlockPattern.init_blocksizespec", teamspec.size());
    if (teamspec.size() == 0) {
      return BlockSizeSpec_t();
    }
    // Extents of a single block:
    std::array<SizeType, NumDimensions> s_blocks{};
    for (auto d = 0; d < NumDimensions; ++d) {
      const Distribution & dist = distspec[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;
    }
    return BlockSizeSpec_t(s_blocks);
  }

  BlockSpec_t initialize_blockspec(
    const SizeSpec_t         & sizespec,
    const DistributionSpec_t & distspec,
    const BlockSizeSpec_t    & blocksizespec) const
  {
    if (blocksizespec.size() == 0) {
      return BlockSpec_t();
    }
    // Number of blocks in all dimensions:
    std::array<SizeType, NumDimensions> n_blocks{};
    for (auto d = 0; d < NumDimensions; ++d) {
      DASH_LOG_TRACE_VAR("BlockPattern.init_blockspec", distspec[d].type);
      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;
    }
    DASH_LOG_TRACE_VAR("BlockPattern.init_blockspec", n_blocks);
    return BlockSpec_t(n_blocks);
  }

  BlockSpec_t initialize_local_blockspec(
    const BlockSizeSpec_t     & blocksizespec,
    const LocalMemoryLayout_t & local_mem_layout) const
  {
    auto num_l_blocks = local_mem_layout.extents();
    for (auto d = 0; d < NumDimensions; ++d) {
      auto blocksize_d = blocksizespec.extent(d);
      if (blocksize_d > 0) {
        num_l_blocks[d] = dash::math::div_ceil(
                            num_l_blocks[d],
                            blocksizespec.extent(d));
      } else {
        num_l_blocks[d] = 0;
      }
    }
    DASH_LOG_TRACE_VAR("BlockPattern.init_local_blockspec", num_l_blocks);
    return BlockSpec_t(num_l_blocks);
  }

  SizeType initialize_local_capacity() const
  {
    SizeType l_capacity = 1;
    if (_teamspec.size() == 0) {
      return 0;
    }
    for (auto d = 0; d < NumDimensions; ++d) {
      auto num_units_d    = _teamspec.extent(d);
      // Maximum number of occurrences of a single unit in given
      // dimension:
      auto num_blocks_d   = dash::math::div_ceil(
                              _memory_layout.extent(d),
                              _blocksize_spec.extent(d));
      auto max_l_blocks_d = dash::math::div_ceil(
                              num_blocks_d,
                              num_units_d);
      DASH_LOG_TRACE_VAR("BlockPattern.init_lcapacity.d", d);
      DASH_LOG_TRACE_VAR("BlockPattern.init_lcapacity.d", num_units_d);
      DASH_LOG_TRACE_VAR("BlockPattern.init_lcapacity.d", max_l_blocks_d);
      l_capacity *= max_l_blocks_d;
    }
    l_capacity *= _blocksize_spec.size();;
    DASH_LOG_DEBUG_VAR("BlockPattern.init_lcapacity >", l_capacity);
    return l_capacity;
  }

  void initialize_local_range()
  {
    auto l_size = _local_memory_layout.size();
    DASH_LOG_DEBUG_VAR("BlockPattern.init_local_range()", l_size);
    if (l_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(l_size - 1) + 1;
    }
    DASH_LOG_DEBUG_VAR("BlockPattern.init_local_range >", _lbegin);
    DASH_LOG_DEBUG_VAR("BlockPattern.init_local_range >", _lend);
  }

  std::array<SizeType, NumDimensions> initialize_local_extents(
      team_unit_t unit) const
  {
    DASH_LOG_DEBUG_VAR("BlockPattern.init_local_extents()", unit);
    DASH_LOG_DEBUG_VAR("BlockPattern.init_local_extents()", _nunits);
    if (_nunits == 0) {
      return ::std::array<SizeType, NumDimensions> {{ }};
    }
    // Coordinates of local unit id in team spec:
    auto unit_ts_coords = _teamspec.coords(unit);
    DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents", unit_ts_coords);
    ::std::array<SizeType, NumDimensions> l_extents{};
    for (auto d = 0; d < NumDimensions; ++d) {
      auto num_elem_d         = _memory_layout.extent(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];
      DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents.d", d);
      DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents.d", unit_ts_coord);
      DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents.d", num_elem_d);
      DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents.d", num_units_d);
      DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents.d", num_blocks_d);
      DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents.d", blocksize_d);
      DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents.d",
                         min_local_blocks_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;
      if (num_blocks_d == 1 && num_units_d == 1) {
        // One block assigned to one unit, use full extent in dimension:
        l_extents[d] = num_elem_d;
      } else {
        // Number of additional blocks for this unit, if any:
        auto num_add_blocks =
            static_cast<IndexType>(num_blocks_d % num_units_d);
        // Unit id assigned to the last block in dimension:
        team_unit_t last_block_unit_d((num_blocks_d % num_units_d == 0)
                                        ? num_units_d - 1
                                        : (num_blocks_d % num_units_d) - 1);
        DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents.d",
                           last_block_unit_d);
        DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents.d", num_add_blocks);
        if (unit_ts_coord < num_add_blocks) {
          // Unit is assigned to an additional block:
          l_extents[d] += blocksize_d;
          DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents.d", l_extents[d]);
        }
        if (unit_ts_coord == last_block_unit_d) {
          // If the last block in the dimension is underfilled and
          // assigned to the local unit, subtract the missing extent:
          SizeType undfill_blocksize_d = underfilled_blocksize(d);
          DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents",
                             undfill_blocksize_d);
          l_extents[d] -= undfill_blocksize_d;
        }
      }
      DASH_LOG_TRACE_VAR("BlockPattern.init_local_extents.d", l_extents[d]);
    }
    DASH_LOG_DEBUG_VAR("BlockPattern.init_local_extents >", l_extents);
    return l_extents;
  }
};

} // namespace dash

#include <dash/pattern/BlockPattern1D.h>

#endif // DASH__BLOCK_PATTERN_H_