blob: 0a3ba47184dce48e66b3c906f4f90502e406e397 (
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
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
|
#pragma once
#include <util/thread/pool.h>
#include <util/datetime/base.h>
#include <util/thread/lfqueue.h>
#include <util/system/thread.h>
#include <util/generic/vector.h>
#include <util/generic/scope.h>
#include <util/stream/str.h>
#include <library/cpp/threading/bounded_queue/bounded_queue.h>
#include <library/cpp/yt/threading/event_count.h>
class TFastElasticQueue
: public TThreadPoolBase
, private IThreadFactory::IThreadAble
{
public:
explicit TFastElasticQueue(const TParams& params = {})
: TThreadPoolBase(params)
{
Y_ENSURE(!params.Blocking_);
}
~TFastElasticQueue() {
Stop();
}
void Start(size_t threadCount, size_t maxQueueSize) override {
Y_ENSURE(Threads_.empty());
Y_ENSURE(maxQueueSize > 0);
Queue_.Reset(new NThreading::TBoundedQueue<IObjectInQueue*>(FastClp2(maxQueueSize + threadCount))); //threadCount is for stop events
MaxQueueSize_ = maxQueueSize;
try {
for (size_t i = 0; i < threadCount; ++i) {
Threads_.push_back(Pool()->Run(this));
}
} catch (...) {
Stop();
throw;
}
Stopped_ = false;
}
size_t ObjectCount() const {
//GuardCount_ can be temporary incremented above real object count in queue
return Min(GuardCount_.load(), MaxQueueSize_);
}
bool Add(IObjectInQueue* obj) override Y_WARN_UNUSED_RESULT {
if (Stopped_ || !obj) {
return false;
}
if (GuardCount_.fetch_add(1) >= MaxQueueSize_) {
GuardCount_.fetch_sub(1);
return false;
}
QueueSize_.fetch_add(1);
if (!Queue_->Enqueue(obj)) {
//Simultaneous Dequeue calls can return not in exact fifo order of items,
//so there can be GuardCount_ < MaxQueueSize_ but Enqueue will fail because of
//the oldest enqueued item is not actually dequeued and ring buffer can't proceed.
GuardCount_.fetch_sub(1);
QueueSize_.fetch_sub(1);
return false;
}
Event_.NotifyOne();
return true;
}
size_t Size() const noexcept override {
return QueueSize_.load();
}
void Stop() noexcept override {
Stopped_ = true;
for (size_t i = 0; i < Threads_.size(); ++i) {
while (!Queue_->Enqueue(nullptr)) {
Sleep(TDuration::MilliSeconds(1));
}
Event_.NotifyOne();
}
while (!Threads_.empty()) {
Threads_.back()->Join();
Threads_.pop_back();
}
Queue_.Reset();
}
void DoExecute() override {
TThread::SetCurrentThreadName(Params.ThreadName_.c_str());
while (true) {
IObjectInQueue* job = nullptr;
Event_.Await([&]() {
return Queue_->Dequeue(job);
});
if (!job) {
break;
}
QueueSize_.fetch_sub(1);
Y_DEFER {
GuardCount_.fetch_sub(1);
};
if (Params.Catching_) {
try {
try {
job->Process(nullptr);
} catch (...) {
Cdbg << "[mtp queue] " << CurrentExceptionMessage() << Endl;
}
} catch (...) {
;
}
} else {
job->Process(nullptr);
}
}
}
private:
std::atomic<bool> Stopped_ = false;
size_t MaxQueueSize_ = 0;
alignas(64) std::atomic<size_t> GuardCount_ = 0;
alignas(64) std::atomic<size_t> QueueSize_ = 0;
TVector<THolder<IThreadFactory::IThread>> Threads_;
THolder<NThreading::TBoundedQueue<IObjectInQueue*>> Queue_;
NYT::NThreading::TEventCount Event_;
};
|
