1 | #include "PerformanceTest.h" |
2 | |
3 | #include <Core/Types.h> |
4 | #include <Common/CpuId.h> |
5 | #include <Common/quoteString.h> |
6 | #include <common/getMemoryAmount.h> |
7 | #include <DataStreams/copyData.h> |
8 | #include <DataStreams/NullBlockOutputStream.h> |
9 | #include <DataStreams/RemoteBlockInputStream.h> |
10 | #include <IO/ConnectionTimeouts.h> |
11 | #include <IO/ReadBufferFromFile.h> |
12 | #include <IO/ReadHelpers.h> |
13 | #include <IO/WriteBufferFromFile.h> |
14 | |
15 | #include <filesystem> |
16 | |
17 | #include "executeQuery.h" |
18 | |
19 | |
20 | namespace DB |
21 | { |
22 | |
23 | namespace ErrorCodes |
24 | { |
25 | extern const int NOT_IMPLEMENTED; |
26 | } |
27 | |
28 | namespace |
29 | { |
30 | void waitQuery(Connection & connection) |
31 | { |
32 | bool finished = false; |
33 | |
34 | while (true) |
35 | { |
36 | if (!connection.poll(1000000)) |
37 | continue; |
38 | |
39 | Packet packet = connection.receivePacket(); |
40 | switch (packet.type) |
41 | { |
42 | case Protocol::Server::EndOfStream: |
43 | finished = true; |
44 | break; |
45 | case Protocol::Server::Exception: |
46 | throw *packet.exception; |
47 | } |
48 | |
49 | if (finished) |
50 | break; |
51 | } |
52 | } |
53 | } |
54 | |
55 | namespace fs = std::filesystem; |
56 | |
57 | PerformanceTest::PerformanceTest( |
58 | const XMLConfigurationPtr & config_, |
59 | Connection & connection_, |
60 | const ConnectionTimeouts & timeouts_, |
61 | InterruptListener & interrupt_listener_, |
62 | const PerformanceTestInfo & test_info_, |
63 | Context & context_, |
64 | const std::vector<size_t> & queries_to_run_) |
65 | : config(config_) |
66 | , connection(connection_) |
67 | , timeouts(timeouts_) |
68 | , interrupt_listener(interrupt_listener_) |
69 | , test_info(test_info_) |
70 | , context(context_) |
71 | , queries_to_run(queries_to_run_) |
72 | , log(&Poco::Logger::get("PerformanceTest" )) |
73 | { |
74 | } |
75 | |
76 | bool PerformanceTest::checkPreconditions() const |
77 | { |
78 | if (!config->has("preconditions" )) |
79 | return true; |
80 | |
81 | Strings preconditions; |
82 | config->keys("preconditions" , preconditions); |
83 | size_t table_precondition_index = 0; |
84 | size_t cpu_precondition_index = 0; |
85 | |
86 | for (const std::string & precondition : preconditions) |
87 | { |
88 | if (precondition == "flush_disk_cache" ) |
89 | { |
90 | if (system( |
91 | "(>&2 echo 'Flushing disk cache...') && (sudo sh -c 'echo 3 > /proc/sys/vm/drop_caches') && (>&2 echo 'Flushed.')" )) |
92 | { |
93 | LOG_WARNING(log, "Failed to flush disk cache" ); |
94 | return false; |
95 | } |
96 | } |
97 | |
98 | if (precondition == "ram_size" ) |
99 | { |
100 | size_t ram_size_needed = config->getUInt64("preconditions.ram_size" ); |
101 | size_t actual_ram = getMemoryAmount(); |
102 | if (!actual_ram) |
103 | throw Exception("ram_size precondition not available on this platform" , ErrorCodes::NOT_IMPLEMENTED); |
104 | |
105 | if (ram_size_needed > actual_ram) |
106 | { |
107 | LOG_WARNING(log, "Not enough RAM: need = " << ram_size_needed << ", present = " << actual_ram); |
108 | return false; |
109 | } |
110 | } |
111 | |
112 | if (precondition == "table_exists" ) |
113 | { |
114 | std::string precondition_key = "preconditions.table_exists[" + std::to_string(table_precondition_index++) + "]" ; |
115 | std::string table_to_check = config->getString(precondition_key); |
116 | std::string query = "EXISTS TABLE " + table_to_check + ";" ; |
117 | |
118 | size_t exist = 0; |
119 | |
120 | connection.sendQuery(timeouts, query, "" , QueryProcessingStage::Complete, &test_info.settings, nullptr, false); |
121 | |
122 | while (true) |
123 | { |
124 | Packet packet = connection.receivePacket(); |
125 | |
126 | if (packet.type == Protocol::Server::Data) |
127 | { |
128 | for (const ColumnWithTypeAndName & column : packet.block) |
129 | { |
130 | if (column.name == "result" && column.column->size() > 0) |
131 | { |
132 | exist = column.column->get64(0); |
133 | if (exist) |
134 | break; |
135 | } |
136 | } |
137 | } |
138 | |
139 | if (packet.type == Protocol::Server::Exception |
140 | || packet.type == Protocol::Server::EndOfStream) |
141 | break; |
142 | } |
143 | |
144 | if (!exist) |
145 | { |
146 | LOG_WARNING(log, "Table " << backQuote(table_to_check) << " doesn't exist" ); |
147 | return false; |
148 | } |
149 | } |
150 | |
151 | if (precondition == "cpu" ) |
152 | { |
153 | std::string precondition_key = "preconditions.cpu[" + std::to_string(cpu_precondition_index++) + "]" ; |
154 | std::string flag_to_check = config->getString(precondition_key); |
155 | |
156 | #define CHECK_CPU_PRECONDITION(OP) \ |
157 | if (flag_to_check == #OP) \ |
158 | { \ |
159 | if (!Cpu::CpuFlagsCache::have_##OP) \ |
160 | { \ |
161 | LOG_WARNING(log, "CPU doesn't support " << #OP); \ |
162 | return false; \ |
163 | } \ |
164 | } else |
165 | |
166 | CPU_ID_ENUMERATE(CHECK_CPU_PRECONDITION) |
167 | { |
168 | LOG_WARNING(log, "CPU doesn't support " << flag_to_check); |
169 | return false; |
170 | } |
171 | |
172 | #undef CHECK_CPU_PRECONDITION |
173 | } |
174 | } |
175 | |
176 | return true; |
177 | } |
178 | |
179 | |
180 | UInt64 PerformanceTest::calculateMaxExecTime() const |
181 | { |
182 | |
183 | UInt64 result = 0; |
184 | for (const auto & stop_conditions : test_info.stop_conditions_by_run) |
185 | { |
186 | UInt64 condition_max_time = stop_conditions.getMaxExecTime(); |
187 | if (condition_max_time == 0) |
188 | return 0; |
189 | result += condition_max_time; |
190 | } |
191 | return result; |
192 | } |
193 | |
194 | |
195 | void PerformanceTest::prepare() const |
196 | { |
197 | for (const auto & query : test_info.create_and_fill_queries) |
198 | { |
199 | LOG_INFO(log, "Executing create or fill query \"" << query << '\"'); |
200 | connection.sendQuery(timeouts, query, "" , QueryProcessingStage::Complete, &test_info.settings, nullptr, false); |
201 | waitQuery(connection); |
202 | LOG_INFO(log, "Query finished" ); |
203 | } |
204 | |
205 | } |
206 | |
207 | void PerformanceTest::finish() const |
208 | { |
209 | for (const auto & query : test_info.drop_queries) |
210 | { |
211 | LOG_INFO(log, "Executing drop query \"" << query << '\"'); |
212 | connection.sendQuery(timeouts, query, "" , QueryProcessingStage::Complete, &test_info.settings, nullptr, false); |
213 | waitQuery(connection); |
214 | LOG_INFO(log, "Query finished" ); |
215 | } |
216 | } |
217 | |
218 | std::vector<TestStats> PerformanceTest::execute() |
219 | { |
220 | std::vector<TestStats> statistics_by_run; |
221 | size_t query_count; |
222 | if (queries_to_run.empty()) |
223 | query_count = test_info.queries.size(); |
224 | else |
225 | query_count = queries_to_run.size(); |
226 | size_t total_runs = test_info.times_to_run * test_info.queries.size(); |
227 | statistics_by_run.resize(total_runs); |
228 | LOG_INFO(log, "Totally will run cases " << test_info.times_to_run * query_count << " times" ); |
229 | UInt64 max_exec_time = calculateMaxExecTime(); |
230 | if (max_exec_time != 0) |
231 | LOG_INFO(log, "Test will be executed for a maximum of " << max_exec_time / 1000. << " seconds" ); |
232 | else |
233 | LOG_INFO(log, "Test execution time cannot be determined" ); |
234 | |
235 | for (size_t number_of_launch = 0; number_of_launch < test_info.times_to_run; ++number_of_launch) |
236 | { |
237 | QueriesWithIndexes queries_with_indexes; |
238 | |
239 | for (size_t query_index = 0; query_index < test_info.queries.size(); ++query_index) |
240 | { |
241 | if (queries_to_run.empty() || std::find(queries_to_run.begin(), queries_to_run.end(), query_index) != queries_to_run.end()) |
242 | { |
243 | size_t statistic_index = number_of_launch * test_info.queries.size() + query_index; |
244 | queries_with_indexes.push_back({test_info.queries[query_index], statistic_index}); |
245 | } |
246 | else |
247 | LOG_INFO(log, "Will skip query " << test_info.queries[query_index] << " by index" ); |
248 | } |
249 | |
250 | if (got_SIGINT) |
251 | break; |
252 | |
253 | runQueries(queries_with_indexes, statistics_by_run); |
254 | } |
255 | |
256 | if (got_SIGINT) |
257 | { |
258 | return statistics_by_run; |
259 | } |
260 | |
261 | // Pull memory usage data from query log. The log is normally filled in |
262 | // background, so we have to flush it synchronously here to see all the |
263 | // previous queries. |
264 | { |
265 | NullBlockOutputStream null_output(Block{}); |
266 | RemoteBlockInputStream flush_log(connection, "system flush logs" , |
267 | {} /* header */, context); |
268 | copyData(flush_log, null_output); |
269 | } |
270 | |
271 | for (auto & statistics : statistics_by_run) |
272 | { |
273 | if (statistics.query_id.empty()) |
274 | { |
275 | // We have statistics structs for skipped queries as well, so we |
276 | // have to filter them out. |
277 | continue; |
278 | } |
279 | |
280 | // We run some test queries several times, specifying the same query id, |
281 | // so this query to the log may return several records. Choose the |
282 | // last one, because this is when the query performance has stabilized. |
283 | RemoteBlockInputStream log_reader(connection, |
284 | "select memory_usage, query_start_time from system.query_log " |
285 | "where type = 2 and query_id = '" + statistics.query_id + "' " |
286 | "order by query_start_time desc" , |
287 | {} /* header */, context); |
288 | |
289 | log_reader.readPrefix(); |
290 | Block block = log_reader.read(); |
291 | if (block.columns() == 0) |
292 | { |
293 | LOG_WARNING(log, "Query '" << statistics.query_id << "' is not found in query log." ); |
294 | continue; |
295 | } |
296 | |
297 | auto column = block.getByName("memory_usage" ).column; |
298 | statistics.memory_usage = column->get64(0); |
299 | |
300 | log_reader.readSuffix(); |
301 | } |
302 | |
303 | return statistics_by_run; |
304 | } |
305 | |
306 | void PerformanceTest::runQueries( |
307 | const QueriesWithIndexes & queries_with_indexes, |
308 | std::vector<TestStats> & statistics_by_run) |
309 | { |
310 | for (const auto & [query, run_index] : queries_with_indexes) |
311 | { |
312 | LOG_INFO(log, "[" << run_index<< "] Run query '" << query << "'" ); |
313 | TestStopConditions & stop_conditions = test_info.stop_conditions_by_run[run_index]; |
314 | TestStats & statistics = statistics_by_run[run_index]; |
315 | statistics.startWatches(); |
316 | try |
317 | { |
318 | executeQuery(connection, query, statistics, stop_conditions, interrupt_listener, context, test_info.settings); |
319 | |
320 | if (test_info.exec_type == ExecutionType::Loop) |
321 | { |
322 | LOG_INFO(log, "Will run query in loop" ); |
323 | for (size_t iteration = 1; !statistics.got_SIGINT; ++iteration) |
324 | { |
325 | stop_conditions.reportIterations(iteration); |
326 | if (stop_conditions.areFulfilled()) |
327 | { |
328 | LOG_INFO(log, "Stop conditions fullfilled" ); |
329 | break; |
330 | } |
331 | |
332 | executeQuery(connection, query, statistics, stop_conditions, interrupt_listener, context, test_info.settings); |
333 | } |
334 | } |
335 | } |
336 | catch (const Exception & e) |
337 | { |
338 | statistics.exception = "Code: " + std::to_string(e.code()) + ", e.displayText() = " + e.displayText(); |
339 | LOG_WARNING(log, "Code: " << e.code() << ", e.displayText() = " << e.displayText() |
340 | << ", Stack trace:\n\n" << e.getStackTrace().toString()); |
341 | } |
342 | |
343 | if (!statistics.got_SIGINT) |
344 | statistics.ready = true; |
345 | else |
346 | { |
347 | got_SIGINT = true; |
348 | LOG_INFO(log, "Got SIGINT, will terminate as soon as possible" ); |
349 | break; |
350 | } |
351 | } |
352 | } |
353 | |
354 | |
355 | } |
356 | |