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
|
