GlobHeapMem.h

#ifndef DASH__GLOB_HEAP_H__INCLUDED
#define DASH__GLOB_HEAP_H__INCLUDED

#include <dash/dart/if/dart.h>

#include <dash/Types.h>
#include <dash/GlobPtr.h>
#include <dash/GlobSharedRef.h>
#include <dash/Allocator.h>
#include <dash/Team.h>
#include <dash/Array.h>
#include <dash/Onesided.h>

#include <dash/algorithm/MinMax.h>
#include <dash/algorithm/Copy.h>

#include <dash/memory/GlobHeapPtr.h>
#include <dash/memory/GlobHeapLocalPtr.h>

#include <dash/internal/Logging.h>

#include <list>
#include <vector>
#include <iterator>
#include <sstream>
#include <iostream>


namespace dash {

// Forward declarations
template<typename T, class MemSpaceT> class GlobHeapPtr;

template <
    typename ElementType,
    class LocalMemorySpace                   = dash::HostSpace,
    global_allocation_policy AllocationPolicy =
        global_allocation_policy::epoch_synchronized,
    template <class> class LocalAlloc = allocator::DefaultAllocator>
class GlobHeapMem {
private:
  typedef GlobHeapMem self_t;

  using memory_traits = dash::memory_space_traits<LocalMemorySpace>;

  using allocator_traits = dash::allocator_traits<dash::EpochSynchronizedAllocator<
      typename std::remove_const<ElementType>::type,
      LocalMemorySpace,
      AllocationPolicy,
      LocalAlloc>>;

  using local_allocator_traits =
      std::allocator_traits<typename allocator_traits::local_allocator>;

public:
  using value_type = typename allocator_traits::value_type;

  using allocator_type = typename allocator_traits::allocator_type;

  using local_memory_space = typename memory_traits::memory_space_type;

  typedef typename allocator_type::size_type       size_type;
  typedef typename allocator_type::difference_type difference_type;
  typedef typename allocator_type::difference_type index_type;

  typedef typename allocator_type::pointer raw_pointer;

  typedef GlobHeapLocalPtr<value_type, index_type> local_pointer;
  typedef GlobHeapLocalPtr<value_type, index_type> const_local_pointer;

  typedef GlobHeapPtr<value_type, GlobHeapMem>       pointer;
  typedef GlobHeapPtr<const value_type, GlobHeapMem> const_pointer;

  typedef       value_type &                                local_reference;
  typedef const value_type &                          const_local_reference;

  typedef typename local_pointer::bucket_type                   bucket_type;

  template <typename U>
  using rebind =
      GlobHeapMem<U, LocalMemorySpace, AllocationPolicy, LocalAlloc>;

  using local_void_pointer = void *;

private:
  typedef typename std::list<bucket_type>                       bucket_list;
  typedef typename bucket_list::iterator                    bucket_iterator;

  typedef dash::Array<
            size_type, int, dash::CSRPattern<1, dash::ROW_MAJOR, int> >
    local_sizes_map;

  typedef std::vector<std::vector<size_type> >       bucket_cumul_sizes_map;

  template<typename T_, class GMem_>
  friend class dash::GlobHeapPtr;

private:
  allocator_type             _allocator;
  dash::Team               * _team;
  dart_team_t                _teamid;
  size_type                  _nunits = 0;
  local_pointer              _lbegin = nullptr;
  local_pointer              _lend   = nullptr;
  team_unit_t                _myid   { DART_UNDEFINED_UNIT_ID };
  bucket_list                _buckets;
  bucket_list                _detach_buckets;
  bucket_iterator            _attach_buckets_first;
  local_sizes_map            _local_sizes;
  bucket_cumul_sizes_map     _bucket_cumul_sizes;
  local_sizes_map            _num_attach_buckets;
  local_sizes_map            _num_detach_buckets;
  size_type                  _remote_size = 0;
  index_type                 _begin_idx;
  index_type                 _end_idx;

public:
  GlobHeapMem(
      size_type n_local_elem = 0)
    : GlobHeapMem(n_local_elem, nullptr)
  {
  }

  GlobHeapMem(
      size_type n_local_elem,
      dash::Team & team)
    : GlobHeapMem(n_local_elem, nullptr, team)
  {
  }

  GlobHeapMem(
      size_type n_local_elem,
      LocalMemorySpace* r)
    : GlobHeapMem(n_local_elem, r, dash::Team::All())
  {
  }

  GlobHeapMem(size_type n_local_elem, LocalMemorySpace* r, Team& team)
  : _allocator(team, r),
    _team(&team),
    _teamid(team.dart_id()),
    _nunits(team.size()),
    _myid(team.myid()),
    _attach_buckets_first(_buckets.end()),
    _local_sizes(team.size(), team),
    _bucket_cumul_sizes(team.size()),
    _num_attach_buckets(team.size(), team),
    _num_detach_buckets(team.size(), team),
    _remote_size(0)
  {
    DASH_LOG_TRACE("GlobHeapMem.(ninit,nunits)",
                   n_local_elem, team.size());

    _local_sizes.local[0]        = 0;
    _num_attach_buckets.local[0] = 0;
    _num_detach_buckets.local[0] = 0;

    DASH_LOG_TRACE("GlobHeapMem.GlobHeapMem",
                   "allocating initial memory space");
    grow(n_local_elem);
    commit();

    DASH_LOG_TRACE("GlobHeapMem.GlobHeapMem >");

  }

  ~GlobHeapMem()
  {
    DASH_LOG_TRACE("GlobHeapMem.~GlobHeapMem()");
  }

  GlobHeapMem()                        = delete;

  GlobHeapMem(const self_t & other)    = default;

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

  constexpr bool operator==(const self_t & rhs) const noexcept
  {
    return (_teamid         == rhs._teamid &&
            _nunits         == rhs._nunits &&
            _lbegin         == rhs._lbegin &&
            _lend           == rhs._lend &&
            _buckets        == rhs._buckets &&
            _detach_buckets == rhs._detach_buckets);
  }

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

  constexpr size_type size() const noexcept
  {
    return _remote_size + local_size();
  }

  constexpr size_type local_size() const noexcept
  {
    return _local_sizes.local[0];
  }

  inline size_type local_size(team_unit_t unit) const
  {
    DASH_LOG_TRACE("GlobHeapMem.local_size(u)", "unit:", unit);
    DASH_ASSERT_RANGE(0, unit, _nunits-1, "unit id out of range");
    DASH_LOG_TRACE_VAR("GlobHeapMem.local_size",
                       _bucket_cumul_sizes[unit]);
    size_type unit_local_size;
    if (unit == _myid) {
      // Value of _local_sizes[u] is the local size as visible by the unit,
      // i.e. including size of unattached buckets.
      unit_local_size = _local_sizes.local[0];
    } else {
      unit_local_size = _bucket_cumul_sizes[unit].back();
    }
    DASH_LOG_TRACE("GlobHeapMem.local_size >", unit_local_size);
    return unit_local_size;
  }

  constexpr dash::Team & team() const noexcept
  {
    return (_team != nullptr) ? *_team : dash::Team::Null();
  }

  local_pointer grow(size_type num_elements)
  {
    DASH_LOG_DEBUG_VAR("GlobHeapMem.grow()", num_elements);
    size_type local_size_old = _local_sizes.local[0];
    DASH_LOG_TRACE("GlobHeapMem.grow",
                   "current local size:", local_size_old);
    if (num_elements == 0) {
      DASH_LOG_DEBUG("GlobHeapMem.grow >", "no grow");
      return _lend;
    }
    // Update size of local memory space:
    _local_sizes.local[0]        += num_elements;
    // Update number of local buckets marked for attach:
    _num_attach_buckets.local[0] += 1;

    // Create new unattached bucket:
    DASH_LOG_TRACE("GlobHeapMem.grow", "creating new unattached bucket:",
                   "size:", num_elements);
    bucket_type bucket;
    bucket.size               = num_elements;
    bucket.allocated_size     = num_elements;
    bucket.lptr               = _allocator.allocate_local(bucket.size);
    bucket.gptr               = DART_GPTR_NULL;
    bucket.attached           = false;
    // Add bucket to local memory space:
    _buckets.push_back(bucket);
    if (_attach_buckets_first == _buckets.end()) {
      // Move iterator to first unattached bucket to position of new bucket:
      _attach_buckets_first = _buckets.begin();
      std::advance(_attach_buckets_first,  _buckets.size() - 1);
    }
    _bucket_cumul_sizes[_myid].push_back(_local_sizes.local[0]);
    DASH_LOG_TRACE("GlobHeapMem.grow", "added unattached bucket:",
                   "allocated size:", bucket.size,
                   "lptr:", bucket.lptr);
    // Update local iteration space:
    update_lbegin();
    update_lend();
    DASH_ASSERT_EQ(_local_sizes.local[0], _lend - _lbegin,
                   "local size differs from local iteration space size");
    DASH_LOG_TRACE("GlobHeapMem.grow",
                   "new local size:",     _local_sizes.local[0]);
    DASH_LOG_TRACE("GlobHeapMem.grow",
                   "local buckets:",      _buckets.size(),
                   "unattached buckets:", _num_attach_buckets.local[0]);
    DASH_LOG_TRACE("GlobHeapMem.grow >");
    // Return local iterator to start of allocated memory:
    return _lbegin + local_size_old;
  }

  void shrink(size_type num_elements)
  {
    // This function updates the size of the local memory space of the
    // calling unit u.
    // The following members are updated:
    //
    // _local_sizes[u]:
    //   Size of local memory space as visible to unit u.
    //
    // _bucket_cumul_sizes:
    //    An array mapping units to a list of their cumulative bucket sizes
    //    (i.e. postfix sum) which is required to iterate over the
    //    non-contigous global dynamic memory space.
    //    For example, if unit 2 allocated buckets with sizes 1, 3 and 5,
    //    _bucket_cumul_sizes[2] is a list { 1, 4, 9 }.
    //
    // _buckets:
    //    List of local buckets that provide the underlying storage of the
    //    active unit's local memory space.
    //
    // Notes:
    //
    // It must be ensured that the updated cumulative bucket sizes of a
    // remote unit can be resolved in \c update_local_size() after any
    // possible combination of grow- and shrink-operations at the remote unit
    // from the following information:
    //
    // - the cumulative bucket sizes of the remote unit at the time of the
    //   last commit
    // - the remote unit's local size at the time of the last commit
    // - the remote unit's current local size (including unattached buckets)
    // - the number of the remote unit's unattached buckets and their size

    DASH_LOG_DEBUG_VAR("GlobHeapMem.shrink()", num_elements);
    DASH_ASSERT_LT(num_elements, local_size() + 1,
                   "cannot shrink size " << local_size() << " "
                   "by " << num_elements << " elements");
    if (num_elements == 0) {
      DASH_LOG_DEBUG("GlobHeapMem.shrink >", "no shrink");
      return;
    }
    DASH_LOG_TRACE("GlobHeapMem.shrink",
                   "current local size:", _local_sizes.local[0]);
    DASH_LOG_TRACE("GlobHeapMem.shrink",
                   "current local buckets:", _buckets.size());
    // Position of iterator to first unattached bucket:
    auto attach_buckets_first_pos = std::distance(_buckets.begin(),
                                                  _attach_buckets_first);
    DASH_LOG_TRACE_VAR("GlobHeapMem.shrink", attach_buckets_first_pos);
    // Number of elements left to deallocate:
    auto num_dealloc = num_elements;
    // Try to reduce local capacity by deallocating un-attached local buckets
    // as they do not have to be detached collectively.
    // Unattached buckets can be removed from memory space immediately as
    // remote units cannot have a pending reference on them.
    while (!_buckets.back().attached && num_dealloc > 0) {
      bucket_type & bucket_last = _buckets.back();
      // Shrink / remove unattached buckets starting at newest bucket:
      if (bucket_last.size <= num_dealloc) {
        DASH_LOG_TRACE("GlobHeapMem.shrink", "remove unattached bucket:",
                       "size:", bucket_last.size);
        // Mark entire bucket for deallocation below:
        num_dealloc           -= bucket_last.size;
        _local_sizes.local[0] -= bucket_last.size;
        _bucket_cumul_sizes[_myid].pop_back();
        // End iterator of _buckets about to change, update iterator to first
        // unattached bucket if it references the removed bucket:
        auto it_last_bucket = _buckets.end();
        std::advance(it_last_bucket, -1);
        if (_attach_buckets_first == it_last_bucket) {
          // Iterator to first unattached bucket references last bucket:
          DASH_LOG_TRACE("GlobHeapMem.shrink",
                         "updating iterator to first unattached bucket");
          std::advance(_attach_buckets_first, -1);
        }
        _allocator.deallocate_local(bucket_last.lptr, bucket_last.allocated_size);
        _buckets.pop_back();
        if (_attach_buckets_first->attached) {
          // Updated iterator to first unattached bucket references attached
          // bucket:
          _attach_buckets_first = _buckets.end();
        }
        // Update number of local buckets marked for attach:
        DASH_ASSERT_GT(_num_attach_buckets.local[0], 0,
                       "Last bucket unattached but number of buckets marked "
                       "for attach is 0");
        _num_attach_buckets.local[0] -= 1;
      } else if (bucket_last.size > num_dealloc) {
        auto const size_new = bucket_last.size - num_dealloc;

        DASH_LOG_TRACE("GlobHeapMem.shrink", "shrink unattached bucket:",
                       "old size:", bucket_last.size,
                       "new size:", size_new);
        _local_sizes.local[0]             -= num_dealloc;
        _bucket_cumul_sizes[_myid].back() -= num_dealloc;
        num_dealloc = 0;

        //Deallocate old local bucket
        _allocator.deallocate_local(bucket_last.lptr, bucket_last.allocated_size);
        //Allocate new bucket with specified size
        bucket_last.lptr = _allocator.allocate_local(size_new);
        if (bucket_last.lptr == nullptr) {
          DASH_THROW(dash::exception::RuntimeError,
                     "GlobHeapMem.shrink: Allocating bucket of size failed");
        }
        bucket_last.size = size_new;
        bucket_last.allocated_size = size_new;
      }
    }
    // Number of elements to deallocate exceeds capacity of un-attached
    // buckets, deallocate attached buckets:
    auto num_dealloc_gbuckets = 0;
    // Shrink attached buckets starting at newest bucket:
    for (auto bucket_it = _buckets.rbegin();
         bucket_it != _buckets.rend();
         ++bucket_it) {
      if (!bucket_it->attached) {
        continue;
      }
      if (num_dealloc == 0) {
        break;
      }
      if (bucket_it->size <= num_dealloc) {
        // mark entire bucket for deallocation:
        num_dealloc_gbuckets++;
        _num_detach_buckets.local[0]      += 1;
        _local_sizes.local[0]             -= bucket_it->size;
        _bucket_cumul_sizes[_myid].back() -= bucket_it->size;
        num_dealloc                       -= bucket_it->size;
      } else if (bucket_it->size > num_dealloc) {
        DASH_LOG_TRACE("GlobHeapMem.shrink", "shrink attached bucket:",
                       "old size:", bucket_it->size,
                       "new size:", bucket_it->size - num_dealloc);
        bucket_it->size                   -= num_dealloc;
        _local_sizes.local[0]             -= num_dealloc;
        _bucket_cumul_sizes[_myid].back() -= num_dealloc;
        num_dealloc = 0;
      }
    }
    // Mark attached buckets for deallocation.
    // Requires separate loop as iterators on _buckets could be invalidated.
    DASH_LOG_DEBUG_VAR("GlobHeapMem.shrink", num_dealloc_gbuckets);
    while (num_dealloc_gbuckets-- > 0) {
      auto dealloc_bucket = _buckets.back();
      DASH_LOG_TRACE("GlobHeapMem.shrink", "deallocate attached bucket:"
                     "size:", dealloc_bucket.size,
                     "lptr:", dealloc_bucket.lptr);
      // Mark bucket to be detached in next call of commit():
      _detach_buckets.push_back(dealloc_bucket);
      // Unregister bucket:
      _buckets.pop_back();
    }
    // Update local iterators as bucket iterators might have changed:
    update_lbegin();
    update_lend();

    DASH_LOG_TRACE("GlobHeapMem.shrink",
                   "cumulative bucket sizes:",  _bucket_cumul_sizes[_myid]);
    DASH_LOG_TRACE("GlobHeapMem.shrink",
                   "new local size:",           _local_sizes.local[0],
                   "new iteration space size:", std::distance(
                                                  _lbegin, _lend));
    DASH_LOG_TRACE("GlobHeapMem.shrink",
                   "total number of buckets:",  _buckets.size(),
                   "unattached buckets:",       std::distance(
                                                  _attach_buckets_first,
                                                  _buckets.end()));
    DASH_LOG_DEBUG("GlobHeapMem.shrink >");
  }

  void commit()
  {
    DASH_LOG_DEBUG("GlobHeapMem.commit()");
    DASH_LOG_TRACE_VAR("GlobHeapMem.commit", _buckets.size());

    // First detach, then attach to minimize number of elements allocated
    // at the same time:
    size_type num_detached_elem = commit_detach();
    size_type num_attached_elem = commit_attach();

    if (num_detached_elem > 0 || num_attached_elem > 0) {
      // Update _begin iterator:
      DASH_LOG_TRACE("GlobHeapMem.commit", "updating _begin");
      _begin_idx = 0;
      DASH_LOG_TRACE("GlobHeapMem.commit", "updating _end");
      _end_idx   = size();
    }
    // Update local iterators as bucket iterators might have changed:
    DASH_LOG_TRACE("GlobHeapMem.commit", "updating _lbegin");
    update_lbegin();
    DASH_LOG_TRACE("GlobHeapMem.commit", "updating _lend");
    update_lend();
    DASH_LOG_DEBUG("GlobHeapMem.commit >", "finished");
  }

  void resize(size_type num_elements)
  {
    DASH_LOG_DEBUG("GlobHeapMem.resize()", "new size:", num_elements);
    index_type diff_capacity = num_elements - size();
    if (diff_capacity > 0) {
      grow(diff_capacity);
    } else if (diff_capacity < 0) {
      shrink(-diff_capacity);
    }
    DASH_LOG_DEBUG("GlobHeapMem.resize >");
  }

  pointer begin() noexcept
  {
    return pointer(this, _begin_idx);
  }

  constexpr const_pointer begin() const noexcept
  {
    return const_pointer(this, _begin_idx);
  }

  pointer end() noexcept
  {
    return pointer(this, _end_idx);
  }

  const_pointer end() const noexcept
  {
    return const_pointer(this, _end_idx);
  }

  inline local_pointer & lbegin() noexcept
  {
    return _lbegin;
  }

  inline const_local_pointer lbegin() const noexcept
  {
    return _lbegin;
  }

  inline local_pointer & lend() noexcept
  {
    return _lend;
  }

  inline const_local_pointer & lend() const noexcept
  {
    return _lend;
  }

  template<typename ValueType = value_type>
  void put_value(
    const ValueType & newval,
    index_type        global_index)
  {
    DASH_LOG_TRACE("GlobHeapMem.put_value(newval, gidx = %d)",
                   global_index);
    auto git = pointer(this, global_index);
    dash::put_value(newval, git);
  }

  template<typename ValueType = value_type>
  void get_value(
    ValueType  * ptr,
    index_type   global_index) const
  {
    DASH_LOG_TRACE("GlobHeapMem.get_value(newval, gidx = %d)",
                   global_index);
    auto git = pointer(this, global_index);
    dash::get_value(ptr, git);
  }

  void barrier() const
  {
    DASH_ASSERT_RETURNS(
      dart_barrier(_teamid),
      DART_OK);
  }

  template<typename IndexT>
  pointer at(
    team_unit_t unit,
    IndexT      local_index)
  {
    DASH_LOG_DEBUG("GlobHeapMem.at()",
                   "unit:", unit, "lidx:", local_index);
    if (_nunits == 0) {
      DASH_THROW(dash::exception::RuntimeError, "No units in team");
    }
    pointer git(this, unit, local_index);
    DASH_LOG_DEBUG_VAR("GlobHeapMem.at >", git);
    return git;
  }

  template<typename IndexT>
  const_pointer at(
    team_unit_t unit,
    IndexT      local_index) const
  {
    DASH_LOG_DEBUG("GlobHeapMem.at() const",
                   "unit:", unit, "lidx:", local_index);
    if (_nunits == 0) {
      DASH_THROW(dash::exception::RuntimeError, "No units in team");
    }
    const_pointer git(this, unit, local_index);
    DASH_LOG_DEBUG_VAR("GlobHeapMem.at const >", git);
    return git;
  }

  inline const bucket_list & local_buckets() const
  {
    return _buckets;
  }

private:

  void update_lbegin() noexcept
  {
    DASH_LOG_TRACE("GlobHeapMem.update_lbegin()");
    local_pointer unit_lbegin(
             // iteration space
             _buckets.begin(), _buckets.end(),
             // position in iteration space
             0,
             // bucket at position in iteration space,
             // offset in bucket
             _buckets.begin(), 0);
    DASH_LOG_TRACE("GlobHeapMem.update_lbegin >", unit_lbegin);
    _lbegin = unit_lbegin;
  }

  void update_lend() noexcept
  {
    DASH_LOG_TRACE("GlobHeapMem.update_lend()");
    local_pointer unit_lend(
             // iteration space
             _buckets.begin(), _buckets.end(),
             // position in iteration space
             local_size(),
             // bucket at position in iteration space,
             // offset in bucket
             _buckets.end(), 0);
    DASH_LOG_TRACE("GlobHeapMem.update_lend >", unit_lend);
    _lend = unit_lend;
  }


  size_type commit_detach()
  {
    DASH_LOG_TRACE("GlobHeapMem.commit_detach()");
    DASH_LOG_TRACE("GlobHeapMem.commit_detach",
                   "local buckets to detach:", _num_detach_buckets.local[0]);
    // Number of elements successfully deallocated from global memory in
    // this commit:
    size_type num_detached_elem = 0;
    for (auto bucket_it = _detach_buckets.begin();
         bucket_it != _detach_buckets.cend(); ++bucket_it) {
      DASH_LOG_TRACE("GlobHeapMem.commit_detach", "detaching bucket:",
                     "size:", bucket_it->size,
                     "lptr:", bucket_it->lptr,
                     "gptr:", bucket_it->gptr);
      // Detach bucket from global memory region and deallocate its local
      // memory segment:
      if (bucket_it->attached) {
        _allocator.deallocate(bucket_it->gptr, bucket_it->allocated_size);
        num_detached_elem   += bucket_it->size;
        bucket_it->attached  = false;
      }
    }
    _detach_buckets.clear();
    DASH_LOG_TRACE("GlobHeapMem.commit_detach >",
                   "globally deallocated elements:", num_detached_elem);
    return num_detached_elem;
  }

  size_type commit_attach()
  {
    DASH_LOG_TRACE("GlobHeapMem.commit_attach()");
    DASH_LOG_TRACE("GlobHeapMem.commit_attach",
                   "local buckets to attach:", _num_attach_buckets.local[0]);
    // Unregister buckets marked for detach in global memory:
    _num_attach_buckets.barrier();
    // Minumum and maximum number of buckets to be attached by any unit:
    auto min_max_attach     = gather_min_max(_num_attach_buckets.begin(),
                                             _num_attach_buckets.end());
    auto min_attach_buckets = min_max_attach.first;
    auto max_attach_buckets = min_max_attach.second;
    DASH_LOG_TRACE("GlobHeapMem.commit_attach",
                   "min. attach buckets:",  min_attach_buckets);
    DASH_LOG_TRACE("GlobHeapMem.commit_attach",
                   "max. attach buckets:",  max_attach_buckets);
    // Number of buckets successfully attached in this commit:
    size_type num_attached_buckets = 0;
    // Number of elements allocated in global memory in this commit:
    size_type num_attached_elem    = 0;
    // Number of elements at remote units before the commit:
    size_type old_remote_size      = _remote_size;
    _remote_size                   = update_remote_size();
    // Whether at least one remote unit needs to attach additional global
    // memory:
    bool has_remote_attach         = _remote_size > old_remote_size;
    DASH_LOG_TRACE_VAR("GlobHeapMem.commit_attach", old_remote_size);
    DASH_LOG_TRACE_VAR("GlobHeapMem.commit_attach", _remote_size);
    DASH_LOG_TRACE_VAR("GlobHeapMem.commit_attach", size());
    DASH_LOG_TRACE_VAR("GlobHeapMem.commit_attach", has_remote_attach);
    // Plausibility check:
    DASH_ASSERT(!has_remote_attach || max_attach_buckets > 0);

    // Attach local unattached buckets in global memory space.
    // As bucket sizes differ between units, units must collect gptr's
    // (dart_gptr_t) and size of buckets attached by other units and store
    // them locally so a remote unit's local index can be mapped to the
    // remote unit's bucket.
    if (min_attach_buckets == 0 && max_attach_buckets == 0) {
      DASH_LOG_TRACE("GlobHeapMem.commit_attach", "no attach");
      DASH_ASSERT(_attach_buckets_first == _buckets.end());
      DASH_ASSERT(_buckets.empty() || _buckets.back().attached);
    }
    DASH_LOG_TRACE("GlobHeapMem.commit_attach", "attaching",
                   std::distance(_attach_buckets_first, _buckets.end()),
                   "buckets");
    for (; _attach_buckets_first != _buckets.end(); ++_attach_buckets_first) {
      bucket_type & bucket = *_attach_buckets_first;
      DASH_ASSERT(!bucket.attached);
      DASH_LOG_TRACE("GlobHeapMem.commit_attach", "attaching bucket");
      DASH_LOG_TRACE_VAR("GlobHeapMem.commit_attach", bucket.size);
      DASH_LOG_TRACE_VAR("GlobHeapMem.commit_attach", bucket.lptr);
      // Attach bucket's local memory segment in global memory:
      bucket.gptr     = _allocator.attach(bucket.lptr, bucket.size);
      bucket.attached = true;
      DASH_LOG_TRACE("GlobHeapMem.commit_attach", "attached bucket:",
                     "gptr:", bucket.gptr);
      num_attached_elem            += bucket.size;
      _num_attach_buckets.local[0] -= 1;
      num_attached_buckets++;
    }
    DASH_ASSERT(_attach_buckets_first == _buckets.end());
    // All units must attach the same number of buckets collectively.
    // Attach empty buckets if this unit attached less than the maximum
    // number of buckets attached by any other unit in this commit:
    DASH_LOG_TRACE("GlobHeapMem.commit_attach",
                   "local buckets attached:", num_attached_buckets);
    DASH_LOG_TRACE("GlobHeapMem.commit_attach",
                   "buckets required to attach:", max_attach_buckets);
    while (num_attached_buckets < max_attach_buckets) {
      DASH_LOG_TRACE("GlobHeapMem.commit_attach", "attaching null bucket");
      bucket_type bucket;
      bucket.size               = 0;
      bucket.allocated_size     = 0;
      bucket.lptr               = _allocator.allocate_local(0);
      bucket.gptr               = _allocator.attach(bucket.lptr, bucket.size);
      bucket.attached           = true;
      DASH_ASSERT(!DART_GPTR_ISNULL(bucket.gptr));
      _buckets.push_back(bucket);
      num_attached_buckets++;
      DASH_LOG_TRACE("GlobHeapMem.commit_attach", "attached null bucket:",
                     "gptr:", bucket.gptr,
                     "left:", max_attach_buckets - num_attached_buckets);
    }
    DASH_LOG_TRACE("GlobHeapMem.commit_attach >",
                   "globally allocated elements:", num_attached_elem);
    return num_attached_elem;
  }

  template<
    typename GlobalIt,
    typename ValueType = typename GlobalIt::value_type >
  std::pair<ValueType, ValueType> gather_min_max(
    const GlobalIt & first,
    const GlobalIt & last)
  {
    // TODO: Unoptimized, use dash::min_max_element once it is available
    //
    DASH_LOG_TRACE("GlobHeapMem.gather_min_max()");
    std::vector<ValueType> lcopy(dash::distance(first, last));
    auto lcopy_end = dash::copy(first, last, lcopy.data());
    auto min_lptr  = std::min_element(lcopy.data(), lcopy_end);
    auto max_lptr  = std::max_element(lcopy.data(), lcopy_end);
    std::pair<ValueType, ValueType> min_max;
    min_max.first  = *min_lptr;
    min_max.second = *max_lptr;
    DASH_LOG_TRACE("GlobHeapMem.gather_min_max >",
                   "min:", min_max.first,
                   "max:", min_max.second);
    return min_max;
  }

  size_type update_remote_size()
  {
    // This function updates local snapshots of the remote unit's local
    // sizes.
    // The following members are updated:
    //
    // _remote_size:
    //    The sum of all remote units' local size, including the sizes of
    //    unattached buckets.
    //
    // _bucket_cumul_sizes:
    //    An array mapping units to a list of their cumulative bucket sizes
    //    (i.e. postfix sum) which is required to iterate over the
    //    non-contigous global dynamic memory space.
    //    For example, if unit 2 allocated buckets with sizes 1, 3 and 5,
    //    _bucket_cumul_sizes[2] is a list { 1, 4, 9 }.
    //
    // Outline:
    //
    // 1. Create local copy of the distributed array _num_attach_buckets that
    //    contains the number of unattached buckets of every unit.
    // 2. Temporarily attach an array attach_bucket_sizes in global memory
    //    that contains the sizes of this unit's unattached buckets.
    // 3. At this point, every unit published the number of buckets it will
    //    attach in the next commit, and their sizes.
    //    The current local size Lu of every unit, including its unattached
    //    buckets, is stored in _local_sizes.
    // 4. For every remote unit u:
    //    - If unit u has one unattached bucket, append the unit's current
    //      local size Lu to the unit's list of cumulative bucket sizes.
    //    - If unit u has more than one unattached bucket, the sizes of the
    //      single buckets must be retrieved from the vector
    //      attach_bucket_sizes temporarily attached by u in step 1.
    // 5. Detach vector attach_bucket_sizes.

    DASH_LOG_TRACE("GlobHeapMem.update_remote_size()");
    size_type new_remote_size = 0;
    // Number of unattached buckets of every unit:
    std::vector<size_type> num_unattached_buckets(_nunits, 0);
    _num_attach_buckets.barrier();
    dash::copy(_num_attach_buckets.begin(), _num_attach_buckets.end(),
               num_unattached_buckets.data());

#ifdef DASH_ENABLE_TRACE_LOGGING
    std::for_each(std::begin(_num_attach_buckets),
                  std::end(_num_attach_buckets),
                  [](size_type const& bsz) {
                    DASH_LOG_TRACE("GlobMemHeap.update_remote_size()",
                                   "num_buckets at unit: ", bsz);
                  });
#endif

    std::vector<size_type> attach_buckets_sizes;
    std::vector<size_type> displs;
    std::vector<size_type> team_unattached_bucket_sizes;

    auto atLeast2 = std::find_if(num_unattached_buckets.begin(),
                                 num_unattached_buckets.end(),
                                 [](size_type const& sz) { return sz > 1; });

    //At least one remote unit has more than two buckets to attach
    //so we make all to all communication
    if (atLeast2 != num_unattached_buckets.end()) {
      // Attach array of local unattached bucket sizes to allow remote units
      // to
      // query the sizes of this unit's unattached buckets.
      for (auto bit = _attach_buckets_first; bit != _buckets.end(); ++bit) {
        attach_buckets_sizes.push_back(bit->size);
      }

      DASH_ASSERT(attach_buckets_sizes.size() ==
                  _num_attach_buckets.local[0]);
      DASH_LOG_TRACE_VAR("GlobHeapMem.update_remote_size",
                         attach_buckets_sizes);

      dash::dart_storage<size_type> ds(attach_buckets_sizes.size());

      // Accumulate number of unattached buckets of each unit
      auto const n_team_unattached_buckets =
          std::accumulate(std::begin(num_unattached_buckets),
                          std::end(num_unattached_buckets), 0);

      team_unattached_bucket_sizes.reserve(n_team_unattached_buckets);

      displs.reserve(_team->size());
      displs[0] = 0;

      //calculate the displs of each unit
      std::partial_sum(std::begin(num_unattached_buckets),
                       // We stop at last element
                       std::end(num_unattached_buckets) - 1,
                       // We start at offset 1, since disp[0] = 0
                       std::begin(displs) + 1);

      DASH_ASSERT_RETURNS(dart_allgatherv(
                              // array of locally unattached bucket sizes
                              attach_buckets_sizes.data(),
                              // number of locally unattached buckets
                              ds.nelem,
                              // DART Datatype
                              ds.dtype,
                              // receive buffer
                              team_unattached_bucket_sizes.data(),
                              // receive counts
                              num_unattached_buckets.data(),
                              // receive displs
                              displs.data(),
                              // DART Team
                              _team->dart_id()),
                          DART_OK);
    }

    for (size_type u = 0; u < _nunits; ++u) {
      if (u == _myid) {
        continue;
      }
      DASH_LOG_TRACE("GlobHeapMem.update_remote_size",
                     "collecting local bucket sizes of unit", u);
      // Last known local attached capacity of remote unit:
      auto& u_bucket_cumul_sizes = _bucket_cumul_sizes[u];
      // Request current locally allocated capacity of remote unit:
      size_type u_local_size_old  = u_bucket_cumul_sizes.size() == 0
                                    ? 0
                                    : u_bucket_cumul_sizes.back();
      size_type u_local_size_new  = _local_sizes[u];
      DASH_LOG_TRACE_VAR("GlobHeapMem.update_remote_size",
                         u_local_size_old);
      DASH_LOG_TRACE_VAR("GlobHeapMem.update_remote_size",
                         u_local_size_old);
      difference_type u_local_size_diff  = u_local_size_new - u_local_size_old;
      new_remote_size       += u_local_size_new;
      // Number of unattached buckets of unit u:
      size_type u_num_attach_buckets = num_unattached_buckets[u];
      DASH_LOG_TRACE_VAR("GlobHeapMem.update_remote_size",
                         u_num_attach_buckets);
      if (u_num_attach_buckets == 1) {
        // One unattached bucket at unit u, no need to request single bucket
        // sizes:
        u_bucket_cumul_sizes.push_back(u_local_size_new);
      }
      else if (u_num_attach_buckets > 1) {
        auto const u_end = displs[u] + u_num_attach_buckets;
        for (int bi = displs[u]; bi < u_end; ++bi) {
          size_type single_bkt_size = team_unattached_bucket_sizes[bi];
          size_type cumul_bkt_size  = single_bkt_size;
          DASH_LOG_TRACE_VAR("GlobHeapMem.update_remote_size",
                             single_bkt_size);
          if (u_bucket_cumul_sizes.size() > 0) {
            cumul_bkt_size += u_bucket_cumul_sizes.back();
          }
          u_bucket_cumul_sizes.push_back(cumul_bkt_size);
        }
      }
      // Local memory space of unit shrunk:
      if (u_local_size_diff < 0 && u_bucket_cumul_sizes.size() > 0) {
        u_bucket_cumul_sizes.back() += u_local_size_diff;
      }
    }

    _team->barrier();
#if DASH_ENABLE_TRACE_LOGGING
    for (size_type u = 0; u < _nunits; ++u) {
      DASH_LOG_TRACE("GlobHeapMem.update_remote_size",
                     "unit", u,
                     "cumulative bucket sizes:", _bucket_cumul_sizes[u]);
    }
#endif
    DASH_LOG_TRACE("GlobHeapMem.update_remote_size >", new_remote_size);
    _remote_size = new_remote_size;
    return _remote_size;
  }

  dart_gptr_t dart_gptr_at(
    team_unit_t unit,
    index_type  bucket_index,
    index_type  bucket_phase) const
  {
    DASH_LOG_DEBUG("GlobHeapMem.dart_gptr_at(u,bi,bp)",
                   unit, bucket_index, bucket_phase);
    if (_nunits == 0) {
      DASH_THROW(dash::exception::RuntimeError, "No units in team");
    }
    // Get the referenced bucket's dart_gptr:
    auto bucket_it = _buckets.begin();
    std::advance(bucket_it, bucket_index);
    auto dart_gptr = bucket_it->gptr;
    DASH_LOG_TRACE_VAR("GlobHeapMem.dart_gptr_at", bucket_it->attached);
    DASH_LOG_TRACE_VAR("GlobHeapMem.dart_gptr_at", bucket_it->gptr);
    if (unit == _myid) {
      DASH_LOG_TRACE_VAR("GlobHeapMem.dart_gptr_at", bucket_it->lptr);
      DASH_LOG_TRACE_VAR("GlobHeapMem.dart_gptr_at", bucket_it->size);
      DASH_ASSERT_LT(bucket_phase, bucket_it->size,
                     "bucket phase out of bounds");
    }
    if (DART_GPTR_ISNULL(dart_gptr)) {
      DASH_LOG_TRACE("GlobHeapMem.dart_gptr_at",
                     "bucket.gptr is DART_GPTR_NULL");
      dart_gptr = DART_GPTR_NULL;
    } else {
      // Move dart_gptr to unit and local offset:
      DASH_ASSERT_RETURNS(
        dart_gptr_setunit(&dart_gptr, unit),
        DART_OK);
      DASH_ASSERT_RETURNS(
        dart_gptr_incaddr(
          &dart_gptr,
          bucket_phase * sizeof(value_type)),
        DART_OK);
    }
    DASH_LOG_DEBUG("GlobHeapMem.dart_gptr_at >", dart_gptr);
    return dart_gptr;
  }

}; // class GlobHeapMem

} // namespace dash

#endif // DASH__GLOB_HEAP_H__INCLUDED