241 lines
8.6 KiB
C
241 lines
8.6 KiB
C
|
/*
|
||
|
|
Copyright (c) 2020 Erik Rigtorp <erik@rigtorp.se>
|
||
|
|
|
||
|
|
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||
|
|
of this software and associated documentation files (the "Software"), to deal
|
||
|
|
in the Software without restriction, including without limitation the rights
|
||
|
|
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||
|
|
copies of the Software, and to permit persons to whom the Software is
|
||
|
|
furnished to do so, subject to the following conditions:
|
||
|
|
|
||
|
|
The above copyright notice and this permission notice shall be included in all
|
||
|
|
copies or substantial portions of the Software.
|
||
|
|
|
||
|
|
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||
|
|
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||
|
|
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||
|
|
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||
|
|
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||
|
|
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||
|
|
SOFTWARE.
|
||
|
|
*/
|
||
|
|
|
||
|
|
#pragma once
|
||
|
|
#ifdef _MSC_VER
|
||
|
|
#pragma warning(push)
|
||
|
|
#pragma warning(disable : 4324)
|
||
|
|
#endif
|
||
|
|
|
||
|
|
#include <atomic>
|
||
|
|
#include <cassert>
|
||
|
|
#include <cstddef> // offsetof
|
||
|
|
#include <memory>
|
||
|
|
#include <new>
|
||
|
|
#include <stdexcept>
|
||
|
|
|
||
|
|
namespace Common {
|
||
|
|
namespace mpmc {
|
||
|
|
static constexpr size_t hardwareInterferenceSize = 64;
|
||
|
|
|
||
|
|
template <typename T>
|
||
|
|
using AlignedAllocator = std::allocator<T>;
|
||
|
|
|
||
|
|
template <typename T>
|
||
|
|
struct Slot {
|
||
|
|
~Slot() noexcept {
|
||
|
|
if (turn & 1) {
|
||
|
|
destroy();
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
template <typename... Args>
|
||
|
|
void construct(Args&&... args) noexcept {
|
||
|
|
static_assert(std::is_nothrow_constructible<T, Args&&...>::value,
|
||
|
|
"T must be nothrow constructible with Args&&...");
|
||
|
|
new (&storage) T(std::forward<Args>(args)...);
|
||
|
|
}
|
||
|
|
|
||
|
|
void destroy() noexcept {
|
||
|
|
static_assert(std::is_nothrow_destructible<T>::value, "T must be nothrow destructible");
|
||
|
|
reinterpret_cast<T*>(&storage)->~T();
|
||
|
|
}
|
||
|
|
|
||
|
|
T&& move() noexcept {
|
||
|
|
return reinterpret_cast<T&&>(storage);
|
||
|
|
}
|
||
|
|
|
||
|
|
// Align to avoid false sharing between adjacent slots
|
||
|
|
alignas(hardwareInterferenceSize) std::atomic<size_t> turn = {0};
|
||
|
|
typename std::aligned_storage<sizeof(T), alignof(T)>::type storage;
|
||
|
|
};
|
||
|
|
|
||
|
|
template <typename T, typename Allocator = AlignedAllocator<Slot<T>>>
|
||
|
|
class Queue {
|
||
|
|
private:
|
||
|
|
static_assert(std::is_nothrow_copy_assignable<T>::value ||
|
||
|
|
std::is_nothrow_move_assignable<T>::value,
|
||
|
|
"T must be nothrow copy or move assignable");
|
||
|
|
|
||
|
|
static_assert(std::is_nothrow_destructible<T>::value, "T must be nothrow destructible");
|
||
|
|
|
||
|
|
public:
|
||
|
|
explicit Queue(const size_t capacity, const Allocator& allocator = Allocator())
|
||
|
|
: capacity_(capacity), allocator_(allocator), head_(0), tail_(0) {
|
||
|
|
if (capacity_ < 1) {
|
||
|
|
throw std::invalid_argument("capacity < 1");
|
||
|
|
}
|
||
|
|
// Allocate one extra slot to prevent false sharing on the last slot
|
||
|
|
slots_ = allocator_.allocate(capacity_ + 1);
|
||
|
|
// Allocators are not required to honor alignment for over-aligned types
|
||
|
|
// (see http://eel.is/c++draft/allocator.requirements#10) so we verify
|
||
|
|
// alignment here
|
||
|
|
if (reinterpret_cast<size_t>(slots_) % alignof(Slot<T>) != 0) {
|
||
|
|
allocator_.deallocate(slots_, capacity_ + 1);
|
||
|
|
throw std::bad_alloc();
|
||
|
|
}
|
||
|
|
for (size_t i = 0; i < capacity_; ++i) {
|
||
|
|
new (&slots_[i]) Slot<T>();
|
||
|
|
}
|
||
|
|
static_assert(alignof(Slot<T>) == hardwareInterferenceSize,
|
||
|
|
"Slot must be aligned to cache line boundary to prevent false sharing");
|
||
|
|
static_assert(sizeof(Slot<T>) % hardwareInterferenceSize == 0,
|
||
|
|
"Slot size must be a multiple of cache line size to prevent "
|
||
|
|
"false sharing between adjacent slots");
|
||
|
|
static_assert(sizeof(Queue) % hardwareInterferenceSize == 0,
|
||
|
|
"Queue size must be a multiple of cache line size to "
|
||
|
|
"prevent false sharing between adjacent queues");
|
||
|
|
static_assert(offsetof(Queue, tail_) - offsetof(Queue, head_) ==
|
||
|
|
static_cast<std::ptrdiff_t>(hardwareInterferenceSize),
|
||
|
|
"head and tail must be a cache line apart to prevent false sharing");
|
||
|
|
}
|
||
|
|
|
||
|
|
~Queue() noexcept {
|
||
|
|
for (size_t i = 0; i < capacity_; ++i) {
|
||
|
|
slots_[i].~Slot();
|
||
|
|
}
|
||
|
|
allocator_.deallocate(slots_, capacity_ + 1);
|
||
|
|
}
|
||
|
|
|
||
|
|
// non-copyable and non-movable
|
||
|
|
Queue(const Queue&) = delete;
|
||
|
|
Queue& operator=(const Queue&) = delete;
|
||
|
|
|
||
|
|
template <typename... Args>
|
||
|
|
void emplace(Args&&... args) noexcept {
|
||
|
|
static_assert(std::is_nothrow_constructible<T, Args&&...>::value,
|
||
|
|
"T must be nothrow constructible with Args&&...");
|
||
|
|
auto const head = head_.fetch_add(1);
|
||
|
|
auto& slot = slots_[idx(head)];
|
||
|
|
while (turn(head) * 2 != slot.turn.load(std::memory_order_acquire))
|
||
|
|
;
|
||
|
|
slot.construct(std::forward<Args>(args)...);
|
||
|
|
slot.turn.store(turn(head) * 2 + 1, std::memory_order_release);
|
||
|
|
}
|
||
|
|
|
||
|
|
template <typename... Args>
|
||
|
|
bool try_emplace(Args&&... args) noexcept {
|
||
|
|
static_assert(std::is_nothrow_constructible<T, Args&&...>::value,
|
||
|
|
"T must be nothrow constructible with Args&&...");
|
||
|
|
auto head = head_.load(std::memory_order_acquire);
|
||
|
|
for (;;) {
|
||
|
|
auto& slot = slots_[idx(head)];
|
||
|
|
if (turn(head) * 2 == slot.turn.load(std::memory_order_acquire)) {
|
||
|
|
if (head_.compare_exchange_strong(head, head + 1)) {
|
||
|
|
slot.construct(std::forward<Args>(args)...);
|
||
|
|
slot.turn.store(turn(head) * 2 + 1, std::memory_order_release);
|
||
|
|
return true;
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
auto const prevHead = head;
|
||
|
|
head = head_.load(std::memory_order_acquire);
|
||
|
|
if (head == prevHead) {
|
||
|
|
return false;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
void push(const T& v) noexcept {
|
||
|
|
static_assert(std::is_nothrow_copy_constructible<T>::value,
|
||
|
|
"T must be nothrow copy constructible");
|
||
|
|
emplace(v);
|
||
|
|
}
|
||
|
|
|
||
|
|
template <typename P, typename = typename std::enable_if<
|
||
|
|
std::is_nothrow_constructible<T, P&&>::value>::type>
|
||
|
|
void push(P&& v) noexcept {
|
||
|
|
emplace(std::forward<P>(v));
|
||
|
|
}
|
||
|
|
|
||
|
|
bool try_push(const T& v) noexcept {
|
||
|
|
static_assert(std::is_nothrow_copy_constructible<T>::value,
|
||
|
|
"T must be nothrow copy constructible");
|
||
|
|
return try_emplace(v);
|
||
|
|
}
|
||
|
|
|
||
|
|
template <typename P, typename = typename std::enable_if<
|
||
|
|
std::is_nothrow_constructible<T, P&&>::value>::type>
|
||
|
|
bool try_push(P&& v) noexcept {
|
||
|
|
return try_emplace(std::forward<P>(v));
|
||
|
|
}
|
||
|
|
|
||
|
|
void pop(T& v) noexcept {
|
||
|
|
auto const tail = tail_.fetch_add(1);
|
||
|
|
auto& slot = slots_[idx(tail)];
|
||
|
|
while (turn(tail) * 2 + 1 != slot.turn.load(std::memory_order_acquire))
|
||
|
|
;
|
||
|
|
v = slot.move();
|
||
|
|
slot.destroy();
|
||
|
|
slot.turn.store(turn(tail) * 2 + 2, std::memory_order_release);
|
||
|
|
}
|
||
|
|
|
||
|
|
bool try_pop(T& v) noexcept {
|
||
|
|
auto tail = tail_.load(std::memory_order_acquire);
|
||
|
|
for (;;) {
|
||
|
|
auto& slot = slots_[idx(tail)];
|
||
|
|
if (turn(tail) * 2 + 1 == slot.turn.load(std::memory_order_acquire)) {
|
||
|
|
if (tail_.compare_exchange_strong(tail, tail + 1)) {
|
||
|
|
v = slot.move();
|
||
|
|
slot.destroy();
|
||
|
|
slot.turn.store(turn(tail) * 2 + 2, std::memory_order_release);
|
||
|
|
return true;
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
auto const prevTail = tail;
|
||
|
|
tail = tail_.load(std::memory_order_acquire);
|
||
|
|
if (tail == prevTail) {
|
||
|
|
return false;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
private:
|
||
|
|
constexpr size_t idx(size_t i) const noexcept {
|
||
|
|
return i % capacity_;
|
||
|
|
}
|
||
|
|
|
||
|
|
constexpr size_t turn(size_t i) const noexcept {
|
||
|
|
return i / capacity_;
|
||
|
|
}
|
||
|
|
|
||
|
|
private:
|
||
|
|
const size_t capacity_;
|
||
|
|
Slot<T>* slots_;
|
||
|
|
[[no_unique_address]] Allocator allocator_;
|
||
|
|
|
||
|
|
// Align to avoid false sharing between head_ and tail_
|
||
|
|
alignas(hardwareInterferenceSize) std::atomic<size_t> head_;
|
||
|
|
alignas(hardwareInterferenceSize) std::atomic<size_t> tail_;
|
||
|
|
};
|
||
|
|
} // namespace mpmc
|
||
|
|
|
||
|
|
template <typename T, typename Allocator = mpmc::AlignedAllocator<mpmc::Slot<T>>>
|
||
|
|
using MPMCQueue = mpmc::Queue<T, Allocator>;
|
||
|
|
|
||
|
|
} // namespace Common
|
||
|
|
|
||
|
|
#ifdef _MSC_VER
|
||
|
|
#pragma warning(pop)
|
||
|
|
#endif
|