blob: 9e18e7bcff3e2d57c1ccbb46c55c407ab320ccea (
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
|
#pragma once
#include <map>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <chrono>
#include <Common/CurrentMetrics.h>
namespace CurrentMetrics
{
extern const Metric QueryPreempted;
}
namespace DB
{
/** Implements query priorities in very primitive way.
* Allows to freeze query execution if at least one query of higher priority is executed.
*
* Priority value is integer, smaller means higher priority.
*
* Priority 0 is special - queries with that priority is always executed,
* not depends on other queries and not affect other queries.
* Thus 0 means - don't use priorities.
*
* NOTE Possibilities for improvement:
* - implement limit on maximum number of running queries with same priority.
*/
class QueryPriorities
{
public:
using Priority = int;
private:
friend struct Handle;
using Count = int;
/// Number of currently running queries for each priority.
using Container = std::map<Priority, Count>;
std::mutex mutex;
std::condition_variable condvar;
Container container;
/** If there are higher priority queries - sleep until they are finish or timeout happens.
*/
template <typename Duration>
void waitIfNeed(Priority priority, Duration timeout)
{
if (0 == priority)
return;
std::unique_lock lock(mutex);
/// Is there at least one more priority query?
bool found = false;
for (const auto & value : container)
{
if (value.first >= priority)
break;
if (value.second > 0)
{
found = true;
break;
}
}
if (!found)
return;
CurrentMetrics::Increment metric_increment{CurrentMetrics::QueryPreempted};
/// Spurious wakeups are Ok. We allow to wait less than requested.
condvar.wait_for(lock, timeout);
}
public:
struct HandleImpl
{
private:
QueryPriorities & parent;
QueryPriorities::Container::value_type & value;
public:
HandleImpl(QueryPriorities & parent_, QueryPriorities::Container::value_type & value_)
: parent(parent_), value(value_) {}
~HandleImpl()
{
{
std::lock_guard lock(parent.mutex);
--value.second;
}
parent.condvar.notify_all();
}
template <typename Duration>
void waitIfNeed(Duration timeout)
{
parent.waitIfNeed(value.first, timeout);
}
};
using Handle = std::shared_ptr<HandleImpl>;
/** Register query with specified priority.
* Returns an object that remove record in destructor.
*/
Handle insert(Priority priority)
{
if (0 == priority)
return {};
std::lock_guard lock(mutex);
auto it = container.emplace(priority, 0).first;
++it->second;
return std::make_shared<HandleImpl>(*this, *it);
}
};
}
|
