8000 re-open the PR, queue length counter still underflows (#10065) · archerli/arangodb@9f51c03 · GitHub
[go: up one dir, main page]
Skip to content

Commit 9f51c03

Browse files
jsteemannjsteemann
authored
re-open the PR, queue length counter still underflows (arangodb#10065)
1 parent 02f9eb0 commit 9f51c03

File tree

4 files changed

+88
-61
lines changed

4 files changed

+88
-61
lines changed

‎arangod/GeneralServer/RequestLane.h

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -98,7 +98,11 @@ enum class RequestLane {
9898
// AGENCY_CALLBACK`
9999
};
100100

101-
enum class RequestPriority { HIGH, MED, LOW };
101+
enum class RequestPriority {
102+
HIGH = 0,
103+
MED = 1,
104+
LOW = 2
105+
};
102106

103107
inline RequestPriority PriorityRequestLane(RequestLane lane) {
104108
switch (lane) {

‎arangod/Scheduler/SchedulerFeature.h

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -48,18 +48,17 @@ class SchedulerFeature final : public application_features::ApplicationFeature {
4848
private:
4949
uint64_t _nrMinimalThreads = 2;
5050
uint64_t _nrMaximalThreads = 0;
51-
uint64_t _queueSize = 128;
52-
uint64_t _fifo1Size = 1024 * 1024;
51+
uint64_t _queueSize = 4096;
52+
uint64_t _fifo1Size = 4096;
5353
uint64_t _fifo2Size = 4096;
5454

5555
std::unique_ptr<Scheduler> _scheduler;
5656

5757
// -------------------------------------------------------------------------
58-
// UNRELATED SECTION STARS HERE: Singals and other things creeped into Sched
58+
// UNRELATED SECTION STARTS HERE: Signals and other things crept into Sched
5959
// -------------------------------------------------------------------------
6060

6161
public:
62-
/*size_t concurrency() const { return static_cast<size_t>(_nrMaximalThreads); }*/
6362
void buildControlCHandler();
6463
void buildHangupHandler();
6564

‎arangod/Scheduler/SupervisedScheduler.cpp

Lines changed: 76 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ bool isDirectDeadlockLane(RequestLane lane) {
6868
namespace {
6969
typedef std::chrono::time_point<std::chrono::steady_clock> time_point;
7070

71-
// value initialise these arrays, otherwise mac will crash
71+
// value-initialize these arrays, otherwise mac will crash
7272
thread_local time_point conditionQueueFullSince{};
7373
thread_local uint_fast32_t queueWarningTick{};
7474

@@ -80,7 +80,7 @@ time_point lastQueueFullWarning[3];
8080
int64_t fullQueueEvents[3] = {0, 0, 0};
8181
std::mutex fullQueueWarningMutex[3];
8282

83-
void logQueueWarningEveryNowAndThen(int64_t events) {
83+
void logQueueWarningEveryNowAndThen(int64_t events, uint64_t maxQueueSize) {
8484
auto const now = std::chrono::steady_clock::now();
8585
uint64_t totalEvents;
8686
bool printLog = false;
@@ -99,13 +99,13 @@ void logQueueWarningEveryNowAndThen(int64_t events) {
9999

100100
if (printLog) {
101101
LOG_TOPIC("dead2", WARN, Logger::THREADS)
102-
<< "Scheduler queue"
102+
<< "Scheduler queue with max capacity " << maxQueueSize
103103
<< " is filled more than 50% in last " << sinceLast.count()
104-
<< "s. (happened " << totalEvents << " times since last message)";
104+
<< "s (happened " << totalEvents << " times since last message)";
105105
}
106106
}
107107

108-
void logQueueFullEveryNowAndThen(int64_t fifo) {
108+
void logQueueFullEveryNowAndThen(int64_t fifo, uint64_t maxQueueSize) {
109109
auto const& now = std::chrono::steady_clock::now();
110110
uint64_t events;
111111
bool printLog = false;
@@ -122,7 +122,8 @@ void logQueueFullEveryNowAndThen(int64_t fifo) {
122122

123123
if (printLog) {
124124
LOG_TOPIC("dead1", WARN, Logger::THREADS)
125-
<< "Scheduler queue " << fifo << " is full. (happened " << events
125+
<< "Scheduler queue " << fifo << " with max capacity " << maxQueueSize
126+
<< " is full (happened " << events
126127
<< " times since last message)";
127128
}
128129
}
@@ -153,7 +154,7 @@ class SupervisedSchedulerWorkerThread final : public SupervisedSchedulerThread {
153154
explicit SupervisedSchedulerWorkerThread(SupervisedScheduler& scheduler)
154155
: Thread("SchedWorker"), SupervisedSchedulerThread(scheduler) {}
155156
~SupervisedSchedulerWorkerThread() { shutdown(); }
156-
void run() override { _scheduler.runWorker(); };
157+
void run() override { _scheduler.runWorker(); }
157158
};
158159

159160
} // namespace arangodb
@@ -172,69 +173,92 @@ SupervisedScheduler::SupervisedScheduler(uint64_t minThreads, uint64_t maxThread
172173
_definitiveWakeupTime_ns(100000),
173174
_maxNumWorker(maxThreads),
174175
_numIdleWorker(minThreads),
175-
_maxFifoSize(maxQueueSize) {
176-
_queue[0].reserve(maxQueueSize);
177-
_queue[1].reserve(fifo1Size);
178-
_queue[2].reserve(fifo2Size);
176+
_maxFifoSize(maxQueueSize),
177+
_fifo1Size(fifo1Size),
178+
_fifo2Size(fifo2Size) {
179+
_queues[0].reserve(maxQueueSize);
180+
_queues[1].reserve(fifo1Size);
181+
_queues[2].reserve(fifo2Size);
179182
}
180183

181184
SupervisedScheduler::~SupervisedScheduler() {}
182185

183186
bool SupervisedScheduler::queue(RequestLane lane, std::function<void()> handler,
184187
bool allowDirectHandling) {
185-
if (!isDirectDeadlockLane(lane) && allowDirectHandling &&
186-
!ServerState::instance()->isClusterRole() && (_jobsSubmitted - _jobsDone) < 2) {
187-
_jobsSubmitted.fetch_add(1, std::memory_order_relaxed);
188-
_jobsDequeued.fetch_add(1, std::memory_order_relaxed);
189-
_jobsDirectExec.fetch_add(1, std::memory_order_release);
190-
try {
191-
handler();
192-
_jobsDone.fetch_add(1, std::memory_order_release);
193-
return true;
194-
} catch (...) {
195-
_jobsDone.fetch_add(1, std::memory_order_release);
196-
throw;
188+
if (!isDirectDeadlockLane(lane) &&
189+
allowDirectHandling &&
190+
!ServerState::instance()->isClusterRole()) {
191+
uint64_t const jobsDone = _jobsDone.load(std::memory_order_acquire);
192+
uint64_t const jobsSubmitted = _jobsSubmitted.load(std::memory_order_relaxed);
193+
if (jobsSubmitted - jobsDone < 2) {
194+
_jobsSubmitted.fetch_add(1, std::memory_order_relaxed);
195+
_jobsDequeued.fetch_add(1, std::memory_order_relaxed);
196+
_jobsDirectExec.fetch_add(1, std::memory_order_relaxed);
197+
try {
198+
handler();
199+
_jobsDone.fetch_add(1, std::memory_order_release);
200+
return true;
201+
} catch (...) {
202+
_jobsDone.fetch_add(1, std::memory_order_release);
203+
throw;
204+
}
197205
}
198206
}
199-
200-
size_t queueNo = static_cast<size_t>(PriorityRequestLane(lane));
207+
208+
auto work = std::make_unique<WorkItem>(std::move(handler));
209+
210+
// use memory order acquire to make sure, pushed item is visible
211+
uint64_t const jobsDone = _jobsDone.load(std::memory_order_acquire);
212+
uint64_t const jobsSubmitted = _jobsSubmitted.fetch_add(1, std::memory_order_relaxed);
213+
214+
// to make sure the queue length hasn't underflowed
215+
TRI_ASSERT(jobsDone <= jobsSubmitted);
216+
217+
uint64_t const approxQueueLength = jobsSubmitted - jobsDone;
218+
219+
size_t const queueNo = static_cast<size_t>(PriorityRequestLane(lane));
201220

202221
TRI_ASSERT(queueNo <= 2);
203222
TRI_ASSERT(isStopping() == false);
204223

205-
auto work = std::make_unique<WorkItem>(std::move(handler));
224+
if (!_queues[queueNo].bounded_push(work.get())) {
225+
_jobsSubmitted.fetch_sub(1, std::memory_order_release);
206226

207-
if (!_queue[queueNo].push(work.get())) {
208-
logQueueFullEveryNowAndThen(queueNo);
227+
uint64_t maxSize = _maxFifoSize;
228+
if (queueNo == 1) {
229+
maxSize = _fifo1Size;
230+
} else if (queueNo == 2) {
231+
maxSize = _fifo2Size;
232+
}
233+
LOG_TOPIC("98d94", DEBUG, Logger::THREADS) << "unable to push job to scheduler queue: queue is full";
234+
logQueueFullEveryNowAndThen(queueNo, maxSize);
209235
return false;
210236
}
237+
211238
// queue now has ownership for the WorkItem
212239
work.release();
213240

214-
static thread_local uint64_t lastSubmitTime_ns;
241+
static thread_local uint64_t lastSubmitTime_ns = 0;
215242

216-
// use memory order release to make sure, pushed item is visible
217-
uint64_t jobsSubmitted = _jobsSubmitted.fetch_add(1, std::memory_order_release);
218-
uint64_t approxQueueLength = jobsSubmitted - _jobsDone;
219243
uint64_t now_ns = getTickCount_ns();
220244
uint64_t sleepyTime_ns = now_ns - lastSubmitTime_ns;
221245
lastSubmitTime_ns = now_ns;
222246

223247
if (approxQueueLength > _maxFifoSize / 2) {
224-
if ((queueWarningTick++ & 0xFF) == 0) {
248+
if ((::queueWarningTick++ & 0xFF) == 0) {
225249
auto const& now = std::chrono::steady_clock::now();
226-
if (conditionQueueFullSince == time_point{}) {
227-
logQueueWarningEveryNowAndThen(queueWarningTick);
228-
conditionQueueFullSince = now;
229-
} else if (now - conditionQueueFullSince > std::chrono::seconds(5)) {
230-
logQueueWarningEveryNowAndThen(queueWarningTick);
231-
queueWarningTick = 0;
232-
conditionQueueFullSince = now;
250+
if (::conditionQueueFullSince == time_point{}) {
251+
logQueueWarningEveryNowAndThen(::queueWarningTick, _maxFifoSize);
252+
::conditionQueueFullSince = now;
253+
} else if (now - ::conditionQueueFullSince > std::chrono::seconds(5)) {
254+
logQueueWarningEveryNowAndThen(::queueWarningTick, _maxFifoSize);
255+
::queueWarningTick = 0;
256+
::conditionQueueFullSince = now;
233257
}
234258
}
235259
} else {
236-
queueWarningTick = 0;
237-
conditionQueueFullSince = time_point{};
260+
::queueWarningTick = 0;
261+
::conditionQueueFullSince = time_point{};
238262
}
239263

240264
bool doNotify = false;
@@ -263,9 +287,6 @@ bool SupervisedScheduler::start() {
263287
}
264288

265289
void SupervisedScheduler::shutdown() {
266-
// THIS IS WHAT WE SHOULD AIM FOR, BUT NOBODY CARES
267-
// TRI_ASSERT(_jobsSubmitted <= _jobsDone);
268-
269290
{
270291
std::unique_lock<std::mutex> guard(_mutex);
271292
_stopping = true;
@@ -275,8 +296,8 @@ void SupervisedScheduler::shutdown() {
275296
Scheduler::shutdown();
276297

277298
while (true) {
278-
auto jobsSubmitted = _jobsSubmitted.load();
279-
auto jobsDone = _jobsDone.load();
299+
auto jobsDone = _jobsDone.load(std::memory_order_acquire);
300+
auto jobsSubmitted = _jobsSubmitted.load(std::memory_order_relaxed);
280301

281302
if (jobsSubmitted <= jobsDone) {
282303
break;
@@ -344,7 +365,7 @@ void SupervisedScheduler::runWorker() {
344365
break;
345366
}
346367

347-
F438 _jobsDequeued++;
368+
_jobsDequeued.fetch_add(1, std::memory_order_relaxed);
348369

349370
state->_lastJobStarted = clock::now();
350371
state->_working = true;
@@ -372,8 +393,8 @@ void SupervisedScheduler::runSupervisor() {
372393

373394
while (!_stopping) {
374395
uint64_t jobsDone = _jobsDone.load(std::memory_order_acquire);
375-
uint64_t jobsSubmitted = _jobsSubmitted.load(std::memory_order_acquire);
376-
uint64_t jobsDequeued = _jobsDequeued.load(std::memory_order_acquire);
396+
uint64_t jobsSubmitted = _jobsSubmitted.load(std::memory_order_relaxed);
397+
uint64_t jobsDequeued = _jobsDequeued.load(std::memory_order_relaxed);
377398

378399
if (jobsDone == lastJobsDone && (jobsDequeued < jobsSubmitted)) {
379400
jobsStallingTick++;
@@ -485,8 +506,9 @@ bool SupervisedScheduler::canPullFromQueue(uint64_t queueIndex) const {
485506
// then a job gets done fast (eg dequeued++, done++)
486507
// and then we read done.
487508
uint64_t jobsDone = _jobsDone.load(std::memory_order_acquire);
488-
uint64_t jobsDequeued = _jobsDequeued.load(std::memory_order_acquire);
509+
uint64_t jobsDequeued = _jobsDequeued.load(std::memory_order_relaxed);
489510
TRI_ASSERT(jobsDequeued >= jobsDone);
511+
490512
switch (queueIndex) {
491513
case 0:
492514
// We can always! pull from high priority
@@ -511,7 +533,7 @@ std::unique_ptr<SupervisedScheduler::WorkItem> SupervisedScheduler::getWork(
511533
auto queueIdx = triesCount % 3;
512534
// Order of this if is important! First check if we are allowed to pull,
513535
// then really pull from queue
514-
if (canPullFromQueue(queueIdx) && _queue[queueIdx].pop(work)) {
536+
if (canPullFromQueue(queueIdx) && _queues[queueIdx].pop(work)) {
515537
return std::unique_ptr<WorkItem>(work);
516538
}
517539

@@ -537,7 +559,7 @@ std::unique_ptr<SupervisedScheduler::WorkItem> SupervisedScheduler::getWork(
537559
void SupervisedScheduler::startOneThread() {
538560
// TRI_ASSERT(_numWorkers < _maxNumWorker);
539561
if (_numWorkers + _abandonedWorkerStates.size() >= _maxNumWorker) {
540-
return; // do not add more threads, than maximum allows
562+
return; // do not add more threads than maximum allows
541563
}
542564

543565
std::unique_lock<std::mutex> guard(_mutexSupervisor);
@@ -622,7 +644,7 @@ Scheduler::QueueStatistics SupervisedScheduler::queueStatistics() const {
622644
uint64_t const numWorkers = _numWorkers.load(std::memory_order_relaxed);
623645

624646
// read _jobsDone first, so the differences of the counters cannot get negative
625-
uint64_t const jobsDone = _jobsDone.load(std::memory_order_relaxed);
647+
uint64_t const jobsDone = _jobsDone.load(std::memory_order_acquire);
626648
uint64_t const jobsDequeued = _jobsDequeued.load(std::memory_order_relaxed);
627649
uint64_t const jobsSubmitted = _jobsSubmitted.load(std::memory_order_relaxed);
628650

‎arangod/Scheduler/SupervisedScheduler.h

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ class SupervisedScheduler final : public Scheduler {
7777

7878
// Since the lockfree queue can only handle PODs, one has to wrap lambdas
7979
// in a container class and store pointers. -- Maybe there is a better way?
80-
boost::lockfree::queue<WorkItem*> _queue[3];
80+
boost::lockfree::queue<WorkItem*> _queues[3];
8181

8282
// aligning required to prevent false sharing - assumes cache line size is 64
8383
alignas(64) std::atomic<uint64_t> _jobsSubmitted;
@@ -144,7 +144,9 @@ class SupervisedScheduler final : public Scheduler {
144144
std::condition_variable _conditionSupervisor;
145145
std::unique_ptr<SupervisedSchedulerManagerThread> _manager;
146146

147-
size_t _maxFifoSize;
147+
uint64_t const _maxFifoSize;
148+
uint64_t const _fifo1Size;
149+
uint64_t const _fifo2Size;
148150

149151
std::unique_ptr<WorkItem> getWork(std::shared_ptr<WorkerState>& state);
150152

0 commit comments

Comments
 (0)
0