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
| //===-- BackgroundQueue.cpp - Task queue for background index -------------===//
//
// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
// See https://llvm.org/LICENSE.txt for license information.
// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
//
//===----------------------------------------------------------------------===//
#include "index/Background.h"
namespace clang {
namespace clangd {
static std::atomic<bool> PreventStarvation = {false};
void BackgroundQueue::preventThreadStarvationInTests() {
PreventStarvation.store(true);
}
void BackgroundQueue::work(std::function<void()> OnIdle) {
while (true) {
llvm::Optional<Task> Task;
{
std::unique_lock<std::mutex> Lock(Mu);
CV.wait(Lock, [&] { return ShouldStop || !Queue.empty(); });
if (ShouldStop) {
Queue.clear();
CV.notify_all();
return;
}
++NumActiveTasks;
std::pop_heap(Queue.begin(), Queue.end());
Task = std::move(Queue.back());
Queue.pop_back();
}
if (Task->ThreadPri != llvm::ThreadPriority::Default &&
!PreventStarvation.load())
llvm::set_thread_priority(Task->ThreadPri);
Task->Run();
if (Task->ThreadPri != llvm::ThreadPriority::Default)
llvm::set_thread_priority(llvm::ThreadPriority::Default);
{
std::unique_lock<std::mutex> Lock(Mu);
if (NumActiveTasks == 1 && Queue.empty() && OnIdle) {
// We just finished the last item, the queue is going idle.
Lock.unlock();
OnIdle();
Lock.lock();
}
assert(NumActiveTasks > 0 && "before decrementing");
--NumActiveTasks;
}
CV.notify_all();
}
}
void BackgroundQueue::stop() {
{
std::lock_guard<std::mutex> QueueLock(Mu);
ShouldStop = true;
}
CV.notify_all();
}
void BackgroundQueue::push(Task T) {
{
std::lock_guard<std::mutex> Lock(Mu);
T.QueuePri = std::max(T.QueuePri, Boosts.lookup(T.Tag));
Queue.push_back(std::move(T));
std::push_heap(Queue.begin(), Queue.end());
}
CV.notify_all();
}
void BackgroundQueue::append(std::vector<Task> Tasks) {
{
std::lock_guard<std::mutex> Lock(Mu);
for (Task &T : Tasks)
T.QueuePri = std::max(T.QueuePri, Boosts.lookup(T.Tag));
std::move(Tasks.begin(), Tasks.end(), std::back_inserter(Queue));
std::make_heap(Queue.begin(), Queue.end());
}
CV.notify_all();
}
void BackgroundQueue::boost(llvm::StringRef Tag, unsigned NewPriority) {
std::lock_guard<std::mutex> Lock(Mu);
unsigned &Boost = Boosts[Tag];
bool Increase = NewPriority > Boost;
Boost = NewPriority;
if (!Increase)
return; // existing tasks unaffected
unsigned Changes = 0;
for (Task &T : Queue)
if (Tag == T.Tag && NewPriority > T.QueuePri) {
T.QueuePri = NewPriority;
++Changes;
}
if (Changes)
std::make_heap(Queue.begin(), Queue.end());
// No need to signal, only rearranged items in the queue.
}
bool BackgroundQueue::blockUntilIdleForTest(
llvm::Optional<double> TimeoutSeconds) {
std::unique_lock<std::mutex> Lock(Mu);
return wait(Lock, CV, timeoutSeconds(TimeoutSeconds),
[&] { return Queue.empty() && NumActiveTasks == 0; });
}
} // namespace clangd
} // namespace clang
|