blob: c5c67140865e8174dac6348e4ced58baef11b3bd (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
|
#pragma once
#include <util/generic/yexception.h>
//https://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue
namespace NThreading {
template<typename T>
class TBoundedQueue {
public:
explicit TBoundedQueue(size_t size)
: Buffer_(new TCell[size])
, Mask_(size - 1)
{
Y_ENSURE(size >= 2 && (size & (size - 1)) == 0);
for (size_t i = 0; i < size; ++i) {
Buffer_[i].Sequence.store(i, std::memory_order_relaxed);
}
}
template <typename T_>
[[nodiscard]] bool Enqueue(T_&& data) noexcept {
TCell* cell;
size_t pos = EnqueuePos_.load(std::memory_order_relaxed);
for (;;) {
cell = &Buffer_[pos & Mask_];
size_t seq = cell->Sequence.load(std::memory_order_acquire);
intptr_t dif = (intptr_t)seq - (intptr_t)pos;
if (dif == 0) {
if (EnqueuePos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) {
break;
}
} else if (dif < 0) {
return false;
} else {
pos = EnqueuePos_.load(std::memory_order_relaxed);
}
}
static_assert(noexcept(cell->Data = std::forward<T_>(data)));
cell->Data = std::forward<T_>(data);
cell->Sequence.store(pos + 1, std::memory_order_release);
return true;
}
[[nodiscard]] bool Dequeue(T& data) noexcept {
TCell* cell;
size_t pos = DequeuePos_.load(std::memory_order_relaxed);
for (;;) {
cell = &Buffer_[pos & Mask_];
size_t seq = cell->Sequence.load(std::memory_order_acquire);
intptr_t dif = (intptr_t)seq - (intptr_t)(pos + 1);
if (dif == 0) {
if (DequeuePos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) {
break;
}
} else if (dif < 0) {
return false;
} else {
pos = DequeuePos_.load(std::memory_order_relaxed);
}
}
static_assert(noexcept(data = std::move(cell->Data)));
data = std::move(cell->Data);
cell->Sequence.store(pos + Mask_ + 1, std::memory_order_release);
return true;
}
private:
struct TCell {
std::atomic<size_t> Sequence;
T Data;
};
std::unique_ptr<TCell[]> Buffer_;
const size_t Mask_;
alignas(64) std::atomic<size_t> EnqueuePos_ = 0;
alignas(64) std::atomic<size_t> DequeuePos_ = 0;
};
}
|
