Line data    Source code

       1           0 : // Distributed under the MIT License.
       2             : // See LICENSE.txt for details.
       3             : 
       4             : #pragma once
       5             : 
       6             : #include <atomic>
       7             : #include <concepts>
       8             : #include <cstddef>
       9             : #include <cstdint>
      10             : #include <memory>
      11             : #include <mutex>
      12             : #include <new>  // for hardware_destructive_interference_size
      13             : #include <stdexcept>
      14             : #include <utility>
      15             : #include <vector>
      16             : 
      17             : #include "Parallel/MultiReaderSpinlock.hpp"
      18             : #include "Utilities/ErrorHandling/Assert.hpp"
      19             : #include "Utilities/Gsl.hpp"
      20             : 
      21             : namespace Parallel {
      22             : /*!
      23             :  * \brief A threadsafe parallel first-in-first-out cache.
      24             :  */
      25             : template <class T>
      26           1 : class FifoCache {
      27           0 :   using stored_type = std::pair<MultiReaderSpinlock, T>;
      28           0 :   using value_type = std::vector<stored_type>;
      29             : 
      30             :  public:
      31             :   /*!
      32             :    * \brief Wrapper type used as the result from `find` to ensure correct
      33             :    * thread safety.
      34             :    *
      35             :    * Use `.value()` to get the stored value.
      36             :    *
      37             :    * Use `.has_value()` to check if a value is stored.
      38             :    */
      39           1 :   struct Cached {
      40             :    public:
      41           0 :     Cached() = delete;
      42           0 :     explicit Cached(const stored_type* t);
      43             :     /// NOLINTNEXTLINE(google-explicit-constructor)
      44           1 :     Cached(const std::nullopt_t /*unused*/) : Cached{nullptr} {}
      45           0 :     Cached(const Cached& rhs);
      46           0 :     Cached& operator=(const Cached& rhs);
      47           0 :     Cached(Cached&& rhs) noexcept(true);
      48           0 :     Cached& operator=(Cached&& rhs) noexcept(true);
      49           0 :     ~Cached() noexcept(true);
      50             : 
      51             :     /// \brief Returns a reference to the held object.
      52             :     ///
      53             :     /// \throws std::runtime_error if no value
      54           1 :     auto value() const -> const T&;
      55             : 
      56             :     /// \brief Returns `true` if a value is stored, otherwise returns `false`.
      57           1 :     bool has_value() const { return t_ != nullptr; }
      58             : 
      59             :    private:
      60           0 :     stored_type* t_;
      61             :   };
      62             : 
      63             :   /// \brief Create a FifoCache that has \p capacity.
      64           1 :   explicit FifoCache(std::unsigned_integral auto capacity);
      65             : 
      66           0 :   FifoCache() = delete;
      67           0 :   FifoCache(const FifoCache& rhs) = delete;
      68           0 :   FifoCache& operator=(const FifoCache& rhs) = delete;
      69           0 :   FifoCache(FifoCache&& rhs) = delete;
      70           0 :   FifoCache& operator=(FifoCache&& rhs) = delete;
      71           0 :   ~FifoCache() = default;
      72             : 
      73             :   /*!
      74             :    * \brief Pushes the entry computed by `compute_value()` to the front of the
      75             :    * queue, ejecting the last entry if the capacity is reached. The inserted
      76             :    * entry is returned.
      77             :    *
      78             :    * This function allows lazy computation of the value, which means the
      79             :    * computation is elided if another thread pushes the new cache entry before
      80             :    * this one.
      81             :    *
      82             :    * The predicate must satisfy `predicate(t) == true` to avoid inserting
      83             :    * duplicates. This is best guaranteed by passing the same predicate that
      84             :    * would be passed to calls to `find()`.
      85             :    */
      86             :   template <class ComputeValue, class UnaryPredicate>
      87           1 :   auto push(ComputeValue&& compute_value, const UnaryPredicate& predicate)
      88             :       -> Cached;
      89             : 
      90             :   /*!
      91             :    * \brief Pushes the entry `t` to the front of the queue,
      92             :    * ejecting the last entry if the capacity is reached. The inserted
      93             :    * entry is returned.
      94             :    *
      95             :    * The predicate must satisfy `predicate(t) == true` to avoid inserting
      96             :    * duplicates. This is best guaranteed by passing the same predicate that
      97             :    * would be passed to calls to `find()`.
      98             :    */
      99             :   template <class UnaryPredicate>
     100           1 :   auto push(T t, const UnaryPredicate& predicate) -> Cached;
     101             : 
     102             :   /*!
     103             :    * \brief Get the first element that matches `predicate`.
     104             :    *
     105             :    * If no value in the cache matches the predicate then `result.has_value()`
     106             :    * is `false` and `result.value()` will throw.
     107             :    *
     108             :    * The return type is designed to handle the locking and unlocking
     109             :    * of the data to ensure thread safety.
     110             :    */
     111             :   template <class UnaryPredicate>
     112           1 :   [[nodiscard]] auto find(const UnaryPredicate& predicate) const -> Cached;
     113             : 
     114             :  private:
     115             : #if defined(__cpp_lib_hardware_interference_size)
     116             :   static constexpr size_t cache_line_size_ =
     117             :       std::hardware_destructive_interference_size;
     118             : #else
     119           0 :   static constexpr size_t cache_line_size_ = 64;
     120             : #endif
     121             : 
     122             :   // std::vector does not have an atomic size so in order to prevent tearing
     123             :   // we have to track the size separately.
     124           0 :   alignas(cache_line_size_) std::atomic<std::uint64_t> size_{0};
     125           0 :   alignas(cache_line_size_) std::mutex write_lock_{};
     126           0 :   alignas(cache_line_size_) value_type data_{};
     127             :   // Ensure we pad the end to avoid false sharing. Unlikely to happen because
     128             :   // of the layout, but better to be safe.
     129             : #if defined(__clang__)
     130             : #pragma GCC diagnostic push
     131             : #pragma GCC diagnostic ignored "-Wunused-private-field"
     132             : #endif
     133             :   // NOLINTNEXTLINE(modernize-avoid-c-arrays)
     134           0 :   char padding_[cache_line_size_ - sizeof(value_type) % cache_line_size_] = {};
     135             : #if defined(__clang__)
     136             : #pragma GCC diagnostic pop
     137             : #endif
     138             : };
     139             : 
     140             : template <class T>
     141             : FifoCache<T>::FifoCache(const std::unsigned_integral auto capacity)
     142             :     : data_(capacity) {
     143             :   ASSERT(capacity > 0, "Must have a positive capacity but got " << capacity);
     144             : }
     145             : 
     146             : template <class T>
     147             : template <class ComputeValue, class UnaryPredicate>
     148             : auto FifoCache<T>::push(ComputeValue&& compute_value,
     149             :                         const UnaryPredicate& predicate) -> Cached {
     150             :   std::lock_guard guard(write_lock_);
     151             :   auto size = size_.load(std::memory_order_acquire);
     152             :   for (size_t i = 0; i < size; ++i) {
     153             :     Cached vt{std::addressof(data_[i])};
     154             :     if (predicate(vt.value())) {
     155             :       return {vt};
     156             :     }
     157             :   }
     158             : 
     159             :   // Compute the new value _before_ locking data to minimize locked time.
     160             :   T new_value = std::forward<ComputeValue>(compute_value)();
     161             :   if (size < data_.capacity()) {
     162             :     ++size;
     163             :   }
     164             :   data_[size - 1].first.write_lock();
     165             :   for (size_t i = size - 1; i > 0; --i) {
     166             :     data_[i - 1].first.write_lock();
     167             :     data_[i].second = std::move(data_[i - 1].second);
     168             :   }
     169             :   data_[0].second = std::move(new_value);
     170             :   for (size_t i = 0; i < size; ++i) {
     171             :     data_[i].first.write_unlock();
     172             :   }
     173             :   size_.store(size, std::memory_order_release);
     174             :   return find(predicate);
     175             : }
     176             : 
     177             : template <class T>
     178             : template <class UnaryPredicate>
     179             : auto FifoCache<T>::push(T t, const UnaryPredicate& predicate) -> Cached {
     180             :   return push(
     181             :       [t_local = std::move(t)]() mutable -> T  // NOLINT(spectre-mutable)
     182             :       { return std::move(t_local); },
     183             :       predicate);
     184             : }
     185             : 
     186             : template <class T>
     187             : template <class UnaryPredicate>
     188             : [[nodiscard]] auto FifoCache<T>::find(const UnaryPredicate& predicate) const
     189             :     -> Cached {
     190             :   const auto size = size_.load(std::memory_order_relaxed);
     191             :   for (size_t i = 0; i < size; ++i) {
     192             :     Cached vt{std::addressof(data_[i])};
     193             :     if (predicate(vt.value())) {
     194             :       return vt;
     195             :     }
     196             :   }
     197             :   return Cached{nullptr};
     198             : }
     199             : 
     200             : template <class T>
     201             : FifoCache<T>::Cached::Cached(const stored_type* t)
     202             :     // NOLINTNEXTLINE(cppcoreguidelines-pro-type-const-cast)
     203             :     : t_(const_cast<stored_type*>(t)) {
     204             :   if (t_ != nullptr) {
     205             :     t_->first.read_lock();
     206             :   }
     207             : }
     208             : 
     209             : template <class T>
     210             : FifoCache<T>::Cached::Cached(const Cached& rhs) : t_(rhs.t_) {
     211             :   if (t_ != nullptr) {
     212             :     t_->first.read_lock();
     213             :   }
     214             : }
     215             : 
     216             : template <class T>
     217             : typename FifoCache<T>::Cached& FifoCache<T>::Cached::operator=(
     218             :     const Cached& rhs) {
     219             :   if (&rhs == this) {
     220             :     return *this;
     221             :   }
     222             :   if (t_ != nullptr) {
     223             :     t_->first.read_unlock();
     224             :   }
     225             :   t_ = rhs.t_;
     226             :   if (t_ != nullptr) {
     227             :     t_->first.read_lock();
     228             :   }
     229             :   return *this;
     230             : }
     231             : 
     232             : template <class T>
     233             : FifoCache<T>::Cached::Cached(Cached&& rhs) noexcept(true) : t_(rhs.t_) {
     234             :   rhs.t_ = nullptr;
     235             : }
     236             : 
     237             : template <class T>
     238             : typename FifoCache<T>::Cached& FifoCache<T>::Cached::operator=(
     239             :     Cached&& rhs) noexcept(true) {
     240             :   if (&rhs == this) {
     241             :     return *this;
     242             :   }
     243             :   if (t_ != nullptr) {
     244             :     t_->first.read_unlock();
     245             :   }
     246             :   t_ = rhs.t_;
     247             :   rhs.t_ = nullptr;
     248             :   return *this;
     249             : }
     250             : 
     251             : template <class T>
     252             : FifoCache<T>::Cached::~Cached() noexcept(true) {
     253             :   if (t_ != nullptr) {
     254             :     t_->first.read_unlock();
     255             :   }
     256             : }
     257             : 
     258             : template <class T>
     259             : auto FifoCache<T>::Cached::value() const -> const T& {
     260             :   if (UNLIKELY(not has_value())) {
     261             :     throw std::runtime_error{"No value in FifoCache."};
     262             :   }
     263             :   return t_->second;
     264             : }
     265             : }  // namespace Parallel