concurrent_vector.h

Go to the documentation of this file.

/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 *
 * This source code is licensed under the MIT license found in the
 * LICENSE file in the root directory of this source tree.
 */
 
#pragma once
 
#include <algorithm>
#include <cassert>
#include <climits>
#include <cstring>
#include <initializer_list>
#include <stdexcept>
#include <utility>
 
#include <dispenso/detail/math.h>
#include <dispenso/platform.h>
#include <dispenso/tsan_annotations.h>
 
namespace dispenso {
 
enum class ConcurrentVectorReallocStrategy { kFullBufferAhead, kHalfBufferAhead, kAsNeeded };
 
struct ReserveTagS {};
constexpr ReserveTagS ReserveTag;
 
// Textual inclusion.  Includes undocumented implementation details, e.g. iterators.
#include <dispenso/detail/concurrent_vector_impl.h>
 
template <typename T>
struct DefaultConcurrentVectorSizeTraits {
  static constexpr size_t kDefaultCapacity = (sizeof(T) >= 256) ? 2 : 512 / sizeof(T);
 
  static constexpr size_t kMaxVectorSize =
      (size_t{1} << (sizeof(size_t) * CHAR_BIT > 47 ? 47 : sizeof(size_t) * CHAR_BIT - 1)) /
      sizeof(T);
};
 
struct DefaultConcurrentVectorTraits {
  static constexpr bool kPreferBuffersInline = true;
 
  static constexpr ConcurrentVectorReallocStrategy kReallocStrategy =
      ConcurrentVectorReallocStrategy::kAsNeeded;
 
  static constexpr bool kIteratorPreferSpeed = true;
};
 
template <
    typename T,
    typename Traits = DefaultConcurrentVectorTraits,
    typename SizeTraits = DefaultConcurrentVectorSizeTraits<T>>
class ConcurrentVector {
 public:
  using value_type = T;
  using reference = T&;
  using const_reference = const T&;
  using size_type = size_t;
  using difference_type = ssize_t;
  using reference_type = T&;
  using const_reference_type = const T&;
  using pointer = T*;
  using const_pointer = const T*;
  using iterator = std::conditional_t<
      Traits::kIteratorPreferSpeed,
      cv::ConcurrentVectorIterator<ConcurrentVector<T, Traits>, T, false>,
      cv::CompactCVecIterator<ConcurrentVector<T, Traits>, T, false>>;
  using const_iterator = std::conditional_t<
      Traits::kIteratorPreferSpeed,
      cv::ConcurrentVectorIterator<ConcurrentVector<T, Traits>, T, true>,
      cv::CompactCVecIterator<ConcurrentVector<T, Traits>, T, true>>;
  using reverse_iterator = std::reverse_iterator<iterator>;
  using const_reverse_iterator = std::reverse_iterator<const_iterator>;
 
  ConcurrentVector() : ConcurrentVector(SizeTraits::kDefaultCapacity / 2, ReserveTag) {}
 
  ConcurrentVector(size_t startCapacity, ReserveTagS)
      : firstBucketShift_(
            detail::log2(
                detail::nextPow2(std::max(startCapacity, SizeTraits::kDefaultCapacity / 2)))),
        firstBucketLen_(size_type{1} << firstBucketShift_) {
    T* firstTwo = cv::alloc<T>(2 * firstBucketLen_);
    buffers_[0].store(firstTwo, std::memory_order_release);
    buffers_[1].store(firstTwo + firstBucketLen_, std::memory_order_release);
  }
 
  explicit ConcurrentVector(size_t startSize) : ConcurrentVector(startSize, ReserveTag) {
    size_.store(startSize, std::memory_order_relaxed);
    T* buf = buffers_[0].load(std::memory_order_relaxed);
    for (size_t i = 0; i < startSize; ++i) {
      new (buf + i) T();
    }
  }
 
  ConcurrentVector(size_t startSize, const T& defaultValue)
      : ConcurrentVector(startSize, ReserveTag) {
    size_.store(startSize, std::memory_order_relaxed);
    T* buf = buffers_[0].load(std::memory_order_relaxed);
    for (size_t i = 0; i < startSize; ++i) {
      new (buf + i) T(defaultValue);
    }
  }
 
  template <typename InIterator>
  ConcurrentVector(InIterator start, InIterator end)
      : ConcurrentVector(std::distance(start, end), start, end) {}
 
  template <typename InIterator>
  ConcurrentVector(size_type startSize, InIterator start, InIterator end)
      : ConcurrentVector(startSize, ReserveTag) {
    size_.store(startSize, std::memory_order_relaxed);
    assert(std::distance(start, end) == static_cast<difference_type>(startSize));
    internalInit(start, end, begin());
  }
 
  ConcurrentVector(std::initializer_list<T> l)
      : ConcurrentVector(l.size(), std::begin(l), std::end(l)) {}
 
  ConcurrentVector(const ConcurrentVector& other)
      : ConcurrentVector(other.size(), other.cbegin(), other.cend()) {}
 
  ConcurrentVector(ConcurrentVector&& other)
      : buffers_(std::move(other.buffers_)),
        firstBucketShift_(other.firstBucketShift_),
        firstBucketLen_(other.firstBucketLen_),
        size_(other.size_.load(std::memory_order_relaxed)) {
    other.size_.store(0, std::memory_order_relaxed);
    // This is possibly unnecessary overhead, but enables the "other" vector to be in a valid,
    // usable state right away, no empty check or clear required, as it is for std::vector.
    T* firstTwo = cv::alloc<T>(2 * firstBucketLen_);
    other.buffers_[0].store(firstTwo, std::memory_order_relaxed);
    other.buffers_[1].store(firstTwo + firstBucketLen_, std::memory_order_relaxed);
  }
 
  ConcurrentVector& operator=(const ConcurrentVector& other) {
    if (&other == this) {
      return *this;
    }
 
    clear();
    reserve(other.size());
    size_.store(other.size(), std::memory_order_relaxed);
    internalInit(other.cbegin(), other.cend(), begin());
 
    return *this;
  }
 
  ConcurrentVector& operator=(ConcurrentVector&& other) {
    using std::swap;
    if (&other == this) {
      return *this;
    }
 
    clear();
    swap(firstBucketShift_, other.firstBucketShift_);
    swap(firstBucketLen_, other.firstBucketLen_);
    buffers_ = std::move(other.buffers_);
    size_t curLen = size_.load(std::memory_order_relaxed);
    size_.store(other.size_.load(std::memory_order_relaxed), std::memory_order_relaxed);
    other.size_.store(curLen, std::memory_order_relaxed);
 
    return *this;
  }
 
  void assign(size_type count, const T& value) {
    clear();
    reserve(count);
    size_.store(count, std::memory_order_relaxed);
    internalFillN(begin(), count, value);
  }
 
  template <
      typename It,
      typename = typename std::iterator_traits<It>::difference_type,
      typename = typename std::iterator_traits<It>::pointer,
      typename = typename std::iterator_traits<It>::reference,
      typename = typename std::iterator_traits<It>::value_type,
      typename = typename std::iterator_traits<It>::iterator_category>
  void assign(It start, It end) {
    clear();
    auto count = std::distance(start, end);
    reserve(count);
    size_.store(count, std::memory_order_relaxed);
    internalInit(start, end, begin());
  }
 
  void reserve(difference_type capacity) {
    buffers_.allocAsNecessary({0, 0, firstBucketLen_}, capacity, bucketAndSubIndex(capacity));
  }
 
  void resize(difference_type len) {
    difference_type curLen = static_cast<difference_type>(size_.load(std::memory_order_relaxed));
    if (curLen < len) {
      grow_to_at_least(len);
    } else if (curLen > len) {
      auto it = end();
      auto newEnd = begin() + len;
      do {
        --it;
        it->~T();
      } while (it != newEnd);
      size_.store(len, std::memory_order_relaxed);
    }
  }
 
  void resize(difference_type len, const T& value) {
    difference_type curLen = static_cast<difference_type>(size_.load(std::memory_order_relaxed));
    if (curLen < len) {
      grow_to_at_least(len, value);
    } else if (curLen > len) {
      auto it = end();
      auto newEnd = begin() + len;
      do {
        --it;
        it->~T();
      } while (it != newEnd);
      size_.store(len, std::memory_order_relaxed);
    }
  }
 
  size_type default_capacity() const {
    return 2 * firstBucketLen_;
  }
 
  size_type capacity() const {
    size_t cap = 2 * firstBucketLen_;
    for (size_t b = 2; b < kMaxBuffers; ++b) {
      if (!buffers_[b].load(std::memory_order_relaxed)) {
        break;
      }
      cap *= 2;
    }
    return cap;
  }
 
  void clear() {
    auto binfo = bucketAndSubIndex(size_.load(std::memory_order_relaxed));
    size_t len = binfo.bucketIndex;
    size_t cap = binfo.bucketCapacity;
    size_t b = binfo.bucket;
    do {
      T* buf = buffers_[b].load(std::memory_order_relaxed);
      T* t = buf + len;
 
      while (t != buf) {
        --t;
        t->~T();
      }
      cap >>= int{b > 1};
      len = cap;
    } while (b--);
    size_.store(0, std::memory_order_release);
  }
 
  void shrink_to_fit() {
    constexpr size_t kMaxExtra = 2;
    auto binfo = bucketAndSubIndex(size_.load(std::memory_order_relaxed));
 
    // We need to at least skip the first two buckets, since we have those as a single allocation.
    size_t startBucket = std::max<size_t>(2, binfo.bucket + kMaxExtra);
 
    for (size_t b = startBucket; b < kMaxBuffers; ++b) {
      T* ptr = buffers_[b].load(std::memory_order_relaxed);
      if (!ptr) {
        break;
      }
      if (buffers_.shouldDealloc(b)) {
        cv::dealloc<T>(ptr);
      }
      buffers_[b].store(nullptr, std::memory_order_release);
    }
  }
 
  ~ConcurrentVector() {
    clear();
    shrink_to_fit();
    cv::dealloc<T>(buffers_[0].load(std::memory_order_acquire));
  }
 
  iterator insert(const_iterator pos, const T& value) {
    auto it = insertPartial(pos);
    new (&*it) T(value);
    return it;
  }
 
  iterator insert(const_iterator pos, T&& value) {
    auto it = insertPartial(pos);
    new (&*it) T(std::move(value));
    return it;
  }
 
  iterator insert(const_iterator pos, size_type count, const T& value) {
    auto it = insertPartial(pos, count);
    std::fill_n(it, count, value);
    return it;
  }
 
  template <
      typename InputIt,
      typename = typename std::iterator_traits<InputIt>::difference_type,
      typename = typename std::iterator_traits<InputIt>::pointer,
      typename = typename std::iterator_traits<InputIt>::reference,
      typename = typename std::iterator_traits<InputIt>::value_type,
      typename = typename std::iterator_traits<InputIt>::iterator_category>
  iterator insert(const_iterator pos, InputIt first, InputIt last) {
    size_t len = std::distance(first, last);
    auto it = insertPartial(pos, len);
    std::copy_n(first, len, it);
    return it;
  }
 
  iterator insert(const_iterator pos, std::initializer_list<T> ilist) {
    return insert(pos, ilist.begin(), ilist.end());
  }
 
  iterator erase(const_iterator pos) {
    auto e = end();
    if (e == pos) {
      return e;
    }
    size_.fetch_sub(1, std::memory_order_relaxed);
    --e;
    if (e == pos) {
      e->~T();
      return e;
    }
    ++e;
    auto it = begin();
    it += (pos - it);
    return std::move(pos + 1, const_iterator(e), it);
  }
 
  iterator erase(const_iterator first, const_iterator last) {
    size_t len = std::distance(first, last);
    auto it = begin();
    size_t startIdx = first - it;
    if (len == 0) {
      return it + (startIdx + len);
    }
    it += startIdx;
 
    auto e_it = std::move(last, cend(), it);
 
    if (e_it < last) {
      // remove any values that were not already moved into
      do {
        --last;
        last->~T();
      } while (e_it != last);
    }
    size_.fetch_sub(len, std::memory_order_relaxed);
    return e_it;
  }
 
  iterator push_back(const T& val) {
    return emplace_back(val);
  }
 
  iterator push_back(T&& val) {
    return emplace_back(std::move(val));
  }
 
  template <typename... Args>
  iterator emplace_back(Args&&... args) {
    auto index = size_.fetch_add(1, std::memory_order_relaxed);
    auto binfo = bucketAndSubIndex(index);
 
    buffers_.allocAsNecessary(binfo);
 
    iterator ret{this, index, binfo};
 
    if (Traits::kIteratorPreferSpeed) {
      new (&*ret) T(std::forward<Args>(args)...);
    } else {
      new (buffers_[binfo.bucket] + binfo.bucketIndex) T(std::forward<Args>(args)...);
    }
 
    return ret;
  }
 
  template <typename Gen>
  iterator grow_by_generator(size_type delta, Gen gen) {
    iterator ret = growByUninitialized(delta);
    for (auto it = ret; delta--; ++it) {
      new (&*it) T(gen());
    }
    return ret;
  }
 
  iterator grow_by(size_type delta, const T& t) {
    iterator ret = growByUninitialized(delta);
    internalFillN(ret, delta, t);
    return ret;
  }
 
  iterator grow_by(size_type delta) {
    iterator ret = growByUninitialized(delta);
    internalFillDefaultN(ret, delta);
    return ret;
  }
 
  template <
      typename It,
      typename = typename std::iterator_traits<It>::difference_type,
      typename = typename std::iterator_traits<It>::pointer,
      typename = typename std::iterator_traits<It>::reference,
      typename = typename std::iterator_traits<It>::value_type,
      typename = typename std::iterator_traits<It>::iterator_category>
  iterator grow_by(It start, It end) {
    iterator ret = growByUninitialized(std::distance(start, end));
    internalInit(start, end, ret);
    return ret;
  }
 
  iterator grow_by(std::initializer_list<T> initList) {
    return grow_by(std::begin(initList), std::end(initList));
  }
 
  iterator grow_to_at_least(size_type n) {
    size_t curSize = size_.load(std::memory_order_relaxed);
    if (curSize < n) {
      return grow_by(n - curSize);
    }
    return {this, n - 1, bucketAndSubIndex(n - 1)};
  }
 
  iterator grow_to_at_least(size_type n, const T& t) {
    size_t curSize = size_.load(std::memory_order_relaxed);
    if (curSize < n) {
      return grow_by(n - curSize, t);
    }
    return {this, n - 1, bucketAndSubIndex(n - 1)};
  }
 
  void pop_back() {
    auto binfo = bucketAndSubIndex(size_.fetch_sub(1, std::memory_order_relaxed) - 1);
    T* elt = buffers_[binfo.bucket].load(std::memory_order_relaxed) + binfo.bucketIndex;
    elt->~T();
  }
 
  const T& operator[](size_type index) const {
    auto binfo = bucketAndSubIndexForIndex(index);
    T* buf = buffers_[binfo.bucket].load(std::memory_order_relaxed);
    return buf[binfo.bucketIndex];
  }
 
  T& operator[](size_type index) {
    auto binfo = bucketAndSubIndexForIndex(index);
    T* buf = buffers_[binfo.bucket].load(std::memory_order_relaxed);
    return buf[binfo.bucketIndex];
  }
 
  const T& at(size_type index) const {
#if defined(__cpp_exceptions)
    if (index >= size_.load(std::memory_order_relaxed)) {
      throw std::out_of_range("Index too large");
    }
#endif
    return operator[](index);
  }
 
  T& at(size_type index) {
#if defined(__cpp_exceptions)
    if (index >= size_.load(std::memory_order_relaxed)) {
      throw std::out_of_range("Index too large");
    }
#endif
    return operator[](index);
  }
 
  iterator begin() {
    return {this, 0, {0, 0, firstBucketLen_}};
  }
 
  iterator end() {
    size_t curSize = size_.load(std::memory_order_relaxed);
    auto binfo = bucketAndSubIndex(curSize);
    return {this, curSize, binfo};
  }
 
  const_iterator begin() const {
    return {this, 0, {0, 0, firstBucketLen_}};
  }
 
  const_iterator end() const {
    size_t curSize = size_.load(std::memory_order_relaxed);
    auto binfo = bucketAndSubIndex(curSize);
    return {this, curSize, binfo};
  }
 
  const_iterator cbegin() const {
    return {this, 0, {0, 0, firstBucketLen_}};
  }
 
  const_iterator cend() const {
    size_t curSize = size_.load(std::memory_order_relaxed);
    auto binfo = bucketAndSubIndex(curSize);
    return {this, curSize, binfo};
  }
 
  reverse_iterator rbegin() {
    return reverse_iterator(end());
  }
 
  reverse_iterator rend() {
    return reverse_iterator(begin());
  }
 
  const_reverse_iterator rbegin() const {
    return const_reverse_iterator(cend());
  }
 
  const_reverse_iterator rend() const {
    return const_reverse_iterator(cbegin());
  }
 
  bool empty() const {
    return size_.load(std::memory_order_relaxed) == 0;
  }
 
  constexpr size_type max_size() const noexcept {
    return Traits::kMaxVectorSize;
  }
 
  size_type size() const {
    return size_.load(std::memory_order_relaxed);
  }
 
  T& front() {
    return *buffers_[0].load(std::memory_order_relaxed);
  }
 
  const T& front() const {
    return *buffers_[0].load(std::memory_order_relaxed);
  }
 
  T& back() {
    return *(end() - 1);
  }
 
  const T& back() const {
    return *(end() - 1);
  }
 
  void swap(ConcurrentVector& oth) {
    using std::swap;
    swap(firstBucketShift_, oth.firstBucketShift_);
    swap(firstBucketLen_, oth.firstBucketLen_);
    size_t othlen = oth.size_.load(std::memory_order_relaxed);
    oth.size_.store(size_.load(std::memory_order_relaxed), std::memory_order_relaxed);
    size_.store(othlen, std::memory_order_relaxed);
 
    // okay, this relies on the fact that we're essentially swapping in the move operator.
    buffers_ = std::move(oth.buffers_);
  }
 
 private:
  DISPENSO_INLINE cv::BucketInfo bucketAndSubIndexForIndex(size_t index) const {
#if defined(__clang__)
    if (index < firstBucketLen_) {
      return {0, index, firstBucketLen_};
    }
 
    size_t l2idx = detail::log2(index);
    size_t bucket = (l2idx + 1) - firstBucketShift_;
    size_t bucketCapacity = size_t{1} << l2idx;
    size_t bucketIndex = index - bucketCapacity;
 
    return {bucket, bucketIndex, bucketCapacity};
#else
    return bucketAndSubIndex(index);
#endif // __clang__
  }
 
  DISPENSO_INLINE cv::BucketInfo bucketAndSubIndex(size_t index) const {
    size_t l2idx = detail::log2(index | 1);
    size_t bucket = (l2idx + 1) - firstBucketShift_;
    size_t bucketCapacity = size_t{1} << l2idx;
    size_t bucketIndex = index - bucketCapacity;
 
    bucket = index < firstBucketLen_ ? 0 : bucket;
    bucketIndex = index < firstBucketLen_ ? index : bucketIndex;
    bucketCapacity = index < firstBucketLen_ ? firstBucketLen_ : bucketCapacity;
 
    return {bucket, bucketIndex, bucketCapacity};
  }
 
  template <typename InIterator>
  void internalInit(InIterator start, InIterator end, iterator it) {
    while (start != end) {
      new (&*it) T(*start);
      ++it;
      ++start;
    }
  }
 
  void internalFillN(iterator it, size_t len, const T& value) {
    for (; len--; ++it) {
      new (&*it) T(value);
    }
  }
 
  void internalFillDefaultN(iterator it, size_t len) {
    for (; len--; ++it) {
      new (&*it) T();
    }
  }
 
  iterator growByUninitialized(size_type delta) {
    auto index = size_.fetch_add(delta, std::memory_order_relaxed);
    auto binfo = bucketAndSubIndex(index);
    buffers_.allocAsNecessary(binfo, delta, bucketAndSubIndex(index + delta));
    return {this, index, binfo};
  }
 
  iterator insertPartial(const_iterator pos) {
    auto e = end();
    auto index = size_.fetch_add(1, std::memory_order_relaxed);
    auto binfo = bucketAndSubIndex(index);
    buffers_.allocAsNecessary(binfo);
    new (&*e) T();
    return std::move_backward(pos, const_iterator(e), e + 1) - 1;
  }
 
  iterator insertPartial(const_iterator pos, size_t len) {
    auto e = end();
    auto index = size_.fetch_add(len, std::memory_order_relaxed);
    buffers_.allocAsNecessary(bucketAndSubIndex(index), len, bucketAndSubIndex(index + len));
    for (auto it = e + len; it != e;) {
      --it;
      new (&*it) T();
    }
    return std::move_backward(pos, const_iterator(e), e + len) - len;
  }
 
  alignas(kCacheLineSize) cv::ConVecBuffer<
      T,
      SizeTraits::kDefaultCapacity / 2,
      SizeTraits::kMaxVectorSize,
      Traits::kPreferBuffersInline,
      Traits::kReallocStrategy> buffers_;
  static constexpr size_t kMaxBuffers = decltype(buffers_)::kMaxBuffers;
 
  size_t firstBucketShift_;
  size_t firstBucketLen_;
 
  alignas(kCacheLineSize) std::atomic<size_t> size_{0};
 
  friend class cv::ConVecIterBase<ConcurrentVector<T, Traits>, T>;
  friend class cv::ConcurrentVectorIterator<ConcurrentVector<T, Traits>, T, false>;
  friend class cv::ConcurrentVectorIterator<ConcurrentVector<T, Traits>, T, true>;
};
 
template <typename T, class Traits1, class Traits2>
inline bool operator==(
    const ConcurrentVector<T, Traits1>& a,
    const ConcurrentVector<T, Traits2>& b) {
  return a.size() == b.size() && std::equal(a.begin(), a.end(), b.begin());
}
 
template <typename T, class Traits1, class Traits2>
inline bool operator!=(
    const ConcurrentVector<T, Traits1>& a,
    const ConcurrentVector<T, Traits2>& b) {
  return !(a == b);
}
 
template <typename T, class Traits1, class Traits2>
inline bool operator<(
    const ConcurrentVector<T, Traits1>& a,
    const ConcurrentVector<T, Traits2>& b) {
  return std::lexicographical_compare(a.begin(), a.end(), b.begin(), b.end());
}
 
template <typename T, class Traits1, class Traits2>
inline bool operator>(
    const ConcurrentVector<T, Traits1>& a,
    const ConcurrentVector<T, Traits2>& b) {
  return b < a;
}
 
template <typename T, class Traits1, class Traits2>
inline bool operator<=(
    const ConcurrentVector<T, Traits1>& a,
    const ConcurrentVector<T, Traits2>& b) {
  return !(b < a);
}
 
template <typename T, class Traits1, class Traits2>
inline bool operator>=(
    const ConcurrentVector<T, Traits1>& a,
    const ConcurrentVector<T, Traits2>& b) {
  return !(a < b);
}
 
template <typename T, class Traits>
inline void swap(ConcurrentVector<T, Traits>& a, ConcurrentVector<T, Traits>& b) {
  a.swap(b);
}
 
// Textual inclusion.  Includes undocumented implementation details, e.g. iterators.
#include <dispenso/detail/concurrent_vector_impl2.h>
 
} // namespace dispenso