Branch data Line data Source code
1 : : /*
2 : : * Copyright (C) 2026 by Thun Lu. All rights reserved.
3 : : * Author: Thun Lu <thun.lu@zohomail.cn>
4 : : * Repo: https://github.com/thun-res/vlink
5 : : * _ __ __ _ __
6 : : * | | / / / / (_) ____ / /__
7 : : * | | / / / / / / / __ \ / //_/
8 : : * | |/ / / /___ / / / / / / / ,<
9 : : * |___/ /_____/ /_/ /_/ /_/ /_/|_|
10 : : *
11 : : * Licensed under the Apache License, Version 2.0 (the "License");
12 : : * you may not use this file except in compliance with the License.
13 : : * You may obtain a copy of the License at
14 : : *
15 : : * http://www.apache.org/licenses/LICENSE-2.0
16 : : *
17 : : * Unless required by applicable law or agreed to in writing, software
18 : : * distributed under the License is distributed on an "AS IS" BASIS,
19 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 : : * See the License for the specific language governing permissions and
21 : : * limitations under the License.
22 : : */
23 : :
24 : : #include "./extension/vdb_reader.h"
25 : :
26 : : #include <algorithm>
27 : : #include <atomic>
28 : : #include <cmath>
29 : : #include <cstdint>
30 : : #include <cstring>
31 : : #include <filesystem>
32 : : #include <fstream>
33 : : #include <memory>
34 : : #include <mutex>
35 : : #include <shared_mutex>
36 : : #include <string>
37 : : #include <string_view>
38 : : #include <unordered_map>
39 : : #include <utility>
40 : : #include <vector>
41 : :
42 : : #include "./base/condition_variable.h"
43 : : #include "./base/elapsed_timer.h"
44 : : #include "./base/helpers.h"
45 : : #include "./base/logger.h"
46 : : #include "./version.h"
47 : :
48 : : // json
49 : : #include <nlohmann/json.hpp>
50 : :
51 : : #ifdef VLINK_ENABLE_SQLITE
52 : : #include <sqlite3.h>
53 : : #endif
54 : :
55 : : #define ENABLE_DATABASE_TABLE_CHECK 0
56 : :
57 : : namespace vlink {
58 : :
59 : 24512 : [[maybe_unused]] static constexpr int get_column(int column) noexcept { return column; }
60 : :
61 : : [[maybe_unused]] static constexpr size_t kMaxTaskSize = 50000U;
62 : :
63 : : #ifdef VLINK_ENABLE_SQLITE
64 : : struct SqliteStmtFinalizer final {
65 : 1377 : void operator()(::sqlite3_stmt* stmt) const noexcept {
66 [ + - ]: 1377 : if (stmt) {
67 : 1377 : ::sqlite3_finalize(stmt);
68 : : }
69 : 1377 : }
70 : : };
71 : :
72 : : using SqliteStmtPtr = std::unique_ptr<::sqlite3_stmt, SqliteStmtFinalizer>;
73 : :
74 : 6271 : static std::string sqlite_column_text_or_empty(::sqlite3_stmt* stmt, int column) {
75 : 6271 : const auto* text = ::sqlite3_column_text(stmt, column);
76 : :
77 [ - + ]: 6271 : if (!text) {
78 : : return {}; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
79 : : }
80 : :
81 [ + - ]: 6271 : return {reinterpret_cast<const char*>(text), static_cast<size_t>(::sqlite3_column_bytes(stmt, column))};
82 : : }
83 : :
84 : 42 : static bool exec_repair_sql(::sqlite3* db, const char* sql, const char* what) {
85 : 42 : char* err_msg = nullptr;
86 [ + - ]: 42 : const int ret = ::sqlite3_exec(db, sql, nullptr, nullptr, &err_msg);
87 : :
88 [ - + ]: 42 : if VUNLIKELY (ret != SQLITE_OK) {
89 [ # # # # : 0 : CLOG_W("VDBReader: Failed to %s: %s.", what, err_msg ? err_msg : ::sqlite3_errmsg(db));
# # # # #
# ]
90 : :
91 [ # # ]: 0 : if (err_msg) {
92 [ # # ]: 0 : ::sqlite3_free(err_msg);
93 : : }
94 : :
95 : 0 : return false;
96 : : }
97 : :
98 : 42 : return true;
99 : : }
100 : :
101 : 6 : static bool repair_bag_metadata(::sqlite3* db) {
102 : : static constexpr const char* kCollectStats =
103 : : "CREATE TEMP TABLE _vlink_fix_stats AS SELECT url AS uid, COUNT(*) AS n, MIN(elapsed) AS mn, "
104 : : "MAX(elapsed) AS mx FROM VLinkDatas GROUP BY url;";
105 : : static constexpr const char* kRepairUrls =
106 : : "UPDATE VLinkUrls SET count = IFNULL((SELECT n FROM _vlink_fix_stats WHERE uid = VLinkUrls.id), 0), "
107 : : "freq = IFNULL((SELECT CASE WHEN mx > mn THEN n / ((mx - mn) / 1000000.0) ELSE 0 END FROM "
108 : : "_vlink_fix_stats WHERE uid = VLinkUrls.id), 0);";
109 : : static constexpr const char* kRepairHeader =
110 : : "UPDATE VLinkHeader SET count = IFNULL((SELECT SUM(n) FROM _vlink_fix_stats), 0), "
111 : : "duration = IFNULL((SELECT MAX(mx) FROM _vlink_fix_stats), 0), "
112 : : "complete = CASE WHEN (SELECT COUNT(*) FROM _vlink_fix_stats) > 0 THEN 1 ELSE complete END;";
113 : : static constexpr const char* kDropStats = "DROP TABLE IF EXISTS temp._vlink_fix_stats;";
114 : :
115 [ - + ]: 6 : if VUNLIKELY (!exec_repair_sql(db, kDropStats, "drop stale repair table")) {
116 : 0 : return false;
117 : : }
118 : :
119 [ - + ]: 6 : if VUNLIKELY (!exec_repair_sql(db, "BEGIN IMMEDIATE;", "begin repair transaction")) {
120 : 0 : return false;
121 : : }
122 : :
123 [ + - - + : 6 : if VUNLIKELY (!exec_repair_sql(db, kCollectStats, "collect repair statistics") ||
+ - - + -
+ ]
124 : : !exec_repair_sql(db, kRepairUrls, "repair url metadata") ||
125 : : !exec_repair_sql(db, kRepairHeader, "repair header metadata")) {
126 : 0 : exec_repair_sql(db, "ROLLBACK;", "rollback repair transaction");
127 : 0 : exec_repair_sql(db, kDropStats, "drop repair table");
128 : 0 : return false;
129 : : }
130 : :
131 [ - + ]: 6 : if VUNLIKELY (!exec_repair_sql(db, "COMMIT;", "commit repair transaction")) {
132 : 0 : exec_repair_sql(db, "ROLLBACK;", "rollback repair transaction");
133 : 0 : exec_repair_sql(db, kDropStats, "drop repair table");
134 : 0 : return false;
135 : : }
136 : :
137 : 6 : return exec_repair_sql(db, kDropStats, "drop repair table");
138 : : }
139 : : #endif
140 : :
141 : : // VDBReader::Impl
142 : : struct VDBReader::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
143 : : std::atomic<BagReader::Status> status{BagReader::kStopped};
144 : : std::atomic_bool stop_flag{false};
145 : : std::atomic_bool pause_flag{false};
146 : : std::atomic_bool pause_next_flag{false};
147 : : std::atomic_bool jump_flag{false};
148 : : std::atomic<int64_t> pause_elapsed{0};
149 : : std::atomic<int64_t> offset_elapsed{0};
150 : : std::atomic<int64_t> real_elapsed{0};
151 : : std::atomic<int64_t> extra_elapsed{0};
152 : : std::atomic<int64_t> begin_time{0};
153 : : std::atomic<double> rate{1.0};
154 : : std::atomic<int> times{1};
155 : : std::atomic_bool is_pending{false};
156 : : std::atomic<int> split_index{0};
157 : :
158 : : bool read_only{false};
159 : : bool try_to_fix{false};
160 : : bool enable_compress{false};
161 : :
162 : : std::string path;
163 : : BagReader::Info info;
164 : : std::vector<BagReader::Info::UrlMeta> raw_url_metas;
165 : : std::mutex mtx;
166 : : ConditionVariable cv;
167 : :
168 : : BagReader::Config config;
169 : : std::mutex config_mtx;
170 : : std::shared_mutex time_mtx;
171 : :
172 : : ElapsedTimer elapsed_timer{ElapsedTimer::kMicro};
173 : : ElapsedTimer pause_elapsed_timer{ElapsedTimer::kMicro};
174 : : ElapsedTimer offset_timer{ElapsedTimer::kMicro};
175 : : ElapsedTimer real_timer{ElapsedTimer::kMicro};
176 : :
177 : : BagReader::StatusCallback status_callback;
178 : : BagReader::ReadyCallback ready_callback;
179 : : BagReader::FinishCallback finish_callback;
180 : :
181 : : std::string update_sql_default_str;
182 : : std::string update_sql_time_str;
183 : :
184 : : int64_t total_start_timestamp_ns{-1};
185 : : bool total_has_completed{false};
186 : :
187 : : // database
188 : : #ifdef VLINK_ENABLE_SQLITE
189 : :
190 : : // WrapperFile
191 : : struct WrapperFile final {
192 : : std::string path;
193 : : ::sqlite3* db{nullptr};
194 : : ::sqlite3_stmt* stmt{nullptr};
195 : : int index{0};
196 : : int64_t start_timestamp_ns{0};
197 : : int64_t begin{0};
198 : : int64_t end{0};
199 : : std::unordered_map<std::string, int> url_to_id_map;
200 : : std::unordered_map<int, std::string> id_to_url_map;
201 : : bool has_idx_elapsed{false};
202 : : bool has_idx_url{false};
203 : : bool has_schema{false};
204 : : bool has_completed{false};
205 : : bool is_channel_broken{false};
206 : :
207 : 846 : WrapperFile() {
208 [ + - ]: 846 : url_to_id_map.reserve(128);
209 [ + - ]: 846 : id_to_url_map.reserve(128);
210 : 846 : }
211 : : };
212 : :
213 : : std::vector<WrapperFile> file_list;
214 : :
215 : : SqliteStmtPtr cursor_stmt;
216 : : int cursor_file_index{0};
217 : : int64_t cursor_begin_us{0};
218 : : int64_t cursor_end_us{0};
219 : : BagReader::Config cursor_config;
220 : : #endif
221 : : };
222 : :
223 : : // VDBReader
224 : 221 : VDBReader::VDBReader(const std::string& path, bool read_only, bool try_to_fix)
225 [ + - ]: 221 : : BagReader(path, read_only, try_to_fix), impl_(std::make_unique<Impl>()) {
226 [ + - + - ]: 221 : set_name("VDBReader");
227 : :
228 [ + - + - ]: 221 : url_ser_map().reserve(128);
229 [ + - + - ]: 221 : url_schema_type_map().reserve(128);
230 [ + - ]: 221 : impl_->update_sql_default_str.reserve(256);
231 [ + - ]: 221 : impl_->update_sql_time_str.reserve(256);
232 : :
233 : 221 : impl_->read_only = read_only;
234 : 221 : impl_->try_to_fix = try_to_fix;
235 : :
236 [ + + ]: 221 : open(path);
237 : :
238 : : #ifndef VLINK_ENABLE_SQLITE
239 : : VLOG_F("VDBReader: The compile macro VLINK_ENABLE_SQLITE is not turned on.");
240 : : #endif
241 : 231 : }
242 : :
243 : 211 : VDBReader::~VDBReader() {
244 [ + + ]: 211 : if (!impl_->stop_flag.load(std::memory_order_relaxed)) {
245 : 210 : do_stop();
246 : : }
247 : :
248 : 211 : quit(true);
249 : :
250 : 211 : impl_->cv.notify_one();
251 : :
252 : 211 : wait_for_quit();
253 : :
254 : 211 : detach_plugin();
255 : :
256 : 211 : close();
257 : 211 : }
258 : :
259 : 8 : void VDBReader::bind_bag_interface(const std::shared_ptr<BagPluginInterface>& bag_interface) {
260 : 8 : BagReader::bind_bag_interface(bag_interface);
261 : 8 : impl_->info.url_metas = impl_->raw_url_metas;
262 : 8 : process_url_metas(impl_->info.url_metas);
263 : 8 : rebuild_url_meta_lookup(impl_->info.url_metas);
264 : 8 : }
265 : :
266 : 2 : void VDBReader::register_status_callback(StatusCallback&& status_callback) {
267 : 2 : impl_->status_callback = std::move(status_callback);
268 : 2 : }
269 : :
270 : 4 : void VDBReader::register_ready_callback(ReadyCallback&& ready_callback) {
271 : 4 : impl_->ready_callback = std::move(ready_callback);
272 : 4 : }
273 : :
274 : 14 : void VDBReader::register_finish_callback(FinishCallback&& finish_callback) {
275 : 14 : impl_->finish_callback = std::move(finish_callback);
276 : 14 : }
277 : :
278 : 15 : void VDBReader::register_output_callback(OutputCallback&& output_callback) {
279 : 15 : BagReader::register_output_callback(std::move(output_callback));
280 : 15 : }
281 : :
282 : 27 : void VDBReader::play(const Config& config) {
283 : : #ifdef VLINK_ENABLE_SQLITE
284 : :
285 [ + - - + ]: 27 : if VUNLIKELY (is_busy()) {
286 [ # # # # : 0 : VLOG_W("VDBReader: Is busy.");
# # ]
287 : : }
288 : :
289 [ + + ]: 27 : if (config.skip_blank) {
290 : 2 : impl_->begin_time.store(std::max(config.begin_time, impl_->info.blank_duration), std::memory_order_relaxed);
291 : : } else {
292 : 25 : impl_->begin_time.store(config.begin_time, std::memory_order_relaxed);
293 : : }
294 : :
295 [ + + ]: 27 : if (config.rate <= 0) {
296 : 1 : impl_->rate.store(1, std::memory_order_relaxed);
297 : : } else {
298 : 26 : impl_->rate.store(config.rate, std::memory_order_relaxed);
299 : : }
300 : :
301 : 27 : impl_->times.store(config.times, std::memory_order_relaxed);
302 : :
303 : 54 : impl_->real_elapsed.store(impl_->begin_time.load(std::memory_order_relaxed) * 1000U, std::memory_order_relaxed);
304 : 27 : impl_->is_pending.store(true, std::memory_order_relaxed);
305 : :
306 : : {
307 [ + - ]: 27 : std::unique_lock lock(impl_->mtx);
308 : 27 : impl_->stop_flag.store(false, std::memory_order_relaxed);
309 : 27 : impl_->pause_flag.store(false, std::memory_order_relaxed);
310 : 27 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
311 : 27 : impl_->jump_flag.store(false, std::memory_order_relaxed);
312 : 27 : }
313 : :
314 : 27 : Config config_copy;
315 : :
316 : : {
317 [ + - ]: 27 : std::unique_lock lock(impl_->config_mtx);
318 [ + - ]: 27 : impl_->config = config;
319 [ + - ]: 27 : config_copy = impl_->config;
320 : 27 : }
321 : :
322 [ + - + - ]: 54 : post_task([this, config_copy = std::move(config_copy)]() { read(config_copy); });
323 : : #else
324 : : (void)config;
325 : : #endif
326 : 27 : }
327 : :
328 : 4 : void VDBReader::stop() { do_stop(); }
329 : :
330 : 2 : void VDBReader::pause() {
331 : : #ifdef VLINK_ENABLE_SQLITE
332 : : {
333 [ + - ]: 2 : std::unique_lock lock(impl_->mtx);
334 : 2 : impl_->pause_flag.store(true, std::memory_order_relaxed);
335 : 2 : }
336 : :
337 : 2 : impl_->cv.notify_one();
338 : : #endif
339 : 2 : }
340 : :
341 : 2 : void VDBReader::resume() {
342 : : #ifdef VLINK_ENABLE_SQLITE
343 : : {
344 [ + - ]: 2 : std::unique_lock lock(impl_->mtx);
345 : 2 : impl_->pause_flag.store(false, std::memory_order_relaxed);
346 : 2 : }
347 : :
348 : 2 : impl_->cv.notify_one();
349 : : #endif
350 : 2 : }
351 : :
352 : 10 : void VDBReader::pause_to_next() {
353 : : #ifdef VLINK_ENABLE_SQLITE
354 : : {
355 [ + - ]: 10 : std::unique_lock lock(impl_->mtx);
356 : :
357 [ + + ]: 10 : if (!impl_->pause_flag.load(std::memory_order_relaxed)) {
358 : 1 : return;
359 : : }
360 : :
361 : 9 : impl_->pause_next_flag.store(true, std::memory_order_relaxed);
362 [ + + ]: 10 : }
363 : :
364 : 9 : impl_->cv.notify_one();
365 : : #endif
366 : : }
367 : :
368 : 7 : void VDBReader::jump(int64_t begin_time, double rate, int times, bool force_to_play) {
369 : : #ifdef VLINK_ENABLE_SQLITE
370 : :
371 [ + + ]: 7 : if (begin_time < 0) {
372 : 1 : begin_time = 0;
373 [ + + ]: 6 : } else if (begin_time > impl_->info.total_duration) {
374 : 1 : begin_time = std::max<int64_t>(0, impl_->info.total_duration - 100);
375 : : }
376 : :
377 : 7 : impl_->real_elapsed.store(begin_time * 1000U, std::memory_order_relaxed);
378 : 7 : impl_->is_pending.store(true, std::memory_order_relaxed);
379 : :
380 : 7 : bool last_pause_flag = impl_->pause_flag.load(std::memory_order_relaxed);
381 : :
382 : : {
383 [ + - ]: 7 : std::unique_lock lock(impl_->mtx);
384 : 7 : impl_->stop_flag.store(false, std::memory_order_relaxed);
385 : 7 : impl_->pause_flag.store(false, std::memory_order_relaxed);
386 : 7 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
387 : 7 : impl_->jump_flag.store(true, std::memory_order_relaxed);
388 : 7 : }
389 : :
390 : 7 : impl_->cv.notify_one();
391 : :
392 [ + + ]: 14 : for (const auto& wrapper_file : impl_->file_list) {
393 [ + - ]: 7 : if (wrapper_file.db) {
394 [ + - ]: 7 : ::sqlite3_interrupt(wrapper_file.db);
395 : : }
396 : : }
397 : :
398 [ + - ]: 7 : wait_for_idle();
399 : :
400 : 7 : impl_->begin_time.store(begin_time, std::memory_order_relaxed);
401 : :
402 [ + + ]: 7 : if (rate <= 0) {
403 : 3 : impl_->rate.store(1, std::memory_order_relaxed);
404 : : } else {
405 : 4 : impl_->rate.store(rate, std::memory_order_relaxed);
406 : : }
407 : :
408 : 7 : impl_->times.store(times, std::memory_order_relaxed);
409 : :
410 : : {
411 [ + - ]: 7 : std::unique_lock lock(impl_->mtx);
412 : 7 : impl_->stop_flag.store(false, std::memory_order_relaxed);
413 [ + + ]: 8 : impl_->pause_flag.store(force_to_play ? false : last_pause_flag, std::memory_order_relaxed);
414 : 7 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
415 : 7 : impl_->jump_flag.store(false, std::memory_order_relaxed);
416 : 7 : }
417 : :
418 : 7 : Config config_copy;
419 : :
420 : : {
421 [ + - ]: 7 : std::unique_lock lock(impl_->config_mtx);
422 [ + - ]: 7 : config_copy = impl_->config;
423 : 7 : }
424 : :
425 : 7 : config_copy.auto_pause = false;
426 [ + - + - ]: 14 : post_task([this, config_copy = std::move(config_copy)]() { read(config_copy); });
427 : : #else
428 : : (void)begin_time;
429 : : (void)rate;
430 : : (void)times;
431 : : (void)force_to_play;
432 : : #endif
433 : 8 : }
434 : :
435 : 59 : std::future<bool> VDBReader::check() {
436 : : #ifdef VLINK_ENABLE_SQLITE
437 : :
438 [ + + ]: 59 : if VUNLIKELY (is_busy()) {
439 [ + - + - : 2 : VLOG_W("VDBReader: Is busy.");
+ - ]
440 : : }
441 : :
442 : 868 : return invoke_task([this]() {
443 : 59 : int ret = 0;
444 : :
445 [ + + ]: 59 : if (!impl_->total_has_completed) {
446 [ + - + - : 2 : VLOG_W("VDBReader: Incomplete data detected.");
+ - ]
447 : 1 : return false;
448 : : }
449 : :
450 [ + + ]: 138 : for (auto& wrapper_file : impl_->file_list) {
451 [ - + ]: 80 : if VUNLIKELY (!wrapper_file.db) {
452 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
453 : : }
454 : :
455 [ + + ]: 80 : if (wrapper_file.stmt) {
456 [ + - ]: 2 : ::sqlite3_finalize(wrapper_file.stmt);
457 : 2 : wrapper_file.stmt = nullptr;
458 : : }
459 : :
460 : 80 : ::sqlite3_stmt* integrity_stmt = nullptr;
461 [ + - ]: 80 : ret = ::sqlite3_prepare_v2(wrapper_file.db, "PRAGMA integrity_check;", -1, &integrity_stmt, nullptr);
462 : 80 : SqliteStmtPtr integrity_stmt_guard(integrity_stmt);
463 : :
464 [ + - - + ]: 80 : if VUNLIKELY (is_ready_to_quit()) {
465 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
466 [ - + ]: 80 : } else if VUNLIKELY (ret != SQLITE_OK) {
467 : : CLOG_W("Failed to prepare integrity check: %s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
468 : : ::sqlite3_errmsg(wrapper_file.db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
469 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
470 : : }
471 : :
472 : 80 : bool has_integrity_result = false;
473 : 80 : int step_ret = SQLITE_OK;
474 : : for (;;) {
475 [ + - ]: 160 : step_ret = ::sqlite3_step(integrity_stmt);
476 : :
477 [ + + ]: 160 : if (step_ret != SQLITE_ROW) {
478 : 80 : break;
479 : : }
480 : :
481 : 80 : has_integrity_result = true;
482 [ + - ]: 80 : const auto integrity_result = sqlite_column_text_or_empty(integrity_stmt, get_column(0));
483 : :
484 [ + - - + ]: 80 : if VUNLIKELY (integrity_result != "ok") {
485 : : CLOG_W("Failed integrity check: %s.", integrity_result.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
486 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
487 : : }
488 [ + - ]: 160 : }
489 : :
490 [ + - - + : 80 : if VUNLIKELY (step_ret != SQLITE_DONE || !has_integrity_result) {
- + ]
491 : : CLOG_W("Failed to read integrity check: %s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
492 : : ::sqlite3_errmsg(wrapper_file.db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
493 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
494 : : }
495 : :
496 : 80 : integrity_stmt_guard.reset();
497 : :
498 [ + - ]: 80 : ret = ::sqlite3_prepare_v2(wrapper_file.db, "SELECT * FROM VLinkDatas;", -1, &wrapper_file.stmt, nullptr);
499 : :
500 [ + - - + ]: 80 : if VUNLIKELY (is_ready_to_quit()) {
501 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
502 [ - + ]: 80 : } else if VUNLIKELY (ret != SQLITE_OK) {
503 : : CLOG_W("Failed to prepare datas table: %s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
504 : : ::sqlite3_errmsg(wrapper_file.db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
505 : :
506 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
507 : : }
508 : :
509 [ + - ]: 80 : ::sqlite3_step(wrapper_file.stmt);
510 [ + - ]: 80 : }
511 : :
512 : 58 : bool is_ok = true;
513 : :
514 [ + + ]: 58 : if VUNLIKELY (impl_->info.total_duration < impl_->info.blank_duration) {
515 [ + - + - : 4 : VLOG_W("VDBReader: Invalid duration, blank=", impl_->info.blank_duration, " total=", impl_->info.total_duration,
+ - ]
516 : : ".");
517 : 2 : is_ok = false;
518 : : }
519 : :
520 [ + + + + : 58 : if VUNLIKELY (impl_->info.message_count > 0 && impl_->info.url_metas.empty()) {
+ + ]
521 [ + - + - : 4 : VLOG_W("VDBReader: Message count is ", impl_->info.message_count, " but url meta list is empty.");
+ - ]
522 : 2 : is_ok = false;
523 : : }
524 : :
525 : 58 : size_t total_count = 0;
526 : 58 : size_t total_raw_size = 0;
527 : :
528 [ + + ]: 162 : for (const auto& url_meta : impl_->info.url_metas) {
529 : 104 : total_count += url_meta.count;
530 : 104 : total_raw_size += url_meta.size;
531 : :
532 [ - + ]: 104 : if VUNLIKELY (!url_meta.valid) {
533 : : CLOG_W("VDBReader: Invalid url meta detected at index=%d.", url_meta.index); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
534 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
535 : : }
536 : :
537 [ + + ]: 104 : if VUNLIKELY (url_meta.url.empty()) {
538 [ + - + - : 4 : CLOG_W("VDBReader: Empty url detected at index=%d.", url_meta.index);
+ - ]
539 : 2 : is_ok = false;
540 : : }
541 : :
542 [ + + ]: 104 : if VUNLIKELY (url_meta.url_type.empty()) {
543 [ + - + - : 4 : CLOG_W("VDBReader: Empty url_type detected for url=%s.", url_meta.url.c_str());
+ - ]
544 : 2 : is_ok = false;
545 : : }
546 : :
547 [ + + + + : 104 : if VUNLIKELY (url_meta.count > 0 && url_meta.ser_type.empty()) {
+ + ]
548 [ + - + - : 6 : CLOG_W("VDBReader: Empty ser_type detected for url=%s.", url_meta.url.c_str());
+ - ]
549 : 3 : is_ok = false;
550 : : }
551 : :
552 [ - + ]: 104 : if VUNLIKELY (!SchemaData::is_valid_type(url_meta.schema_type)) {
553 : : CLOG_W("VDBReader: Invalid schema_type=%d detected for url=%s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
554 : : static_cast<int>(url_meta.schema_type), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
555 : : url_meta.url.c_str());
556 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
557 : : }
558 : :
559 : 104 : auto inferred_schema_type = SchemaData::infer_ser_type(url_meta.ser_type);
560 : :
561 [ + + + + : 104 : if VUNLIKELY (url_meta.schema_type == SchemaType::kUnknown && inferred_schema_type != SchemaType::kUnknown) {
+ + ]
562 : 1 : const auto schema_label = SchemaData::convert_type(inferred_schema_type);
563 [ + - + - : 2 : CLOG_W("VDBReader: Missing schema_type for url=%s, inferred=%.*s.", url_meta.url.c_str(),
+ - ]
564 : : static_cast<int>(schema_label.size()), schema_label.data());
565 : 1 : is_ok = false;
566 : : }
567 : :
568 [ + - + + : 104 : if VUNLIKELY (!std::isfinite(url_meta.loss) ||
+ + + + +
+ + + +
+ ]
569 : : (url_meta.loss != -1.0 && (url_meta.loss < 0.0 || url_meta.loss > 1.0))) {
570 [ + - + - : 8 : CLOG_W("VDBReader: Invalid loss=%f detected for url=%s.", url_meta.loss, url_meta.url.c_str());
+ - ]
571 : 4 : is_ok = false;
572 : : }
573 : :
574 [ + + ]: 104 : if VUNLIKELY (url_meta.freq < 0.0) {
575 [ + - + - : 6 : CLOG_W("VDBReader: Invalid freq=%f detected for url=%s.", url_meta.freq, url_meta.url.c_str());
+ - ]
576 : 3 : is_ok = false;
577 : : }
578 : : }
579 : :
580 [ + + + + : 115 : if ((!impl_->info.url_metas.empty() || impl_->info.message_count != 0) &&
+ + ]
581 [ + + ]: 57 : total_count != static_cast<size_t>(impl_->info.message_count)) {
582 [ + - + - : 10 : VLOG_W("VDBReader: Message count mismatch, header=", impl_->info.message_count, " metas=", total_count, ".");
+ - ]
583 : 5 : is_ok = false;
584 : : }
585 : :
586 [ + + - + : 113 : if ((!impl_->info.url_metas.empty() || impl_->info.total_raw_size != 0) &&
- + ]
587 [ - + ]: 55 : total_raw_size != static_cast<size_t>(impl_->info.total_raw_size)) {
588 [ # # # # : 0 : VLOG_W("VDBReader: Raw size mismatch, header=", impl_->info.total_raw_size,
# # ]
589 : : " metas=", total_raw_size, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
590 : : "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
591 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
592 : : }
593 : :
594 [ + - + + ]: 88 : for (const auto& schema_data : detect_schema()) {
595 [ - + ]: 30 : if VUNLIKELY (schema_data.name.empty()) {
596 : : CLOG_W("VDBReader: Empty schema name detected."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
597 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
598 : : }
599 : :
600 [ + + ]: 30 : if VUNLIKELY (schema_data.encoding.empty()) {
601 [ + - + - : 2 : CLOG_W("VDBReader: Empty schema encoding detected for name=%s.", schema_data.name.c_str());
+ - ]
602 : 1 : is_ok = false;
603 : : }
604 : :
605 [ + - + + : 30 : if VUNLIKELY (!SchemaData::is_valid_type(schema_data.schema_type) ||
+ + ]
606 : : schema_data.schema_type == SchemaType::kUnknown) {
607 [ + - + - : 4 : CLOG_W("VDBReader: Invalid schema_type=%d detected for schema=%s.", static_cast<int>(schema_data.schema_type),
+ - ]
608 : : schema_data.name.c_str());
609 : 2 : is_ok = false;
610 : : }
611 : 58 : }
612 : :
613 : 58 : return is_ok;
614 [ + - ]: 59 : });
615 : : #else
616 : : return std::future<bool>();
617 : : #endif
618 : : }
619 : :
620 : 5 : std::future<bool> VDBReader::reindex() {
621 : : #ifdef VLINK_ENABLE_SQLITE
622 : :
623 [ + + ]: 5 : if VUNLIKELY (is_busy()) {
624 [ + - + - : 2 : VLOG_W("VDBReader: Is busy.");
+ - ]
625 : : }
626 : :
627 : 41 : return invoke_task([this]() {
628 : 5 : int ret = 0;
629 : 5 : char* err_msg = nullptr;
630 : :
631 [ + + ]: 11 : for (auto& wrapper_file : impl_->file_list) {
632 [ - + ]: 6 : if VUNLIKELY (!wrapper_file.db) {
633 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
634 : : }
635 : :
636 [ + + ]: 6 : if (wrapper_file.stmt) {
637 [ + - ]: 5 : ::sqlite3_finalize(wrapper_file.stmt);
638 : 5 : wrapper_file.stmt = nullptr;
639 : : }
640 : :
641 [ + - ]: 6 : ret = ::sqlite3_exec(wrapper_file.db, "DROP INDEX IF EXISTS idx_elapsed;", nullptr, nullptr, &err_msg);
642 : :
643 [ + - + - : 6 : if VUNLIKELY (!is_ready_to_quit() && ret != SQLITE_OK) {
- + - + ]
644 : : // LCOV_EXCL_START GCOVR_EXCL_START
645 : : CLOG_W("Failed to drop idx_elapsed: %s.", err_msg);
646 : :
647 : : if (err_msg) {
648 : : ::sqlite3_free(err_msg);
649 : : err_msg = nullptr;
650 : : }
651 : :
652 : : return false;
653 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
654 : : }
655 : :
656 [ + - ]: 6 : ret = ::sqlite3_exec(wrapper_file.db, "DROP INDEX IF EXISTS idx_url;", nullptr, nullptr, &err_msg);
657 : :
658 [ + - + - : 6 : if VUNLIKELY (!is_ready_to_quit() && ret != SQLITE_OK) {
- + - + ]
659 : : // LCOV_EXCL_START GCOVR_EXCL_START
660 : : CLOG_W("Failed to drop idx_url: %s.", err_msg);
661 : :
662 : : if (err_msg) {
663 : : ::sqlite3_free(err_msg);
664 : : err_msg = nullptr;
665 : : }
666 : :
667 : : return false;
668 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
669 : : }
670 : :
671 [ + - ]: 6 : ret = ::sqlite3_exec(wrapper_file.db, "DROP INDEX IF EXISTS idx_elapsed_url;", nullptr, nullptr, &err_msg);
672 : :
673 [ + - + - : 6 : if VUNLIKELY (!is_ready_to_quit() && ret != SQLITE_OK) {
- + - + ]
674 : : // LCOV_EXCL_START GCOVR_EXCL_START
675 : : CLOG_W("Failed to drop idx_elapsed_url: %s.", err_msg);
676 : :
677 : : if (err_msg) {
678 : : ::sqlite3_free(err_msg);
679 : : err_msg = nullptr;
680 : : }
681 : :
682 : : return false;
683 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
684 : : }
685 : :
686 [ + - ]: 6 : ret = ::sqlite3_exec(wrapper_file.db, "CREATE INDEX idx_elapsed_url ON VLinkDatas(elapsed, url);", nullptr,
687 : : nullptr, &err_msg);
688 : :
689 [ + - - + ]: 6 : if VUNLIKELY (is_ready_to_quit()) {
690 : : // LCOV_EXCL_START GCOVR_EXCL_START
691 : : if (err_msg) {
692 : : ::sqlite3_free(err_msg);
693 : : err_msg = nullptr;
694 : : }
695 : :
696 : : return false;
697 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
698 [ - + ]: 6 : } else if VUNLIKELY (ret != SQLITE_OK) {
699 : : // LCOV_EXCL_START GCOVR_EXCL_START
700 : : CLOG_W("Failed to create idx_elapsed_url: %s.", err_msg);
701 : :
702 : : if (err_msg) {
703 : : ::sqlite3_free(err_msg);
704 : : err_msg = nullptr;
705 : : }
706 : :
707 : : return false;
708 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
709 : : }
710 : :
711 [ + - ]: 6 : ret = ::sqlite3_exec(wrapper_file.db, "PRAGMA optimize;", nullptr, nullptr, &err_msg);
712 : :
713 [ + - - + ]: 6 : if VUNLIKELY (is_ready_to_quit()) {
714 : : // LCOV_EXCL_START GCOVR_EXCL_START
715 : : if (err_msg) {
716 : : ::sqlite3_free(err_msg);
717 : : err_msg = nullptr;
718 : : }
719 : :
720 : : return false;
721 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
722 [ - + ]: 6 : } else if VUNLIKELY (ret != SQLITE_OK) {
723 : : // LCOV_EXCL_START GCOVR_EXCL_START
724 : : CLOG_W("Failed to optimize: %s.", err_msg);
725 : :
726 : : if (err_msg) {
727 : : ::sqlite3_free(err_msg);
728 : : err_msg = nullptr;
729 : : }
730 : :
731 : : return false;
732 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
733 : : }
734 : :
735 [ + - ]: 6 : ret = ::sqlite3_prepare_v2(wrapper_file.db, "SELECT * FROM VLinkDatas;", -1, &wrapper_file.stmt, nullptr);
736 : :
737 [ + - - + ]: 6 : if VUNLIKELY (is_ready_to_quit()) {
738 : : // LCOV_EXCL_START GCOVR_EXCL_START
739 : : if (err_msg) {
740 : : ::sqlite3_free(err_msg);
741 : : err_msg = nullptr;
742 : : }
743 : :
744 : : return false;
745 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
746 [ - + ]: 6 : } else if VUNLIKELY (ret != SQLITE_OK) {
747 : : // LCOV_EXCL_START GCOVR_EXCL_START
748 : : CLOG_W("Failed to prepare datas table: %s.", ::sqlite3_errmsg(wrapper_file.db));
749 : :
750 : : if (err_msg) {
751 : : ::sqlite3_free(err_msg);
752 : : err_msg = nullptr;
753 : : }
754 : :
755 : : return false;
756 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
757 : : }
758 : :
759 [ + - ]: 6 : ::sqlite3_step(wrapper_file.stmt);
760 : :
761 [ - + ]: 6 : if (err_msg) {
762 : : ::sqlite3_free(err_msg); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
763 : : err_msg = nullptr; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
764 : : }
765 : : }
766 : :
767 : 5 : return true;
768 [ + - ]: 5 : });
769 : : #else
770 : : return std::future<bool>();
771 : : #endif
772 : : }
773 : :
774 : 9 : std::future<bool> VDBReader::fix(bool rebuild) {
775 : : #ifdef VLINK_ENABLE_SQLITE
776 : :
777 [ + + ]: 9 : if VUNLIKELY (is_busy()) {
778 [ + - + - : 2 : VLOG_W("VDBReader: Is busy.");
+ - ]
779 : : }
780 : :
781 [ + - + + ]: 9 : const bool can_repair_metadata = !impl_->read_only && !is_split_mode();
782 : :
783 [ - + ]: 9 : if (impl_->read_only) {
784 [ # # # # : 0 : VLOG_W("VDBReader: Read-only reader, metadata repair skipped.");
# # ]
785 [ + + ]: 9 : } else if (is_split_mode()) {
786 [ + - + - : 6 : VLOG_W("VDBReader: Split bag, metadata repair skipped; the index file is not rewritten.");
+ - ]
787 : : }
788 : :
789 : 168 : return invoke_task([this, rebuild, can_repair_metadata]() {
790 : 9 : int ret = 0;
791 : 9 : char* err_msg = nullptr;
792 : :
793 [ + + ]: 21 : for (auto& wrapper_file : impl_->file_list) {
794 [ - + ]: 12 : if VUNLIKELY (!wrapper_file.db) {
795 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
796 : : }
797 : :
798 [ + + ]: 12 : if (wrapper_file.stmt) {
799 [ + - ]: 8 : ::sqlite3_finalize(wrapper_file.stmt);
800 : 8 : wrapper_file.stmt = nullptr;
801 : : }
802 : :
803 [ + + + - : 12 : if (can_repair_metadata && !repair_bag_metadata(wrapper_file.db)) {
- + - + ]
804 : 0 : return false;
805 : : }
806 : :
807 [ + - - + ]: 12 : if VUNLIKELY (is_ready_to_quit()) {
808 : 0 : return false;
809 : : }
810 : :
811 : : // opt
812 : : {
813 [ + - ]: 12 : ret = ::sqlite3_exec(wrapper_file.db, "PRAGMA synchronous = OFF;", nullptr, nullptr, &err_msg);
814 : :
815 [ + - + - : 12 : if VUNLIKELY (!is_ready_to_quit() && ret != SQLITE_OK) {
- + - + ]
816 : : // LCOV_EXCL_START GCOVR_EXCL_START
817 : : CLOG_W("Failed to set synchronous: %s.", err_msg);
818 : :
819 : : if (err_msg) {
820 : : ::sqlite3_free(err_msg);
821 : : err_msg = nullptr;
822 : : }
823 : :
824 : : return false;
825 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
826 : : }
827 : :
828 [ + - ]: 12 : ret = ::sqlite3_exec(wrapper_file.db, "PRAGMA temp_store = MEMORY;", nullptr, nullptr, &err_msg);
829 : :
830 [ + - + - : 12 : if VUNLIKELY (!is_ready_to_quit() && ret != SQLITE_OK) {
- + - + ]
831 : : // LCOV_EXCL_START GCOVR_EXCL_START
832 : : CLOG_W("Failed to set temp_store: %s.", err_msg);
833 : :
834 : : if (err_msg) {
835 : : ::sqlite3_free(err_msg);
836 : : err_msg = nullptr;
837 : : }
838 : :
839 : : return false;
840 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
841 : : }
842 : :
843 [ + - ]: 12 : ret = ::sqlite3_exec(wrapper_file.db, "PRAGMA page_size = 16384;", nullptr, nullptr, &err_msg);
844 : :
845 [ + - + - : 12 : if VUNLIKELY (!is_ready_to_quit() && ret != SQLITE_OK) {
- + - + ]
846 : : // LCOV_EXCL_START GCOVR_EXCL_START
847 : : CLOG_W("Failed to set page_size: %s.", err_msg);
848 : :
849 : : if (err_msg) {
850 : : ::sqlite3_free(err_msg);
851 : : err_msg = nullptr;
852 : : }
853 : :
854 : : return false;
855 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
856 : : }
857 : :
858 [ + - ]: 12 : ret = ::sqlite3_exec(wrapper_file.db, "PRAGMA cache_size = 8192;", nullptr, nullptr, &err_msg);
859 : :
860 [ + - + - : 12 : if VUNLIKELY (!is_ready_to_quit() && ret != SQLITE_OK) {
- + - + ]
861 : : // LCOV_EXCL_START GCOVR_EXCL_START
862 : : CLOG_W("Failed to set cache_size: %s.", err_msg);
863 : :
864 : : if (err_msg) {
865 : : ::sqlite3_free(err_msg);
866 : : err_msg = nullptr;
867 : : }
868 : :
869 : : return false;
870 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
871 : : }
872 : :
873 [ + - ]: 12 : ret = ::sqlite3_exec(wrapper_file.db, "PRAGMA journal_mode = OFF;", nullptr, nullptr, &err_msg);
874 : :
875 [ + - + - : 12 : if VUNLIKELY (!is_ready_to_quit() && ret != SQLITE_OK) {
- + - + ]
876 : : // LCOV_EXCL_START GCOVR_EXCL_START
877 : : CLOG_W("Failed to restore journal mode: %s.", err_msg);
878 : :
879 : : if (err_msg) {
880 : : ::sqlite3_free(err_msg);
881 : : err_msg = nullptr;
882 : : }
883 : :
884 : : return false;
885 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
886 : : }
887 : :
888 [ + - ]: 12 : ret = ::sqlite3_exec(wrapper_file.db, "PRAGMA automatic_index = OFF;", nullptr, nullptr, &err_msg);
889 : :
890 [ + - + - : 12 : if VUNLIKELY (!is_ready_to_quit() && ret != SQLITE_OK) {
- + - + ]
891 : : // LCOV_EXCL_START GCOVR_EXCL_START
892 : : CLOG_W("Failed to set automatic_index: %s.", err_msg);
893 : :
894 : : if (err_msg) {
895 : : ::sqlite3_free(err_msg);
896 : : err_msg = nullptr;
897 : : }
898 : :
899 : : return false;
900 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
901 : : }
902 : :
903 [ + - ]: 12 : ret = ::sqlite3_exec(wrapper_file.db, "PRAGMA locking_mode = EXCLUSIVE;", nullptr, nullptr, &err_msg);
904 : :
905 [ + - - + ]: 12 : if VUNLIKELY (is_ready_to_quit()) {
906 : : // LCOV_EXCL_START GCOVR_EXCL_START
907 : : if (err_msg) {
908 : : ::sqlite3_free(err_msg);
909 : : err_msg = nullptr;
910 : : }
911 : :
912 : : return false;
913 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
914 [ - + ]: 12 : } else if VUNLIKELY (ret != SQLITE_OK) {
915 : : // LCOV_EXCL_START GCOVR_EXCL_START
916 : : CLOG_W("Failed to set locking_mode: %s.", err_msg);
917 : :
918 : : if (err_msg) {
919 : : ::sqlite3_free(err_msg);
920 : : err_msg = nullptr;
921 : : }
922 : :
923 : : return false;
924 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
925 : : }
926 : : }
927 : :
928 [ + + ]: 12 : if (rebuild) {
929 [ + - ]: 3 : ret = ::sqlite3_exec(wrapper_file.db, "VACUUM;", nullptr, nullptr, &err_msg);
930 : :
931 [ + - - + ]: 3 : if VUNLIKELY (is_ready_to_quit()) {
932 : : // LCOV_EXCL_START GCOVR_EXCL_START
933 : : if (err_msg) {
934 : : ::sqlite3_free(err_msg);
935 : : err_msg = nullptr;
936 : : }
937 : :
938 : : return false;
939 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
940 [ - + ]: 3 : } else if VUNLIKELY (ret != SQLITE_OK) {
941 : : // LCOV_EXCL_START GCOVR_EXCL_START
942 : : CLOG_W("Failed to vacuum: %s.", err_msg);
943 : :
944 : : if (err_msg) {
945 : : ::sqlite3_free(err_msg);
946 : : err_msg = nullptr;
947 : : }
948 : :
949 : : return false;
950 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
951 : : }
952 : :
953 [ + - ]: 3 : ret = ::sqlite3_exec(wrapper_file.db, "DROP INDEX IF EXISTS idx_elapsed;", nullptr, nullptr, &err_msg);
954 : :
955 [ + - + - : 3 : if VUNLIKELY (!is_ready_to_quit() && ret != SQLITE_OK) {
- + - + ]
956 : : // LCOV_EXCL_START GCOVR_EXCL_START
957 : : CLOG_W("Failed to drop idx_elapsed: %s.", err_msg);
958 : :
959 : : if (err_msg) {
960 : : ::sqlite3_free(err_msg);
961 : : err_msg = nullptr;
962 : : }
963 : :
964 : : return false;
965 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
966 : : }
967 : :
968 [ + - ]: 3 : ret = ::sqlite3_exec(wrapper_file.db, "DROP INDEX IF EXISTS idx_url;", nullptr, nullptr, &err_msg);
969 : :
970 [ + - + - : 3 : if VUNLIKELY (!is_ready_to_quit() && ret != SQLITE_OK) {
- + - + ]
971 : : // LCOV_EXCL_START GCOVR_EXCL_START
972 : : CLOG_W("Failed to drop idx_url: %s.", err_msg);
973 : :
974 : : if (err_msg) {
975 : : ::sqlite3_free(err_msg);
976 : : err_msg = nullptr;
977 : : }
978 : :
979 : : return false;
980 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
981 : : }
982 : :
983 [ + - ]: 3 : ret = ::sqlite3_exec(wrapper_file.db, "DROP INDEX IF EXISTS idx_elapsed_url;", nullptr, nullptr, &err_msg);
984 : :
985 [ + - + - : 3 : if VUNLIKELY (!is_ready_to_quit() && ret != SQLITE_OK) {
- + - + ]
986 : : // LCOV_EXCL_START GCOVR_EXCL_START
987 : : CLOG_W("Failed to drop idx_elapsed_url: %s.", err_msg);
988 : :
989 : : if (err_msg) {
990 : : ::sqlite3_free(err_msg);
991 : : err_msg = nullptr;
992 : : }
993 : :
994 : : return false;
995 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
996 : : }
997 : :
998 [ + - ]: 3 : ret = ::sqlite3_exec(wrapper_file.db, "CREATE INDEX idx_elapsed_url ON VLinkDatas(elapsed, url);", nullptr,
999 : : nullptr, &err_msg);
1000 : :
1001 [ + - - + ]: 3 : if VUNLIKELY (is_ready_to_quit()) {
1002 : : // LCOV_EXCL_START GCOVR_EXCL_START
1003 : : if (err_msg) {
1004 : : ::sqlite3_free(err_msg);
1005 : : err_msg = nullptr;
1006 : : }
1007 : :
1008 : : return false;
1009 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1010 [ - + ]: 3 : } else if VUNLIKELY (ret != SQLITE_OK) {
1011 : : // LCOV_EXCL_START GCOVR_EXCL_START
1012 : : CLOG_W("Failed to create idx_elapsed_url: %s.", err_msg);
1013 : :
1014 : : if (err_msg) {
1015 : : ::sqlite3_free(err_msg);
1016 : : err_msg = nullptr;
1017 : : }
1018 : :
1019 : : return false;
1020 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1021 : : }
1022 : : }
1023 : :
1024 [ + - ]: 12 : ret = ::sqlite3_exec(wrapper_file.db, "PRAGMA optimize;", nullptr, nullptr, &err_msg);
1025 : :
1026 [ + - - + ]: 12 : if VUNLIKELY (is_ready_to_quit()) {
1027 : : // LCOV_EXCL_START GCOVR_EXCL_START
1028 : : if (err_msg) {
1029 : : ::sqlite3_free(err_msg);
1030 : : err_msg = nullptr;
1031 : : }
1032 : :
1033 : : return false;
1034 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1035 [ - + ]: 12 : } else if VUNLIKELY (ret != SQLITE_OK) {
1036 : : // LCOV_EXCL_START GCOVR_EXCL_START
1037 : : CLOG_W("Failed to optimize: %s.", err_msg);
1038 : :
1039 : : if (err_msg) {
1040 : : ::sqlite3_free(err_msg);
1041 : : err_msg = nullptr;
1042 : : }
1043 : :
1044 : : return false;
1045 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1046 : : }
1047 : :
1048 [ + - ]: 12 : ret = ::sqlite3_prepare_v2(wrapper_file.db, "SELECT * FROM VLinkDatas;", -1, &wrapper_file.stmt, nullptr);
1049 : :
1050 [ + - - + ]: 12 : if VUNLIKELY (is_ready_to_quit()) {
1051 : : // LCOV_EXCL_START GCOVR_EXCL_START
1052 : : if (err_msg) {
1053 : : ::sqlite3_free(err_msg);
1054 : : err_msg = nullptr;
1055 : : }
1056 : :
1057 : : return false;
1058 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1059 [ - + ]: 12 : } else if VUNLIKELY (ret != SQLITE_OK) {
1060 : : // LCOV_EXCL_START GCOVR_EXCL_START
1061 : : CLOG_W("Failed to prepare datas table: %s.", ::sqlite3_errmsg(wrapper_file.db));
1062 : :
1063 : : if (err_msg) {
1064 : : ::sqlite3_free(err_msg);
1065 : : err_msg = nullptr;
1066 : : }
1067 : :
1068 : : return false;
1069 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1070 : : }
1071 : :
1072 [ + - ]: 12 : ::sqlite3_step(wrapper_file.stmt);
1073 : :
1074 [ - + ]: 12 : if (err_msg) {
1075 : : ::sqlite3_free(err_msg); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1076 : : err_msg = nullptr; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1077 : : }
1078 : : }
1079 : :
1080 : 9 : return true;
1081 [ + - ]: 9 : });
1082 : : #else
1083 : : (void)rebuild;
1084 : : return std::future<bool>();
1085 : : #endif
1086 : : }
1087 : :
1088 : 10 : void VDBReader::tag(const std::string& tag_name) {
1089 : : #ifdef VLINK_ENABLE_SQLITE
1090 : :
1091 [ + + ]: 10 : if VUNLIKELY (is_busy()) {
1092 [ + - + - : 2 : VLOG_W("VDBReader: Is busy.");
+ - ]
1093 : : }
1094 : :
1095 [ + - + - ]: 10 : post_task([this, tag_name]() {
1096 : 10 : int ret = 0;
1097 : 10 : std::string update_tag_sql;
1098 : :
1099 [ + + ]: 26 : for (auto& wrapper_file : impl_->file_list) {
1100 [ - + ]: 16 : if VUNLIKELY (!wrapper_file.db) {
1101 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1102 : : }
1103 : :
1104 [ + + ]: 16 : if (wrapper_file.stmt) {
1105 [ + - ]: 14 : ::sqlite3_finalize(wrapper_file.stmt);
1106 : 14 : wrapper_file.stmt = nullptr;
1107 : : }
1108 : :
1109 [ + - ]: 16 : update_tag_sql = "ALTER TABLE VLinkHeader ADD COLUMN tag TEXT;";
1110 : :
1111 [ + - ]: 16 : ret = sqlite3_exec(wrapper_file.db, update_tag_sql.c_str(), nullptr, nullptr, nullptr);
1112 : :
1113 : : (void)ret;
1114 : :
1115 : 16 : ::sqlite3_stmt* update_tag_stmt = nullptr;
1116 [ + - ]: 16 : ret = ::sqlite3_prepare_v2(wrapper_file.db, "UPDATE VLinkHeader SET tag = ?;", -1, &update_tag_stmt, nullptr);
1117 : 16 : SqliteStmtPtr update_tag_stmt_guard(update_tag_stmt);
1118 : :
1119 [ + - ]: 16 : if VLIKELY (ret == SQLITE_OK) {
1120 [ + - ]: 16 : ret = ::sqlite3_bind_text(update_tag_stmt, 1, tag_name.c_str(), static_cast<int>(tag_name.size()),
1121 : : SQLITE_TRANSIENT); // NOLINT(performance-no-int-to-ptr)
1122 : : }
1123 : :
1124 [ + - ]: 16 : if VLIKELY (ret == SQLITE_OK) {
1125 [ + - ]: 16 : ret = ::sqlite3_step(update_tag_stmt);
1126 : : }
1127 : :
1128 [ - + ]: 16 : if VUNLIKELY (ret != SQLITE_DONE) {
1129 : : CLOG_W("Failed to set tag: %s.", ::sqlite3_errmsg(wrapper_file.db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1130 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1131 : : }
1132 : :
1133 : 16 : update_tag_stmt_guard.reset();
1134 : :
1135 [ + - ]: 16 : ret = ::sqlite3_prepare_v2(wrapper_file.db, "SELECT * FROM VLinkDatas;", -1, &wrapper_file.stmt, nullptr);
1136 : :
1137 [ - + ]: 16 : if VUNLIKELY (ret != SQLITE_OK) {
1138 : : CLOG_W("Failed to prepare datas table: %s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1139 : : ::sqlite3_errmsg(wrapper_file.db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1140 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1141 : : }
1142 : :
1143 [ + - ]: 16 : ::sqlite3_step(wrapper_file.stmt);
1144 : :
1145 [ + - ]: 16 : impl_->info.tag_name = tag_name;
1146 [ + - ]: 16 : }
1147 : :
1148 : : try {
1149 : : #ifdef _WIN32
1150 : : std::filesystem::path file_path(Helpers::string_to_wstring(impl_->path));
1151 : : std::string suffix = Helpers::path_to_string(file_path.extension());
1152 : : #else
1153 [ + - ]: 10 : std::filesystem::path file_path(impl_->path);
1154 [ + - + - ]: 10 : std::string suffix = file_path.extension().string();
1155 : : #endif
1156 : :
1157 : 55 : std::transform(suffix.begin(), suffix.end(), suffix.begin(), [](unsigned char c) { return std::tolower(c); });
1158 : :
1159 [ + + ]: 10 : if (suffix == ".vdbx") {
1160 : : try {
1161 : 5 : nlohmann::ordered_json root_json;
1162 : 5 : nlohmann::ordered_json header_json;
1163 : :
1164 : : {
1165 [ + - ]: 5 : std::ifstream file(file_path);
1166 : :
1167 [ + + ]: 5 : file >> root_json;
1168 : :
1169 [ + - ]: 4 : file.close();
1170 : 5 : }
1171 : :
1172 [ + - + - ]: 4 : header_json = root_json["VLinkHeader"];
1173 : :
1174 [ + - + - ]: 4 : header_json["tag"] = tag_name;
1175 : :
1176 [ + - ]: 4 : root_json["VLinkHeader"] = std::move(header_json);
1177 : :
1178 : : {
1179 [ + - ]: 4 : std::ofstream filex(impl_->path, std::ios::out | std::ios::trunc);
1180 : :
1181 [ + - + - ]: 4 : if VLIKELY (filex.is_open()) {
1182 [ + - + - ]: 4 : filex << root_json.dump(4);
1183 [ + - ]: 4 : filex.close();
1184 : : }
1185 : 4 : }
1186 [ - + ]: 7 : } catch (nlohmann::json::exception& e) {
1187 [ + - + - : 2 : VLOG_W("VDBReader: JSON parse error, ", e.what(), ".");
+ - ]
1188 : 1 : }
1189 : : }
1190 [ - - ]: 10 : } catch (std::filesystem::filesystem_error& e) {
1191 : : VLOG_F("VDBReader: Filesystem error, ", e.what(), "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1192 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1193 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1194 [ + - ]: 10 : });
1195 : : #else
1196 : : (void)tag_name;
1197 : : #endif
1198 : 10 : }
1199 : :
1200 : 5 : int64_t VDBReader::get_timestamp() const {
1201 [ + - ]: 5 : std::shared_lock time_lock(impl_->time_mtx);
1202 : :
1203 [ + + ]: 5 : if (impl_->status.load(std::memory_order_relaxed) == kPlaying) {
1204 [ - + ]: 1 : if (impl_->is_pending.load(std::memory_order_relaxed)) {
1205 : : return impl_->real_elapsed.load(std::memory_order_relaxed) / 1000U; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1206 : : } else {
1207 : 1 : return (impl_->real_elapsed.load(std::memory_order_relaxed) +
1208 : 1 : (impl_->real_timer.get() * impl_->rate.load(std::memory_order_relaxed))) /
1209 : 1 : 1000U;
1210 : : }
1211 [ + + ]: 4 : } else if (impl_->status.load(std::memory_order_relaxed) == kPaused) {
1212 : 2 : return (impl_->real_elapsed.load(std::memory_order_relaxed) +
1213 : 2 : ((impl_->real_timer.get() - impl_->pause_elapsed_timer.get()) *
1214 : 2 : impl_->rate.load(std::memory_order_relaxed))) /
1215 : 2 : 1000U;
1216 : : } else {
1217 : 2 : return 0;
1218 : : }
1219 : 5 : }
1220 : :
1221 : 5 : int64_t VDBReader::get_real_timestamp() const {
1222 [ + + + + : 9 : if (impl_->status.load(std::memory_order_relaxed) == kPlaying ||
+ + ]
1223 : 4 : impl_->status.load(std::memory_order_relaxed) == kPaused) {
1224 : 6 : return impl_->real_elapsed.load(std::memory_order_relaxed) / 1000U;
1225 : : } else {
1226 : 2 : return 0;
1227 : : }
1228 : : }
1229 : :
1230 : 20 : VDBReader::Status VDBReader::get_status() const { return impl_->status.load(std::memory_order_relaxed); }
1231 : :
1232 : 279 : const BagReader::Info& VDBReader::get_info() const { return impl_->info; }
1233 : :
1234 : 64 : std::vector<SchemaData> VDBReader::detect_schema() {
1235 : : #ifdef VLINK_ENABLE_SQLITE
1236 : 64 : std::vector<SchemaData> schema_list;
1237 : 64 : std::unordered_map<std::string, size_t> schema_index_map;
1238 : :
1239 [ + + ]: 64 : if (!impl_->info.has_schema) {
1240 : 32 : return schema_list;
1241 : : }
1242 : :
1243 [ + - ]: 32 : schema_index_map.reserve(impl_->info.url_metas.size());
1244 : :
1245 [ + + ]: 68 : for (auto& wrapper_file : impl_->file_list) {
1246 : 36 : ::sqlite3_stmt* schema_stmt = nullptr;
1247 : :
1248 [ + - ]: 36 : int ret = ::sqlite3_prepare_v2(wrapper_file.db, "SELECT ser, encoding, data FROM VLinkSchemas;", -1, &schema_stmt,
1249 : : nullptr);
1250 : :
1251 [ - + ]: 36 : if VUNLIKELY (ret != SQLITE_OK) {
1252 : : // LCOV_EXCL_START GCOVR_EXCL_START
1253 : : CLOG_E("Failed to prepare schema table: %s.", ::sqlite3_errmsg(wrapper_file.db));
1254 : :
1255 : : if (schema_stmt) {
1256 : : ::sqlite3_finalize(schema_stmt);
1257 : : schema_stmt = nullptr;
1258 : : }
1259 : :
1260 : : return schema_list;
1261 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1262 : : }
1263 : :
1264 : 36 : const uint8_t* data = nullptr;
1265 : 36 : size_t size = 0;
1266 : :
1267 [ + - + + ]: 74 : while (::sqlite3_step(schema_stmt) == SQLITE_ROW) {
1268 : 38 : SchemaData schema;
1269 [ + - ]: 38 : schema.name = sqlite_column_text_or_empty(schema_stmt, get_column(0));
1270 [ + - ]: 38 : schema.encoding = sqlite_column_text_or_empty(schema_stmt, get_column(1));
1271 : 38 : schema.schema_type = SchemaData::resolve_type(SchemaType::kUnknown, schema.name, schema.encoding);
1272 : :
1273 [ + - ]: 38 : data = static_cast<const uint8_t*>(::sqlite3_column_blob(schema_stmt, get_column(2)));
1274 [ + - ]: 38 : size = ::sqlite3_column_bytes(schema_stmt, get_column(2));
1275 : :
1276 [ + + + - : 38 : if (!schema.name.empty() && data && size > 0) {
+ - + + ]
1277 [ + - ]: 37 : std::string schema_key = schema.name;
1278 [ + - ]: 37 : schema_key.push_back('\x1F');
1279 [ + - ]: 37 : schema_key.append(SchemaData::convert_type(schema.schema_type));
1280 : :
1281 [ + - ]: 37 : auto schema_index_iter = schema_index_map.find(schema_key);
1282 : :
1283 [ + + ]: 37 : if (schema_index_iter == schema_index_map.end()) {
1284 : 35 : schema.data = Bytes::deep_copy(data, size);
1285 [ + - ]: 35 : schema_index_map.emplace(schema_key, schema_list.size());
1286 [ + - ]: 35 : schema_list.emplace_back(std::move(schema));
1287 : : } else {
1288 : 2 : auto& current_schema = schema_list[schema_index_iter->second];
1289 : :
1290 [ - + - - : 2 : if (current_schema.encoding.empty() && !schema.encoding.empty()) {
- + ]
1291 : : current_schema.encoding = schema.encoding; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1292 : : }
1293 : :
1294 [ - + ]: 2 : if (current_schema.data.empty()) {
1295 : : current_schema.data = Bytes::deep_copy(data, size); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1296 : : }
1297 : : }
1298 : 37 : }
1299 : 38 : }
1300 : :
1301 [ + - ]: 36 : if VLIKELY (schema_stmt) {
1302 [ + - ]: 36 : ::sqlite3_finalize(schema_stmt);
1303 : 36 : schema_stmt = nullptr;
1304 : : }
1305 : : }
1306 : :
1307 : 32 : return schema_list;
1308 : : #else
1309 : : return std::vector<SchemaData>();
1310 : : #endif
1311 : 64 : }
1312 : :
1313 : 29 : bool VDBReader::is_split_mode() const { return impl_->info.split_count > 0; }
1314 : :
1315 : 4 : int VDBReader::get_split_index() const { return impl_->split_index.load(std::memory_order_relaxed); }
1316 : :
1317 : 6 : bool VDBReader::is_jumping() const { return impl_->jump_flag.load(std::memory_order_relaxed); }
1318 : :
1319 : 117 : size_t VDBReader::get_max_task_count() const { return kMaxTaskSize; }
1320 : :
1321 : 153 : void VDBReader::on_begin() { MessageLoop::on_begin(); }
1322 : :
1323 : 153 : void VDBReader::on_end() { MessageLoop::on_end(); }
1324 : :
1325 : 118 : bool VDBReader::do_open_cursor(const Config& config) {
1326 : : #ifdef VLINK_ENABLE_SQLITE
1327 : 118 : impl_->cursor_stmt.reset();
1328 : 118 : impl_->cursor_config = config;
1329 [ + + ]: 118 : impl_->cursor_begin_us = config.begin_time > 0 ? config.begin_time * 1000 : 0;
1330 [ + + ]: 118 : impl_->cursor_end_us = config.end_time > 0 ? config.end_time * 1000 : 0;
1331 : 118 : impl_->cursor_file_index = 0;
1332 : :
1333 [ - + ]: 118 : if VUNLIKELY (impl_->file_list.empty()) {
1334 : : VLOG_W("VDBReader: Cursor cannot find any data."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1335 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1336 : : }
1337 : :
1338 : 118 : return prepare_cursor_stmt(0);
1339 : : #else
1340 : : (void)config;
1341 : : return false;
1342 : : #endif
1343 : : }
1344 : :
1345 : 597 : bool VDBReader::do_read_next(Frame& out, bool& is_error) {
1346 : : #ifdef VLINK_ENABLE_SQLITE
1347 : 597 : is_error = false;
1348 : :
1349 : : while (true) {
1350 [ - + ]: 918 : if VUNLIKELY (!impl_->cursor_stmt) {
1351 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1352 : : }
1353 : :
1354 : 918 : const int step = ::sqlite3_step(impl_->cursor_stmt.get());
1355 : :
1356 [ + + ]: 918 : if (step == SQLITE_ROW) {
1357 [ + - ]: 481 : const int64_t timestamp = ::sqlite3_column_int64(impl_->cursor_stmt.get(), get_column(0));
1358 : :
1359 [ + - ]: 481 : const int url_id = ::sqlite3_column_int(impl_->cursor_stmt.get(), get_column(1));
1360 [ + - ]: 481 : auto& wrapper_file = impl_->file_list.at(impl_->cursor_file_index);
1361 [ + - ]: 481 : auto iter = wrapper_file.id_to_url_map.find(url_id);
1362 : :
1363 [ + + - + : 481 : if VUNLIKELY (iter == wrapper_file.id_to_url_map.end() || iter->second.empty()) {
+ + ]
1364 : 2 : continue;
1365 : : }
1366 : :
1367 : 480 : const auto& url = iter->second;
1368 : :
1369 : 480 : std::string output_url;
1370 : :
1371 [ + - + + ]: 480 : if VUNLIKELY (!convert_playback_url(url, output_url)) {
1372 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1373 : : }
1374 : :
1375 [ + + + - : 479 : if (!impl_->cursor_config.filter_urls.empty() && impl_->cursor_config.filter_urls.count(output_url) == 0U) {
- + - + ]
1376 : 0 : continue;
1377 : : }
1378 : :
1379 : 479 : std::string_view action_str;
1380 : : const auto* action_ptr =
1381 [ + - ]: 479 : reinterpret_cast<const char*>(::sqlite3_column_text(impl_->cursor_stmt.get(), get_column(2)));
1382 : :
1383 [ + + ]: 479 : if (action_ptr != nullptr) {
1384 : 477 : action_str = std::string_view(
1385 [ + - ]: 477 : action_ptr, static_cast<size_t>(::sqlite3_column_bytes(impl_->cursor_stmt.get(), get_column(2))));
1386 : : }
1387 : :
1388 [ + - ]: 479 : const auto* data = static_cast<const uint8_t*>(::sqlite3_column_blob(impl_->cursor_stmt.get(), get_column(3)));
1389 [ + - ]: 479 : const int size = ::sqlite3_column_bytes(impl_->cursor_stmt.get(), get_column(3));
1390 : :
1391 : 479 : out.timestamp = timestamp;
1392 : 479 : out.url = std::move(output_url);
1393 : 479 : out.ser_type.clear();
1394 : 479 : out.schema_type = SchemaType::kUnknown;
1395 [ + - ]: 479 : out.action_type = convert_action(action_str);
1396 : :
1397 [ + + + + : 479 : if VUNLIKELY (impl_->enable_compress && Bytes::is_compress_data(data, size)) {
+ + ]
1398 : 11 : out.data = Bytes::uncompress_data(data, size, false);
1399 : : } else {
1400 : 468 : out.data = Bytes::shallow_copy(data, size);
1401 : : }
1402 : :
1403 [ + - ]: 479 : fill_frame_meta(out);
1404 : :
1405 : 479 : return true;
1406 [ + + + - ]: 917 : } else if (step == SQLITE_DONE) {
1407 [ + + ]: 437 : if (impl_->cursor_file_index + 1 < static_cast<int>(impl_->file_list.size())) {
1408 [ - + ]: 319 : if VUNLIKELY (!prepare_cursor_stmt(impl_->cursor_file_index + 1)) {
1409 : : is_error = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1410 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1411 : : }
1412 : :
1413 : 319 : continue;
1414 : : }
1415 : :
1416 : 118 : return false;
1417 : : } else {
1418 : : // LCOV_EXCL_START GCOVR_EXCL_START
1419 : : CLOG_W("VDBReader: Cursor step failed: %s.", ::sqlite3_errmsg(::sqlite3_db_handle(impl_->cursor_stmt.get())));
1420 : : is_error = true;
1421 : : return false;
1422 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1423 : : }
1424 : 321 : }
1425 : : #else
1426 : : (void)out;
1427 : : is_error = false;
1428 : : return false;
1429 : : #endif
1430 : : }
1431 : :
1432 : 437 : bool VDBReader::prepare_cursor_stmt(int file_index) {
1433 : : #ifdef VLINK_ENABLE_SQLITE
1434 : 437 : impl_->cursor_stmt.reset();
1435 : :
1436 [ + - - + : 437 : if VUNLIKELY (file_index < 0 || file_index >= static_cast<int>(impl_->file_list.size())) {
- + ]
1437 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1438 : : }
1439 : :
1440 [ + - ]: 437 : auto& wrapper_file = impl_->file_list.at(file_index);
1441 : :
1442 [ - + ]: 437 : if VUNLIKELY (!wrapper_file.db) {
1443 : : VLOG_W("VDBReader: Cursor target db is empty."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1444 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1445 : : }
1446 : :
1447 : 437 : std::string where;
1448 : :
1449 [ + + ]: 437 : if (!impl_->cursor_config.filter_urls.empty()) {
1450 [ + - ]: 11 : std::string id_list = "url IN (";
1451 : 11 : bool id_appended = false;
1452 : :
1453 [ + + ]: 32 : for (const auto& [url, id] : wrapper_file.url_to_id_map) {
1454 [ + - + + ]: 21 : if (!match_playback_url_filter(url, impl_->cursor_config.filter_urls)) {
1455 : 16 : continue;
1456 : : }
1457 : :
1458 [ + - + - ]: 5 : id_list.append(std::to_string(id));
1459 [ + - ]: 5 : id_list.append(",");
1460 : 5 : id_appended = true;
1461 : : }
1462 : :
1463 [ + + ]: 11 : if (id_appended) {
1464 : 5 : id_list.pop_back();
1465 [ + - ]: 5 : id_list.append(")");
1466 : : } else {
1467 [ + - ]: 6 : id_list = "url IN (NULL)";
1468 : : }
1469 : :
1470 : 11 : where = std::move(id_list);
1471 : 11 : }
1472 : :
1473 [ + + ]: 437 : if (impl_->cursor_begin_us > 0) {
1474 [ + + ]: 15 : if (!where.empty()) {
1475 [ + - ]: 5 : where.append(" AND ");
1476 : : }
1477 : :
1478 [ + - ]: 15 : where.append("elapsed >= ");
1479 [ + - + - ]: 15 : where.append(std::to_string(impl_->cursor_begin_us));
1480 : : }
1481 : :
1482 [ + + ]: 437 : if (impl_->cursor_end_us > 0) {
1483 [ + + ]: 17 : if (!where.empty()) {
1484 [ + - ]: 10 : where.append(" AND ");
1485 : : }
1486 : :
1487 [ + - ]: 17 : where.append("elapsed <= ");
1488 [ + - + - ]: 17 : where.append(std::to_string(impl_->cursor_end_us));
1489 : : }
1490 : :
1491 [ + - ]: 437 : std::string select_sql = "SELECT elapsed, url, action, data FROM VLinkDatas";
1492 : :
1493 [ + + ]: 437 : if (!where.empty()) {
1494 [ + - ]: 28 : select_sql.append(" WHERE ");
1495 [ + - ]: 28 : select_sql.append(where);
1496 : : }
1497 : :
1498 : 437 : const bool order_by_elapsed = wrapper_file.has_idx_elapsed;
1499 : :
1500 [ + + + - ]: 437 : select_sql.append(order_by_elapsed ? " ORDER BY elapsed, rowid;" : " ORDER BY rowid;");
1501 : :
1502 : 437 : ::sqlite3_stmt* stmt = nullptr;
1503 [ + - ]: 437 : const int ret = ::sqlite3_prepare_v2(wrapper_file.db, select_sql.c_str(), -1, &stmt, nullptr);
1504 : :
1505 [ - + ]: 437 : if VUNLIKELY (ret != SQLITE_OK) {
1506 : : CLOG_W("VDBReader: Failed to prepare cursor stmt: %s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1507 : : ::sqlite3_errmsg(wrapper_file.db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1508 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1509 : : }
1510 : :
1511 : 437 : impl_->cursor_stmt.reset(stmt);
1512 : 437 : impl_->cursor_file_index = file_index;
1513 : :
1514 : 437 : return true;
1515 : : #else
1516 : : (void)file_index;
1517 : : return false;
1518 : : #endif
1519 : 437 : }
1520 : :
1521 : 92 : void VDBReader::update_status(Status status) {
1522 : 92 : bool has_changed = false;
1523 : :
1524 [ + + ]: 92 : if (status == kStopped) {
1525 [ + + ]: 31 : if (impl_->status.load(std::memory_order_relaxed) != kStopped) {
1526 : 30 : impl_->status.store(kStopped, std::memory_order_relaxed);
1527 : 30 : has_changed = true;
1528 : : }
1529 [ + + ]: 61 : } else if (status == kPaused) {
1530 [ + + ]: 19 : if (impl_->status.load(std::memory_order_relaxed) != kPaused) {
1531 : 14 : impl_->status.store(kPaused, std::memory_order_relaxed);
1532 : 14 : has_changed = true;
1533 : : }
1534 [ + - ]: 42 : } else if (status == kPlaying) {
1535 [ + + ]: 42 : if (impl_->status.load(std::memory_order_relaxed) != kPlaying) {
1536 : 39 : impl_->status.store(kPlaying, std::memory_order_relaxed);
1537 : 39 : has_changed = true;
1538 : : }
1539 : : }
1540 : :
1541 [ + + ]: 92 : if (has_changed) {
1542 [ + + ]: 83 : if VLIKELY (impl_->status_callback) {
1543 : 19 : impl_->status_callback(impl_->status.load(std::memory_order_relaxed));
1544 : : }
1545 : : }
1546 : 92 : }
1547 : :
1548 : 214 : void VDBReader::do_stop() {
1549 : : #ifdef VLINK_ENABLE_SQLITE
1550 : : {
1551 [ + - ]: 214 : std::unique_lock lock(impl_->mtx);
1552 : 214 : impl_->stop_flag.store(true, std::memory_order_relaxed);
1553 : 214 : impl_->pause_flag.store(false, std::memory_order_relaxed);
1554 : 214 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
1555 : 214 : impl_->jump_flag.store(false, std::memory_order_relaxed);
1556 : 214 : }
1557 : :
1558 : 214 : impl_->cv.notify_one();
1559 : :
1560 [ + + ]: 1056 : for (const auto& wrapper_file : impl_->file_list) {
1561 [ + - ]: 842 : if (wrapper_file.db) {
1562 [ + - ]: 842 : ::sqlite3_interrupt(wrapper_file.db);
1563 : : }
1564 : : }
1565 : : #endif
1566 : 214 : }
1567 : :
1568 : 14 : void VDBReader::do_pause() {
1569 [ + - ]: 14 : std::unique_lock lock(impl_->mtx);
1570 : :
1571 [ + - ]: 14 : while (impl_->pause_flag.load(std::memory_order_relaxed)) {
1572 : 14 : impl_->pause_elapsed_timer.restart();
1573 [ + - ]: 14 : update_status(kPaused);
1574 : :
1575 [ + - ]: 14 : impl_->cv.wait(lock, [this]() -> bool {
1576 [ + + + + ]: 67 : return impl_->stop_flag.load(std::memory_order_relaxed) || !impl_->pause_flag.load(std::memory_order_relaxed) ||
1577 [ + - ]: 30 : impl_->pause_next_flag.load(std::memory_order_relaxed) ||
1578 [ + - - + ]: 55 : impl_->jump_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
1579 : : });
1580 : :
1581 : 14 : impl_->pause_elapsed.fetch_add(impl_->pause_elapsed_timer.get(), std::memory_order_relaxed);
1582 : :
1583 : : {
1584 [ + - ]: 14 : std::lock_guard time_lock(impl_->time_mtx);
1585 : 14 : impl_->real_timer.restart();
1586 : :
1587 [ - + ]: 28 : if (impl_->offset_elapsed.load(std::memory_order_relaxed) > 0) {
1588 : : // LCOV_EXCL_START GCOVR_EXCL_START
1589 : : impl_->real_elapsed.fetch_add((impl_->offset_timer.get() - impl_->pause_elapsed_timer.get() -
1590 : : impl_->extra_elapsed.load(std::memory_order_relaxed)) *
1591 : : impl_->rate.load(std::memory_order_relaxed),
1592 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1593 : : std::memory_order_relaxed);
1594 : : }
1595 : 14 : }
1596 : :
1597 [ + - ]: 14 : update_status(kPlaying);
1598 : :
1599 [ + - ]: 14 : if (impl_->pause_next_flag.load(std::memory_order_relaxed)) {
1600 : 28 : impl_->pause_elapsed.fetch_sub(impl_->offset_elapsed.load(std::memory_order_relaxed), std::memory_order_relaxed);
1601 : 14 : break;
1602 : : } else if (impl_->offset_elapsed.load(std::memory_order_relaxed) > 0) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1603 : : impl_->offset_timer.restart(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1604 : :
1605 : : impl_->cv.wait_for( // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1606 : : lock,
1607 : : // LCOV_EXCL_START GCOVR_EXCL_START
1608 : : std::chrono::microseconds(impl_->offset_elapsed.load(std::memory_order_relaxed)), [this]() -> bool {
1609 : : return impl_->stop_flag.load(std::memory_order_relaxed) ||
1610 : : impl_->pause_flag.load(std::memory_order_relaxed) ||
1611 : : impl_->pause_next_flag.load(std::memory_order_relaxed) ||
1612 : : impl_->jump_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
1613 : : });
1614 : :
1615 : : if VUNLIKELY (impl_->pause_flag.load(std::memory_order_relaxed)) {
1616 : : impl_->offset_elapsed.fetch_sub(impl_->offset_timer.get(), std::memory_order_relaxed);
1617 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1618 : : } else {
1619 : : impl_->offset_elapsed.store(0, std::memory_order_relaxed); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1620 : : }
1621 : : }
1622 : : }
1623 : 14 : }
1624 : :
1625 : 846 : bool VDBReader::prepare_file(void* file) {
1626 : : #ifdef VLINK_ENABLE_SQLITE
1627 : 846 : auto* wrapper_file = static_cast<Impl::WrapperFile*>(file);
1628 : :
1629 : 846 : wrapper_file->has_completed = true;
1630 : :
1631 : 846 : int ret = 0;
1632 : :
1633 : : // opt busy_timeout
1634 [ + - ]: 846 : ::sqlite3_busy_timeout(wrapper_file->db, 100);
1635 : :
1636 : : #if ENABLE_DATABASE_TABLE_CHECK
1637 : : {
1638 : : // search count
1639 : : ::sqlite3_stmt* search_stmt = nullptr;
1640 : : bool need_rebuild_header = false;
1641 : : ret = ::sqlite3_prepare_v2(wrapper_file->db, "PRAGMA table_info(VLinkHeader);", -1, &search_stmt, nullptr);
1642 : :
1643 : : if VUNLIKELY (ret != SQLITE_OK) {
1644 : : wrapper_file->has_completed = false;
1645 : :
1646 : : if (!impl_->try_to_fix || impl_->read_only) {
1647 : : CLOG_F("Failed to search header table: %s.", ::sqlite3_errmsg(wrapper_file->db));
1648 : : return false;
1649 : : }
1650 : :
1651 : : need_rebuild_header = true;
1652 : : } else {
1653 : : const char* expected_columns[] = {
1654 : : "major", "minor", "patch", "count", "duration", "accuracy", "compress",
1655 : : "process", "date", "tag", "complete", "timezone", "start_timestamp",
1656 : : };
1657 : :
1658 : : const char* expected_types[] = {
1659 : : "INTEGER", "INTEGER", "INTEGER", "INTEGER", "INTEGER", "TEXT", "TEXT",
1660 : : "TEXT", "TEXT", "TEXT", "INTEGER", "INTEGER", "INTEGER",
1661 : : };
1662 : :
1663 : : size_t column_index = 0;
1664 : : int invalid_header_num = 0;
1665 : : int step_ret = SQLITE_OK;
1666 : : for (;;) {
1667 : : step_ret = ::sqlite3_step(search_stmt);
1668 : :
1669 : : if (step_ret != SQLITE_ROW) {
1670 : : break;
1671 : : }
1672 : :
1673 : : if (column_index >= sizeof(expected_columns) / sizeof(expected_columns[0])) {
1674 : : need_rebuild_header = true;
1675 : : invalid_header_num = 1;
1676 : : break;
1677 : : }
1678 : :
1679 : : const auto* column_name = reinterpret_cast<const char*>(::sqlite3_column_text(search_stmt, get_column(1)));
1680 : : const auto* column_type = reinterpret_cast<const char*>(::sqlite3_column_text(search_stmt, get_column(2)));
1681 : :
1682 : : if (!column_name || std::strcmp(column_name, expected_columns[column_index]) != 0) {
1683 : : need_rebuild_header = true;
1684 : : invalid_header_num = 2;
1685 : : break;
1686 : : }
1687 : :
1688 : : if (!column_type || std::strcmp(column_type, expected_types[column_index]) != 0) {
1689 : : need_rebuild_header = true;
1690 : : invalid_header_num = 3;
1691 : : break;
1692 : : }
1693 : :
1694 : : ++column_index;
1695 : : }
1696 : :
1697 : : if (!need_rebuild_header && step_ret != SQLITE_DONE) {
1698 : : if VLIKELY (search_stmt) {
1699 : : ::sqlite3_finalize(search_stmt);
1700 : : search_stmt = nullptr;
1701 : : }
1702 : :
1703 : : wrapper_file->has_completed = false;
1704 : : CLOG_F("Failed to inspect header table: %s.", ::sqlite3_errmsg(wrapper_file->db));
1705 : : return false;
1706 : : }
1707 : :
1708 : : if (!need_rebuild_header) {
1709 : : need_rebuild_header = (column_index != sizeof(expected_columns) / sizeof(expected_columns[0]));
1710 : : if (need_rebuild_header) {
1711 : : invalid_header_num = 4;
1712 : : }
1713 : : }
1714 : :
1715 : : if (need_rebuild_header) {
1716 : : if (!impl_->try_to_fix || impl_->read_only) {
1717 : : CLOG_F("VDBReader: Table [VLinkHeader] is incompatible, num=%d.", invalid_header_num);
1718 : : return false;
1719 : : }
1720 : :
1721 : : CLOG_W("VDBReader: Table [VLinkHeader] is incompatible, num=%d. Try to rebuild.", invalid_header_num);
1722 : : }
1723 : : }
1724 : :
1725 : : if VLIKELY (search_stmt) {
1726 : : ::sqlite3_finalize(search_stmt);
1727 : : search_stmt = nullptr;
1728 : : }
1729 : :
1730 : : if (need_rebuild_header) {
1731 : : wrapper_file->has_completed = false;
1732 : : char* err_msg = nullptr;
1733 : :
1734 : : ret = ::sqlite3_exec(wrapper_file->db, "DROP TABLE IF EXISTS VLinkHeader;", nullptr, nullptr, &err_msg);
1735 : : if VUNLIKELY (ret != SQLITE_OK) {
1736 : : CLOG_F("Failed to drop VLinkHeader: %s.", err_msg);
1737 : :
1738 : : if (err_msg) {
1739 : : ::sqlite3_free(err_msg);
1740 : : err_msg = nullptr;
1741 : : }
1742 : :
1743 : : return false;
1744 : : }
1745 : :
1746 : : ret = ::sqlite3_exec(
1747 : : wrapper_file->db,
1748 : : "CREATE TABLE IF NOT EXISTS VLinkHeader(major INTEGER, minor INTEGER, patch INTEGER, count INTEGER, "
1749 : : "duration INTEGER, accuracy TEXT, compress TEXT, process TEXT, date TEXT, tag TEXT, complete INTEGER, "
1750 : : "timezone INTEGER, start_timestamp INTEGER);",
1751 : : nullptr, nullptr, &err_msg);
1752 : : if VUNLIKELY (ret != SQLITE_OK) {
1753 : : CLOG_F("Failed to create header table: %s.", err_msg);
1754 : :
1755 : : if (err_msg) {
1756 : : ::sqlite3_free(err_msg);
1757 : : err_msg = nullptr;
1758 : : }
1759 : :
1760 : : return false;
1761 : : }
1762 : :
1763 : : ::sqlite3_stmt* create_header_stmt = nullptr;
1764 : : ret = ::sqlite3_prepare_v2(
1765 : : wrapper_file->db,
1766 : : "INSERT INTO VLinkHeader (major, minor, patch, count, duration, accuracy, compress, "
1767 : : "process, date, tag, complete, timezone, start_timestamp) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);",
1768 : : -1, &create_header_stmt, nullptr);
1769 : :
1770 : : if VUNLIKELY (ret != SQLITE_OK) {
1771 : : CLOG_F("Failed to prepare header table: %s.", ::sqlite3_errmsg(wrapper_file->db));
1772 : :
1773 : : if (err_msg) {
1774 : : ::sqlite3_free(err_msg);
1775 : : err_msg = nullptr;
1776 : : }
1777 : :
1778 : : return false;
1779 : : }
1780 : :
1781 : : ::sqlite3_bind_int(create_header_stmt, get_column(0), VLINK_VERSION_MAJOR);
1782 : : ::sqlite3_bind_int(create_header_stmt, get_column(1), VLINK_VERSION_MINOR);
1783 : : ::sqlite3_bind_int(create_header_stmt, get_column(2), VLINK_VERSION_PATCH);
1784 : : ::sqlite3_bind_int64(create_header_stmt, get_column(3), 0);
1785 : : ::sqlite3_bind_int64(create_header_stmt, get_column(4), 0);
1786 : : ::sqlite3_bind_text(create_header_stmt, get_column(5), "MicroSecond", -1, SQLITE_STATIC);
1787 : : ::sqlite3_bind_text(create_header_stmt, get_column(6), "None", -1, SQLITE_STATIC);
1788 : : ::sqlite3_bind_text(create_header_stmt, get_column(7), "None", -1, SQLITE_STATIC);
1789 : : ::sqlite3_bind_text(create_header_stmt, get_column(8), "None", -1, SQLITE_STATIC);
1790 : : ::sqlite3_bind_text(create_header_stmt, get_column(9), "None", -1, SQLITE_STATIC);
1791 : : ::sqlite3_bind_int(create_header_stmt, get_column(10), 0);
1792 : : ::sqlite3_bind_int(create_header_stmt, get_column(11), 0);
1793 : : ::sqlite3_bind_int64(create_header_stmt, get_column(12), 0);
1794 : :
1795 : : ret = ::sqlite3_step(create_header_stmt);
1796 : :
1797 : : if VUNLIKELY (ret != SQLITE_DONE) {
1798 : : CLOG_F("Failed to insert header table: %s.", ::sqlite3_errmsg(wrapper_file->db));
1799 : :
1800 : : if (err_msg) {
1801 : : ::sqlite3_free(err_msg);
1802 : : err_msg = nullptr;
1803 : : }
1804 : :
1805 : : return false;
1806 : : }
1807 : :
1808 : : if VLIKELY (create_header_stmt) {
1809 : : ::sqlite3_finalize(create_header_stmt);
1810 : : create_header_stmt = nullptr;
1811 : : }
1812 : :
1813 : : if (err_msg) {
1814 : : ::sqlite3_free(err_msg);
1815 : : err_msg = nullptr;
1816 : : }
1817 : : }
1818 : :
1819 : : ::sqlite3_stmt* schema_info_stmt = nullptr;
1820 : : ret = ::sqlite3_prepare_v2(wrapper_file->db, "PRAGMA table_info(VLinkSchemas);", -1, &schema_info_stmt, nullptr);
1821 : :
1822 : : if VUNLIKELY (ret != SQLITE_OK) {
1823 : : wrapper_file->has_completed = false;
1824 : : CLOG_F("Failed to search schema table: %s.", ::sqlite3_errmsg(wrapper_file->db));
1825 : : return false;
1826 : : }
1827 : :
1828 : : {
1829 : : const char* expected_columns[] = {"ser", "encoding", "data"};
1830 : :
1831 : : const char* expected_types[] = {"TEXT", "TEXT", "BLOB"};
1832 : :
1833 : : size_t column_index = 0;
1834 : : bool invalid_schema_table = false;
1835 : : int invalid_schema_num = 0;
1836 : : int step_ret = SQLITE_OK;
1837 : :
1838 : : for (;;) {
1839 : : step_ret = ::sqlite3_step(schema_info_stmt);
1840 : :
1841 : : if (step_ret != SQLITE_ROW) {
1842 : : break;
1843 : : }
1844 : :
1845 : : if (column_index >= sizeof(expected_columns) / sizeof(expected_columns[0])) {
1846 : : invalid_schema_table = true;
1847 : : invalid_schema_num = 1;
1848 : : break;
1849 : : }
1850 : :
1851 : : const auto* column_name = reinterpret_cast<const char*>(::sqlite3_column_text(schema_info_stmt, get_column(1)));
1852 : : const auto* column_type = reinterpret_cast<const char*>(::sqlite3_column_text(schema_info_stmt, get_column(2)));
1853 : :
1854 : : if (!column_name || std::strcmp(column_name, expected_columns[column_index]) != 0) {
1855 : : invalid_schema_table = true;
1856 : : invalid_schema_num = 2;
1857 : : break;
1858 : : }
1859 : :
1860 : : if (!column_type || std::strcmp(column_type, expected_types[column_index]) != 0) {
1861 : : invalid_schema_table = true;
1862 : : invalid_schema_num = 3;
1863 : : break;
1864 : : }
1865 : :
1866 : : ++column_index;
1867 : : }
1868 : :
1869 : : if (!invalid_schema_table && step_ret != SQLITE_DONE) {
1870 : : if VLIKELY (schema_info_stmt) {
1871 : : ::sqlite3_finalize(schema_info_stmt);
1872 : : schema_info_stmt = nullptr;
1873 : : }
1874 : :
1875 : : wrapper_file->has_completed = false;
1876 : : CLOG_F("Failed to inspect schema table: %s.", ::sqlite3_errmsg(wrapper_file->db));
1877 : : return false;
1878 : : }
1879 : :
1880 : : if (!invalid_schema_table) {
1881 : : invalid_schema_table = (column_index != sizeof(expected_columns) / sizeof(expected_columns[0]));
1882 : : if (invalid_schema_table) {
1883 : : invalid_schema_num = 4;
1884 : : }
1885 : : }
1886 : :
1887 : : if VLIKELY (schema_info_stmt) {
1888 : : ::sqlite3_finalize(schema_info_stmt);
1889 : : schema_info_stmt = nullptr;
1890 : : }
1891 : :
1892 : : if VUNLIKELY (invalid_schema_table) {
1893 : : wrapper_file->has_completed = false;
1894 : : CLOG_F("VDBReader: Table [VLinkSchemas] is incompatible, num=%d.", invalid_schema_num);
1895 : : return false;
1896 : : }
1897 : : }
1898 : :
1899 : : ::sqlite3_stmt* urls_info_stmt = nullptr;
1900 : : ret = ::sqlite3_prepare_v2(wrapper_file->db, "PRAGMA table_info(VLinkUrls);", -1, &urls_info_stmt, nullptr);
1901 : :
1902 : : if VUNLIKELY (ret != SQLITE_OK) {
1903 : : wrapper_file->has_completed = false;
1904 : : CLOG_F("Failed to search urls table: %s.", ::sqlite3_errmsg(wrapper_file->db));
1905 : : return false;
1906 : : }
1907 : :
1908 : : {
1909 : : const char* expected_columns[] = {"id", "url", "type", "ser", "encoding", "count", "loss", "size", "freq"};
1910 : :
1911 : : const char* expected_types[] = {"INTEGER", "TEXT", "TEXT", "TEXT", "TEXT", "INTEGER", "REAL", "INTEGER", "REAL"};
1912 : :
1913 : : size_t column_index = 0;
1914 : : bool invalid_urls_table = false;
1915 : : int invalid_urls_num = 0;
1916 : : int step_ret = SQLITE_OK;
1917 : :
1918 : : for (;;) {
1919 : : step_ret = ::sqlite3_step(urls_info_stmt);
1920 : :
1921 : : if (step_ret != SQLITE_ROW) {
1922 : : break;
1923 : : }
1924 : :
1925 : : if (column_index >= sizeof(expected_columns) / sizeof(expected_columns[0])) {
1926 : : invalid_urls_table = true;
1927 : : invalid_urls_num = 1;
1928 : : break;
1929 : : }
1930 : :
1931 : : const auto* column_name = reinterpret_cast<const char*>(::sqlite3_column_text(urls_info_stmt, get_column(1)));
1932 : : const auto* column_type = reinterpret_cast<const char*>(::sqlite3_column_text(urls_info_stmt, get_column(2)));
1933 : :
1934 : : if (!column_name || std::strcmp(column_name, expected_columns[column_index]) != 0) {
1935 : : invalid_urls_table = true;
1936 : : invalid_urls_num = 2;
1937 : : break;
1938 : : }
1939 : :
1940 : : if (!column_type || std::strcmp(column_type, expected_types[column_index]) != 0) {
1941 : : invalid_urls_table = true;
1942 : : invalid_urls_num = 3;
1943 : : break;
1944 : : }
1945 : :
1946 : : ++column_index;
1947 : : }
1948 : :
1949 : : if (!invalid_urls_table && step_ret != SQLITE_DONE) {
1950 : : if VLIKELY (urls_info_stmt) {
1951 : : ::sqlite3_finalize(urls_info_stmt);
1952 : : urls_info_stmt = nullptr;
1953 : : }
1954 : :
1955 : : wrapper_file->has_completed = false;
1956 : : CLOG_F("Failed to inspect urls table: %s.", ::sqlite3_errmsg(wrapper_file->db));
1957 : : return false;
1958 : : }
1959 : :
1960 : : if (!invalid_urls_table) {
1961 : : invalid_urls_table = (column_index != sizeof(expected_columns) / sizeof(expected_columns[0]));
1962 : : if (invalid_urls_table) {
1963 : : invalid_urls_num = 4;
1964 : : }
1965 : : }
1966 : :
1967 : : if VLIKELY (urls_info_stmt) {
1968 : : ::sqlite3_finalize(urls_info_stmt);
1969 : : urls_info_stmt = nullptr;
1970 : : }
1971 : :
1972 : : if VUNLIKELY (invalid_urls_table) {
1973 : : wrapper_file->has_completed = false;
1974 : : CLOG_F("VDBReader: Table [VLinkUrls] is incompatible, num=%d.", invalid_urls_num);
1975 : : return false;
1976 : : }
1977 : : }
1978 : :
1979 : : ::sqlite3_stmt* datas_info_stmt = nullptr;
1980 : : ret = ::sqlite3_prepare_v2(wrapper_file->db, "PRAGMA table_info(VLinkDatas);", -1, &datas_info_stmt, nullptr);
1981 : :
1982 : : if VUNLIKELY (ret != SQLITE_OK) {
1983 : : wrapper_file->has_completed = false;
1984 : : CLOG_F("Failed to search datas table: %s.", ::sqlite3_errmsg(wrapper_file->db));
1985 : : return false;
1986 : : }
1987 : :
1988 : : {
1989 : : const char* expected_columns[] = {"elapsed", "url", "action", "data"};
1990 : :
1991 : : const char* expected_types[] = {"INTEGER", "INTEGER", "TEXT", "BLOB"};
1992 : :
1993 : : size_t column_index = 0;
1994 : : bool invalid_datas_table = false;
1995 : : int invalid_datas_num = 0;
1996 : : int step_ret = SQLITE_OK;
1997 : :
1998 : : for (;;) {
1999 : : step_ret = ::sqlite3_step(datas_info_stmt);
2000 : :
2001 : : if (step_ret != SQLITE_ROW) {
2002 : : break;
2003 : : }
2004 : :
2005 : : if (column_index >= sizeof(expected_columns) / sizeof(expected_columns[0])) {
2006 : : invalid_datas_table = true;
2007 : : invalid_datas_num = 1;
2008 : : break;
2009 : : }
2010 : :
2011 : : const auto* column_name = reinterpret_cast<const char*>(::sqlite3_column_text(datas_info_stmt, get_column(1)));
2012 : : const auto* column_type = reinterpret_cast<const char*>(::sqlite3_column_text(datas_info_stmt, get_column(2)));
2013 : :
2014 : : if (!column_name || std::strcmp(column_name, expected_columns[column_index]) != 0) {
2015 : : invalid_datas_table = true;
2016 : : invalid_datas_num = 2;
2017 : : break;
2018 : : }
2019 : :
2020 : : if (!column_type || std::strcmp(column_type, expected_types[column_index]) != 0) {
2021 : : invalid_datas_table = true;
2022 : : invalid_datas_num = 3;
2023 : : break;
2024 : : }
2025 : :
2026 : : ++column_index;
2027 : : }
2028 : :
2029 : : if (!invalid_datas_table && step_ret != SQLITE_DONE) {
2030 : : if VLIKELY (datas_info_stmt) {
2031 : : ::sqlite3_finalize(datas_info_stmt);
2032 : : datas_info_stmt = nullptr;
2033 : : }
2034 : :
2035 : : wrapper_file->has_completed = false;
2036 : : CLOG_F("Failed to inspect datas table: %s.", ::sqlite3_errmsg(wrapper_file->db));
2037 : : return false;
2038 : : }
2039 : :
2040 : : if (!invalid_datas_table) {
2041 : : invalid_datas_table = (column_index != sizeof(expected_columns) / sizeof(expected_columns[0]));
2042 : : if (invalid_datas_table) {
2043 : : invalid_datas_num = 4;
2044 : : }
2045 : : }
2046 : :
2047 : : if VLIKELY (datas_info_stmt) {
2048 : : ::sqlite3_finalize(datas_info_stmt);
2049 : : datas_info_stmt = nullptr;
2050 : : }
2051 : :
2052 : : if VUNLIKELY (invalid_datas_table) {
2053 : : wrapper_file->has_completed = false;
2054 : : CLOG_F("VDBReader: Table [VLinkDatas] is incompatible, num=%d.", invalid_datas_num);
2055 : : return false;
2056 : : }
2057 : : }
2058 : : }
2059 : : #endif
2060 : :
2061 : : // prepare header table
2062 : 846 : ::sqlite3_stmt* header_stmt = nullptr;
2063 [ + - ]: 846 : ret = ::sqlite3_prepare_v2(
2064 : : wrapper_file->db,
2065 : : "SELECT major, minor, patch, count, duration, accuracy, compress, process, date, tag, complete, timezone, "
2066 : : "start_timestamp FROM VLinkHeader LIMIT 1;",
2067 : : -1, &header_stmt, nullptr);
2068 : :
2069 [ + + ]: 846 : if VUNLIKELY (ret != SQLITE_OK) {
2070 : 2 : wrapper_file->has_completed = false;
2071 [ + - + - : 4 : CLOG_F("Failed to prepare header table: %s.", ::sqlite3_errmsg(wrapper_file->db));
+ - - + ]
2072 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2073 : : }
2074 : :
2075 : 844 : SqliteStmtPtr header_stmt_guard(header_stmt);
2076 : :
2077 [ + - ]: 844 : ret = ::sqlite3_step(header_stmt);
2078 : :
2079 [ - + ]: 844 : if VUNLIKELY (ret != SQLITE_ROW) {
2080 : : wrapper_file->has_completed = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2081 : : CLOG_F("Failed to get header table: %s.", ::sqlite3_errmsg(wrapper_file->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2082 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2083 : : }
2084 : :
2085 [ + - ]: 844 : int major = ::sqlite3_column_int(header_stmt, get_column(0));
2086 [ + - ]: 844 : int minor = ::sqlite3_column_int(header_stmt, get_column(1));
2087 [ + - ]: 844 : int patch = ::sqlite3_column_int(header_stmt, get_column(2));
2088 : :
2089 [ - + ]: 844 : if VUNLIKELY (major != VLINK_VERSION_MAJOR) {
2090 : : wrapper_file->has_completed = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2091 : : VLOG_F("VDBReader: Database version is incompatible."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2092 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2093 : : }
2094 : :
2095 [ + - + - : 844 : impl_->info.version = std::to_string(major) + "." + std::to_string(minor) + "." + std::to_string(patch);
+ - + - +
- + - +
- ]
2096 [ + - ]: 844 : impl_->info.storage_type = "SQLite3";
2097 [ + - ]: 844 : impl_->info.message_count = ::sqlite3_column_int64(header_stmt, get_column(3));
2098 [ + - ]: 844 : impl_->info.total_duration = ::sqlite3_column_int64(header_stmt, get_column(4)) / 1000U;
2099 [ + - ]: 844 : impl_->info.time_accuracy = sqlite_column_text_or_empty(header_stmt, get_column(5));
2100 [ + - ]: 844 : impl_->info.compression_type = sqlite_column_text_or_empty(header_stmt, get_column(6));
2101 [ + - ]: 844 : impl_->info.process_name = sqlite_column_text_or_empty(header_stmt, get_column(7));
2102 [ + - ]: 844 : impl_->info.date_time = sqlite_column_text_or_empty(header_stmt, get_column(8));
2103 : :
2104 [ + - ]: 844 : const char* tag_name_str = reinterpret_cast<const char*>(::sqlite3_column_text(header_stmt, get_column(9)));
2105 : :
2106 [ + + ]: 844 : if (tag_name_str) {
2107 [ + - ]: 843 : impl_->info.tag_name = tag_name_str;
2108 : : } else {
2109 [ + - ]: 1 : impl_->info.tag_name = "Empty";
2110 : : }
2111 : :
2112 [ + - ]: 844 : impl_->info.has_completed = (::sqlite3_column_int(header_stmt, get_column(10)) == 1);
2113 [ + - ]: 844 : impl_->info.timezone = ::sqlite3_column_int(header_stmt, get_column(11));
2114 [ + - ]: 844 : impl_->info.start_timestamp = ::sqlite3_column_int64(header_stmt, get_column(12));
2115 : :
2116 [ + + ]: 844 : if VUNLIKELY (impl_->info.start_timestamp < 0) {
2117 : 1 : impl_->info.start_timestamp = 0;
2118 : 1 : wrapper_file->has_completed = false;
2119 : :
2120 [ + - ]: 1 : if (impl_->read_only) {
2121 [ + - + - : 2 : VLOG_E("VDBReader: Invalid start_timestamp.");
+ - ]
2122 : : }
2123 : : }
2124 : :
2125 [ + + + + ]: 1700 : if (impl_->info.compression_type.empty() || impl_->info.compression_type == "None" ||
2126 [ + - + + : 1700 : impl_->info.compression_type == "NONE" || impl_->info.compression_type == "none") {
+ + ]
2127 : 834 : impl_->enable_compress = false;
2128 : : } else {
2129 : 10 : impl_->enable_compress = true;
2130 : : }
2131 : :
2132 [ + - + + ]: 844 : if VUNLIKELY (impl_->info.time_accuracy != "MicroSecond") {
2133 : 1 : wrapper_file->has_completed = false;
2134 [ + - + - : 2 : VLOG_F("VDBReader: Database accuracy is not supported.");
- + ]
2135 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2136 : : }
2137 : :
2138 : 843 : header_stmt_guard.reset();
2139 : 843 : header_stmt = nullptr;
2140 : :
2141 : : // prepare schema table
2142 : 843 : ::sqlite3_stmt* schema_stmt = nullptr;
2143 : :
2144 [ + - ]: 843 : ret = ::sqlite3_prepare_v2(wrapper_file->db,
2145 : : "SELECT 1 FROM sqlite_master WHERE type='table' AND name='VLinkSchemas' LIMIT 1;", -1,
2146 : : &schema_stmt, nullptr);
2147 : :
2148 [ + - ]: 843 : if (ret == SQLITE_OK) {
2149 [ + - + - ]: 843 : if (::sqlite3_step(schema_stmt) == SQLITE_ROW) {
2150 : 843 : sqlite3_stmt* count_stmt = nullptr;
2151 : :
2152 [ + - ]: 843 : ret = ::sqlite3_prepare_v2(wrapper_file->db, "SELECT EXISTS(SELECT 1 FROM VLinkSchemas LIMIT 1);", -1,
2153 : : &count_stmt, nullptr);
2154 : :
2155 [ + - ]: 843 : if (ret == SQLITE_OK) {
2156 [ + - + - ]: 843 : if (::sqlite3_step(count_stmt) == SQLITE_ROW) {
2157 [ + - ]: 843 : int exists = ::sqlite3_column_int(count_stmt, 0);
2158 : :
2159 [ + + ]: 843 : if (exists) {
2160 : 43 : wrapper_file->has_schema = true;
2161 : : }
2162 : : }
2163 : : }
2164 : :
2165 [ + - ]: 843 : if VLIKELY (count_stmt) {
2166 [ + - ]: 843 : ::sqlite3_finalize(count_stmt);
2167 : 843 : count_stmt = nullptr;
2168 : : }
2169 : : }
2170 : : }
2171 : :
2172 [ + - ]: 843 : if VLIKELY (schema_stmt) {
2173 [ + - ]: 843 : ::sqlite3_finalize(schema_stmt);
2174 : 843 : schema_stmt = nullptr;
2175 : : }
2176 : :
2177 : : // prepare urls table
2178 : 843 : ::sqlite3_stmt* urls_stmt = nullptr;
2179 [ + - ]: 843 : ret = ::sqlite3_prepare_v2(wrapper_file->db,
2180 : : "SELECT id, url, type, ser, encoding, count, loss, size, freq FROM VLinkUrls;", -1,
2181 : : &urls_stmt, nullptr);
2182 : :
2183 [ + + ]: 843 : if VUNLIKELY (ret != SQLITE_OK) {
2184 : 1 : wrapper_file->has_completed = false;
2185 [ + - + - : 2 : CLOG_F("Failed to prepare urls table: %s.", ::sqlite3_errmsg(wrapper_file->db));
+ - - + ]
2186 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2187 : : }
2188 : :
2189 [ + - ]: 842 : url_ser_map().clear();
2190 [ + - ]: 842 : url_schema_type_map().clear();
2191 : 842 : impl_->info.url_metas.clear();
2192 : 842 : impl_->raw_url_metas.clear();
2193 : :
2194 : 842 : impl_->info.total_raw_size = 0;
2195 : :
2196 [ + - + + ]: 1755 : while (::sqlite3_step(urls_stmt) == SQLITE_ROW) {
2197 : 913 : Info::UrlMeta url_meta;
2198 : 913 : url_meta.valid = true;
2199 [ + - ]: 913 : url_meta.index = ::sqlite3_column_int(urls_stmt, get_column(0));
2200 [ + - ]: 913 : url_meta.url = sqlite_column_text_or_empty(urls_stmt, get_column(1));
2201 [ + - ]: 913 : url_meta.url_type = sqlite_column_text_or_empty(urls_stmt, get_column(2));
2202 [ + - ]: 913 : url_meta.ser_type = sqlite_column_text_or_empty(urls_stmt, get_column(3));
2203 [ + - ]: 913 : const auto* encoding_label = reinterpret_cast<const char*>(::sqlite3_column_text(urls_stmt, get_column(4)));
2204 [ + - ]: 913 : url_meta.schema_type = SchemaData::resolve_type(SchemaData::convert_encoding(encoding_label ? encoding_label : ""),
2205 : : url_meta.ser_type); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2206 : :
2207 [ + - ]: 913 : url_meta.count = static_cast<size_t>(::sqlite3_column_int64(urls_stmt, get_column(5)));
2208 [ + - ]: 913 : url_meta.loss = ::sqlite3_column_double(urls_stmt, get_column(6));
2209 [ + - ]: 913 : url_meta.size = ::sqlite3_column_int64(urls_stmt, get_column(7));
2210 [ + - ]: 913 : url_meta.freq = ::sqlite3_column_double(urls_stmt, get_column(8));
2211 : :
2212 : 913 : impl_->info.total_raw_size += url_meta.size;
2213 : :
2214 [ + - ]: 913 : wrapper_file->id_to_url_map.emplace(url_meta.index, url_meta.url);
2215 [ + - ]: 913 : wrapper_file->url_to_id_map.emplace(url_meta.url, url_meta.index);
2216 : :
2217 [ + - + - ]: 913 : url_ser_map().emplace(url_meta.url, url_meta.ser_type);
2218 [ + - + - ]: 913 : url_schema_type_map().emplace(url_meta.url, url_meta.schema_type);
2219 [ + - ]: 913 : impl_->info.url_metas.emplace_back(std::move(url_meta));
2220 : 913 : }
2221 : :
2222 [ + - ]: 842 : std::sort(impl_->info.url_metas.begin(), impl_->info.url_metas.end());
2223 [ + - ]: 842 : impl_->raw_url_metas = impl_->info.url_metas;
2224 [ + - ]: 842 : rebuild_url_meta_lookup(impl_->info.url_metas);
2225 : :
2226 [ + - ]: 842 : if VLIKELY (urls_stmt) {
2227 [ + - ]: 842 : ::sqlite3_finalize(urls_stmt);
2228 : 842 : urls_stmt = nullptr;
2229 : : }
2230 : :
2231 : : // get idx_elapsed
2232 : 842 : ::sqlite3_stmt* idx_elapsed_stmt = nullptr;
2233 : :
2234 [ + - ]: 842 : ret = ::sqlite3_prepare_v2(
2235 : : wrapper_file->db,
2236 : : "SELECT 1 FROM sqlite_master WHERE tbl_name='VLinkDatas' AND type='index' AND name='idx_elapsed' LIMIT 1;", -1,
2237 : : &idx_elapsed_stmt, nullptr);
2238 : :
2239 [ + - ]: 842 : if (ret == SQLITE_OK) {
2240 [ + - - + ]: 842 : if (::sqlite3_step(idx_elapsed_stmt) == SQLITE_ROW) {
2241 : : wrapper_file->has_idx_elapsed = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2242 : : }
2243 : : }
2244 : :
2245 [ + - ]: 842 : if VLIKELY (idx_elapsed_stmt) {
2246 [ + - ]: 842 : ::sqlite3_finalize(idx_elapsed_stmt);
2247 : 842 : idx_elapsed_stmt = nullptr;
2248 : : }
2249 : :
2250 : : // get idx_elapsed_url
2251 : 842 : ::sqlite3_stmt* idx_elapsed_url_stmt = nullptr;
2252 : :
2253 [ + - ]: 842 : ret = ::sqlite3_prepare_v2(
2254 : : wrapper_file->db,
2255 : : "SELECT 1 FROM sqlite_master WHERE tbl_name='VLinkDatas' AND type='index' AND name='idx_elapsed_url' LIMIT 1;",
2256 : : -1, &idx_elapsed_url_stmt, nullptr);
2257 : :
2258 [ + - ]: 842 : if (ret == SQLITE_OK) {
2259 [ + - + + ]: 842 : if (::sqlite3_step(idx_elapsed_url_stmt) == SQLITE_ROW) {
2260 : 839 : wrapper_file->has_idx_elapsed = true;
2261 : 839 : wrapper_file->has_idx_url = true;
2262 : : }
2263 : : }
2264 : :
2265 [ + - ]: 842 : if VLIKELY (idx_elapsed_url_stmt) {
2266 [ + - ]: 842 : ::sqlite3_finalize(idx_elapsed_url_stmt);
2267 : 842 : idx_elapsed_url_stmt = nullptr;
2268 : : }
2269 : :
2270 : 842 : impl_->info.has_idx_elapsed = wrapper_file->has_idx_elapsed;
2271 : 842 : impl_->info.has_idx_url = wrapper_file->has_idx_url;
2272 : 842 : impl_->info.has_schema = wrapper_file->has_schema;
2273 : :
2274 : : // prepare datas
2275 : :
2276 [ + + ]: 842 : if (wrapper_file->has_idx_elapsed) {
2277 [ + - ]: 839 : ret = ::sqlite3_prepare_v2(wrapper_file->db, "SELECT * FROM VLinkDatas ORDER BY elapsed LIMIT 1;", -1,
2278 : : &wrapper_file->stmt, nullptr);
2279 : : } else {
2280 [ + - ]: 3 : ret = ::sqlite3_prepare_v2(wrapper_file->db, "SELECT * FROM VLinkDatas LIMIT 1;", -1, &wrapper_file->stmt, nullptr);
2281 : : }
2282 : :
2283 [ + + ]: 842 : if VUNLIKELY (ret != SQLITE_OK) {
2284 : 1 : wrapper_file->has_completed = false;
2285 [ + - + - : 2 : CLOG_F("Failed to prepare datas table: %s.", ::sqlite3_errmsg(wrapper_file->db));
+ - - + ]
2286 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2287 : : }
2288 : :
2289 : : // get blank time
2290 [ + - ]: 841 : ::sqlite3_step(wrapper_file->stmt);
2291 [ + - ]: 841 : impl_->info.blank_duration = ::sqlite3_column_int64(wrapper_file->stmt, get_column(0)) / 1000U;
2292 : :
2293 [ - + ]: 841 : if VUNLIKELY (impl_->info.blank_duration < 0) {
2294 : : impl_->info.blank_duration = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2295 : : }
2296 : :
2297 [ + - ]: 841 : if VLIKELY (wrapper_file->stmt) {
2298 [ + - ]: 841 : ::sqlite3_finalize(wrapper_file->stmt);
2299 : 841 : wrapper_file->stmt = nullptr;
2300 : : }
2301 : :
2302 : : // update duration
2303 : :
2304 [ + + + + : 841 : if (wrapper_file->has_idx_elapsed && impl_->info.has_completed) {
+ + ]
2305 [ + - ]: 834 : ret = ::sqlite3_prepare_v2(wrapper_file->db, "SELECT * FROM VLinkDatas ORDER BY elapsed DESC LIMIT 1;", -1,
2306 : : &wrapper_file->stmt, nullptr);
2307 : :
2308 [ - + ]: 834 : if VUNLIKELY (ret != SQLITE_OK) {
2309 : : // LCOV_EXCL_START GCOVR_EXCL_START
2310 : : wrapper_file->has_completed = false;
2311 : : CLOG_F("Failed to prepare datas table: %s.", ::sqlite3_errmsg(wrapper_file->db));
2312 : : return false;
2313 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
2314 : : }
2315 : :
2316 : : // get duration time
2317 [ + - ]: 834 : ::sqlite3_step(wrapper_file->stmt);
2318 [ + - ]: 834 : int64_t total_duration = ::sqlite3_column_int64(wrapper_file->stmt, get_column(0)) / 1000U;
2319 : :
2320 [ + + ]: 834 : if (total_duration > impl_->info.blank_duration) {
2321 : 51 : impl_->info.total_duration = total_duration;
2322 : : }
2323 : :
2324 [ + - ]: 834 : if VLIKELY (wrapper_file->stmt) {
2325 [ + - ]: 834 : ::sqlite3_finalize(wrapper_file->stmt);
2326 : 834 : wrapper_file->stmt = nullptr;
2327 : : }
2328 : : }
2329 : 841 : return true;
2330 : : #else
2331 : : (void)file;
2332 : : VLOG_F("VDBReader: The compile macro VLINK_ENABLE_SQLITE is not turned on.");
2333 : : return false;
2334 : : #endif
2335 : 844 : }
2336 : :
2337 : 221 : void VDBReader::open(const std::string& path) {
2338 : : #ifdef VLINK_ENABLE_SQLITE
2339 [ + - ]: 221 : close();
2340 : :
2341 : 1692 : auto to_open = [this](Impl::WrapperFile& wrapper_file) -> bool {
2342 [ + + ]: 846 : if (impl_->read_only) {
2343 : 377 : int ret = ::sqlite3_open_v2(wrapper_file.path.c_str(), &wrapper_file.db, SQLITE_OPEN_READONLY, nullptr);
2344 : :
2345 [ - + ]: 377 : if VUNLIKELY (ret != SQLITE_OK) {
2346 : : // LCOV_EXCL_START GCOVR_EXCL_START
2347 : : CLOG_F("Failed to open database [%s].", wrapper_file.path.c_str());
2348 : :
2349 : : if (wrapper_file.db) {
2350 : : ::sqlite3_close_v2(wrapper_file.db);
2351 : : wrapper_file.db = nullptr;
2352 : : }
2353 : :
2354 : : return false;
2355 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
2356 : : }
2357 : : } else {
2358 : 469 : int ret = ::sqlite3_open_v2(wrapper_file.path.c_str(), &wrapper_file.db,
2359 : : SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE, nullptr);
2360 : :
2361 [ - + ]: 469 : if VUNLIKELY (ret != SQLITE_OK) {
2362 : : // LCOV_EXCL_START GCOVR_EXCL_START
2363 : : CLOG_F("Failed to open database [%s].", wrapper_file.path.c_str());
2364 : :
2365 : : if (wrapper_file.db) {
2366 : : ::sqlite3_close_v2(wrapper_file.db);
2367 : : wrapper_file.db = nullptr;
2368 : : }
2369 : :
2370 : : return false;
2371 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
2372 : : }
2373 : : }
2374 : :
2375 [ - + ]: 846 : if VUNLIKELY (!prepare_file(&wrapper_file)) {
2376 : : // LCOV_EXCL_START GCOVR_EXCL_START
2377 : : if (wrapper_file.stmt) {
2378 : : ::sqlite3_finalize(wrapper_file.stmt);
2379 : : wrapper_file.stmt = nullptr;
2380 : : }
2381 : :
2382 : : if (wrapper_file.db) {
2383 : : ::sqlite3_close_v2(wrapper_file.db);
2384 : : wrapper_file.db = nullptr;
2385 : : }
2386 : :
2387 : : return false;
2388 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
2389 : : }
2390 : :
2391 : 841 : return true;
2392 : 221 : };
2393 : :
2394 [ + - ]: 221 : impl_->path = path;
2395 : :
2396 : 221 : impl_->total_start_timestamp_ns = -1;
2397 : 221 : impl_->total_has_completed = true;
2398 : :
2399 : : try {
2400 : : #ifdef _WIN32
2401 : : std::filesystem::path file_path(Helpers::string_to_wstring(path));
2402 : :
2403 : : impl_->info.file_name = Helpers::path_to_string(file_path.filename());
2404 : :
2405 : : impl_->info.file_size = 0;
2406 : :
2407 : : std::string suffix = Helpers::path_to_string(file_path.extension());
2408 : : #else
2409 [ + - ]: 221 : std::filesystem::path file_path(path);
2410 : :
2411 [ + - + - ]: 221 : impl_->info.file_name = file_path.filename().string();
2412 : :
2413 : 221 : impl_->info.file_size = 0;
2414 : :
2415 [ + - + - ]: 221 : std::string suffix = file_path.extension().string();
2416 : : #endif
2417 : :
2418 [ + - + + ]: 221 : if VUNLIKELY (!std::filesystem::exists(file_path)) {
2419 [ + - + - : 2 : CLOG_F("Database [%s] does not exist.", path.c_str());
- + ]
2420 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2421 : : }
2422 : :
2423 : 220 : std::filesystem::path parent_path;
2424 : :
2425 : : try {
2426 [ + - ]: 220 : parent_path = file_path.parent_path();
2427 : : } catch (std::filesystem::filesystem_error&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2428 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2429 : :
2430 : 1234 : std::transform(suffix.begin(), suffix.end(), suffix.begin(), [](unsigned char c) { return std::tolower(c); });
2431 : :
2432 [ + + ]: 220 : if (suffix == ".vdbx") {
2433 : : try {
2434 : 134 : int64_t blank_duration = -1;
2435 : :
2436 : 134 : nlohmann::json root_json;
2437 : :
2438 : : {
2439 [ + - ]: 134 : std::ifstream filex(file_path);
2440 : :
2441 [ + + ]: 134 : filex >> root_json;
2442 : :
2443 [ + - ]: 133 : filex.close();
2444 : 134 : }
2445 : :
2446 [ + - + - ]: 133 : nlohmann::json header_json = root_json["VLinkHeader"];
2447 [ + - + - ]: 133 : nlohmann::json urls_json = root_json["VLinkUrls"];
2448 [ + - + - ]: 133 : nlohmann::json files_json = root_json["VLinkFiles"];
2449 : :
2450 : 133 : impl_->info.file_size = 0;
2451 : :
2452 : 133 : int file_index = 0;
2453 : :
2454 [ + + ]: 133 : if (!files_json.empty()) {
2455 [ + - ]: 132 : impl_->file_list.reserve(files_json.size());
2456 : :
2457 : 132 : impl_->info.has_idx_elapsed = true;
2458 : 132 : impl_->info.has_idx_url = true;
2459 : 132 : impl_->info.has_schema = false;
2460 : : } else {
2461 : 1 : impl_->info.has_idx_elapsed = false;
2462 : 1 : impl_->info.has_idx_url = false;
2463 : 1 : impl_->info.has_schema = false;
2464 : : }
2465 : :
2466 : 133 : std::filesystem::path file_db;
2467 : 133 : std::string file_db_str;
2468 : 133 : bool has_schema = false;
2469 : :
2470 [ + - + - : 893 : for (const auto& file_info : files_json) {
+ - + + ]
2471 : : #ifdef _WIN32
2472 : :
2473 : : if (parent_path.empty()) {
2474 : : file_db = std::filesystem::path(Helpers::string_to_wstring(file_info));
2475 : : } else {
2476 : : file_db = parent_path / std::filesystem::path(Helpers::string_to_wstring(file_info));
2477 : : }
2478 : : #else
2479 : :
2480 [ + + ]: 761 : if (parent_path.empty()) {
2481 [ + - ]: 6 : file_db = std::filesystem::path(file_info);
2482 : : } else {
2483 [ + - + - ]: 755 : file_db = parent_path / std::filesystem::path(file_info);
2484 : : }
2485 : : #endif
2486 : :
2487 [ + - ]: 761 : file_db_str = file_db.string();
2488 : :
2489 [ + - + + ]: 761 : if VUNLIKELY (!std::filesystem::exists(file_db)) {
2490 [ + - + - : 2 : CLOG_F("Database [%s] does not exist.", file_db_str.c_str());
- + ]
2491 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2492 : : }
2493 : :
2494 [ + - ]: 760 : Impl::WrapperFile wrapper_file;
2495 [ + - ]: 760 : wrapper_file.path = file_db_str;
2496 : 760 : wrapper_file.index = file_index;
2497 : :
2498 [ + - - + ]: 760 : if VUNLIKELY (!to_open(wrapper_file)) {
2499 : : // LCOV_EXCL_START GCOVR_EXCL_START
2500 : : CLOG_W("VDBReader: Skipping invalid database [%s].", file_db_str.c_str());
2501 : : continue;
2502 : : }
2503 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
2504 : :
2505 [ - + ]: 760 : if (!wrapper_file.has_idx_elapsed) {
2506 : : impl_->info.has_idx_elapsed = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2507 : : }
2508 : :
2509 [ - + ]: 760 : if (!wrapper_file.has_idx_url) {
2510 : : impl_->info.has_idx_url = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2511 : : }
2512 : :
2513 [ + + ]: 760 : if (wrapper_file.has_schema) {
2514 : 2 : has_schema = true;
2515 : : }
2516 : :
2517 : 760 : std::error_code db_size_ec;
2518 : 760 : std::uintmax_t file_size = std::filesystem::file_size(file_db, db_size_ec);
2519 : :
2520 [ - + ]: 760 : if VUNLIKELY (db_size_ec) {
2521 : : CLOG_W("VDBReader: file_size failed for [%s]: %s.", file_db_str.c_str(), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2522 : : db_size_ec.message().c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2523 : : file_size = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2524 : : }
2525 : :
2526 : 760 : impl_->info.file_size += file_size;
2527 : :
2528 : 760 : wrapper_file.start_timestamp_ns = impl_->info.start_timestamp * 1000'000;
2529 : 760 : wrapper_file.begin = impl_->info.blank_duration;
2530 : 760 : wrapper_file.end = impl_->info.total_duration;
2531 : :
2532 [ + + ]: 760 : if (impl_->total_start_timestamp_ns < 0) {
2533 : 131 : impl_->total_start_timestamp_ns = wrapper_file.start_timestamp_ns;
2534 : : }
2535 : :
2536 [ - + ]: 760 : if (!wrapper_file.has_completed) {
2537 : : impl_->total_has_completed = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2538 : : }
2539 : :
2540 [ + + + + : 760 : if (blank_duration < 0 || impl_->info.blank_duration < blank_duration) {
+ + ]
2541 : 132 : blank_duration = impl_->info.blank_duration;
2542 : : }
2543 : :
2544 [ + - ]: 760 : impl_->file_list.emplace_back(std::move(wrapper_file));
2545 : 760 : ++file_index;
2546 [ + - ]: 760 : }
2547 : :
2548 : 132 : impl_->info.has_schema = has_schema;
2549 : 132 : impl_->info.split_count = impl_->file_list.size();
2550 : :
2551 [ + + ]: 132 : if VUNLIKELY (impl_->file_list.empty()) {
2552 [ + - + - : 2 : VLOG_F("VDBReader: DB list is empty.");
- + ]
2553 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2554 : : }
2555 : :
2556 [ + - + - ]: 131 : int version_major = header_json["major"];
2557 [ + - + - ]: 131 : int version_minor = header_json["minor"];
2558 [ + - + - ]: 131 : int version_patch = header_json["patch"];
2559 : :
2560 : 131 : impl_->info.version =
2561 [ + - + - : 262 : std::to_string(version_major) + "." + std::to_string(version_minor) + "." + std::to_string(version_patch);
+ - + - +
- + - +
- ]
2562 [ + - ]: 131 : impl_->info.storage_type = "SQLite3";
2563 [ + - + - ]: 131 : impl_->info.message_count = header_json["count"];
2564 [ + - + - ]: 131 : impl_->info.total_duration = header_json["duration"];
2565 : 131 : impl_->info.total_duration /= 1000U;
2566 [ + - + - ]: 131 : impl_->info.time_accuracy = header_json["accuracy"];
2567 [ + - + - ]: 131 : impl_->info.compression_type = header_json["compress"];
2568 [ + - + - ]: 131 : impl_->info.process_name = header_json["process"];
2569 [ + - + - ]: 131 : impl_->info.date_time = header_json["date"];
2570 : :
2571 [ + - + + ]: 131 : if (header_json.contains("start_timestamp")) {
2572 [ + - + - ]: 130 : impl_->info.start_timestamp = header_json["start_timestamp"];
2573 : : } else {
2574 : 1 : impl_->info.start_timestamp = Helpers::convert_date_to_timestamp(impl_->info.date_time) / 1000'000;
2575 : : }
2576 : :
2577 [ + + ]: 131 : if VUNLIKELY (impl_->info.start_timestamp < 0) {
2578 : 1 : impl_->info.start_timestamp = 0;
2579 : :
2580 [ + - ]: 1 : if (impl_->read_only) {
2581 [ + - + - : 2 : VLOG_E("VDBReader: Invalid start_timestamp.");
+ - ]
2582 : : }
2583 : : }
2584 : :
2585 [ + - + + ]: 131 : if (header_json.contains("tag")) {
2586 [ + - + - ]: 130 : impl_->info.tag_name = header_json["tag"];
2587 : : } else {
2588 [ + - ]: 1 : impl_->info.tag_name = "Empty";
2589 : : }
2590 : :
2591 [ + - + + ]: 131 : if (header_json.contains("complete")) {
2592 [ + - + - ]: 130 : impl_->info.has_completed = header_json["complete"];
2593 : : } else {
2594 : 1 : impl_->info.has_completed = true;
2595 : : }
2596 : :
2597 [ + - + + ]: 131 : if (header_json.contains("timezone")) {
2598 [ + - + - ]: 130 : impl_->info.timezone = header_json["timezone"];
2599 : : } else {
2600 : 1 : impl_->info.timezone = 480;
2601 : : }
2602 : :
2603 [ + - + + ]: 131 : if (header_json.contains("split_by_size")) {
2604 [ + - + - ]: 130 : impl_->info.split_by_size = header_json["split_by_size"];
2605 : : }
2606 : :
2607 [ + - + + ]: 131 : if (header_json.contains("split_by_time")) {
2608 [ + - + - ]: 130 : impl_->info.split_by_time = header_json["split_by_time"];
2609 : : }
2610 : :
2611 : 131 : impl_->info.blank_duration = blank_duration;
2612 : :
2613 [ + + + + ]: 267 : if (impl_->info.compression_type.empty() || impl_->info.compression_type == "None" ||
2614 [ + - + + : 267 : impl_->info.compression_type == "NONE" || impl_->info.compression_type == "none") {
+ + ]
2615 : 128 : impl_->enable_compress = false;
2616 : : } else {
2617 : 3 : impl_->enable_compress = true;
2618 : : }
2619 : :
2620 [ + - + + ]: 131 : if VUNLIKELY (impl_->info.time_accuracy != "MicroSecond") {
2621 [ + - + - : 2 : VLOG_F("VDBReader: Database accuracy is not supported.");
- + ]
2622 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2623 : : }
2624 : :
2625 [ + - ]: 130 : url_ser_map().clear();
2626 [ + - ]: 130 : url_schema_type_map().clear();
2627 : 130 : impl_->info.url_metas.clear();
2628 : 130 : impl_->raw_url_metas.clear();
2629 : :
2630 [ + - ]: 130 : impl_->info.url_metas.reserve(urls_json.size());
2631 : :
2632 : 130 : impl_->info.total_raw_size = 0;
2633 : :
2634 [ + - + - : 261 : for (const auto& url_info : urls_json) {
+ - + + ]
2635 : 131 : Info::UrlMeta url_meta;
2636 : :
2637 : 131 : url_meta.valid = true;
2638 [ + - + - ]: 131 : url_meta.index = url_info["index"];
2639 [ + - + - ]: 131 : url_meta.url = url_info["url"];
2640 [ + - + - ]: 131 : url_meta.url_type = url_info["type"];
2641 [ + - + - ]: 131 : url_meta.ser_type = url_info["ser"];
2642 : :
2643 [ + - + + ]: 131 : if (url_info.contains("encoding")) {
2644 : 260 : url_meta.schema_type = SchemaData::resolve_type(
2645 [ + - + - ]: 260 : SchemaData::convert_encoding(url_info["encoding"].get<std::string>()), url_meta.ser_type);
2646 : : } else {
2647 : 1 : url_meta.schema_type = SchemaType::kUnknown;
2648 : : }
2649 : :
2650 [ + - + - ]: 131 : url_meta.count = url_info["count"];
2651 [ + - + - ]: 131 : url_meta.loss = url_info["loss"];
2652 : :
2653 [ + - + + ]: 131 : if (url_info.contains("size")) {
2654 [ + - + - ]: 130 : url_meta.size = url_info["size"];
2655 : : }
2656 : :
2657 [ + - + + ]: 131 : if (url_info.contains("freq")) {
2658 [ + - + - ]: 130 : url_meta.freq = url_info["freq"];
2659 : : }
2660 : :
2661 : 131 : impl_->info.total_raw_size += url_meta.size;
2662 : :
2663 [ + - + - ]: 131 : url_ser_map().emplace(url_meta.url, url_meta.ser_type);
2664 [ + - + - ]: 131 : url_schema_type_map().emplace(url_meta.url, url_meta.schema_type);
2665 [ + - ]: 131 : impl_->info.url_metas.emplace_back(std::move(url_meta));
2666 : 131 : }
2667 : :
2668 [ + - ]: 130 : std::sort(impl_->info.url_metas.begin(), impl_->info.url_metas.end());
2669 [ + - ]: 130 : impl_->raw_url_metas = impl_->info.url_metas;
2670 [ + - ]: 130 : rebuild_url_meta_lookup(impl_->info.url_metas);
2671 [ + - + - : 153 : } catch (nlohmann::json::exception& e) {
+ - + - +
- + - +
+ ]
2672 [ + - + - : 2 : VLOG_F("VDBReader: JSON parse error, ", e.what(), ".");
- + ]
2673 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2674 : 1 : }
2675 : : } else {
2676 [ + - ]: 86 : Impl::WrapperFile wrapper_file;
2677 [ + - ]: 86 : wrapper_file.path = impl_->path;
2678 : 86 : wrapper_file.index = 0;
2679 : :
2680 [ + + - + ]: 86 : if VUNLIKELY (!to_open(wrapper_file)) {
2681 : : CLOG_F("VDBReader: Failed to prepare database [%s].", path.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2682 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2683 : : }
2684 : :
2685 : 81 : impl_->info.file_size = 0;
2686 : :
2687 : 81 : std::error_code db_size_ec;
2688 : 81 : std::uintmax_t file_size = std::filesystem::file_size(file_path, db_size_ec);
2689 : :
2690 [ - + ]: 81 : if VUNLIKELY (db_size_ec) {
2691 : : CLOG_W("VDBReader: file_size failed for [%s]: %s.", path.c_str(), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2692 : : db_size_ec.message().c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2693 : : file_size = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2694 : : }
2695 : :
2696 : 81 : impl_->info.file_size += file_size;
2697 : :
2698 : 81 : wrapper_file.start_timestamp_ns = impl_->info.start_timestamp * 1000'000;
2699 : 81 : wrapper_file.begin = impl_->info.blank_duration;
2700 : 81 : wrapper_file.end = impl_->info.total_duration;
2701 : :
2702 [ + - ]: 81 : if (impl_->total_start_timestamp_ns < 0) {
2703 : 81 : impl_->total_start_timestamp_ns = wrapper_file.start_timestamp_ns;
2704 : : }
2705 : :
2706 [ + + ]: 81 : if (!wrapper_file.has_completed) {
2707 : 1 : impl_->total_has_completed = false;
2708 : : }
2709 : :
2710 [ + - ]: 81 : impl_->file_list.emplace_back(std::move(wrapper_file));
2711 [ + - ]: 86 : }
2712 [ + - + - : 250 : } catch (std::filesystem::filesystem_error& e) {
+ - + - ]
2713 : : VLOG_F("VDBReader: Filesystem error, ", e.what(), "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2714 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2715 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2716 : : #else
2717 : : (void)path;
2718 : : VLOG_F("VDBReader: The compile macro VLINK_ENABLE_SQLITE is not turned on.");
2719 : : #endif
2720 : : }
2721 : :
2722 : 432 : void VDBReader::close() {
2723 : : #ifdef VLINK_ENABLE_SQLITE
2724 : 432 : impl_->cursor_stmt.reset();
2725 : :
2726 [ + + ]: 1271 : for (auto& wrapper_file : impl_->file_list) {
2727 [ + + ]: 839 : if (wrapper_file.stmt) {
2728 [ + - ]: 104 : ::sqlite3_finalize(wrapper_file.stmt);
2729 : 104 : wrapper_file.stmt = nullptr;
2730 : : }
2731 : :
2732 [ + - ]: 839 : if (wrapper_file.db) {
2733 [ + - ]: 839 : int ret = ::sqlite3_close_v2(wrapper_file.db);
2734 : :
2735 [ - + ]: 839 : if VUNLIKELY (ret != SQLITE_OK) {
2736 : : CLOG_W("Failed to close database: %s.", ::sqlite3_errmsg(wrapper_file.db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2737 : : }
2738 : :
2739 : 839 : wrapper_file.db = nullptr;
2740 : : }
2741 : : }
2742 : :
2743 : 432 : impl_->file_list.clear();
2744 : : #endif
2745 : 432 : }
2746 : :
2747 : 36 : int VDBReader::get_reset_index(const Config& config) {
2748 : : #ifdef VLINK_ENABLE_SQLITE
2749 : 36 : impl_->is_pending.store(true, std::memory_order_relaxed);
2750 : :
2751 : 36 : int ret = 0;
2752 : 36 : int start_index = -1;
2753 : :
2754 : 36 : int64_t last_time = impl_->begin_time.load(std::memory_order_relaxed);
2755 : :
2756 : 36 : impl_->update_sql_default_str = "SELECT elapsed, url, action, data FROM VLinkDatas ORDER BY rowid;";
2757 : :
2758 : 36 : impl_->update_sql_time_str = "SELECT elapsed, url, action, data FROM VLinkDatas WHERE elapsed >= ";
2759 [ + - ]: 72 : impl_->update_sql_time_str.append(std::to_string(impl_->begin_time.load(std::memory_order_relaxed) * 1000));
2760 : 36 : impl_->update_sql_time_str.append(" ORDER BY rowid;");
2761 : :
2762 : 36 : std::string* select_sql = nullptr;
2763 : :
2764 [ + + ]: 85 : for (auto& wrapper_file : impl_->file_list) {
2765 [ + - + - ]: 49 : if (wrapper_file.has_idx_elapsed && wrapper_file.has_idx_url) {
2766 [ + + ]: 49 : if (config.filter_urls.empty()) {
2767 [ + - ]: 44 : impl_->update_sql_default_str = "SELECT elapsed, url, action, data FROM VLinkDatas ORDER BY elapsed, rowid;";
2768 : :
2769 [ + - ]: 44 : impl_->update_sql_time_str = "SELECT elapsed, url, action, data FROM VLinkDatas WHERE elapsed >= ";
2770 [ + - + - ]: 88 : impl_->update_sql_time_str.append(std::to_string(impl_->begin_time.load(std::memory_order_relaxed) * 1000));
2771 [ + - ]: 44 : impl_->update_sql_time_str.append(" ORDER BY elapsed, rowid;");
2772 : : } else {
2773 [ + - ]: 5 : std::string id_list_str = " url IN (";
2774 : 5 : bool id_appended = false;
2775 : :
2776 [ + + ]: 20 : for (const auto& [url, id] : wrapper_file.url_to_id_map) {
2777 [ + - + + ]: 15 : if (!match_playback_url_filter(url, config.filter_urls)) {
2778 : 11 : continue;
2779 : : }
2780 : :
2781 [ + - + - ]: 4 : id_list_str.append(std::to_string(id));
2782 [ + - ]: 4 : id_list_str.append(",");
2783 : 4 : id_appended = true;
2784 : : }
2785 : :
2786 [ + + ]: 5 : if (id_appended) {
2787 : 4 : id_list_str.pop_back();
2788 [ + - ]: 4 : id_list_str.append(")");
2789 : : } else {
2790 [ + - ]: 1 : id_list_str = " url IN (NULL)";
2791 : : }
2792 : :
2793 [ + - ]: 5 : impl_->update_sql_default_str = "SELECT elapsed, url, action, data FROM VLinkDatas WHERE";
2794 [ + - ]: 5 : impl_->update_sql_default_str.append(id_list_str);
2795 [ + - + - ]: 5 : impl_->update_sql_default_str.append(wrapper_file.has_idx_elapsed
2796 : : ? " ORDER BY elapsed, rowid;"
2797 : : : " ORDER BY rowid;"); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2798 : :
2799 [ + - ]: 5 : impl_->update_sql_time_str = "SELECT elapsed, url, action, data FROM VLinkDatas WHERE";
2800 [ + - ]: 5 : impl_->update_sql_time_str.append(id_list_str);
2801 [ + - ]: 5 : impl_->update_sql_time_str.append(" AND elapsed >= ");
2802 [ + - + - ]: 10 : impl_->update_sql_time_str.append(std::to_string(impl_->begin_time.load(std::memory_order_relaxed) * 1000));
2803 [ + - + - ]: 5 : impl_->update_sql_time_str.append(wrapper_file.has_idx_elapsed
2804 : : ? " ORDER BY elapsed, rowid;"
2805 : : : " ORDER BY rowid;"); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2806 : 5 : }
2807 : : }
2808 : :
2809 [ + + + - : 125 : if (start_index < 0 && impl_->begin_time.load(std::memory_order_relaxed) >= last_time &&
+ + ]
2810 [ + + ]: 76 : impl_->begin_time.load(std::memory_order_relaxed) <= wrapper_file.end) {
2811 [ + + ]: 72 : if (impl_->begin_time.load(std::memory_order_relaxed) > 0) {
2812 : 7 : select_sql = &impl_->update_sql_time_str;
2813 : : } else {
2814 : 29 : select_sql = &impl_->update_sql_default_str;
2815 : : }
2816 : :
2817 : 36 : start_index = wrapper_file.index;
2818 : : } else {
2819 : : select_sql = &impl_->update_sql_default_str; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2820 : : }
2821 : :
2822 [ + + ]: 49 : if VLIKELY (wrapper_file.stmt) {
2823 [ + - ]: 30 : ::sqlite3_finalize(wrapper_file.stmt);
2824 : 30 : wrapper_file.stmt = nullptr;
2825 : : }
2826 : :
2827 [ - + ]: 49 : if VLIKELY (!select_sql) {
2828 : : VLOG_E("VDBReader: Failed to prepare select sql str."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2829 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2830 : : }
2831 : :
2832 [ + - ]: 49 : ret = ::sqlite3_prepare_v2(wrapper_file.db, select_sql->c_str(), -1, &wrapper_file.stmt, nullptr);
2833 : :
2834 [ - + ]: 49 : if VUNLIKELY (ret != SQLITE_OK) {
2835 : : // LCOV_EXCL_START GCOVR_EXCL_START
2836 : : CLOG_W("Failed to prepare datas table: %s.", ::sqlite3_errmsg(wrapper_file.db));
2837 : :
2838 : : start_index = -1;
2839 : :
2840 : : break;
2841 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
2842 : : }
2843 : :
2844 [ + - ]: 49 : ::sqlite3_step(wrapper_file.stmt);
2845 [ + - ]: 49 : ::sqlite3_reset(wrapper_file.stmt);
2846 : :
2847 : 49 : last_time = wrapper_file.end;
2848 : : }
2849 : :
2850 : 36 : impl_->is_pending.store(false, std::memory_order_relaxed);
2851 : :
2852 : 36 : return start_index;
2853 : : #else
2854 : : (void)config;
2855 : : return -1;
2856 : : #endif
2857 : : }
2858 : :
2859 : 34 : void VDBReader::read(const Config& config) {
2860 : : #ifdef VLINK_ENABLE_SQLITE
2861 : 34 : int loop_times = 0;
2862 : :
2863 : 34 : reset_plugin();
2864 : :
2865 [ + + ]: 34 : if (config.auto_pause) {
2866 : 5 : impl_->pause_flag.store(true, std::memory_order_relaxed);
2867 : : }
2868 : :
2869 : 34 : bool is_interrupted = false;
2870 : :
2871 : : do {
2872 : : // prepare
2873 [ + - ]: 36 : int start_index = get_reset_index(config);
2874 : :
2875 [ + + ]: 36 : if (impl_->ready_callback) {
2876 [ + - ]: 13 : impl_->ready_callback();
2877 : : }
2878 : :
2879 [ + - - + : 36 : if VUNLIKELY (start_index < 0 || start_index > static_cast<int>(impl_->file_list.size()) - 1) {
- + ]
2880 : : // LCOV_EXCL_START GCOVR_EXCL_START
2881 : : VLOG_W("VDBReader: Cannot find any data for play.");
2882 : :
2883 : : update_status(kStopped);
2884 : :
2885 : : if (config.auto_quit) {
2886 : : quit();
2887 : : }
2888 : :
2889 : : return;
2890 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
2891 : : }
2892 : :
2893 : : {
2894 [ + - ]: 36 : std::lock_guard time_lock(impl_->time_mtx);
2895 : 36 : impl_->pause_elapsed.store(0, std::memory_order_relaxed);
2896 : 36 : impl_->offset_elapsed.store(0, std::memory_order_relaxed);
2897 : 72 : impl_->real_elapsed.store(impl_->begin_time.load(std::memory_order_relaxed) * 1000U, std::memory_order_relaxed);
2898 : :
2899 : 36 : impl_->elapsed_timer.restart();
2900 : 36 : impl_->pause_elapsed_timer.restart();
2901 : 36 : impl_->offset_timer.restart();
2902 : 36 : impl_->real_timer.restart();
2903 : 36 : }
2904 : :
2905 [ + + ]: 36 : if (impl_->stop_flag.load(std::memory_order_relaxed)) {
2906 : 1 : is_interrupted = true;
2907 [ + - ]: 1 : update_status(kStopped);
2908 : 9 : break;
2909 [ + + ]: 35 : } else if (impl_->jump_flag.load(std::memory_order_relaxed)) {
2910 : 2 : break;
2911 [ + + ]: 33 : } else if (impl_->pause_flag.load(std::memory_order_relaxed)) {
2912 : 5 : impl_->pause_elapsed_timer.restart();
2913 [ + - ]: 5 : update_status(kPaused);
2914 [ + - ]: 5 : do_pause();
2915 : : {
2916 [ + - ]: 5 : std::lock_guard time_lock(impl_->time_mtx);
2917 : 5 : impl_->pause_elapsed.store(0, std::memory_order_relaxed);
2918 : 5 : impl_->offset_elapsed.store(0, std::memory_order_relaxed);
2919 : 10 : impl_->real_elapsed.store(impl_->begin_time.load(std::memory_order_relaxed) * 1000U, std::memory_order_relaxed);
2920 : :
2921 : 5 : impl_->elapsed_timer.restart();
2922 : 5 : impl_->pause_elapsed_timer.restart();
2923 : 5 : impl_->offset_timer.restart();
2924 : 5 : impl_->real_timer.restart();
2925 : 5 : }
2926 : :
2927 : : } else {
2928 [ + - ]: 28 : update_status(kPlaying);
2929 : : }
2930 : :
2931 : 33 : int64_t elapsed = 0;
2932 : 33 : int64_t timestamp = 0;
2933 : 33 : int64_t last_timestamp = 0;
2934 : 33 : int url_id = -1;
2935 : 33 : const uint8_t* data = nullptr;
2936 : 33 : int size = 0;
2937 : 33 : Bytes decompressed_data;
2938 : :
2939 : : // process files
2940 [ + + ]: 73 : for (int index = start_index; index < static_cast<int>(impl_->file_list.size()); ++index) {
2941 : 44 : impl_->split_index.store(index, std::memory_order_relaxed);
2942 : :
2943 [ + - ]: 88 : auto& wrapper_file = impl_->file_list.at(impl_->split_index.load(std::memory_order_relaxed));
2944 : :
2945 [ + - - + : 44 : if VUNLIKELY (!wrapper_file.db || !wrapper_file.stmt) {
- + ]
2946 : : VLOG_W("VDBReader: Target db or stmt is empty."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2947 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2948 : : }
2949 : :
2950 : : // process datas
2951 [ + - + + ]: 128 : for ([[maybe_unused]] int row = 0; ::sqlite3_step(wrapper_file.stmt) == SQLITE_ROW; ++row) {
2952 [ + - ]: 100 : timestamp = ::sqlite3_column_int64(wrapper_file.stmt, get_column(0));
2953 : :
2954 : 100 : impl_->offset_timer.restart();
2955 : :
2956 [ - + ]: 100 : if VUNLIKELY (last_timestamp > timestamp + 10'000U) {
2957 : : VLOG_W("VDBReader: The database timestamp is incorrect."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2958 : : }
2959 : :
2960 : 100 : last_timestamp = timestamp;
2961 : :
2962 [ - + ]: 200 : if (timestamp < impl_->begin_time.load(std::memory_order_relaxed) * 1000U) {
2963 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2964 : : }
2965 : :
2966 [ + + + + ]: 100 : if (config.end_time > 0 && timestamp > config.end_time * 1000U) {
2967 [ + - ]: 24 : is_interrupted = impl_->stop_flag.load(std::memory_order_relaxed) ||
2968 [ + - + - : 24 : impl_->jump_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
- + ]
2969 : 16 : break;
2970 : : }
2971 : :
2972 [ + - ]: 88 : url_id = ::sqlite3_column_int(wrapper_file.stmt, get_column(1));
2973 : :
2974 [ + - ]: 88 : auto iter = wrapper_file.id_to_url_map.find(url_id);
2975 : :
2976 [ - + ]: 88 : if VUNLIKELY (iter == wrapper_file.id_to_url_map.end()) {
2977 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2978 : : }
2979 : :
2980 : 88 : const auto& url = iter->second;
2981 : :
2982 [ - + ]: 88 : if VUNLIKELY (url.empty()) {
2983 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2984 : : }
2985 : :
2986 [ + - - + ]: 88 : if VUNLIKELY (!match_playback_url_filter(url, config.filter_urls)) {
2987 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2988 : : }
2989 : :
2990 [ + - + - : 176 : if (impl_->stop_flag.load(std::memory_order_relaxed) || impl_->jump_flag.load(std::memory_order_relaxed) ||
- + ]
2991 [ + - - + ]: 88 : is_ready_to_quit()) {
2992 : : is_interrupted = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2993 : 0 : break;
2994 : : }
2995 : :
2996 : 88 : std::string_view action_str;
2997 [ + - ]: 88 : const auto* action_ptr = reinterpret_cast<const char*>(::sqlite3_column_text(wrapper_file.stmt, get_column(2)));
2998 : :
2999 [ + - ]: 88 : if (action_ptr != nullptr) {
3000 : 88 : action_str = std::string_view(action_ptr,
3001 [ + - ]: 88 : static_cast<size_t>(::sqlite3_column_bytes(wrapper_file.stmt, get_column(2))));
3002 : : }
3003 : :
3004 [ + - ]: 88 : data = static_cast<const uint8_t*>(::sqlite3_column_blob(wrapper_file.stmt, get_column(3)));
3005 [ + - ]: 88 : size = ::sqlite3_column_bytes(wrapper_file.stmt, get_column(3));
3006 : :
3007 [ + + + + : 88 : if VUNLIKELY (impl_->enable_compress && Bytes::is_compress_data(data, size)) {
+ + ]
3008 : 1 : decompressed_data = Bytes::uncompress_data(data, size, false);
3009 : :
3010 : 1 : data = decompressed_data.data();
3011 : 1 : size = decompressed_data.size();
3012 : : }
3013 : :
3014 : 88 : elapsed =
3015 : 88 : (timestamp / impl_->rate.load(std::memory_order_relaxed)) -
3016 : 88 : (impl_->elapsed_timer.get() - impl_->pause_elapsed.load(std::memory_order_relaxed)) -
3017 : 176 : (impl_->begin_time.load(std::memory_order_relaxed) * 1000U / impl_->rate.load(std::memory_order_relaxed));
3018 : :
3019 : 88 : impl_->extra_elapsed.store(impl_->offset_timer.get(), std::memory_order_relaxed);
3020 : :
3021 : : {
3022 [ + - ]: 88 : std::unique_lock lock(impl_->mtx);
3023 : :
3024 [ + + ]: 88 : if (config.force_delay > 0) {
3025 [ + - ]: 35 : impl_->cv.wait_for(lock, std::chrono::milliseconds(config.force_delay), [this]() -> bool {
3026 [ + - ]: 140 : return impl_->stop_flag.load(std::memory_order_relaxed) ||
3027 [ + + ]: 138 : impl_->pause_next_flag.load(std::memory_order_relaxed) ||
3028 [ + + - + ]: 209 : impl_->jump_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
3029 : : });
3030 [ + + + - ]: 53 : } else if (config.force_delay < 0 && elapsed > 0) {
3031 : 1 : impl_->offset_timer.restart();
3032 : :
3033 [ + - ]: 1 : impl_->cv.wait_for(lock, std::chrono::microseconds(elapsed), [this]() -> bool {
3034 [ + - ]: 4 : return impl_->stop_flag.load(std::memory_order_relaxed) ||
3035 [ + - ]: 4 : impl_->pause_next_flag.load(std::memory_order_relaxed) ||
3036 [ + - ]: 4 : impl_->jump_flag.load(std::memory_order_relaxed) ||
3037 [ + - - + ]: 6 : impl_->pause_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
3038 : : });
3039 : :
3040 [ - + ]: 1 : if VUNLIKELY (impl_->pause_flag.load(std::memory_order_relaxed)) {
3041 : : impl_->offset_elapsed.store(elapsed - impl_->offset_timer.get(), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
3042 : : std::memory_order_relaxed); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
3043 : : }
3044 : : }
3045 : 88 : }
3046 : :
3047 [ + + + + : 172 : if (impl_->stop_flag.load(std::memory_order_relaxed) || impl_->jump_flag.load(std::memory_order_relaxed) ||
+ + ]
3048 [ + - - + ]: 84 : is_ready_to_quit()) {
3049 : 4 : is_interrupted = true;
3050 : 4 : break;
3051 [ + + ]: 84 : } else if (impl_->pause_flag.load(std::memory_order_relaxed)) {
3052 [ + - ]: 9 : do_pause();
3053 : 9 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
3054 : :
3055 [ + - + - : 18 : if (impl_->stop_flag.load(std::memory_order_relaxed) || impl_->jump_flag.load(std::memory_order_relaxed) ||
- + ]
3056 [ + - - + ]: 9 : is_ready_to_quit()) {
3057 : : is_interrupted = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
3058 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
3059 : : }
3060 : : }
3061 : :
3062 : : {
3063 [ + - ]: 84 : std::lock_guard time_lock(impl_->time_mtx);
3064 : 84 : impl_->real_timer.restart();
3065 : 84 : impl_->real_elapsed.store(timestamp, std::memory_order_relaxed);
3066 : 84 : }
3067 : :
3068 : 84 : Frame frame;
3069 : 84 : frame.timestamp = timestamp;
3070 [ + - ]: 84 : frame.url = url;
3071 [ + - ]: 84 : frame.action_type = convert_action(action_str);
3072 : 84 : frame.data = Bytes::shallow_copy(data, size);
3073 : :
3074 [ + - ]: 84 : BagReader::process_output(frame);
3075 : 84 : }
3076 : :
3077 [ + + ]: 44 : if (is_interrupted) {
3078 : 4 : break;
3079 : : }
3080 : : }
3081 : :
3082 [ + + ]: 33 : if (is_interrupted) {
3083 [ + + ]: 4 : if (impl_->stop_flag.load(std::memory_order_relaxed)) {
3084 [ + - ]: 2 : update_status(kStopped);
3085 : : }
3086 : :
3087 : 4 : break;
3088 : : }
3089 : :
3090 [ + + ]: 29 : if (!impl_->jump_flag.load(std::memory_order_relaxed)) {
3091 [ + - ]: 28 : update_status(kStopped);
3092 : :
3093 [ + + ]: 28 : if (config.skip_blank) {
3094 : 3 : impl_->begin_time.store(std::max(config.begin_time, impl_->info.blank_duration), std::memory_order_relaxed);
3095 : : } else {
3096 : 25 : impl_->begin_time.store(config.begin_time, std::memory_order_relaxed);
3097 : : }
3098 : : }
3099 : :
3100 [ + + + + : 56 : if (impl_->stop_flag.load(std::memory_order_relaxed) || impl_->jump_flag.load(std::memory_order_relaxed) ||
+ + ]
3101 [ + - - + ]: 27 : is_ready_to_quit()) {
3102 : 2 : is_interrupted = true;
3103 : 2 : break;
3104 : : }
3105 : :
3106 [ + - ]: 27 : flush_plugin();
3107 [ + - + + : 114 : } while (impl_->times.load(std::memory_order_relaxed) <= 0 ||
- + + ]
3108 [ + - ]: 54 : (impl_->times.load(std::memory_order_relaxed) > 0 &&
3109 [ + + ]: 54 : ++loop_times < impl_->times.load(std::memory_order_relaxed)));
3110 : :
3111 [ + + ]: 34 : if (impl_->stop_flag.load(std::memory_order_relaxed)) {
3112 : 4 : is_interrupted = true;
3113 : : }
3114 : :
3115 [ + + + + : 34 : if (!impl_->jump_flag.load(std::memory_order_relaxed) && impl_->finish_callback) {
+ + ]
3116 : 28 : impl_->finish_callback(is_interrupted);
3117 : : }
3118 : :
3119 [ + + + + : 34 : if (!impl_->jump_flag.load(std::memory_order_relaxed) && config.auto_quit) {
+ + ]
3120 : 1 : quit();
3121 : : }
3122 : : #else
3123 : : (void)config;
3124 : : #endif
3125 : : }
3126 : :
3127 : : } // namespace vlink
|
