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
20namespace DB
21{
22
23namespace ErrorCodes
24{
25extern const int NOT_IMPLEMENTED;
26}
27
28namespace
29{
30void 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
55namespace fs = std::filesystem;
56
57PerformanceTest::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
76bool 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
180UInt64 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
195void 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
207void 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
218std::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
306void 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