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