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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
|
#pragma once
#include <util/system/atomic.h>
#include <util/generic/noncopyable.h>
template <typename ElementType>
class TFunnelQueue: private TNonCopyable {
public:
TFunnelQueue() noexcept
: Front(nullptr)
, Back(nullptr)
{
}
virtual ~TFunnelQueue() noexcept {
for (auto entry = Front; entry; entry = DeleteEntry(entry))
continue;
}
/// Push element. Can be used from many threads. Return true if is first element.
bool
Push(ElementType&& element) noexcept {
TEntry* const next = NewEntry(static_cast<ElementType&&>(element));
TEntry* const prev = AtomicSwap(&Back, next);
AtomicSet(prev ? prev->Next : Front, next);
return !prev;
}
/// Extract top element. Must be used only from one thread. Return true if have more.
bool
Pop() noexcept {
if (TEntry* const top = AtomicGet(Front)) {
const auto last = AtomicCas(&Back, nullptr, top);
if (last) // This is last element in queue. Queue is empty now.
AtomicCas(&Front, nullptr, top);
else // This element is not last.
for (;;) {
if (const auto next = AtomicGet(top->Next)) {
AtomicSet(Front, next);
break;
}
// But Next is null. Wait next assignment in spin lock.
}
DeleteEntry(top);
return !last;
}
return false;
}
/// Peek top element. Must be used only from one thread.
ElementType&
Top() const noexcept {
return AtomicGet(Front)->Data;
}
bool
IsEmpty() const noexcept {
return !AtomicGet(Front);
}
protected:
class TEntry: private TNonCopyable {
friend class TFunnelQueue;
private:
explicit TEntry(ElementType&& element) noexcept
: Data(static_cast<ElementType&&>(element))
, Next(nullptr)
{
}
~TEntry() noexcept {
}
public:
ElementType Data;
TEntry* volatile Next;
};
TEntry* volatile Front;
TEntry* volatile Back;
virtual TEntry* NewEntry(ElementType&& element) noexcept {
return new TEntry(static_cast<ElementType&&>(element));
}
virtual TEntry* DeleteEntry(TEntry* entry) noexcept {
const auto next = entry->Next;
delete entry;
return next;
}
protected:
struct TEntryIter {
TEntry* ptr;
ElementType& operator*() {
return ptr->Data;
}
ElementType* operator->() {
return &ptr->Data;
}
TEntryIter& operator++() {
ptr = AtomicGet(ptr->Next);
return *this;
}
bool operator!=(const TEntryIter& other) const {
return ptr != other.ptr;
}
bool operator==(const TEntryIter& other) const {
return ptr == other.ptr;
}
};
struct TConstEntryIter {
const TEntry* ptr;
const ElementType& operator*() {
return ptr->Data;
}
const ElementType* operator->() {
return &ptr->Data;
}
TEntryIter& operator++() {
ptr = AtomicGet(ptr->Next);
return *this;
}
bool operator!=(const TConstEntryIter& other) const {
return ptr != other.ptr;
}
bool operator==(const TConstEntryIter& other) const {
return ptr == other.ptr;
}
};
public:
using const_iterator = TConstEntryIter;
using iterator = TEntryIter;
iterator begin() {
return {AtomicGet(Front)};
}
const_iterator cbegin() {
return {AtomicGet(Front)};
}
const_iterator begin() const {
return {AtomicGet(Front)};
}
iterator end() {
return {nullptr};
}
const_iterator cend() {
return {nullptr};
}
const_iterator end() const {
return {nullptr};
}
};
template <typename ElementType>
class TPooledFunnelQueue: public TFunnelQueue<ElementType> {
public:
TPooledFunnelQueue() noexcept
: Stack(nullptr)
{
}
virtual ~TPooledFunnelQueue() noexcept override {
for (auto entry = TBase::Front; entry; entry = TBase::DeleteEntry(entry))
continue;
for (auto entry = Stack; entry; entry = TBase::DeleteEntry(entry))
continue;
TBase::Back = TBase::Front = Stack = nullptr;
}
private:
typedef TFunnelQueue<ElementType> TBase;
typename TBase::TEntry* volatile Stack;
protected:
virtual typename TBase::TEntry* NewEntry(ElementType&& element) noexcept override {
while (const auto top = AtomicGet(Stack))
if (AtomicCas(&Stack, top->Next, top)) {
top->Data = static_cast<ElementType&&>(element);
AtomicSet(top->Next, nullptr);
return top;
}
return TBase::NewEntry(static_cast<ElementType&&>(element));
}
virtual typename TBase::TEntry* DeleteEntry(typename TBase::TEntry* entry) noexcept override {
entry->Data = ElementType();
const auto next = entry->Next;
do
AtomicSet(entry->Next, AtomicGet(Stack));
while (!AtomicCas(&Stack, entry, entry->Next));
return next;
}
};
template <typename ElementType, template <typename T> class TQueueType = TFunnelQueue>
class TCountedFunnelQueue: public TQueueType<ElementType> {
public:
TCountedFunnelQueue() noexcept
: Count(0)
{
}
TAtomicBase GetSize() const noexcept {
return AtomicGet(Count);
}
private:
typedef TQueueType<ElementType> TBase;
virtual typename TBase::TEntry* NewEntry(ElementType&& element) noexcept override {
AtomicAdd(Count, 1);
return TBase::NewEntry(static_cast<ElementType&&>(element));
}
virtual typename TBase::TEntry* DeleteEntry(typename TBase::TEntry* entry) noexcept override {
AtomicSub(Count, 1);
return TBase::DeleteEntry(entry);
}
TAtomic Count;
};
|
