LocalRange.h

#ifndef DASH__ALGORITHM__LOCAL_RANGE_H__
#define DASH__ALGORITHM__LOCAL_RANGE_H__

#include <dash/view/IndexSet.h>
#include <dash/view/Sub.h>
#include <dash/view/ViewMod.h>

#include <dash/Range.h>

#include <dash/internal/Logging.h>


namespace dash {

template <
  dim_t SubDim,
  class DomainT,
  class OffsetT >
constexpr auto
sub(
    OffsetT         begin,
    OffsetT         end,
    const DomainT & domain)
  -> typename std::enable_if<
       dash::view_traits<DomainT>::rank::value == 1,
       ViewSubMod<DomainT, SubDim>
     >::type;


template<typename ElementType>
struct LocalRange {
  ElementType * begin;
  ElementType * end;
};

template<typename IndexType>
struct LocalIndexRange {
  IndexType begin;
  IndexType end;
};

#if 0
template<class GlobInputIter>
typename std::enable_if<
  ( !GlobInputIter::has_view::value &&
    typename GlobInputIter::pattern_type::ndim() == 1 ),
  LocalIndexRange<typename GlobInputIter::pattern_type::index_type>
>::type
local_index_range(
  const GlobInputIter & first,
  const GlobInputIter & last)
{
  typedef typename GlobInputIter::pattern_type::index_type idx_t;

  auto grange    = dash::make_range(first, last);
  auto lrange    = dash::index(
                     dash::local(
                       dash::sub(     // < sub needed as temporary
                         first.pos(), //   workaround as
                         last.pos(),  //   index(local(container.local)
                         grange) ) ); //   is not defined in some cases

  idx_t lrange_begin = *dash::begin(lrange);
  idx_t lrange_end   = *dash::end(lrange);

  return LocalIndexRange<idx_t> { lrange_begin, lrange_end };
}
#endif

template<class GlobInputIter>
typename std::enable_if<
  !GlobInputIter::has_view::value,
  LocalIndexRange<typename GlobInputIter::pattern_type::index_type>
>::type
local_index_range(
  const GlobInputIter & first,
  const GlobInputIter & last)
{
  typedef typename GlobInputIter::pattern_type::index_type idx_t;

  typedef typename GlobInputIter::pattern_type pattern_t;
  typedef typename pattern_t::index_type       idx_t;
  // Get offsets of iterators within global memory, O(1):
  auto begin_gindex = static_cast<idx_t>(first.pos());
  auto end_gindex   = static_cast<idx_t>(last.pos());
  DASH_LOG_TRACE("local_index_range(GlobIt,GlobIt)",
                 begin_gindex, end_gindex);
  // Get pattern from global iterators, O(1):
  const auto& pattern = first.pattern();
  DASH_LOG_TRACE_VAR("local_index_range", pattern.local_size());
  if (pattern.local_size() == 0) {
    // Local index range is empty
    DASH_LOG_TRACE("local_index_range (lsize:0) ->", 0, 0);
    return LocalIndexRange<idx_t> { 0, 0 };
  }
  // Global index of first element in pattern, O(1):
  idx_t lbegin_gindex = pattern.lbegin();
  // Global index of last element in pattern, O(1):
  idx_t lend_gindex   = pattern.lend();
  DASH_LOG_TRACE_VAR("local_index_range", lbegin_gindex);
  DASH_LOG_TRACE_VAR("local_index_range", lend_gindex);
  if (lend_gindex   <= begin_gindex ||  // local end before global begin
      lbegin_gindex >= end_gindex) {    // local begin after global end
    // No overlap, intersection is empty
    DASH_LOG_TRACE("local_index_range (intersect:0) >", 0, 0);
    return LocalIndexRange<idx_t> { 0, 0 };
  }
  // Intersect local range and global range, in global index domain:
  auto goffset_lbegin = std::max<idx_t>(lbegin_gindex, begin_gindex);
  auto goffset_lend   = std::min<idx_t>(lend_gindex, end_gindex);
  // Global positions of local range to global coordinates, O(d):
  auto lbegin_gcoords = pattern.coords(goffset_lbegin);
  // Subtract 1 from global end offset as it points one coordinate
  // past the last index which is out of the valid coordinates range:
  auto lend_gcoords   = pattern.coords(goffset_lend-1);
  // Global coordinates of local range to local indices, O(d):
  auto lbegin_index   = pattern.at(lbegin_gcoords);
  // Add 1 to local end index to it points one coordinate past the
  // last index:
  auto lend_index     = pattern.at(lend_gcoords);
  if (lend_index
      == std::numeric_limits<typename pattern_t::index_type>::max()) {
    DASH_LOG_ERROR("local_index_range !",
                   "index type too small for for local index range");
  }
  lend_index += 1;
  // Return local index range
  DASH_LOG_TRACE("local_index_range >", lbegin_index, lend_index);
  return LocalIndexRange<idx_t> { lbegin_index, lend_index };
}

template<class GlobInputIter>
typename std::enable_if<
  GlobInputIter::has_view::value,
  LocalIndexRange<typename GlobInputIter::pattern_type::index_type>
>::type
local_index_range(
  const GlobInputIter & first,
  const GlobInputIter & last)
{
  typedef typename GlobInputIter::pattern_type pattern_t;
  typedef typename pattern_t::index_type       idx_t;
  // Get offsets of iterators within global memory, O(1):
  auto begin_gindex = static_cast<idx_t>(first.pos());
  auto end_gindex   = static_cast<idx_t>(last.pos());
  DASH_LOG_TRACE("local_index_range(ViewIt,ViewIt)",
                 begin_gindex, end_gindex);
  DASH_ASSERT_GT(begin_gindex+1, 0,
                 "local_index_range: begin.gidx must not be negative");
  DASH_ASSERT_GT(end_gindex+1, 0,
                 "local_index_range: end.gidx must not be negative");
  // Check if input range is relative to a view spec (e.g. a block):
  if (first.is_relative() && last.is_relative()) {
    DASH_LOG_TRACE("local_index_range", "input iterators are relative");
    if (first.viewspec() == last.viewspec()) {
      DASH_LOG_TRACE("local_index_range", "input iterators in same view");
      auto l_first        = first.lpos();
      bool first_is_local = l_first.unit == first.team().myid();
      // No need to check if last is local as both are relative to the
      // same view.
      if (first_is_local) {
        auto l_last_idx  = last.lpos().index;
        auto l_first_idx = l_first.index;
        DASH_LOG_TRACE("local_index_range >", l_first_idx, l_last_idx);
        return LocalIndexRange<idx_t> { l_first_idx, l_last_idx };
      }
      DASH_LOG_TRACE("local_index_range >", "not local -> (0,0)");
      return LocalIndexRange<idx_t>{0, 0};
    }
    DASH_THROW(
        dash::exception::InvalidArgument,
        "dash::local_index_range: views of first and last iterators "
        "differ");
  }
  // Get pattern from global iterators, O(1):
  const auto& pattern = first.pattern();
  DASH_LOG_TRACE_VAR("local_index_range", pattern.local_size());
  if (pattern.local_size() == 0) {
    // Local index range is empty
    DASH_LOG_TRACE("local_index_range (lsize:0) ->", 0, 0);
    return LocalIndexRange<idx_t> { 0, 0 };
  }
  // Global index of first element in pattern, O(1):
  idx_t lbegin_gindex = pattern.lbegin();
  // Global index of last element in pattern, O(1):
  idx_t lend_gindex   = pattern.lend();
  DASH_LOG_TRACE_VAR("local_index_range", lbegin_gindex);
  DASH_LOG_TRACE_VAR("local_index_range", lend_gindex);
  if (lend_gindex   <= begin_gindex ||  // local end before global begin
      lbegin_gindex >= end_gindex) {    // local begin after global end
    // No overlap, intersection is empty
    DASH_LOG_TRACE("local_index_range (intersect:0)->", 0, 0);
    return LocalIndexRange<idx_t> { 0, 0 };
  }
  // Intersect local range and global range, in global index domain:
  auto goffset_lbegin = std::max<idx_t>(lbegin_gindex, begin_gindex);
  auto goffset_lend   = std::min<idx_t>(lend_gindex, end_gindex);
  // Global positions of local range to global coordinates, O(d):
  auto lbegin_gcoords = pattern.coords(goffset_lbegin);
  // Subtract 1 from global end offset as it points one coordinate
  // past the last index which is out of the valid coordinates range:
  auto lend_gcoords   = pattern.coords(goffset_lend-1);
  // Global coordinates of local range to local indices, O(d):
  auto lbegin_index   = pattern.at(lbegin_gcoords);
  // Add 1 to local end index to it points one coordinate past the
  // last index:
  auto lend_index     = pattern.at(lend_gcoords);
  if (lend_index
      == std::numeric_limits<typename pattern_t::index_type>::max()) {
    DASH_LOG_ERROR("local_index_range !",
                   "index type too small for for local index range");
  }
  lend_index += 1;
  // Return local index range
  DASH_LOG_TRACE("local_index_range ->", lbegin_index, lend_index);
  return LocalIndexRange<idx_t> { lbegin_index, lend_index };
}

template<class GlobIterType>
LocalRange<const typename GlobIterType::value_type>
local_range(
  const GlobIterType & first,
  const GlobIterType & last)
{
  typedef typename GlobIterType::pattern_type pattern_t;
  typedef typename GlobIterType::value_type   value_t;
  typedef typename pattern_t::index_type      idx_t;
  DASH_LOG_TRACE("local_range()",
                 "gfirst.pos:", first.pos(),
                 "glast.pos:",  last.pos());
  // Global iterators to local index range, O(d):
  auto index_range   = dash::local_index_range(first, last);
  idx_t lbegin_index = index_range.begin;
  idx_t lend_index   = index_range.end;
  if (lbegin_index == lend_index) {
    // Local range is empty
    DASH_LOG_TRACE("local_range >", "empty local range",
                   lbegin_index, lend_index);
    return LocalRange<const value_t> { nullptr, nullptr };
  }
  // Local start address from global memory:
  const auto& pattern = first.pattern();

  auto* lbegin = dash::local_begin(
      static_cast<typename GlobIterType::pointer>(first.globmem().begin()),
      first.team().myid());
  // Add local offsets to local start address:
  if (lbegin == nullptr) {
    DASH_LOG_TRACE("local_range >", "lbegin null");
    return LocalRange<const value_t> { nullptr, nullptr };
  }
  DASH_LOG_TRACE("local_range >",
                 "lbegin:",      lbegin,
                 "l_idx_range:", lbegin_index, "-", lend_index);
  return LocalRange<const value_t> {
           lbegin + lbegin_index,
           lbegin + lend_index };
}

template<class GlobIterType>
LocalRange<typename GlobIterType::value_type>
local_range(
  GlobIterType & first,
  GlobIterType & last)
{
  typedef typename GlobIterType::pattern_type pattern_t;
  typedef typename GlobIterType::value_type   value_t;
  typedef typename pattern_t::index_type      idx_t;
  DASH_LOG_TRACE("local_range()",
                 "gfirst.pos:", first.pos(),
                 "glast.pos:",  last.pos());
  auto index_range   = dash::local_index_range(first, last);
  idx_t lbegin_index = index_range.begin;
  idx_t lend_index   = index_range.end;
  if (lbegin_index == lend_index) {
    // Local range is empty
    DASH_LOG_TRACE("local_range >", "empty local range",
                   lbegin_index, lend_index);
    return LocalRange<value_t> { nullptr, nullptr };
  }
  // Local start address from global memory:

  auto* lbegin = dash::local_begin(
      static_cast<typename GlobIterType::pointer>(first.globmem().begin()),
      first.team().myid());

  // Add local offsets to local start address:
  if (lbegin == nullptr) {
    DASH_LOG_TRACE("local_range >", "lbegin null");
    return LocalRange<value_t> { nullptr, nullptr };
  }
  DASH_LOG_TRACE("local_range >",
                 "lbegin:",      lbegin,
                 "l_idx_range:", lbegin_index, "-", lend_index);
  return LocalRange<value_t> {
           lbegin + lbegin_index,
           lbegin + lend_index };
}

} // namespace dash

#include <dash/algorithm/LocalRanges.h>

#endif  // DASH__ALGORITHM__LOCAL_RANGE_H__