forked from ClickHouse/ClickHouse
-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathExecutionSpeedLimits.cpp
More file actions
148 lines (126 loc) · 6.42 KB
/
Copy pathExecutionSpeedLimits.cpp
File metadata and controls
148 lines (126 loc) · 6.42 KB
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
#include <QueryPipeline/ExecutionSpeedLimits.h>
#include <Common/ProfileEvents.h>
#include <Common/CurrentThread.h>
#include <IO/WriteHelpers.h>
#include <Common/Stopwatch.h>
#include <base/sleep.h>
namespace ProfileEvents
{
extern const Event ThrottlerSleepMicroseconds;
extern const Event OverflowBreak;
extern const Event OverflowThrow;
}
namespace DB
{
namespace ErrorCodes
{
extern const int TOO_SLOW;
extern const int LOGICAL_ERROR;
extern const int TIMEOUT_EXCEEDED;
}
static void limitProgressingSpeed(size_t total_progress_size, size_t max_speed_in_seconds, UInt64 total_elapsed_microseconds)
{
/// How much time to wait for the average speed to become `max_speed_in_seconds`.
UInt64 desired_microseconds = total_progress_size * 1000000 / max_speed_in_seconds;
if (desired_microseconds > total_elapsed_microseconds)
{
UInt64 sleep_microseconds = desired_microseconds - total_elapsed_microseconds;
/// Never sleep more than one second (it should be enough to limit speed for a reasonable amount,
/// and otherwise it's too easy to make query hang).
sleep_microseconds = std::min(static_cast<UInt64>(1000000), sleep_microseconds);
sleepForMicroseconds(sleep_microseconds);
ProfileEvents::increment(ProfileEvents::ThrottlerSleepMicroseconds, sleep_microseconds);
}
}
void ExecutionSpeedLimits::throttle(
size_t read_rows, size_t read_bytes,
size_t total_rows_to_read, UInt64 total_elapsed_microseconds,
OverflowMode timeout_overflow_mode) const
{
if ((min_execution_rps != 0 || max_execution_rps != 0
|| min_execution_bps != 0 || max_execution_bps != 0
|| (total_rows_to_read != 0 && timeout_before_checking_execution_speed != 0))
&& (static_cast<Int64>(total_elapsed_microseconds) > timeout_before_checking_execution_speed.totalMicroseconds()))
{
/// Do not count sleeps in throttlers
UInt64 throttler_sleep_microseconds = CurrentThread::getProfileEvents()[ProfileEvents::ThrottlerSleepMicroseconds];
double elapsed_seconds = 0;
if (total_elapsed_microseconds > throttler_sleep_microseconds)
elapsed_seconds = static_cast<double>(total_elapsed_microseconds - throttler_sleep_microseconds) / 1000000.0;
if (elapsed_seconds > 0)
{
auto rows_per_second = static_cast<double>(read_rows) / elapsed_seconds;
if (min_execution_rps && rows_per_second < static_cast<double>(min_execution_rps))
throw Exception(
ErrorCodes::TOO_SLOW,
"Query is executing too slow: {:.3f} rows/sec., minimum: {}",
rows_per_second,
min_execution_rps);
auto bytes_per_second = static_cast<double>(read_bytes) / elapsed_seconds;
if (min_execution_bps && bytes_per_second < static_cast<double>(min_execution_bps))
throw Exception(
ErrorCodes::TOO_SLOW,
"Query is executing too slow: {:.3f} bytes/sec., minimum: {}",
bytes_per_second,
min_execution_bps);
/// If the predicted execution time is longer than `max_estimated_execution_time`.
if (timeout_overflow_mode == OverflowMode::THROW && max_estimated_execution_time != 0 && total_rows_to_read && read_rows)
{
/// We impose additional constraints to make estimation more reliable.
/// At least 1% of the rows must be read, or at least 1% of the time must be spent.
const auto rows_threshold = total_rows_to_read / 100;
const double time_threshold = max_estimated_execution_time.totalSeconds() / 100.0;
if (read_rows > rows_threshold || elapsed_seconds > time_threshold)
{
double estimated_execution_time_seconds = elapsed_seconds * (static_cast<double>(total_rows_to_read) / static_cast<double>(read_rows));
if (estimated_execution_time_seconds > max_estimated_execution_time.totalSeconds())
throw Exception(
ErrorCodes::TOO_SLOW,
"Estimated query execution time ({:.3f} seconds) is too long. Maximum: {}. Estimated rows to process: {} ({} read in {:.3f} seconds).",
estimated_execution_time_seconds,
max_estimated_execution_time.totalSeconds(),
total_rows_to_read,
read_rows,
elapsed_seconds);
}
}
if (max_execution_rps && rows_per_second >= static_cast<double>(max_execution_rps))
limitProgressingSpeed(read_rows, max_execution_rps, total_elapsed_microseconds);
if (max_execution_bps && bytes_per_second >= static_cast<double>(max_execution_bps))
limitProgressingSpeed(read_bytes, max_execution_bps, total_elapsed_microseconds);
}
}
}
template <typename... Args>
static bool handleOverflowMode(OverflowMode mode, int code, FormatStringHelper<Args...> fmt, Args &&... args)
{
switch (mode)
{
case OverflowMode::THROW:
ProfileEvents::increment(ProfileEvents::OverflowThrow);
throw Exception(code, std::move(fmt), std::forward<Args>(args)...);
case OverflowMode::BREAK:
ProfileEvents::increment(ProfileEvents::OverflowBreak);
return false;
default:
throw Exception(ErrorCodes::LOGICAL_ERROR, "Unknown overflow mode");
}
}
bool ExecutionSpeedLimits::checkTimeLimit(const UInt64 & elapsed_ns, OverflowMode overflow_mode) const
{
if (max_execution_time != 0)
{
/// Compare in whole microseconds: converting the timeout to nanoseconds overflows for values
/// above ~584 years, which `max_execution_time` accepts. Dividing the elapsed time instead can
/// only fire the timeout up to a microsecond later - never before it.
if (elapsed_ns / 1000 > static_cast<UInt64>(max_execution_time.totalMicroseconds()))
return handleOverflowMode(
overflow_mode,
ErrorCodes::TIMEOUT_EXCEEDED,
"Timeout exceeded: elapsed {:.3f} ms, maximum: {:.3f} ms",
static_cast<double>(elapsed_ns) / 1000000ULL,
static_cast<double>(max_execution_time.totalMicroseconds()) / 1000);
}
return true;
}
}