chase_lev_deque.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 <atomic>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <new>
#include <type_traits>
#include <utility>
 
#include <dispenso/platform.h>
#include <dispenso/tsan_annotations.h>
 
namespace dispenso {
 
template <typename T, size_t Capacity = 32>
#if DISPENSO_HAS_CONCEPTS
  requires std::is_trivially_copyable_v<T>
#endif
class ChaseLevDeque {
  static_assert(Capacity >= 1, "ChaseLevDeque capacity must be at least 1");
  static_assert((Capacity & (Capacity - 1)) == 0, "ChaseLevDeque capacity must be a power of two");
#if !DISPENSO_HAS_CONCEPTS
  static_assert(
      std::is_trivially_copyable<T>::value,
      "ChaseLevDeque requires a trivially copyable element type. "
      "Wrap move-only payloads in a raw pointer or POD container.");
#endif
 
 public:
  using value_type = T;
  using size_type = size_t;
 
  ChaseLevDeque() = default;
 
  ChaseLevDeque(const ChaseLevDeque&) = delete;
  ChaseLevDeque& operator=(const ChaseLevDeque&) = delete;
  ChaseLevDeque(ChaseLevDeque&&) = delete;
  ChaseLevDeque& operator=(ChaseLevDeque&&) = delete;
 
  ~ChaseLevDeque() = default;
 
  // ---------------------------------------------------------------------------
  // Owner-only operations: push and pop.
  // ---------------------------------------------------------------------------
 
  bool try_push(const T& item) {
    const int64_t b = bottom_.load(std::memory_order_relaxed);
    const int64_t t = top_.load(std::memory_order_acquire);
    if (b - t >= static_cast<int64_t>(Capacity)) {
      return false;
    }
    // The slot write is ordered after a prior stealer's CAS on top_ (see try_steal):
    // a stealer that read this slot tentatively must have CAS'd top_ before any
    // subsequent push could wrap around to this index. TSAN doesn't model the
    // through-CAS HB chain on non-atomic slot accesses, so annotate as benign.
    DISPENSO_TSAN_ANNOTATE_IGNORE_WRITES_BEGIN();
    *slotPtr(b) = item;
    DISPENSO_TSAN_ANNOTATE_IGNORE_WRITES_END();
    // Release on bottom_ publishes the slot write to any acquire load of bottom_
    // (i.e., the load in try_steal).
    bottom_.store(b + 1, std::memory_order_release);
    return true;
  }
 
  bool try_pop(T& out) {
    const int64_t b = bottom_.load(std::memory_order_relaxed) - 1;
    bottom_.store(b, std::memory_order_relaxed);
    // Seq-cst fence orders our bottom store before the top load, matching steal's fence.
    std::atomic_thread_fence(std::memory_order_seq_cst);
    int64_t t = top_.load(std::memory_order_relaxed);
 
    if (t > b) {
      // Empty. Restore bottom.
      bottom_.store(b + 1, std::memory_order_relaxed);
      return false;
    }
    // Read directly into out. Safe because T is trivially copyable; on CAS loss,
    // out is unspecified per the try_* contract.
    out = *slotPtr(b);
    if (t < b) {
      // Multiple elements remained; no race possible at this slot.
      return true;
    }
    // t == b: last element. Race with stealers via CAS on top.
    bottom_.store(b + 1, std::memory_order_relaxed);
    return top_.compare_exchange_strong(
        t, t + 1, std::memory_order_seq_cst, std::memory_order_relaxed);
  }
 
  bool try_pop_into(T* storage) {
    const int64_t b = bottom_.load(std::memory_order_relaxed) - 1;
    bottom_.store(b, std::memory_order_relaxed);
    std::atomic_thread_fence(std::memory_order_seq_cst);
    int64_t t = top_.load(std::memory_order_relaxed);
 
    if (t > b) {
      bottom_.store(b + 1, std::memory_order_relaxed);
      return false;
    }
    if (t < b) {
      std::memcpy(storage, slotPtr(b), sizeof(T));
      return true;
    }
    bottom_.store(b + 1, std::memory_order_relaxed);
    alignas(T) char tmp[sizeof(T)];
    std::memcpy(tmp, slotPtr(b), sizeof(T));
    if (!top_.compare_exchange_strong(
            t, t + 1, std::memory_order_seq_cst, std::memory_order_relaxed)) {
      return false;
    }
    std::memcpy(storage, tmp, sizeof(T));
    return true;
  }
 
  // ---------------------------------------------------------------------------
  // Stealer operations: callable from any thread, including the owner.
  // ---------------------------------------------------------------------------
 
  bool try_steal(T& out) {
    int64_t t = top_.load(std::memory_order_acquire);
    // Seq-cst fence pairs with the fence in try_pop: ensures we observe an updated
    // bottom if the owner has decremented past us.
    std::atomic_thread_fence(std::memory_order_seq_cst);
    const int64_t b = bottom_.load(std::memory_order_acquire);
    if (t >= b) {
      return false;
    }
    // Read directly into out before CAS. Safe because T is trivially copyable —
    // slot[t] is unchanged whether we win or lose. On CAS loss, out is unspecified
    // per the try_* contract. The non-atomic slot read is ordered against the
    // owner's earlier slot write via the chain: owner writes slot[b] → owner
    // bottom_.store(b+1, release) → our bottom_.load(acquire) above synchronizes
    // with that release → slot read happens-before stealer CAS. The seq_cst CAS
    // on top_ orders us against other concurrent stealers. TSAN doesn't model
    // this chain through non-atomic reads, so annotate as benign.
    DISPENSO_TSAN_ANNOTATE_IGNORE_READS_BEGIN();
    out = *slotPtr(t);
    DISPENSO_TSAN_ANNOTATE_IGNORE_READS_END();
    return top_.compare_exchange_strong(
        t, t + 1, std::memory_order_seq_cst, std::memory_order_relaxed);
  }
 
  bool try_steal_into(T* storage) {
    int64_t t = top_.load(std::memory_order_acquire);
    std::atomic_thread_fence(std::memory_order_seq_cst);
    const int64_t b = bottom_.load(std::memory_order_acquire);
    if (t >= b) {
      return false;
    }
    // Tentative copy via memcpy (T is trivially copyable). Discarded on CAS loss.
    // See try_steal for the reasoning behind the TSAN annotations.
    alignas(T) char tmp[sizeof(T)];
    DISPENSO_TSAN_ANNOTATE_IGNORE_READS_BEGIN();
    std::memcpy(tmp, slotPtr(t), sizeof(T));
    DISPENSO_TSAN_ANNOTATE_IGNORE_READS_END();
    if (!top_.compare_exchange_strong(
            t, t + 1, std::memory_order_seq_cst, std::memory_order_relaxed)) {
      return false;
    }
    std::memcpy(storage, tmp, sizeof(T));
    return true;
  }
 
  // ---------------------------------------------------------------------------
  // Observers. Snapshots only; results may be immediately stale.
  // ---------------------------------------------------------------------------
 
  bool empty() const {
    const int64_t b = bottom_.load(std::memory_order_acquire);
    const int64_t t = top_.load(std::memory_order_acquire);
    return b <= t;
  }
 
  size_type size() const {
    const int64_t b = bottom_.load(std::memory_order_acquire);
    const int64_t t = top_.load(std::memory_order_acquire);
    return b > t ? static_cast<size_type>(b - t) : 0;
  }
 
  static constexpr size_type capacity() noexcept {
    return Capacity;
  }
 
 private:
  static constexpr size_t kMask = Capacity - 1;
  static constexpr size_t kStorageAlign =
      (alignof(T) > kCacheLineSize) ? alignof(T) : kCacheLineSize;
 
  T* slotPtr(int64_t index) {
    return reinterpret_cast<T*>(&storage_[(static_cast<size_t>(index) & kMask) * sizeof(T)]);
  }
 
  alignas(kCacheLineSize) std::atomic<int64_t> top_{0};
 
  alignas(kCacheLineSize) std::atomic<int64_t> bottom_{0};
 
  alignas(kStorageAlign) char storage_[sizeof(T) * Capacity];
};
 
} // namespace dispenso