.. _program_listing_file_include_shad_data_structures_vector.h:

Program Listing for File vector.h
=================================

|exhale_lsh| :ref:`Return to documentation for file <file_include_shad_data_structures_vector.h>` (``include/shad/data_structures/vector.h``)

.. |exhale_lsh| unicode:: U+021B0 .. UPWARDS ARROW WITH TIP LEFTWARDS

.. code-block:: cpp

   //===------------------------------------------------------------*- C++ -*-===//
   //
   //                                     SHAD
   //
   //      The Scalable High-performance Algorithms and Data Structure Library
   //
   //===----------------------------------------------------------------------===//
   //
   // Copyright 2018 Battelle Memorial Institute
   //
   // Licensed under the Apache License, Version 2.0 (the "License"); you may not
   // use this file except in compliance with the License. You may obtain a copy
   // of the License at
   //
   //     http://www.apache.org/licenses/LICENSE-2.0
   //
   // Unless required by applicable law or agreed to in writing, software
   // distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   // WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   // License for the specific language governing permissions and limitations
   // under the License.
   //
   //===----------------------------------------------------------------------===//
   
   #ifndef INCLUDE_SHAD_DATA_STRUCTURES_VECTOR_H_
   #define INCLUDE_SHAD_DATA_STRUCTURES_VECTOR_H_
   
   #include <algorithm>
   #include <atomic>
   #include <cstdint>
   #include <iterator>
   #include <limits>
   #include <memory>
   #include <stdexcept>
   #include <tuple>
   #include <utility>
   #include <vector>
   
   #include "shad/data_structures/abstract_data_structure.h"
   #include "shad/data_structures/buffer.h"
   #include "shad/runtime/runtime.h"
   
   namespace shad {
   
   template <typename T, typename Allocator = std::allocator<T>>
   class Vector : public AbstractDataStructure<Vector<T, Allocator>> {
     template <typename ValueType>
     class Iterator;
   
    public:
     using allocator_type = Allocator;
     using value_type = T;
     using difference_type = typename allocator_type::difference_type;
     using size_type = typename allocator_type::size_type;
     using iterator = Iterator<T>;
     using const_iterator = Iterator<const T>;
     using ObjectID =
         typename AbstractDataStructure<Vector<T, Allocator>>::ObjectID;
   
     static_assert(
         (std::is_same<typename allocator_type::value_type, value_type>::value),
         "Allocator::value_type must be the same as value_type");
   
   
     size_type Size() const noexcept;
   
     size_type MaxSize() const noexcept;
   
     size_type Capacity() const noexcept;
   
     bool Empty() const noexcept;
   
     void Reserve(size_type n);
   
     void Resize(size_type n);
   
   
   
     value_type At(size_type n) const;
   
     value_type operator[](size_type n) const;
   
     value_type Front() const;
   
     value_type Back() const;
   
     void AsyncAt(rt::Handle &handle, size_type n, T *result) const;
   
   
   
     void Clear() noexcept;
   
     void PushBack(const T &value);
   
     iterator InsertAt(shad::Vector<T, Allocator>::size_type position,
                       const shad::Vector<T, Allocator>::value_type &value);
   
     template <typename InputIterator>
     iterator InsertAt(size_type position, InputIterator begin, InputIterator end);
   
     void AsyncInsertAt(rt::Handle &handle,
                        shad::Vector<T, Allocator>::size_type position,
                        const shad::Vector<T, Allocator>::value_type &value);
   
     template <typename InputIterator>
     void AsyncInsertAt(rt::Handle &handle, size_type position,
                        InputIterator begin, InputIterator end);
   
     void BufferedInsertAt(const size_type pos, const value_type &value);
   
     void BufferedAsyncInsertAt(rt::Handle &handle, const size_type pos,
                                const value_type &value);
   
     void WaitForBufferedInsert() { buffers_.FlushAll(); }
   
   
   
     template <typename ApplyFunT, typename... Args>
     void Apply(const size_type position, ApplyFunT &&function, Args &... args);
   
     template <typename ApplyFunT, typename... Args>
     void AsyncApply(rt::Handle &handle, const size_type position,
                     ApplyFunT &&function, Args &... args);
   
     template <typename ApplyFunT, typename... Args>
     void ForEachInRange(const size_type first, const size_type last,
                         ApplyFunT &&function, Args &... args);
   
     template <typename ApplyFunT, typename... Args>
     void AsyncForEachInRange(rt::Handle &handle, const size_type first,
                              const size_type last, ApplyFunT &&function,
                              Args &... args);
   
   
     ObjectID GetGlobalID() const { return oid_; }
   
     ~Vector() { _clear(); }
   
     void BufferEntryInsert(const std::tuple<size_type, value_type> entry) {
       auto blockOffsetPair = _blockOffsetFromPosition(std::get<0>(entry));
       size_type localBlock = _globlalBlockToLocalBlock(blockOffsetPair.first);
       dataBlocks_.at(localBlock)[blockOffsetPair.second] = std::get<1>(entry);
     }
   
    protected:
     Vector(ObjectID oid, size_type n)
         : oid_(oid),
           dataBlocks_(),
           mainLocality_(static_cast<uint64_t>(oid) % rt::numLocalities()),
           sizeCapacityLock_(),
           size_(n),
           capacity_(0),
           allocator_(),
           buffers_(oid) {
       size_t blocksToAllocate = std::max(_sizeToLocalBlocks(n, kBlockSize), 1UL);
       capacity_ =
           std::max(kBlockSize * _blockOffsetFromPosition(n).first, kBlockSize);
   
       if (n == 0 && static_cast<uint32_t>(rt::thisLocality()) != 0) return;
   
       for (size_t i = 0; i < blocksToAllocate; ++i)
         dataBlocks_.emplace_back(std::allocator_traits<allocator_type>::allocate(
             allocator_, kBlockSize));
     }
   
    private:
     friend class AbstractDataStructure<Vector<T, Allocator>>;
     static const size_type kBlockSize;
   
     std::tuple<rt::Locality, size_t, size_t> _targetFromPosition(
         size_type position, size_type blockSize) const {
       size_t blockNumber = position / blockSize;
       size_t destination = blockNumber % rt::numLocalities();
       size_t offset = position % blockSize;
       return std::make_tuple(rt::Locality(destination), blockNumber, offset);
     }
   
     size_t _sizeToLocalBlocks(size_type n, size_type blockSize) const {
       size_t fullyUsedBlocks(0);
       rt::Locality pivot(0);
   
       if (n == 0) return 0;
   
       std::tie(pivot, fullyUsedBlocks, std::ignore) =
           _targetFromPosition(n - 1, blockSize);
   
       size_t localBlocks = fullyUsedBlocks / rt::numLocalities();
       if (rt::thisLocality() <= pivot) localBlocks += 1;
   
       return localBlocks;
     }
   
     std::pair<size_type, size_type> _blockOffsetFromPosition(size_type n) const {
       size_type blockNumber = n / kBlockSize;
       size_type offset = n % kBlockSize;
       return std::make_pair(blockNumber, offset);
     }
   
     size_type _globlalBlockToLocalBlock(size_type n) const {
       size_type i = 0;
       while (static_cast<uint32_t>(rt::thisLocality()) < n) {
         n -= rt::numLocalities();
         ++i;
       }
       return i;
     }
   
     void _reserve(size_type n) {
       size_type currentCapacity = Capacity();
       if (currentCapacity >= n) return;
   
       rt::Locality insertLocality(0);
       size_t lastBlock(0);
       std::tie(insertLocality, lastBlock, std::ignore) =
           _targetFromPosition(currentCapacity, kBlockSize);
       size_t newLastBlock = n / kBlockSize;
   
       size_type blocksToAllocate = 1;
       blocksToAllocate += newLastBlock - lastBlock;
   
       std::vector<size_type> newBlocks(rt::numLocalities(), 0);
       for (size_t i = static_cast<uint32_t>(insertLocality), j = blocksToAllocate;
            j > 0; ++i, --j) {
         newBlocks[i % rt::numLocalities()]++;
       }
   
       rt::Handle handle;
       for (size_t i = 0; i < rt::numLocalities(); ++i) {
         if (newBlocks[i] == 0) continue;
   
         rt::asyncExecuteAt(
             handle, rt::Locality(i),
             [](rt::Handle &, const std::pair<ObjectID, size_type> &args) {
               auto This = Vector<T, Allocator>::GetPtr(args.first);
   
               for (size_t i = 0; i < args.second; ++i) {
                 This->dataBlocks_.emplace_back(
                     std::allocator_traits<allocator_type>::allocate(
                         This->allocator_, kBlockSize));
               }
             },
             std::make_pair(oid_, newBlocks[i]));
       }
   
       rt::waitForCompletion(handle);
   
       capacity_ += kBlockSize * blocksToAllocate;
     }
   
     void _clear() {
       for (auto block : dataBlocks_) {
         for (T *toDestroy = block; toDestroy < block + kBlockSize; ++toDestroy) {
           std::allocator_traits<allocator_type>::destroy(allocator_, toDestroy);
         }
         std::allocator_traits<allocator_type>::deallocate(allocator_, block,
                                                           kBlockSize);
       }
       dataBlocks_.clear();
     }
   
     template <typename ApplyFunT, typename... Args, std::size_t... is>
     static void CallApplyFun(const ObjectID &oid, const size_type position,
                              ApplyFunT function, std::tuple<Args...> &args,
                              std::index_sequence<is...>) {
       auto This = Vector<T, Allocator>::GetPtr(oid);
       auto blockOffsetPair = This->_blockOffsetFromPosition(position);
       size_type localBlock =
           This->_globlalBlockToLocalBlock(blockOffsetPair.first);
       value_type &element =
           This->dataBlocks_.at(localBlock)[blockOffsetPair.second];
       function(position, element, std::get<is>(args)...);
     }
   
     template <typename Tuple, typename... Args>
     static void ApplyFunWrapper(const Tuple &args) {
       constexpr auto Size = std::tuple_size<
           typename std::decay<decltype(std::get<3>(args))>::type>::value;
       Tuple &tuple = const_cast<Tuple &>(args);
       CallApplyFun(std::get<0>(tuple), std::get<1>(tuple), std::get<2>(tuple),
                    std::get<3>(tuple), std::make_index_sequence<Size>{});
     }
   
     template <typename ApplyFunT, typename... Args, std::size_t... is>
     static void AsyncCallApplyFun(rt::Handle &handle, const ObjectID &oid,
                                   const size_type position, ApplyFunT function,
                                   std::tuple<Args...> &args,
                                   std::index_sequence<is...>) {
       auto This = Vector<T, Allocator>::GetPtr(oid);
       auto blockOffsetPair = This->_blockOffsetFromPosition(position);
       size_type localBlock =
           This->_globlalBlockToLocalBlock(blockOffsetPair.first);
       value_type &element = This->dataBlocks_[localBlock][blockOffsetPair.second];
       function(handle, position, element, std::get<is>(args)...);
     }
   
     template <typename Tuple, typename... Args>
     static void AsyncApplyFunWrapper(rt::Handle &handle, const Tuple &args) {
       constexpr auto Size = std::tuple_size<
           typename std::decay<decltype(std::get<3>(args))>::type>::value;
       Tuple &tuple = const_cast<Tuple &>(args);
       AsyncCallApplyFun(handle, std::get<0>(tuple), std::get<1>(tuple),
                         std::get<2>(tuple), std::get<3>(tuple),
                         std::make_index_sequence<Size>{});
     }
   
     template <typename ApplyFunT, typename... Args, std::size_t... is>
     static void CallForEachInRangeFun(const size_t i, const ObjectID &oid,
                                       const size_t position, ApplyFunT function,
                                       std::tuple<Args...> &args,
                                       std::index_sequence<is...>) {
       // Get a local instance on the remote node.
       auto This = Vector<T, Allocator>::GetPtr(oid);
       auto blockOffsetPair = This->_blockOffsetFromPosition(position + i);
       size_type localBlock =
           This->_globlalBlockToLocalBlock(blockOffsetPair.first);
       value_type &element =
           This->dataBlocks_.at(localBlock)[blockOffsetPair.second];
       function(position + i, element, std::get<is>(args)...);
     }
   
     template <typename Tuple, typename... Args>
     static void ForEachInRangeFunWrapper(const Tuple &args, size_t i) {
       constexpr auto Size = std::tuple_size<
           typename std::decay<decltype(std::get<3>(args))>::type>::value;
       Tuple &tuple = const_cast<Tuple &>(args);
       CallForEachInRangeFun(i, std::get<0>(tuple), std::get<1>(tuple),
                             std::get<2>(tuple), std::get<3>(tuple),
                             std::make_index_sequence<Size>{});
     }
   
     template <typename ApplyFunT, typename... Args, std::size_t... is>
     static void AsyncCallForEachInRangeFun(rt::Handle &handle, const size_t i,
                                            const ObjectID &oid,
                                            const size_t position,
                                            ApplyFunT function,
                                            std::tuple<Args...> &args,
                                            std::index_sequence<is...>) {
       // Get a local instance on the remote node.
       auto This = Vector<T, Allocator>::GetPtr(oid);
       auto blockOffsetPair = This->_blockOffsetFromPosition(position + i);
       size_type localBlock =
           This->_globlalBlockToLocalBlock(blockOffsetPair.first);
       value_type &element =
           This->dataBlocks_.at(localBlock)[blockOffsetPair.second];
       function(handle, position + i, element, std::get<is>(args)...);
     }
   
     template <typename Tuple, typename... Args>
     static void AsyncForEachInRangeFunWrapper(rt::Handle &handle,
                                               const Tuple &args, size_t i) {
       constexpr auto Size = std::tuple_size<
           typename std::decay<decltype(std::get<3>(args))>::type>::value;
       Tuple &tuple = const_cast<Tuple &>(args);
       AsyncCallForEachInRangeFun(
           handle, i, std::get<0>(tuple), std::get<1>(tuple), std::get<2>(tuple),
           std::get<3>(tuple), std::make_index_sequence<Size>{});
     }
   
     using BuffersVector =
         impl::BuffersVector<std::tuple<size_t, T>, Vector<T, Allocator>>;
     friend class impl::BuffersVector<std::tuple<size_t, T>, Vector<T, Allocator>>;
   
     ObjectID oid_;
     rt::Locality mainLocality_;
     std::vector<value_type *> dataBlocks_;
     rt::Lock sizeCapacityLock_;
     size_type size_;
     size_type capacity_;
     allocator_type allocator_;
     BuffersVector buffers_;
   };
   
   template <typename T, typename Allocator>
   const typename Vector<T, Allocator>::size_type
       Vector<T, Allocator>::kBlockSize = constants::max((1024 << 6) / sizeof(T),
                                                         1lu);
   
   template <typename T, typename Allocator>
   template <typename ValueType>
   class Vector<T, Allocator>::Iterator
       : std::iterator<std::random_access_iterator_tag, ValueType> {
    public:
     Iterator(Vector<T, Allocator>::size_type n,
              Vector<T, Allocator>::ObjectID oid)
         : position_(n), oid_(oid) {}
   
     Iterator(const Iterator &itr) = default;
   
     Iterator &operator=(const Iterator &itr) = default;
   
     explicit operator bool() const {
       if (position_ == Vector<T, Allocator>::MaxSize())
         return false;
       else
         return true;
     }
   
     bool operator==(const Iterator &rhs) const {
       if (oid_ != rhs.oid_) return false;
   
       if (position_ != rhs.position_)
         return false;
       else
         return true;
     }
   
     bool operator!=(const Iterator &rhs) const { return !(*this == rhs); }
   
     Iterator &operator+=(const ptrdiff_t &movement) {
       position_ += movement;
       return *this;
     }
   
     Iterator &operator-=(const ptrdiff_t &movement) {
       position_ -= movement;
       return *this;
     }
   
     Iterator &operator++() {
       ++position_;
       return *this;
     }
     Iterator &operator--() {
       --position_;
       return *this;
     }
   
     Iterator operator++(int) {
       auto tmp(*this);
       ++position_;
       return tmp;
     }
   
     Iterator operator--(int) {
       auto tmp(*this);
       --position_;
       tmp;
     }
   
     Iterator operator+(const ptrdiff_t &movement) {
       Iterator tmp(*this);
       tmp.position_ += movement;
       return tmp;
     }
   
     Iterator operator-(const ptrdiff_t &movement) {
       Iterator tmp(*this);
       tmp.position_ -= movement;
       return tmp;
     }
   
     ptrdiff_t operator-(const Iterator &rhs) {
       return std::distance(rhs.position_, this->position_);
     }
   
     const value_type operator*() const {
       auto ptr = Vector<T, Allocator>::GetPtr(oid_);
       return ptr->At(position_);
     }
   
     const value_type *operator->() const {
       auto ptr = Vector<T, Allocator>::GetPtr(oid_);
       return &*(*this);
     }
   
    private:
     Vector<T, Allocator>::size_type position_;
     Vector<T, Allocator>::ObjectID oid_;
   };
   
   template <typename T, typename Allocator>
   typename Vector<T, Allocator>::size_type Vector<T, Allocator>::Size() const
       noexcept {
     if (rt::thisLocality() == mainLocality_) return size_;
   
     size_type size = 0;
     rt::executeAtWithRet(mainLocality_,
                          [](const ObjectID &oid, size_type *size) {
                            auto This = Vector<T, Allocator>::GetPtr(oid);
                            *size = This->size_;
                          },
                          oid_, &size);
     return size;
   }
   
   template <typename T, typename Allocator>
   typename Vector<T, Allocator>::size_type Vector<T, Allocator>::MaxSize() const
       noexcept {
     return std::numeric_limits<Vector<T, Allocator>::size_type>::max() - 1;
   }
   
   template <typename T, typename Allocator>
   typename Vector<T, Allocator>::size_type Vector<T, Allocator>::Capacity() const
       noexcept {
     if (rt::thisLocality() == mainLocality_) return capacity_;
   
     size_type capacity = 0;
     rt::executeAtWithRet(mainLocality_,
                          [](const ObjectID &oid, size_type *capacity) {
                            auto ptr = Vector<T, Allocator>::GetPtr(oid);
                            *capacity = ptr->capacity_;
                          },
                          oid_, &capacity);
     return capacity;
   }
   
   template <typename T, typename Allocator>
   bool Vector<T, Allocator>::Empty() const noexcept {
     return Size() == 0;
   }
   
   template <typename T, typename Allocator>
   void Vector<T, Allocator>::Reserve(Vector<T, Allocator>::size_type n) {
     rt::executeAt(mainLocality_,
                   [](const std::pair<ObjectID, size_type> &args) {
                     auto This = Vector<T, Allocator>::GetPtr(args.first);
                     auto n = args.second;
   
                     std::lock_guard<rt::Lock> _(This->sizeCapacityLock_);
                     This->_reserve(n);
                   },
                   std::make_pair(oid_, n));
   }
   
   template <typename T, typename Allocator>
   void Vector<T, Allocator>::Resize(Vector<T, Allocator>::size_type n) {
     rt::executeAt(mainLocality_,
                   [](const std::pair<ObjectID, size_type> &args) {
                     auto This = Vector<T, Allocator>::GetPtr(args.first);
                     auto n = args.second;
   
                     std::lock_guard<rt::Lock> _(This->sizeCapacityLock_);
                     if (n <= This->size_) return;
   
                     This->_reserve(n);
                     This->size_ = n;
                   },
                   std::make_pair(oid_, n));
   }
   
   template <typename T, typename Allocator>
   typename Vector<T, Allocator>::value_type Vector<T, Allocator>::At(
       Vector<T, Allocator>::size_type n) const {
     rt::Locality target(0);
     size_t blockNumber(0);
     size_t offset(0);
     std::tie(target, blockNumber, offset) = _targetFromPosition(n, kBlockSize);
   
     if (target == rt::thisLocality()) {
       size_type localBlock = _globlalBlockToLocalBlock(blockNumber);
       return dataBlocks_[localBlock][offset];
     } else {
       T value;
       rt::executeAtWithRet(
           target,
           [](const std::tuple<ObjectID, size_type, size_type> &args, T *result) {
             auto This = Vector<T, Allocator>::GetPtr(std::get<0>(args));
             size_type localBlock =
                 This->_globlalBlockToLocalBlock(std::get<1>(args));
   
             *result = This->dataBlocks_[localBlock][std::get<2>(args)];
           },
           std::make_tuple(oid_, blockNumber, offset), &value);
       return value;
     }
   }
   
   template <typename T, typename Allocator>
   typename Vector<T, Allocator>::value_type Vector<T, Allocator>::operator[](
       Vector<T, Allocator>::size_type n) const {
     return At(n);
   }
   
   template <typename T, typename Allocator>
   typename Vector<T, Allocator>::value_type Vector<T, Allocator>::Front() const {
     return At(0);
   }
   
   template <typename T, typename Allocator>
   typename Vector<T, Allocator>::value_type Vector<T, Allocator>::Back() const {
     size_type last_position = Size() - 1;
     return At(last_position);
   }
   
   template <typename T, typename Allocator>
   void Vector<T, Allocator>::AsyncAt(
       rt::Handle &handle, Vector<T, Allocator>::size_type n,
       Vector<T, Allocator>::value_type *result) const {
     rt::Locality target(0);
     size_type blockNumber(0);
     size_type offset(0);
     std::tie(target, blockNumber, offset) = _targetFromPosition(n, kBlockSize);
   
     rt::asyncExecuteAtWithRet(
         handle, target,
         [](rt::Handle &handle,
            const std::tuple<ObjectID, size_type, size_type> &args, T *result) {
           auto This = Vector<T, Allocator>::GetPtr(std::get<0>(args));
           size_type localBlock =
               This->_globlalBlockToLocalBlock(std::get<1>(args));
   
           *result = This->dataBlocks_[localBlock][std::get<2>(args)];
         },
         std::make_tuple(oid_, blockNumber, offset), result);
   }
   
   template <typename T, typename Allocator>
   void Vector<T, Allocator>::Clear() noexcept {
     rt::executeAt(mainLocality_,
                   [](const ObjectID &args) {
                     auto This = Vector<T, Allocator>::GetPtr(args);
                     std::lock_guard<rt::Lock> _(This->sizeCapacityLock_);
   
                     This->size_ = 0;
                     This->capacity_ = 0;
   
                     rt::executeOnAll(
                         [](const ObjectID &args) {
                           auto This = Vector<T, Allocator>::GetPtr(args);
                           This->_clear();
                         },
                         args);
                   },
                   oid_);
   }
   
   template <typename T, typename Allocator>
   void Vector<T, Allocator>::PushBack(
       const typename Vector<T, Allocator>::value_type &value) {
     size_type newSize(0);
   
     rt::executeAtWithRet(mainLocality_,
                          [](const ObjectID &args, size_type *size) {
                            auto This = Vector<T, Allocator>::GetPtr(args);
                            std::lock_guard<rt::Lock> _(This->sizeCapacityLock_);
   
                            *size = ++This->size_;
                            if (This->size_ > This->capacity_)
                              This->_reserve(This->size_);
                          },
                          oid_, &newSize);
   
     size_type position = newSize - 1;
   
     rt::Locality target(0);
     size_t blockNumber(0);
     size_t offset(0);
     std::tie(target, blockNumber, offset) =
         _targetFromPosition(position, kBlockSize);
   
     if (target == rt::thisLocality()) {
       size_type localBlock = _globlalBlockToLocalBlock(blockNumber);
       dataBlocks_[localBlock][offset] = value;
     } else {
       using MessageTuple = std::tuple<ObjectID, size_type, size_type, value_type>;
       rt::executeAt(target,
                     [](const MessageTuple &args) {
                       auto This = Vector<T, Allocator>::GetPtr(std::get<0>(args));
                       size_type localBlock =
                           This->_globlalBlockToLocalBlock(std::get<1>(args));
   
                       This->dataBlocks_[localBlock][std::get<2>(args)] =
                           std::get<3>(args);
                     },
                     std::make_tuple(oid_, blockNumber, offset, value));
     }
   }
   
   template <typename T, typename Allocator>
   typename Vector<T, Allocator>::iterator Vector<T, Allocator>::InsertAt(
       Vector<T, Allocator>::size_type position,
       const Vector<T, Allocator>::value_type &value) {
     rt::Locality target(0);
     size_t blockNumber(0);
     size_t offset(0);
     std::tie(target, blockNumber, offset) =
         _targetFromPosition(position, kBlockSize);
   
     if (target == rt::thisLocality()) {
       size_type localBlock = _globlalBlockToLocalBlock(blockNumber);
       dataBlocks_[localBlock][offset] = value;
     } else {
       using MessageTuple = std::tuple<ObjectID, size_type, size_type, value_type>;
       rt::executeAt(target,
                     [](const MessageTuple &args) {
                       auto This = Vector<T, Allocator>::GetPtr(std::get<0>(args));
                       size_type localBlock =
                           This->_globlalBlockToLocalBlock(std::get<1>(args));
   
                       This->dataBlocks_[localBlock][std::get<2>(args)] =
                           std::get<3>(args);
                     },
                     std::make_tuple(oid_, blockNumber, offset, value));
     }
   
     return Vector<T, Allocator>::iterator(position, GetGlobalID());
   }
   
   template <typename T, typename Allocator>
   template <typename IteratorType>
   typename Vector<T, Allocator>::iterator Vector<T, Allocator>::InsertAt(
       Vector<T, Allocator>::size_type position, IteratorType begin,
       IteratorType end) {
     rt::Handle handle;
   
     AsyncInsertAt(handle, position, begin, end);
   
     rt::waitForCompletion(handle);
   
     return Vector<T, Allocator>::iterator(position, GetGlobalID());
   }
   
   template <typename T, typename Allocator>
   void Vector<T, Allocator>::AsyncInsertAt(
       rt::Handle &handle, Vector<T, Allocator>::size_type position,
       const Vector<T, Allocator>::value_type &value) {
     size_type currentSize(0);
   
     rt::Locality target(0);
     size_t blockNumber(0);
     size_t offset(0);
     std::tie(target, blockNumber, offset) =
         _targetFromPosition(position, kBlockSize);
   
     if (target == rt::thisLocality()) {
       size_type localBlock = _globlalBlockToLocalBlock(blockNumber);
       dataBlocks_[localBlock][offset] = value;
     } else {
       using MessageTuple = std::tuple<ObjectID, size_type, size_type, value_type>;
       rt::asyncExecuteAt(
           handle, target,
           [](rt::Handle &, const MessageTuple &args) {
             auto This = Vector<T, Allocator>::GetPtr(std::get<0>(args));
             size_type localBlock =
                 This->_globlalBlockToLocalBlock(std::get<1>(args));
   
             This->dataBlocks_[localBlock][std::get<2>(args)] = std::get<3>(args);
           },
           std::make_tuple(oid_, blockNumber, offset, value));
     }
   }
   
   template <typename T, typename Allocator>
   template <typename IteratorType>
   void Vector<T, Allocator>::AsyncInsertAt(
       rt::Handle &handle, Vector<T, Allocator>::size_type position,
       IteratorType begin, IteratorType end) {
     size_type newElements = std::distance(begin, end);
     size_type newSize(0);
   
     rt::executeAtWithRet(
         mainLocality_,
         [](const std::tuple<ObjectID, size_type, size_type> &args,
            size_type *size) {
           auto This = Vector<T, Allocator>::GetPtr(std::get<0>(args));
           std::lock_guard<rt::Lock> _(This->sizeCapacityLock_);
           auto position = std::get<1>(args);
           auto newElements = std::get<2>(args);
   
           if (position > This->size_ || position + newElements <= This->size_) {
             *size = This->size_;
             return;
           }
   
           size_type end = position + newElements;
           size_type growth = This->size_ > end ? 0 : end - This->size_;
   
           This->size_ += growth;
           if (This->size_ > This->capacity_) This->_reserve(This->size_);
   
           *size = This->size_;
         },
         std::make_tuple(oid_, position, newElements), &newSize);
   
     // Trying to insert in an invalid position.
     if (position > newSize) {
       throw std::out_of_range("AsyncInsertAt: position out of range");
     }
   
     size_type startingPoint = newSize - newElements;
   
     constexpr size_t kNumElements = 4000 / sizeof(value_type);
     static_assert(kNumElements >= 1,
                   "We can't do this due to a series of unfortunate events");
   
     struct InsertMessage {
       ObjectID objID;
       size_type startPosition;
       size_type numElements;
       value_type elements[kNumElements];
   
       InsertMessage()
           : objID(ObjectID::kNullID), startPosition(0), numElements(0) {}
     };
   
     auto insertFunction = [](rt::Handle &, const InsertMessage &args) {
       auto This = Vector<T, Allocator>::GetPtr(args.objID);
       auto blockOffsetPair = This->_blockOffsetFromPosition(args.startPosition);
       size_type localBlock =
           This->_globlalBlockToLocalBlock(blockOffsetPair.first);
       std::copy(&args.elements[0], &args.elements[args.numElements],
                 &This->dataBlocks_.at(localBlock)[blockOffsetPair.second]);
     };
   
     InsertMessage args;
     size_t spaceLeftInBlock = kBlockSize - (startingPoint % kBlockSize);
     // first block
     args.objID = oid_;
     args.startPosition = startingPoint;
     args.numElements =
         std::min(newElements, std::min(kNumElements, spaceLeftInBlock));
   
     std::copy(begin, begin + args.numElements, args.elements);
   
     rt::Locality target(0);
     size_t blockNumber(0);
     size_t offset(0);
     std::tie(target, blockNumber, offset) =
         _targetFromPosition(position, kBlockSize);
   
     rt::asyncExecuteAt(handle, target, insertFunction, args);
   
     begin += args.numElements;
     newElements -= args.numElements;
     startingPoint += args.numElements;
   
     // inner blocks
     while (begin != end && newElements > 0) {
       std::tie(target, blockNumber, offset) =
           _targetFromPosition(startingPoint, kBlockSize);
       args.startPosition = startingPoint;
   
       spaceLeftInBlock = kBlockSize - (startingPoint % kBlockSize);
       args.numElements =
           std::min(newElements, std::min(kNumElements, spaceLeftInBlock));
   
       std::copy(begin, begin + args.numElements, args.elements);
       rt::asyncExecuteAt(handle, target, insertFunction, args);
   
       begin += args.numElements;
       newElements -= args.numElements;
       startingPoint += args.numElements;
     }
   }
   
   template <typename T, typename Allocator>
   void Vector<T, Allocator>::BufferedInsertAt(const size_type position,
                                               const value_type &value) {
     rt::Locality target(0);
     size_t blockNumber(0);
     size_t offset(0);
     std::tie(target, blockNumber, offset) =
         _targetFromPosition(position, kBlockSize);
   
     if (target == rt::thisLocality()) {
       size_type localBlock = _globlalBlockToLocalBlock(blockNumber);
       dataBlocks_[localBlock][offset] = value;
     } else {
       buffers_.Insert(std::make_tuple(position, value), target);
     }
   }
   
   template <typename T, typename Allocator>
   void Vector<T, Allocator>::BufferedAsyncInsertAt(rt::Handle &handle,
                                                    const size_type position,
                                                    const value_type &value) {
     rt::Locality target(0);
     size_t blockNumber(0);
     size_t offset(0);
     std::tie(target, blockNumber, offset) =
         _targetFromPosition(position, kBlockSize);
   
     if (target == rt::thisLocality()) {
       size_type localBlock = _globlalBlockToLocalBlock(blockNumber);
       dataBlocks_[localBlock][offset] = value;
     } else {
       buffers_.AsyncInsert(handle, std::make_tuple(position, value), target);
     }
   }
   
   template <typename T, typename Allocator>
   template <typename ApplyFunT, typename... Args>
   void Vector<T, Allocator>::Apply(const Vector<T, Allocator>::size_type position,
                                    ApplyFunT &&function, Args &... args) {
     rt::Locality target(0);
     size_t blockNumber(0);
     size_t offset(0);
     std::tie(target, blockNumber, offset) =
         _targetFromPosition(position, kBlockSize);
   
     using FunctionTy = void (*)(size_type, T &, Args & ...);
     FunctionTy fn = std::forward<decltype(function)>(function);
     using ArgsTuple =
         std::tuple<ObjectID, size_t, FunctionTy, std::tuple<Args...>>;
     ArgsTuple argsTuple{oid_, position, fn, std::tuple<Args...>(args...)};
   
     rt::executeAt(target, ApplyFunWrapper<ArgsTuple, Args...>, argsTuple);
   }
   
   template <typename T, typename Allocator>
   template <typename ApplyFunT, typename... Args>
   void Vector<T, Allocator>::AsyncApply(
       rt::Handle &handle, const Vector<T, Allocator>::size_type position,
       ApplyFunT &&function, Args &... args) {
     rt::Locality target(0);
     size_t blockNumber(0);
     size_t offset(0);
     std::tie(target, blockNumber, offset) =
         _targetFromPosition(position, kBlockSize);
   
     using FunctionTy = void (*)(rt::Handle &, size_type, T &, Args & ...);
     FunctionTy fn = std::forward<decltype(function)>(function);
     using ArgsTuple =
         std::tuple<ObjectID, size_t, FunctionTy, std::tuple<Args...>>;
     ArgsTuple argsTuple{oid_, position, fn, std::tuple<Args...>(args...)};
   
     rt::asyncExecuteAt(handle, target, AsyncApplyFunWrapper<ArgsTuple, Args...>,
                        argsTuple);
   }
   
   template <typename T, typename Allocator>
   template <typename ApplyFunT, typename... Args>
   void Vector<T, Allocator>::ForEachInRange(const size_type begin,
                                             const size_type end,
                                             ApplyFunT &&function,
                                             Args &... args) {
     using FunctionTy = void (*)(size_type, value_type &, Args & ...);
   
     FunctionTy fn = std::forward<decltype(function)>(function);
     using ArgsTuple =
         std::tuple<ObjectID, size_t, FunctionTy, std::tuple<Args...>>;
   
     size_type numElements = end - begin;
     size_type start = begin;
   
     // first block
     size_type blockSize =
         std::min(numElements, kBlockSize - (start % kBlockSize));
   
     rt::Locality target(0);
     std::tie(target, std::ignore, std::ignore) =
         _targetFromPosition(start, kBlockSize);
   
     ArgsTuple argsTuple{oid_, start, fn, std::tuple<Args...>(args...)};
   
     rt::forEachAt(target, ForEachInRangeFunWrapper<ArgsTuple, Args...>, argsTuple,
                   blockSize);
   
     start += blockSize;
     numElements -= blockSize;
   
     // inner blocks
     while (start != end && numElements > 0) {
       std::tie(target, std::ignore, std::ignore) =
           _targetFromPosition(start, kBlockSize);
       blockSize = std::min(kBlockSize, numElements);
   
       std::get<1>(argsTuple) = start;
       rt::forEachAt(target, ForEachInRangeFunWrapper<ArgsTuple, Args...>,
                     argsTuple, blockSize);
   
       start += blockSize;
       numElements -= blockSize;
     }
   }
   
   template <typename T, typename Allocator>
   template <typename ApplyFunT, typename... Args>
   void Vector<T, Allocator>::AsyncForEachInRange(rt::Handle &handle,
                                                  const size_type begin,
                                                  const size_type end,
                                                  ApplyFunT &&function,
                                                  Args &... args) {
     using FunctionTy =
         void (*)(rt::Handle &, size_type, value_type &, Args & ...);
   
     FunctionTy fn = std::forward<decltype(function)>(function);
     using ArgsTuple =
         std::tuple<ObjectID, size_t, FunctionTy, std::tuple<Args...>>;
   
     size_type numElements = end - begin;
     size_type start = begin;
   
     // first block
     size_type blockSize =
         std::min(numElements, kBlockSize - (start % kBlockSize));
   
     rt::Locality target(0);
     std::tie(target, std::ignore, std::ignore) =
         _targetFromPosition(start, kBlockSize);
   
     ArgsTuple argsTuple{oid_, start, fn, std::tuple<Args...>(args...)};
   
     rt::asyncForEachAt(handle, target,
                        AsyncForEachInRangeFunWrapper<ArgsTuple, Args...>,
                        argsTuple, blockSize);
   
     start += blockSize;
     numElements -= blockSize;
   
     // inner blocks
     while (start != end && numElements > 0) {
       std::tie(target, std::ignore, std::ignore) =
           _targetFromPosition(start, kBlockSize);
       blockSize = std::min(kBlockSize, numElements);
   
       std::get<1>(argsTuple) = start;
       rt::asyncForEachAt(handle, target,
                          AsyncForEachInRangeFunWrapper<ArgsTuple, Args...>,
                          argsTuple, blockSize);
   
       start += blockSize;
       numElements -= blockSize;
     }
   }
   
   }  // namespace shad
   
   #endif  // INCLUDE_SHAD_DATA_STRUCTURES_VECTOR_H_