Line data    Source code

       1           0 : // Distributed under the MIT License.
       2             : // See LICENSE.txt for details.
       3             : 
       4             : /*
       5             :  *The original code is distributed under the following copyright and license:
       6             :  *
       7             :  * Copyright (c) 2020 Erik Rigtorp <erik@rigtorp.se>
       8             :  *
       9             :  * Permission is hereby granted, free of charge, to any person obtaining a copy
      10             :  * of this software and associated documentation files (the "Software"), to deal
      11             :  * in the Software without restriction, including without limitation the rights
      12             :  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
      13             :  * copies of the Software, and to permit persons to whom the Software is
      14             :  * furnished to do so, subject to the following conditions:
      15             :  *
      16             :  * The above copyright notice and this permission notice shall be included in
      17             :  * all copies or substantial portions of the Software.
      18             :  *
      19             :  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
      20             :  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
      21             :  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
      22             :  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
      23             :  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
      24             :  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
      25             :  * SOFTWARE.
      26             :  *
      27             :  *
      28             :  * SXS Modifications:
      29             :  * 1. Casing to match SpECTRE conventions
      30             :  * 2. Static capacity
      31             :  * 3. Storage is std::array
      32             :  * 4. Switch to west-const
      33             :  *
      34             :  */
      35             : 
      36             : #pragma once
      37             : 
      38             : #include <atomic>
      39             : #include <cassert>
      40             : #include <cstddef>
      41             : #include <new>  // Placement new
      42             : #include <stdexcept>
      43             : #include <type_traits>
      44             : 
      45             : #include "Utilities/ErrorHandling/Assert.hpp"
      46             : #include "Utilities/Requires.hpp"
      47             : 
      48             : namespace Parallel {
      49             : /*!
      50             :  * \brief A static capacity runtime-sized single-producer single-consumer
      51             :  * lockfree queue.
      52             :  *
      53             :  * As long as only one thread reads and writes simultaneously the queue is
      54             :  * threadsafe. Which threads read and write can change throughout program
      55             :  * execution, the important thing is that there is no instance during the
      56             :  * execution where more than one thread tries to read and where more than one
      57             :  * thread tries to write.
      58             :  *
      59             :  * \note This class is intentionally not serializable since handling
      60             :  * threadsafety around serialization requires careful thought of the individual
      61             :  * circumstances.
      62             :  */
      63             : template <typename T, size_t Capacity>
      64           1 : class StaticSpscQueue {
      65             :  private:
      66             : #ifdef __cpp_lib_hardware_interference_size
      67             :   static constexpr size_t cache_line_size_ =
      68             :       std::hardware_destructive_interference_size;
      69             : #else
      70           0 :   static constexpr size_t cache_line_size_ = 64;
      71             : #endif
      72             : 
      73             :   // Padding to avoid false sharing between slots_ and adjacent allocations
      74           0 :   static constexpr size_t padding_ = (cache_line_size_ - 1) / sizeof(T) + 1;
      75             : 
      76             :  public:
      77           0 :   StaticSpscQueue() = default;
      78           0 :   ~StaticSpscQueue() {
      79             :     // Destruct objects in the buffer.
      80             :     while (front()) {
      81             :       pop();
      82             :     }
      83             :   }
      84             : 
      85           0 :   StaticSpscQueue(const StaticSpscQueue&) = delete;
      86           0 :   StaticSpscQueue& operator=(const StaticSpscQueue&) = delete;
      87           0 :   StaticSpscQueue(StaticSpscQueue&&) = delete;
      88           0 :   StaticSpscQueue& operator=(StaticSpscQueue&&) = delete;
      89             : 
      90             :   /// Construct a new element at the end of the queue in place.
      91             :   ///
      92             :   /// Uses placement new for in-place construction.
      93             :   ///
      94             :   /// \warning This may overwrite existing elements if `capacity()` is
      95             :   /// exceeded without warning.
      96             :   template <typename... Args>
      97           1 :   void emplace(Args&&... args) noexcept(
      98             :       std::is_nothrow_constructible_v<T, Args&&...>) {
      99             :     static_assert(std::is_constructible_v<T, Args&&...>,
     100             :                   "T must be constructible with Args&&...");
     101             :     const auto write_index = write_index_.load(std::memory_order_relaxed);
     102             :     auto next_write_index = write_index + 1;
     103             :     if (next_write_index == capacity_) {
     104             :       next_write_index = 0;
     105             :     }
     106             :     while (next_write_index == read_index_cache_) {
     107             :       read_index_cache_ = read_index_.load(std::memory_order_acquire);
     108             :     }
     109             :     // Destroy the object this slot already holds (see `data_`).
     110             :     data_[write_index + padding_].~T();
     111             :     new (&data_[write_index + padding_]) T(std::forward<Args>(args)...);
     112             :     write_index_.store(next_write_index, std::memory_order_release);
     113             :   }
     114             : 
     115             :   /// Construct a new element at the end of the queue in place.
     116             :   ///
     117             :   /// Uses placement new for in-place construction.
     118             :   ///
     119             :   /// Returns `true` if the emplacement succeeded and `false` if it did
     120             :   /// not. If it failed then the queue is currently full.
     121             :   template <typename... Args>
     122           1 :   [[nodiscard]] bool try_emplace(Args&&... args) noexcept(
     123             :       std::is_nothrow_constructible_v<T, Args&&...>) {
     124             :     static_assert(std::is_constructible_v<T, Args&&...>,
     125             :                   "T must be constructible with Args&&...");
     126             :     const auto write_index = write_index_.load(std::memory_order_relaxed);
     127             :     auto next_write_index = write_index + 1;
     128             :     if (next_write_index == capacity_) {
     129             :       next_write_index = 0;
     130             :     }
     131             :     if (next_write_index == read_index_cache_) {
     132             :       read_index_cache_ = read_index_.load(std::memory_order_acquire);
     133             :       if (next_write_index == read_index_cache_) {
     134             :         return false;
     135             :       }
     136             :     }
     137             :     // Destroy the object this slot already holds (see `data_`).
     138             :     data_[write_index + padding_].~T();
     139             :     new (&data_[write_index + padding_]) T(std::forward<Args>(args)...);
     140             :     write_index_.store(next_write_index, std::memory_order_release);
     141             :     return true;
     142             :   }
     143             : 
     144             :   /// Push a new element to the end of the queue.
     145             :   ///
     146             :   /// Uses `emplace()` internally.
     147             :   ///
     148             :   /// \warning This may overwrite existing elements if `capacity()` is
     149             :   /// exceeded without warning.
     150           1 :   void push(const T& v) noexcept(std::is_nothrow_copy_constructible_v<T>) {
     151             :     static_assert(std::is_copy_constructible_v<T>,
     152             :                   "T must be copy constructible");
     153             :     emplace(v);
     154             :   }
     155             : 
     156             :   /// Push a new element to the end of the queue.
     157             :   ///
     158             :   /// Uses `emplace()` internally.
     159             :   ///
     160             :   /// \warning This may overwrite existing elements if `capacity()` is
     161             :   /// exceeded without warning.
     162             :   template <typename P, Requires<std::is_constructible_v<T, P&&>> = nullptr>
     163           1 :   void push(P&& v) noexcept(std::is_nothrow_constructible_v<T, P&&>) {
     164             :     emplace(std::forward<P>(v));
     165             :   }
     166             : 
     167             :   /// Push a new element to the end of the queue. Returns `false` if the queue
     168             :   /// is at capacity and does not push the new object, otherwise returns `true`.
     169             :   ///
     170             :   /// Uses `try_emplace()` internally.
     171           1 :   [[nodiscard]] bool try_push(const T& v) noexcept(
     172             :       std::is_nothrow_copy_constructible_v<T>) {
     173             :     static_assert(std::is_copy_constructible_v<T>,
     174             :                   "T must be copy constructible");
     175             :     return try_emplace(v);
     176             :   }
     177             : 
     178             :   /// Push a new element to the end of the queue. Returns `false` if the queue
     179             :   /// is at capacity and does not push the new object, otherwise returns `true`.
     180             :   ///
     181             :   /// Uses `try_emplace()` internally.
     182             :   template <typename P, Requires<std::is_constructible_v<T, P&&>> = nullptr>
     183           1 :   [[nodiscard]] bool try_push(P&& v) noexcept(
     184             :       std::is_nothrow_constructible_v<T, P&&>) {
     185             :     return try_emplace(std::forward<P>(v));
     186             :   }
     187             : 
     188             :   /// Returns the first element from the queue.
     189             :   ///
     190             :   /// \note Returns `nullptr` if the queue is empty.
     191           1 :   [[nodiscard]] T* front() noexcept {
     192             :     const auto read_index = read_index_.load(std::memory_order_relaxed);
     193             :     if (read_index == write_index_cache_) {
     194             :       write_index_cache_ = write_index_.load(std::memory_order_acquire);
     195             :       if (write_index_cache_ == read_index) {
     196             :         return nullptr;
     197             :       }
     198             :     }
     199             :     return &data_[read_index + padding_];
     200             :   }
     201             : 
     202             :   /// Removes the first element from the queue.
     203           1 :   void pop() {
     204             :     static_assert(std::is_nothrow_destructible_v<T>,
     205             :                   "T must be nothrow destructible");
     206             :     const auto read_index = read_index_.load(std::memory_order_relaxed);
     207             : #ifdef SPECTRE_DEBUG
     208             :     const auto write_index = write_index_.load(std::memory_order_acquire);
     209             :     ASSERT(write_index != read_index,
     210             :            "Can't pop an element from an empty queue. read_index: "
     211             :                << read_index << " write_index " << write_index);
     212             : #endif  // SPECTRE_DEBUG
     213             :     // Leave a live object behind (see `data_`).
     214             :     data_[read_index + padding_].~T();
     215             :     new (&data_[read_index + padding_]) T{};
     216             :     auto next_read_index = read_index + 1;
     217             :     if (next_read_index == capacity_) {
     218             :       next_read_index = 0;
     219             :     }
     220             :     if (read_index == write_index_cache_) {
     221             :       write_index_cache_ = next_read_index;
     222             :     }
     223             :     read_index_.store(next_read_index, std::memory_order_release);
     224             :   }
     225             : 
     226             :   /// Returns the size of the queue at a particular hardware state.
     227             :   ///
     228             :   /// Note that while this can be checked in a threadsafe manner, it is up to
     229             :   /// the user to guarantee that another thread does not change the queue
     230             :   /// between when `size()` is called and how the result is used.
     231           1 :   [[nodiscard]] size_t size() const noexcept {
     232             :     std::ptrdiff_t diff = static_cast<std::ptrdiff_t>(
     233             :                               write_index_.load(std::memory_order_acquire)) -
     234             :                           static_cast<std::ptrdiff_t>(
     235             :                               read_index_.load(std::memory_order_acquire));
     236             :     if (diff < 0) {
     237             :       diff += static_cast<std::ptrdiff_t>(capacity_);
     238             :     }
     239             :     return static_cast<size_t>(diff);
     240             :   }
     241             : 
     242             :   /// Returns `true` if the queue may be empty, otherwise `false`.
     243             :   ///
     244             :   /// Note that while this can be checked in a threadsafe manner, it is up to
     245             :   /// the user to guarantee that another thread does not change the queue
     246             :   /// between when `empty()` is called and how the result is used.
     247           1 :   [[nodiscard]] bool empty() const noexcept {
     248             :     return write_index_.load(std::memory_order_acquire) ==
     249             :            read_index_.load(std::memory_order_acquire);
     250             :   }
     251             : 
     252             :   /// Returns the capacity of the queue.
     253           1 :   [[nodiscard]] size_t capacity() const noexcept { return capacity_ - 1; }
     254             : 
     255             :  private:
     256           0 :   static constexpr size_t capacity_ = Capacity + 1;
     257             :   // Every slot holds a live object, so the array's destructor destroys each
     258             :   // element exactly once. `emplace` and `pop` maintain this.
     259           0 :   std::array<T, capacity_ + 2 * padding_> data_{};
     260             : 
     261             :   // Align to cache line size in order to avoid false sharing
     262             :   // read_index_cache_ and write_index_cache_ is used to reduce the amount of
     263             :   // cache coherency traffic
     264           0 :   alignas(cache_line_size_) std::atomic<size_t> write_index_{0};
     265           0 :   alignas(cache_line_size_) size_t read_index_cache_{0};
     266           0 :   alignas(cache_line_size_) std::atomic<size_t> read_index_{0};
     267           0 :   alignas(cache_line_size_) size_t write_index_cache_{0};
     268             : 
     269             :   // Padding to avoid adjacent allocations from sharing a cache line with
     270             :   // write_index_cache_
     271             :   // NOLINTNEXTLINE(modernize-avoid-c-arrays)
     272           0 :   char padding_data_[cache_line_size_ - sizeof(write_index_cache_)]{};
     273             : };
     274             : }  // namespace Parallel