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_writer.h"
25 : :
26 : : #include <algorithm>
27 : : #include <array>
28 : : #include <cstdint>
29 : : #include <cstdio>
30 : : #include <filesystem>
31 : : #include <fstream>
32 : : #include <memory>
33 : : #include <mutex>
34 : : #include <string>
35 : : #include <string_view>
36 : : #include <unordered_map>
37 : : #include <unordered_set>
38 : : #include <utility>
39 : : #include <vector>
40 : :
41 : : #include "./base/elapsed_timer.h"
42 : : #include "./base/helpers.h"
43 : : #include "./base/logger.h"
44 : : #include "./base/utils.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 : : // schema_plugin
55 : : #include "./extension/schema_plugin_interface.h"
56 : :
57 : : namespace vlink {
58 : :
59 : 25822 : [[maybe_unused]] static constexpr int get_column(int column) noexcept { return column + 1; }
60 : :
61 : : static constexpr int kSyncWriteInterval = 1000; // ms
62 : : static constexpr int kCompressMaxIgnoreCnt = 5;
63 : : static constexpr double kCompressMaxRatio = 0.95;
64 : :
65 : : // VDBWriter::Impl
66 : : struct VDBWriter::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
67 : : // UrlMsgInfo
68 : : struct UrlMsgInfo final {
69 : : int index{0};
70 : : size_t count{0};
71 : : size_t size{0};
72 : : int64_t first_timestamp{-1};
73 : : int64_t last_timestamp{-1};
74 : : int64_t previous_segment_last_timestamp{-1};
75 : : double freq{0};
76 : : double loss{0};
77 : : std::string url_type;
78 : : std::string ser_type;
79 : : SchemaType schema_type{SchemaType::kUnknown};
80 : :
81 : : bool operator<(const UrlMsgInfo& target) const noexcept { return index < target.index; }
82 : : };
83 : :
84 : : struct SplitInfo final {
85 : : int64_t row{0};
86 : : std::unordered_map<std::string, UrlMsgInfo> url_map;
87 : : };
88 : :
89 : : struct WriteStateSnapshot final {
90 : : int64_t current_row{0};
91 : : int64_t current_size{0};
92 : : bool has_oversize{false};
93 : : int64_t last_timestamp{0};
94 : : std::vector<std::string> total_url_list;
95 : : int64_t total_current_row{0};
96 : : int64_t total_current_size{0};
97 : : int64_t total_timestamp{0};
98 : : std::unordered_map<std::string, UrlMsgInfo> url_map;
99 : : std::unordered_set<std::string> ser_map;
100 : : std::unordered_map<std::string, UrlMsgInfo> total_url_map;
101 : : std::unordered_map<std::string, SchemaData> total_schema_map;
102 : : std::unordered_map<std::string, int64_t> compress_ignore_map;
103 : : std::string write_url_type;
104 : :
105 : 2787 : explicit WriteStateSnapshot(const VDBWriter::Impl& impl)
106 : 2787 : : current_row(impl.current_row),
107 : 2787 : current_size(impl.current_size),
108 : 2787 : has_oversize(impl.has_oversize),
109 : 2787 : last_timestamp(impl.last_timestamp),
110 : 2787 : total_url_list(impl.total_url_list),
111 : 2787 : total_current_row(impl.total_current_row),
112 : 2787 : total_current_size(impl.total_current_size),
113 : 2787 : total_timestamp(impl.total_timestamp),
114 [ + - ]: 2787 : url_map(impl.url_map),
115 [ + - ]: 2787 : ser_map(impl.ser_map),
116 [ + - ]: 2787 : total_url_map(impl.total_url_map),
117 [ + - ]: 2787 : total_schema_map(impl.total_schema_map),
118 [ + - ]: 2787 : compress_ignore_map(impl.compress_ignore_map),
119 [ + - ]: 2787 : write_url_type(impl.write_url_type) {}
120 : :
121 : 0 : void restore(VDBWriter::Impl& impl) const {
122 : 0 : impl.current_row = current_row;
123 : 0 : impl.current_size = current_size;
124 : 0 : impl.has_oversize = has_oversize;
125 : 0 : impl.last_timestamp = last_timestamp;
126 : 0 : impl.total_url_list = total_url_list;
127 : 0 : impl.total_current_row = total_current_row;
128 : 0 : impl.total_current_size = total_current_size;
129 : 0 : impl.total_timestamp = total_timestamp;
130 : 0 : impl.url_map = url_map;
131 : 0 : impl.ser_map = ser_map;
132 : 0 : impl.total_url_map = total_url_map;
133 : 0 : impl.total_schema_map = total_schema_map;
134 : 0 : impl.compress_ignore_map = compress_ignore_map;
135 : 0 : impl.write_url_type = write_url_type;
136 : 0 : }
137 : : };
138 : :
139 : : struct MemoryCharge final {
140 : 302 : MemoryCharge(std::atomic<int64_t>& counter, int64_t bytes) : value(&counter), size(bytes) {}
141 : :
142 : 604 : MemoryCharge(MemoryCharge&& other) noexcept : value(std::exchange(other.value, nullptr)), size(other.size) {}
143 : :
144 : 906 : ~MemoryCharge() {
145 [ + + ]: 906 : if (value) {
146 : 302 : value->fetch_sub(size, std::memory_order_relaxed);
147 : : }
148 : 906 : }
149 : :
150 : : std::atomic<int64_t>* value;
151 : : int64_t size;
152 : :
153 : : VLINK_DISALLOW_COPY_AND_ASSIGN(MemoryCharge)
154 : : };
155 : :
156 : : std::atomic_bool is_dumping{false};
157 : : std::atomic_bool is_split_mode{false};
158 : : std::atomic<int> split_index{0};
159 : : std::atomic<int64_t> memory_size{0};
160 : : std::atomic_bool in_cached{false};
161 : : std::atomic<int64_t> cached_size{0};
162 : : std::atomic_bool quit_flag{false};
163 : :
164 : : std::string path;
165 : : std::filesystem::path split_output_dir;
166 : : std::string base_dir;
167 : : std::string base_name;
168 : : BagWriter::Config config;
169 : : ElapsedTimer elapsed_timer{ElapsedTimer::kMicro};
170 : :
171 : : int64_t current_row{0};
172 : : int64_t current_size{0};
173 : : bool has_oversize{false};
174 : :
175 : : int64_t last_timestamp{0};
176 : :
177 : : BagWriter::SystemClock time_start;
178 : : BagWriter::SystemClock time_current;
179 : : int64_t start_timestamp{0};
180 : :
181 : : std::vector<std::string> split_file_list;
182 : : std::vector<SplitInfo> split_info_list;
183 : : bool split_before{false};
184 : : bool split_first{false};
185 : :
186 : : std::vector<std::string> total_url_list;
187 : : int64_t total_current_row{0};
188 : : int64_t total_current_size{0};
189 : : int64_t total_timestamp{0};
190 : :
191 : : std::unordered_map<std::string, UrlMsgInfo> url_map;
192 : : std::unordered_set<std::string> ser_map;
193 : : std::unordered_map<std::string, UrlMsgInfo> total_url_map;
194 : : std::unordered_map<std::string, SchemaData> total_schema_map;
195 : :
196 : : BagWriter::SplitCallback split_callback;
197 : : BagWriter::SchemaCallback schema_callback;
198 : : std::string split_filename;
199 : : std::mutex split_mtx;
200 : :
201 : : std::string app_name;
202 : : std::string tag_name;
203 : : int32_t timezone_diff{0};
204 : :
205 : : bool enable_compressed{false};
206 : : std::mutex write_mtx;
207 : :
208 : : ElapsedTimer cache_timer;
209 : : Timer check_timer;
210 : :
211 : : ElapsedTimer sync_timer;
212 : :
213 : : std::unordered_map<std::string, int64_t> compress_ignore_map;
214 : : std::unique_ptr<WriteStateSnapshot> cache_snapshot;
215 : :
216 : : std::string write_url_type;
217 : :
218 : : // database
219 : : #ifdef VLINK_ENABLE_SQLITE
220 : : ::sqlite3* db{nullptr};
221 : : ::sqlite3_stmt* schemas_stmt{nullptr};
222 : : ::sqlite3_stmt* datas_stmt{nullptr};
223 : : ::sqlite3_stmt* urls_stmt{nullptr};
224 : : ::sqlite3_stmt* update_complete_stmt{nullptr};
225 : : ::sqlite3_stmt* update_header_stmt{nullptr};
226 : : ::sqlite3_stmt* update_url_loss_stmt{nullptr};
227 : : ::sqlite3_stmt* update_url_meta_stmt{nullptr};
228 : : ::sqlite3_stmt* update_urls_stmt{nullptr};
229 : : #endif
230 : :
231 : : // schema plugin interface
232 : : SchemaPluginInterface* schema_plugin_interface{nullptr};
233 : : };
234 : :
235 : : // VDBWriter
236 : 182 : VDBWriter::VDBWriter(const std::string& path, const Config& config)
237 [ + - ]: 182 : : BagWriter(path, config), impl_(std::make_unique<Impl>()) {
238 [ + - + - ]: 182 : set_name("VDBWriter");
239 : :
240 [ + - ]: 182 : impl_->url_map.reserve(128);
241 [ + - ]: 182 : impl_->ser_map.reserve(128);
242 [ + - + - ]: 182 : url_loss_map_ref().reserve(128);
243 [ + - ]: 182 : impl_->total_url_map.reserve(128);
244 [ + - + - ]: 182 : total_url_loss_map_ref().reserve(128);
245 [ + - ]: 182 : impl_->total_schema_map.reserve(128);
246 : :
247 [ + - ]: 182 : impl_->schema_plugin_interface = get_schema_interface();
248 : :
249 [ + - + - ]: 182 : impl_->app_name = get_default_app_name();
250 : :
251 [ + + ]: 182 : if (config.tag_name.empty()) {
252 [ + - + - ]: 81 : impl_->tag_name = get_default_tag_name();
253 : : } else {
254 [ + - ]: 101 : impl_->tag_name = config.tag_name;
255 : : }
256 : :
257 [ + - ]: 182 : impl_->timezone_diff = get_default_timezone_diff();
258 : :
259 [ + - ]: 182 : impl_->path = path;
260 [ + - ]: 182 : impl_->config = config;
261 : :
262 [ + + + + ]: 182 : impl_->enable_compressed = impl_->config.compress == kCompressAuto || impl_->config.compress == kCompressLzav;
263 : :
264 [ + + ]: 182 : if VUNLIKELY (impl_->config.max_task_depth <= 0) {
265 : 2 : impl_->config.max_task_depth = BagWriter::Config().max_task_depth;
266 : : }
267 : :
268 [ + - ]: 182 : reset_lockfree_capacity();
269 : :
270 [ + + ]: 182 : if (!config.sync_mode) {
271 [ + - ]: 40 : impl_->check_timer.attach(this);
272 [ + - ]: 40 : impl_->check_timer.set_interval(kSyncWriteInterval);
273 [ + - + - ]: 40 : impl_->check_timer.start([this]() {
274 : : // LCOV_EXCL_START GCOVR_EXCL_START
275 : : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
276 : : return;
277 : : }
278 : :
279 : : if (impl_->cache_timer.get() > 3000) {
280 : : std::lock_guard lock(impl_->write_mtx);
281 : : sync_cache();
282 : : }
283 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
284 : : });
285 : : }
286 : :
287 : : try {
288 : : #ifdef _WIN32
289 : : std::filesystem::path file_path(Helpers::string_to_wstring(path));
290 : : std::string suffix = Helpers::path_to_string(file_path.extension());
291 : : #else
292 [ + - ]: 182 : std::filesystem::path file_path(path);
293 [ + - + - ]: 182 : std::string suffix = file_path.extension().string();
294 : : #endif
295 : :
296 [ + - ]: 182 : const auto parent_path = file_path.parent_path();
297 : :
298 : 996 : std::transform(suffix.begin(), suffix.end(), suffix.begin(), [](unsigned char c) { return std::tolower(c); });
299 : :
300 [ + + ]: 182 : if (suffix == ".vdbx") {
301 [ + - + + ]: 86 : if (std::filesystem::exists(file_path)) {
302 : : try {
303 : 4 : nlohmann::json root_json;
304 : :
305 : : {
306 [ + - ]: 4 : std::ifstream filex(file_path);
307 : :
308 [ + + ]: 4 : filex >> root_json;
309 : :
310 [ + - ]: 3 : filex.close();
311 : 4 : }
312 : :
313 [ + - + - ]: 3 : nlohmann::json files_json = root_json["VLinkFiles"];
314 : :
315 [ + - + - : 12 : for (const auto& file_info : files_json) {
+ - + + ]
316 [ + + ]: 9 : if (!file_info.is_string()) {
317 : 4 : continue;
318 : : }
319 : :
320 [ + - ]: 8 : const auto stale_file_name = file_info.get<std::string>();
321 : : #ifdef _WIN32
322 : : const std::filesystem::path stale_file_path(Helpers::string_to_wstring(stale_file_name));
323 : : #else
324 [ + - ]: 8 : const std::filesystem::path stale_file_path(stale_file_name);
325 : : #endif
326 : :
327 [ + - + - : 24 : if (stale_file_path.empty() || stale_file_path == "." || stale_file_path == ".." ||
+ - + - +
- + + + -
+ + - - -
- ]
328 [ + - + - : 16 : stale_file_path != stale_file_path.filename()) {
+ - - - ]
329 [ + - + - : 6 : CLOG_W("VDBWriter: Ignore unsafe split file path [%s].", stale_file_name.c_str());
+ - ]
330 : 3 : continue;
331 : 3 : }
332 : :
333 [ + - ]: 5 : const auto stale_output_path = parent_path / stale_file_path;
334 : 5 : std::error_code remove_ec;
335 : 5 : std::filesystem::remove(stale_output_path, remove_ec);
336 : :
337 [ + + ]: 5 : if VUNLIKELY (remove_ec) {
338 [ + - + - : 2 : CLOG_W("VDBWriter: Failed to remove stale split path [%s]: %s.", stale_file_name.c_str(),
+ - + - +
- - - ]
339 : : remove_ec.message().c_str());
340 : : }
341 [ + + + + ]: 11 : }
342 : :
343 [ + - ]: 3 : std::filesystem::remove(file_path);
344 [ - + ]: 5 : } catch (const nlohmann::json::exception& e) {
345 [ + - + - : 2 : CLOG_W("VDBWriter: Failed to parse stale split manifest [%s]: %s.", path.c_str(), e.what());
+ - ]
346 : 1 : }
347 : : }
348 : :
349 : 86 : impl_->is_split_mode.store(true, std::memory_order_release);
350 : 86 : impl_->split_index.store(0, std::memory_order_relaxed);
351 : :
352 : : #ifdef _WIN32
353 : :
354 : : std::error_code absolute_ec;
355 : : auto absolute_path = std::filesystem::absolute(file_path, absolute_ec);
356 : :
357 : : if VUNLIKELY (absolute_ec) {
358 : : absolute_path = file_path;
359 : : }
360 : :
361 : : impl_->path = Helpers::path_to_string(absolute_path);
362 : : impl_->split_output_dir = absolute_path.parent_path();
363 : :
364 : : if (parent_path.empty()) {
365 : : impl_->base_dir.clear();
366 : : impl_->base_name = Helpers::path_to_string(file_path.stem());
367 : : } else {
368 : : impl_->base_dir = Helpers::path_to_string(parent_path);
369 : : impl_->base_name = Helpers::path_to_string(std::filesystem::path(parent_path / file_path.stem()));
370 : : }
371 : : #else
372 : :
373 : 86 : std::error_code absolute_ec;
374 [ + - ]: 86 : auto absolute_path = std::filesystem::absolute(file_path, absolute_ec);
375 : :
376 [ - + ]: 86 : if VUNLIKELY (absolute_ec) {
377 [ # # ]: 0 : absolute_path = file_path;
378 : : }
379 : :
380 [ + - ]: 86 : impl_->path = absolute_path.string();
381 [ + - ]: 86 : impl_->split_output_dir = absolute_path.parent_path();
382 : :
383 [ + + ]: 86 : if (parent_path.empty()) {
384 : 3 : impl_->base_dir.clear();
385 [ + - + - ]: 3 : impl_->base_name = file_path.stem().string();
386 : : } else {
387 [ + - ]: 83 : impl_->base_dir = parent_path.string();
388 [ + - + - : 83 : impl_->base_name = std::filesystem::path(parent_path / file_path.stem()).string();
+ - ]
389 : : }
390 : : #endif
391 : :
392 [ + - ]: 86 : impl_->time_start = std::chrono::time_point_cast<std::chrono::milliseconds>(std::chrono::system_clock::now());
393 : 86 : impl_->time_current = impl_->time_start;
394 : :
395 [ + + ]: 86 : if (impl_->config.start_timestamp > 0) {
396 : 30 : impl_->start_timestamp = impl_->config.start_timestamp;
397 : : } else {
398 : 56 : impl_->start_timestamp = impl_->time_start.time_since_epoch().count();
399 : : }
400 : :
401 [ + - ]: 86 : write_filex(false);
402 : :
403 [ + + ]: 86 : if (impl_->config.split_name_by_time) {
404 [ + + ]: 27 : if (impl_->base_dir.empty()) {
405 [ + - + - ]: 2 : impl_->split_filename = get_format_date(&impl_->time_current, true) + ".vdb";
406 : : } else {
407 [ + - + - : 25 : impl_->split_filename = impl_->base_dir + "/" + get_format_date(&impl_->time_current, true) + ".vdb";
+ - + - ]
408 : : }
409 : : } else {
410 : 59 : impl_->split_filename =
411 [ + - + - : 177 : impl_->base_name + "." + std::to_string(impl_->split_index.load(std::memory_order_relaxed) + 1) + ".vdb";
+ - + - ]
412 : : }
413 : :
414 [ + - ]: 86 : open_split(impl_->split_filename);
415 : 86 : } else {
416 [ + - ]: 96 : impl_->time_start = std::chrono::time_point_cast<std::chrono::milliseconds>(std::chrono::system_clock::now());
417 : 96 : impl_->time_current = impl_->time_start;
418 : :
419 [ + + ]: 96 : if (impl_->config.start_timestamp > 0) {
420 : 79 : impl_->start_timestamp = impl_->config.start_timestamp;
421 : : } else {
422 : 17 : impl_->start_timestamp = impl_->time_start.time_since_epoch().count();
423 : : }
424 : :
425 : 96 : impl_->is_split_mode.store(false, std::memory_order_release);
426 : 96 : impl_->split_index.store(0, std::memory_order_relaxed);
427 : :
428 [ + - ]: 96 : open(path);
429 : : }
430 [ - - ]: 182 : } catch (std::filesystem::filesystem_error& e) {
431 : : VLOG_F("VDBWriter: Filesystem error during init, ", e.what(), "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
432 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
433 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
434 : :
435 : 182 : impl_->elapsed_timer.start();
436 : 182 : impl_->sync_timer.start();
437 : :
438 : : #ifndef VLINK_ENABLE_SQLITE
439 : : VLOG_F("VDBWriter: The compile macro VLINK_ENABLE_SQLITE is not turned on.");
440 : : #endif
441 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
442 : :
443 : 185 : VDBWriter::~VDBWriter() {
444 : : #ifdef _WIN32
445 : : if (Utils::is_terminating()) {
446 : : (void)impl_.release();
447 : : return;
448 : : }
449 : : #endif
450 : :
451 : 182 : detach_plugin();
452 : :
453 : 182 : impl_->quit_flag.store(true, std::memory_order_release);
454 : :
455 : 182 : impl_->check_timer.stop();
456 : :
457 [ - + ]: 182 : if VUNLIKELY (!wait_for_idle(30000U)) {
458 : : VLOG_W("VDBWriter: Force to quit."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
459 : : }
460 : :
461 : : #ifdef VLINK_ENABLE_SQLITE
462 : :
463 [ + + ]: 182 : if VLIKELY (impl_->db) {
464 : 112 : ::sqlite3_interrupt(impl_->db);
465 : : }
466 : : #endif
467 : :
468 : 182 : quit(true);
469 : :
470 : 182 : wait_for_quit();
471 : :
472 : 182 : close();
473 : 185 : }
474 : :
475 : 252 : void VDBWriter::close() {
476 : 252 : close_segment();
477 : :
478 [ + + - + : 252 : if VUNLIKELY (impl_->is_split_mode.load(std::memory_order_acquire) && !write_filex(true)) {
- + ]
479 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
480 : : }
481 : 252 : }
482 : :
483 : 75 : void VDBWriter::register_split_callback(SplitCallback&& callback, bool before) {
484 [ + - ]: 75 : std::lock_guard lock(impl_->split_mtx);
485 : 75 : impl_->split_before = before;
486 : 75 : impl_->split_callback = std::move(callback);
487 : 75 : }
488 : :
489 : 5 : void VDBWriter::register_schema_callback(SchemaCallback&& callback) {
490 [ + - ]: 5 : std::lock_guard lock(impl_->write_mtx);
491 : 5 : impl_->schema_callback = std::move(callback);
492 : 5 : }
493 : :
494 : 46 : bool VDBWriter::push_schema(const SchemaData& schema_data) {
495 [ + - ]: 46 : SchemaData stored_schema = schema_data;
496 : :
497 [ + + ]: 46 : if (impl_->config.sync_mode) {
498 [ + - ]: 41 : std::lock_guard lock(impl_->write_mtx);
499 [ + - ]: 41 : return merge_schema(stored_schema);
500 : 41 : }
501 : :
502 [ + - + - ]: 5 : bool posted = post_persistent_task([this, stored_schema = std::move(stored_schema)]() mutable {
503 [ + - ]: 5 : std::lock_guard lock(impl_->write_mtx);
504 : :
505 [ + - - + ]: 5 : if VUNLIKELY (!merge_schema(stored_schema)) {
506 [ # # # # : 0 : CLOG_E("VDBWriter: Deferred merge_schema failed for [%s] in async push_schema path.", stored_schema.name.c_str());
# # ]
507 : 0 : set_fail();
508 : : }
509 : 5 : });
510 : :
511 : 5 : return posted;
512 : 46 : }
513 : :
514 : 31 : bool VDBWriter::is_dumping() const { return impl_->is_dumping.load(std::memory_order_relaxed); }
515 : :
516 : 62 : bool VDBWriter::is_split_mode() const { return impl_->is_split_mode.load(std::memory_order_acquire); }
517 : :
518 : 146 : int VDBWriter::get_split_index() const { return impl_->split_index.load(std::memory_order_relaxed); }
519 : :
520 : 965 : int64_t VDBWriter::record(const Frame& frame, int64_t timestamp) {
521 : : #ifdef VLINK_ENABLE_SQLITE
522 : :
523 : 965 : const std::string& url = frame.url;
524 : 965 : const std::string& ser_type = frame.ser_type;
525 : 965 : const SchemaType schema_type = frame.schema_type;
526 : 965 : const ActionType action_type = frame.action_type;
527 : 965 : const Bytes& data = frame.data;
528 : 965 : const int64_t microseconds_timestamp = timestamp;
529 : :
530 [ + + ]: 965 : if (impl_->config.sync_mode) {
531 [ + - ]: 662 : std::lock_guard lock(impl_->write_mtx);
532 : :
533 [ + - + + ]: 662 : if VUNLIKELY (!write(url, ser_type, schema_type, action_type, data, microseconds_timestamp)) {
534 : 7 : return -1;
535 : : }
536 [ + + ]: 662 : } else {
537 [ + + ]: 606 : if VUNLIKELY (impl_->memory_size.load(std::memory_order_relaxed) + static_cast<int64_t>(data.size()) >
538 : : impl_->config.max_memory_size) {
539 [ + - + - : 2 : CLOG_E("The memory data in the queue exceeds %.1fGB and the task is automatically discarded.",
+ - ]
540 : : impl_->config.max_memory_size / 1024.0 / 1024.0 / 1024.0);
541 : :
542 : 2 : return -1;
543 : : }
544 : :
545 : 302 : int url_index = -1;
546 : 302 : int ser_index = -1;
547 : :
548 [ + - ]: 302 : get_url_meta(url, ser_type, url_index, ser_index);
549 : :
550 : 302 : const auto queued_size = static_cast<int64_t>(data.size());
551 : :
552 : 302 : impl_->memory_size.fetch_add(queued_size, std::memory_order_relaxed);
553 : 302 : Impl::MemoryCharge memory_charge(impl_->memory_size, queued_size);
554 : :
555 [ + - + - ]: 604 : bool posted = post_persistent_task([this, url_index, ser_index, schema_type, action_type, data,
556 : 302 : memory_charge = std::move(memory_charge),
557 : : microseconds_timestamp]() { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
558 : : (void)memory_charge;
559 : :
560 : 301 : std::string url;
561 : 301 : std::string ser_type;
562 : :
563 [ + - ]: 301 : std::lock_guard lock(impl_->write_mtx);
564 [ + - ]: 301 : get_url_meta(url_index, ser_index, url, ser_type);
565 : :
566 [ + - + + ]: 301 : if VUNLIKELY (!write(url, ser_type, schema_type, action_type, data, microseconds_timestamp)) {
567 : 1 : set_fail();
568 : : }
569 : 301 : });
570 : :
571 [ + + ]: 302 : if VUNLIKELY (!posted) {
572 : : return -1; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
573 : : }
574 [ + + ]: 302 : }
575 : :
576 : 956 : return microseconds_timestamp;
577 : : #else
578 : : (void)frame;
579 : : (void)timestamp;
580 : : return -1;
581 : : #endif
582 : : }
583 : :
584 : 1 : int64_t VDBWriter::get_record_timestamp() const { return impl_->elapsed_timer.get(); }
585 : :
586 : 308 : size_t VDBWriter::get_max_task_count() const { return impl_->config.max_task_depth; }
587 : :
588 : 37 : void VDBWriter::on_begin() {
589 : 37 : MessageLoop::on_begin();
590 : :
591 : 37 : impl_->elapsed_timer.restart();
592 : 37 : impl_->sync_timer.restart();
593 : 37 : }
594 : :
595 : 37 : void VDBWriter::on_end() { MessageLoop::on_end(); }
596 : :
597 : 749 : void VDBWriter::open(const std::string& path) {
598 : : #ifdef VLINK_ENABLE_SQLITE
599 : : try {
600 : : #ifdef _WIN32
601 : : impl_->split_file_list.emplace_back(Helpers::path_to_string(std::filesystem::path(path).filename()));
602 : : std::filesystem::path file_path(Helpers::string_to_wstring(path));
603 : : #else
604 [ + - + - : 749 : impl_->split_file_list.emplace_back(std::filesystem::path(path).filename().string());
+ - + - ]
605 [ + - ]: 749 : std::filesystem::path file_path(path);
606 : : #endif
607 : :
608 [ + - + + ]: 749 : if (std::filesystem::exists(file_path)) {
609 [ + - ]: 1 : std::filesystem::remove(file_path);
610 : : } else {
611 [ + - ]: 748 : auto parent_path = file_path.parent_path();
612 : :
613 [ + - + - : 748 : if (!parent_path.empty() && !std::filesystem::exists(parent_path)) {
+ + + + ]
614 [ + - ]: 1 : std::filesystem::create_directories(parent_path);
615 : : }
616 : 748 : }
617 [ - - ]: 749 : } catch (std::filesystem::filesystem_error& e) {
618 : : VLOG_F("VDBWriter: Filesystem error during file preparation, ", e.what(), "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
619 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
620 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
621 : :
622 : 749 : int ret = 0;
623 : 749 : char* err_msg = nullptr;
624 : :
625 : 749 : auto free_err_msg = [&err_msg]() noexcept {
626 [ - + ]: 749 : if (err_msg) {
627 : : ::sqlite3_free(err_msg); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
628 : : err_msg = nullptr; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
629 : : }
630 : 1498 : };
631 : :
632 : 749 : auto finalize_stmt = [](sqlite3_stmt*& stmt) noexcept {
633 [ + - ]: 749 : if (stmt) {
634 : 749 : ::sqlite3_finalize(stmt);
635 : 749 : stmt = nullptr;
636 : : }
637 : 749 : };
638 : :
639 : 0 : auto close_db = [this, &finalize_stmt]() noexcept {
640 : : // LCOV_EXCL_START GCOVR_EXCL_START
641 : : if (!impl_->db) {
642 : : return;
643 : : }
644 : :
645 : : if (impl_->in_cached.load(std::memory_order_relaxed)) {
646 : : ::sqlite3_exec(impl_->db, "ROLLBACK;", nullptr, nullptr, nullptr);
647 : : impl_->in_cached.store(false, std::memory_order_relaxed);
648 : : impl_->cached_size.store(0, std::memory_order_relaxed);
649 : : impl_->cache_timer.stop();
650 : : }
651 : :
652 : : finalize_stmt(impl_->schemas_stmt);
653 : : finalize_stmt(impl_->datas_stmt);
654 : : finalize_stmt(impl_->urls_stmt);
655 : : finalize_stmt(impl_->update_complete_stmt);
656 : : finalize_stmt(impl_->update_header_stmt);
657 : : finalize_stmt(impl_->update_url_loss_stmt);
658 : : finalize_stmt(impl_->update_url_meta_stmt);
659 : : finalize_stmt(impl_->update_urls_stmt);
660 : :
661 : : const char* err_ptr = ::sqlite3_errmsg(impl_->db);
662 : : std::string close_err = err_ptr ? err_ptr : std::string{};
663 : : int close_ret = ::sqlite3_close_v2(impl_->db);
664 : :
665 : : if VUNLIKELY (close_ret != SQLITE_OK) {
666 : : CLOG_W("Failed to close database (rc=%d): %s.", close_ret, close_err.c_str());
667 : : }
668 : :
669 : : impl_->db = nullptr;
670 : : };
671 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
672 : :
673 : : // open db
674 [ + - ]: 749 : ret = ::sqlite3_open_v2(path.c_str(), &impl_->db, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE, nullptr);
675 : :
676 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
677 : : CLOG_F("Failed to open database [%s].", path.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
678 : : close_db(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
679 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
680 : : }
681 : :
682 : : // opt busy_timeout
683 [ + - ]: 749 : ::sqlite3_busy_timeout(impl_->db, 100);
684 : :
685 : : // opt sqlite temp_store
686 [ + - ]: 749 : ret = ::sqlite3_exec(impl_->db, "PRAGMA temp_store = MEMORY;", nullptr, nullptr, &err_msg);
687 : :
688 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
689 : : // LCOV_EXCL_START GCOVR_EXCL_START
690 : : CLOG_F("Failed to set temp_store: %s.", err_msg);
691 : :
692 : : free_err_msg();
693 : : close_db();
694 : :
695 : : return;
696 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
697 : : }
698 : :
699 : : // opt sqlite page_size
700 [ + - ]: 749 : ret = ::sqlite3_exec(impl_->db, "PRAGMA page_size = 16384;", nullptr, nullptr, &err_msg);
701 : :
702 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
703 : : // LCOV_EXCL_START GCOVR_EXCL_START
704 : : CLOG_F("Failed to set page_size: %s.", err_msg);
705 : :
706 : : free_err_msg();
707 : : close_db();
708 : :
709 : : return;
710 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
711 : : }
712 : :
713 : : // opt sqlite cache_size
714 [ + - ]: 749 : ret = ::sqlite3_exec(impl_->db, "PRAGMA cache_size = 8192;", nullptr, nullptr, &err_msg);
715 : :
716 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
717 : : // LCOV_EXCL_START GCOVR_EXCL_START
718 : : CLOG_F("Failed to set cache_size: %s.", err_msg);
719 : :
720 : : free_err_msg();
721 : : close_db();
722 : :
723 : : return;
724 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
725 : : }
726 : :
727 : : // opt sqlite synchronous
728 [ + - ]: 749 : ret = ::sqlite3_exec(impl_->db, "PRAGMA synchronous = OFF;", nullptr, nullptr, &err_msg);
729 : :
730 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
731 : : // LCOV_EXCL_START GCOVR_EXCL_START
732 : : CLOG_F("Failed to set synchronous: %s.", err_msg);
733 : :
734 : : free_err_msg();
735 : : close_db();
736 : :
737 : : return;
738 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
739 : : }
740 : :
741 : : // opt sqlite journal_mode
742 : :
743 [ + + ]: 749 : if (impl_->config.wal_mode) {
744 [ + - ]: 3 : ret = ::sqlite3_exec(impl_->db, "PRAGMA journal_mode = WAL;", nullptr, nullptr, &err_msg);
745 [ - + ]: 3 : if VUNLIKELY (ret != SQLITE_OK) {
746 : : // LCOV_EXCL_START GCOVR_EXCL_START
747 : : CLOG_F("Failed to set journal_mode: %s.", err_msg);
748 : :
749 : : free_err_msg();
750 : : close_db();
751 : :
752 : : return;
753 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
754 : : }
755 : : } else {
756 [ + - ]: 746 : ret = ::sqlite3_exec(impl_->db, "PRAGMA journal_mode = OFF;", nullptr, nullptr, &err_msg);
757 [ - + ]: 746 : if VUNLIKELY (ret != SQLITE_OK) {
758 : : // LCOV_EXCL_START GCOVR_EXCL_START
759 : : CLOG_F("Failed to set journal_mode: %s.", err_msg);
760 : :
761 : : free_err_msg();
762 : : close_db();
763 : :
764 : : return;
765 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
766 : : }
767 : : }
768 : :
769 : : // opt sqlite automatic_index
770 [ + - ]: 749 : ret = ::sqlite3_exec(impl_->db, "PRAGMA automatic_index = OFF;", nullptr, nullptr, &err_msg);
771 : :
772 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
773 : : // LCOV_EXCL_START GCOVR_EXCL_START
774 : : CLOG_F("Failed to set automatic_index: %s.", err_msg);
775 : :
776 : : free_err_msg();
777 : : close_db();
778 : :
779 : : return;
780 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
781 : : }
782 : :
783 : : // opt sqlite locking_mode
784 [ + - ]: 749 : ret = ::sqlite3_exec(impl_->db, "PRAGMA locking_mode = EXCLUSIVE;", nullptr, nullptr, &err_msg);
785 : :
786 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
787 : : // LCOV_EXCL_START GCOVR_EXCL_START
788 : : CLOG_F("Failed to set locking_mode: %s.", err_msg);
789 : :
790 : : free_err_msg();
791 : : close_db();
792 : :
793 : : return;
794 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
795 : : }
796 : :
797 : : // set busy_timeout
798 [ + - ]: 749 : ::sqlite3_busy_timeout(impl_->db, 500);
799 : :
800 [ + - ]: 749 : begin_cache();
801 : :
802 : : // create header table
803 [ + - ]: 749 : ret = ::sqlite3_exec(impl_->db,
804 : : "CREATE TABLE IF NOT EXISTS VLinkHeader(major INTEGER, minor INTEGER, patch INTEGER, "
805 : : "count INTEGER, duration INTEGER, accuracy TEXT, compress TEXT, process TEXT, date TEXT, "
806 : : "tag TEXT, complete INTEGER, timezone INTEGER, start_timestamp INTEGER);",
807 : : nullptr, nullptr, &err_msg);
808 : :
809 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
810 : : // LCOV_EXCL_START GCOVR_EXCL_START
811 : : CLOG_F("Failed to create header table: %s.", err_msg);
812 : :
813 : : free_err_msg();
814 : : close_db();
815 : :
816 : : return;
817 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
818 : : }
819 : :
820 : : // prepare header table
821 : 749 : ::sqlite3_stmt* header_stmt = nullptr;
822 [ + - ]: 749 : ret = ::sqlite3_prepare_v2(
823 : 749 : impl_->db,
824 : : "INSERT INTO VLinkHeader (major, minor, patch, count, duration, accuracy, compress, "
825 : : "process, date, tag, complete, timezone, start_timestamp) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);",
826 : : -1, &header_stmt, nullptr);
827 : :
828 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
829 : : // LCOV_EXCL_START GCOVR_EXCL_START
830 : : CLOG_F("Failed to prepare header table: %s.", ::sqlite3_errmsg(impl_->db));
831 : :
832 : : free_err_msg();
833 : : finalize_stmt(header_stmt);
834 : : close_db();
835 : :
836 : : return;
837 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
838 : : }
839 : :
840 : : // get date
841 [ + - ]: 749 : const std::string& date_str = get_format_date(&impl_->time_current);
842 : :
843 : : // insert header table
844 [ + - ]: 749 : ::sqlite3_bind_int(header_stmt, get_column(0), VLINK_VERSION_MAJOR);
845 [ + - ]: 749 : ::sqlite3_bind_int(header_stmt, get_column(1), VLINK_VERSION_MINOR);
846 [ + - ]: 749 : ::sqlite3_bind_int(header_stmt, get_column(2), VLINK_VERSION_PATCH);
847 [ + - ]: 749 : ::sqlite3_bind_int64(header_stmt, get_column(3), 0);
848 [ + - ]: 749 : ::sqlite3_bind_int64(header_stmt, get_column(4), 0);
849 [ + - ]: 749 : ::sqlite3_bind_text(header_stmt, get_column(5), "MicroSecond", -1, SQLITE_STATIC);
850 [ + + + - ]: 749 : ::sqlite3_bind_text(header_stmt, get_column(6), impl_->enable_compressed ? "lzav" : "None", -1, SQLITE_STATIC);
851 [ + - ]: 749 : ::sqlite3_bind_text(header_stmt, get_column(7), impl_->app_name.data(), impl_->app_name.size(), SQLITE_STATIC);
852 [ + - ]: 749 : ::sqlite3_bind_text(header_stmt, get_column(8), date_str.data(), date_str.size(), SQLITE_STATIC);
853 [ + - ]: 749 : ::sqlite3_bind_text(header_stmt, get_column(9), impl_->tag_name.data(), impl_->tag_name.size(), SQLITE_STATIC);
854 [ + - ]: 749 : ::sqlite3_bind_int(header_stmt, get_column(10), 0);
855 [ + - ]: 749 : ::sqlite3_bind_int(header_stmt, get_column(11), impl_->timezone_diff);
856 [ + - ]: 749 : ::sqlite3_bind_int64(header_stmt, get_column(12), impl_->start_timestamp);
857 : :
858 [ + - ]: 749 : ret = ::sqlite3_step(header_stmt);
859 : :
860 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_DONE) {
861 : : // LCOV_EXCL_START GCOVR_EXCL_START
862 : : CLOG_F("Failed to insert header table: %s.", ::sqlite3_errmsg(impl_->db));
863 : :
864 : : free_err_msg();
865 : : finalize_stmt(header_stmt);
866 : : close_db();
867 : :
868 : : return;
869 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
870 : : }
871 : :
872 : 749 : finalize_stmt(header_stmt);
873 : :
874 : : // create schema table
875 [ + - ]: 749 : ret = ::sqlite3_exec(impl_->db, "CREATE TABLE IF NOT EXISTS VLinkSchemas(ser TEXT, encoding TEXT, data BLOB);",
876 : : nullptr, nullptr, &err_msg);
877 : :
878 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
879 : : // LCOV_EXCL_START GCOVR_EXCL_START
880 : : CLOG_F("Failed to create schema table: %s.", err_msg);
881 : :
882 : : free_err_msg();
883 : : close_db();
884 : :
885 : : return;
886 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
887 : : }
888 : :
889 : : // prepare schema table
890 [ + - ]: 749 : ret = ::sqlite3_prepare_v2(impl_->db, "INSERT INTO VLinkSchemas(ser, encoding, data) VALUES (?, ?, ?);", -1,
891 : 749 : &impl_->schemas_stmt, nullptr);
892 : :
893 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
894 : : // LCOV_EXCL_START GCOVR_EXCL_START
895 : : CLOG_F("Failed to prepare schema table: %s.", ::sqlite3_errmsg(impl_->db));
896 : :
897 : : free_err_msg();
898 : : close_db();
899 : :
900 : : return;
901 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
902 : : }
903 : :
904 : : // create url table
905 [ + - ]: 749 : ret = ::sqlite3_exec(impl_->db,
906 : : "CREATE TABLE IF NOT EXISTS VLinkUrls(id INTEGER, url TEXT, type TEXT, ser TEXT, encoding TEXT, "
907 : : "count INTEGER, loss REAL, size INTEGER, freq REAL);",
908 : : nullptr, nullptr, &err_msg);
909 : :
910 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
911 : : // LCOV_EXCL_START GCOVR_EXCL_START
912 : : CLOG_F("Failed to create urls table: %s.", err_msg);
913 : :
914 : : free_err_msg();
915 : : close_db();
916 : :
917 : : return;
918 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
919 : : }
920 : :
921 : : // prepare url table
922 [ + - ]: 749 : ret = ::sqlite3_prepare_v2(impl_->db,
923 : : "INSERT INTO VLinkUrls (id, url, type, ser, encoding, count, loss, size, freq) VALUES "
924 : : "(?, ?, ?, ?, ?, ?, ?, ?, ?);",
925 : 749 : -1, &impl_->urls_stmt, nullptr);
926 : :
927 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
928 : : // LCOV_EXCL_START GCOVR_EXCL_START
929 : : CLOG_F("Failed to prepare urls table: %s.", ::sqlite3_errmsg(impl_->db));
930 : :
931 : : free_err_msg();
932 : : close_db();
933 : :
934 : : return;
935 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
936 : : }
937 : :
938 : : // create data table
939 [ + - ]: 749 : ret = ::sqlite3_exec(impl_->db,
940 : : "CREATE TABLE IF NOT EXISTS VLinkDatas(elapsed INTEGER, url INTEGER, action TEXT, data BLOB);",
941 : : nullptr, nullptr, &err_msg);
942 : :
943 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
944 : : // LCOV_EXCL_START GCOVR_EXCL_START
945 : : CLOG_F("Failed to create datas table: %s.", err_msg);
946 : :
947 : : free_err_msg();
948 : : close_db();
949 : :
950 : : return;
951 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
952 : : }
953 : :
954 : : // prepare data table
955 [ + - ]: 749 : ret = ::sqlite3_prepare_v2(impl_->db, "INSERT INTO VLinkDatas (elapsed, url, action, data) VALUES (?, ?, ?, ?);", -1,
956 : 749 : &impl_->datas_stmt, nullptr);
957 : :
958 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
959 : : // LCOV_EXCL_START GCOVR_EXCL_START
960 : : CLOG_F("Failed to prepare datas table: %s.", ::sqlite3_errmsg(impl_->db));
961 : :
962 : : free_err_msg();
963 : : close_db();
964 : :
965 : : return;
966 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
967 : : }
968 : :
969 : : // prepare update_header table
970 [ + - ]: 749 : ret = sqlite3_prepare_v2(impl_->db, "UPDATE VLinkHeader SET count=?, duration=? WHERE ROWID=1;", -1,
971 : 749 : &impl_->update_header_stmt, nullptr);
972 : :
973 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
974 : : // LCOV_EXCL_START GCOVR_EXCL_START
975 : : CLOG_F("Failed to prepare update_header: %s.", ::sqlite3_errmsg(impl_->db));
976 : :
977 : : free_err_msg();
978 : : close_db();
979 : :
980 : : return;
981 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
982 : : }
983 : :
984 : : // prepare update_url table
985 [ + - ]: 749 : ret = sqlite3_prepare_v2(impl_->db, "UPDATE VLinkUrls SET count=?, size=?, freq=? WHERE url=?;", -1,
986 : 749 : &impl_->update_urls_stmt, nullptr);
987 : :
988 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
989 : : // LCOV_EXCL_START GCOVR_EXCL_START
990 : : CLOG_F("Failed to prepare update_urls: %s.", ::sqlite3_errmsg(impl_->db));
991 : :
992 : : free_err_msg();
993 : : close_db();
994 : :
995 : : return;
996 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
997 : : }
998 : :
999 : : // create idx_elapsed_url
1000 [ + - ]: 749 : ret = ::sqlite3_exec(impl_->db, "CREATE INDEX IF NOT EXISTS idx_elapsed_url ON VLinkDatas(elapsed, url);", nullptr,
1001 : : nullptr, &err_msg);
1002 : :
1003 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
1004 : : // LCOV_EXCL_START GCOVR_EXCL_START
1005 : : CLOG_F("Failed to create datas idx_elapsed_url: %s.", err_msg);
1006 : :
1007 : : free_err_msg();
1008 : : close_db();
1009 : :
1010 : : return;
1011 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1012 : : }
1013 : :
1014 [ + - ]: 749 : ret = sqlite3_prepare_v2(impl_->db, "UPDATE VLinkUrls SET ser=?, encoding=? WHERE url=?;", -1,
1015 : 749 : &impl_->update_url_meta_stmt, nullptr);
1016 : :
1017 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
1018 : : // LCOV_EXCL_START GCOVR_EXCL_START
1019 : : CLOG_F("Failed to prepare update_url_meta: %s.", ::sqlite3_errmsg(impl_->db));
1020 : :
1021 : : free_err_msg();
1022 : : close_db();
1023 : :
1024 : : return;
1025 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1026 : : }
1027 : :
1028 [ + - ]: 749 : ret = sqlite3_prepare_v2(impl_->db, "UPDATE VLinkUrls SET loss=? WHERE url=?;", -1, &impl_->update_url_loss_stmt,
1029 : : nullptr);
1030 : :
1031 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
1032 : : // LCOV_EXCL_START GCOVR_EXCL_START
1033 : : CLOG_F("Failed to prepare update_url_loss: %s.", ::sqlite3_errmsg(impl_->db));
1034 : :
1035 : : free_err_msg();
1036 : : close_db();
1037 : :
1038 : : return;
1039 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1040 : : }
1041 : :
1042 [ + - ]: 749 : ret = sqlite3_prepare_v2(impl_->db, "UPDATE VLinkHeader SET complete=1 WHERE ROWID=1;", -1,
1043 : 749 : &impl_->update_complete_stmt, nullptr);
1044 : :
1045 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
1046 : : // LCOV_EXCL_START GCOVR_EXCL_START
1047 : : CLOG_F("Failed to prepare update_complete: %s.", ::sqlite3_errmsg(impl_->db));
1048 : :
1049 : : free_err_msg();
1050 : : close_db();
1051 : :
1052 : : return;
1053 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1054 : : }
1055 : :
1056 : 749 : free_err_msg();
1057 : :
1058 [ + - ]: 749 : sync_cache();
1059 : :
1060 : : #else
1061 : : (void)path;
1062 : : VLOG_F("VDBWriter: The compile macro VLINK_ENABLE_SQLITE is not turned on.");
1063 : : #endif
1064 : :
1065 : 749 : impl_->last_timestamp = 0;
1066 [ + - ]: 749 : }
1067 : :
1068 : 653 : bool VDBWriter::open_split(const std::string& path) {
1069 : 653 : const auto split_file_count = impl_->split_file_list.size();
1070 : : #ifdef _WIN32
1071 : : const auto file_name = std::filesystem::path(Helpers::string_to_wstring(path)).filename();
1072 : : const auto split_path = impl_->split_output_dir / file_name;
1073 : : open(Helpers::path_to_string(split_path));
1074 : : #else
1075 [ + - + - : 1306 : const auto split_path = impl_->split_output_dir / std::filesystem::path(path).filename();
+ - ]
1076 [ + - + - ]: 653 : open(split_path.string());
1077 : : #endif
1078 : :
1079 : : #ifdef VLINK_ENABLE_SQLITE
1080 [ + - ]: 653 : if VLIKELY (impl_->db) {
1081 : 653 : return true;
1082 : : }
1083 : : #endif
1084 : :
1085 [ # # ]: 0 : impl_->split_file_list.resize(split_file_count);
1086 : 0 : std::error_code remove_ec;
1087 : 0 : std::filesystem::remove(split_path, remove_ec);
1088 : 0 : return false;
1089 : 653 : }
1090 : :
1091 : 820 : void VDBWriter::close_segment() {
1092 : : #ifdef VLINK_ENABLE_SQLITE
1093 : :
1094 [ + + ]: 820 : if (!impl_->db) {
1095 : 71 : return;
1096 : : }
1097 : :
1098 : 749 : bool close_success = sync_cache();
1099 : :
1100 [ - + ]: 749 : if (!close_success) {
1101 : : if (impl_->in_cached.load(std::memory_order_relaxed)) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1102 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1103 : : }
1104 : : } else {
1105 : 749 : close_success = begin_cache();
1106 : : }
1107 : :
1108 [ + - ]: 749 : if (close_success) {
1109 : 749 : int ret = SQLITE_OK;
1110 : :
1111 [ + + ]: 1558 : for (const auto& [url, info] : impl_->url_map) {
1112 [ + - ]: 809 : ::sqlite3_bind_int64(impl_->update_urls_stmt, get_column(0), info.count);
1113 [ + - ]: 809 : ::sqlite3_bind_int64(impl_->update_urls_stmt, get_column(1), info.size);
1114 [ + - ]: 809 : ::sqlite3_bind_double(impl_->update_urls_stmt, get_column(2), info.freq);
1115 [ + - ]: 809 : ::sqlite3_bind_text(impl_->update_urls_stmt, get_column(3), url.c_str(), url.size(), SQLITE_STATIC);
1116 : :
1117 [ + - ]: 809 : ret = ::sqlite3_step(impl_->update_urls_stmt);
1118 : :
1119 [ - + ]: 809 : if VUNLIKELY (ret != SQLITE_DONE) {
1120 : : CLOG_W("Failed to update urls table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1121 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1122 : : }
1123 : :
1124 [ + - ]: 809 : ret = ::sqlite3_reset(impl_->update_urls_stmt);
1125 : :
1126 [ - + ]: 809 : if VUNLIKELY (ret != SQLITE_OK) {
1127 : : CLOG_W("Failed to reset urls table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1128 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1129 : : }
1130 : :
1131 [ - + ]: 809 : if VUNLIKELY (!close_success) {
1132 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1133 : : }
1134 : : }
1135 : :
1136 [ + - ]: 749 : if (close_success) {
1137 : 749 : ::sqlite3_bind_int64(impl_->update_header_stmt, get_column(0), impl_->current_row);
1138 : 749 : ::sqlite3_bind_int64(impl_->update_header_stmt, get_column(1), impl_->last_timestamp);
1139 : :
1140 : 749 : ret = ::sqlite3_step(impl_->update_header_stmt);
1141 : :
1142 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_DONE) {
1143 : : CLOG_W("Failed to update header table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1144 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1145 : : }
1146 : :
1147 : 749 : ret = ::sqlite3_reset(impl_->update_header_stmt);
1148 : :
1149 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
1150 : : CLOG_W("Failed to reset header table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1151 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1152 : : }
1153 : : }
1154 : :
1155 [ + - ]: 749 : if (close_success) {
1156 [ + - + - ]: 749 : std::lock_guard lock(sample_mutex());
1157 : :
1158 [ + - + + ]: 793 : for (const auto& [url, loss] : url_loss_map_ref()) {
1159 [ + - + + ]: 88 : for (const auto& recorded_url : recorded_urls_for_origin(url)) {
1160 [ + - ]: 44 : ::sqlite3_bind_double(impl_->update_url_loss_stmt, get_column(0), loss);
1161 [ + - ]: 44 : ::sqlite3_bind_text(impl_->update_url_loss_stmt, get_column(1), recorded_url.c_str(), recorded_url.size(),
1162 : : SQLITE_STATIC);
1163 : :
1164 [ + - ]: 44 : ret = ::sqlite3_step(impl_->update_url_loss_stmt);
1165 : :
1166 [ - + ]: 44 : if VUNLIKELY (ret != SQLITE_DONE) {
1167 : : CLOG_W("Failed to update url loss: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1168 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1169 : : }
1170 : :
1171 [ + - ]: 44 : ret = ::sqlite3_reset(impl_->update_url_loss_stmt);
1172 : :
1173 [ - + ]: 44 : if VUNLIKELY (ret != SQLITE_OK) {
1174 : : CLOG_W("Failed to reset url loss: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1175 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1176 : : }
1177 : :
1178 [ - + ]: 44 : if VUNLIKELY (!close_success) {
1179 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1180 : : }
1181 : 44 : }
1182 : :
1183 [ - + ]: 44 : if VUNLIKELY (!close_success) {
1184 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1185 : : }
1186 : : }
1187 : 749 : }
1188 : :
1189 [ + - ]: 749 : if (close_success) {
1190 : 749 : ret = ::sqlite3_step(impl_->update_complete_stmt);
1191 : :
1192 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_DONE) {
1193 : : CLOG_W("Failed to mark database complete: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1194 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1195 : : }
1196 : :
1197 : 749 : ret = ::sqlite3_reset(impl_->update_complete_stmt);
1198 : :
1199 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
1200 : : CLOG_W("Failed to reset complete flag: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1201 : : close_success = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1202 : : }
1203 : : }
1204 : :
1205 [ + - ]: 749 : if (close_success) {
1206 : 749 : close_success = sync_cache();
1207 : : }
1208 : : }
1209 : :
1210 [ - + ]: 749 : if VUNLIKELY (!close_success) {
1211 : 0 : set_fail();
1212 : :
1213 [ # # ]: 0 : if (impl_->in_cached.load(std::memory_order_relaxed)) {
1214 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1215 : : }
1216 : : }
1217 : :
1218 [ + - ]: 749 : if VLIKELY (impl_->datas_stmt) {
1219 : 749 : ::sqlite3_finalize(impl_->datas_stmt);
1220 : 749 : impl_->datas_stmt = nullptr;
1221 : : }
1222 : :
1223 [ + - ]: 749 : if VLIKELY (impl_->urls_stmt) {
1224 : 749 : ::sqlite3_finalize(impl_->urls_stmt);
1225 : 749 : impl_->urls_stmt = nullptr;
1226 : : }
1227 : :
1228 [ + - ]: 749 : if VLIKELY (impl_->schemas_stmt) {
1229 : 749 : ::sqlite3_finalize(impl_->schemas_stmt);
1230 : 749 : impl_->schemas_stmt = nullptr;
1231 : : }
1232 : :
1233 [ + - ]: 749 : if VLIKELY (impl_->update_urls_stmt) {
1234 : 749 : ::sqlite3_finalize(impl_->update_urls_stmt);
1235 : 749 : impl_->update_urls_stmt = nullptr;
1236 : : }
1237 : :
1238 [ + - ]: 749 : if VLIKELY (impl_->update_url_meta_stmt) {
1239 : 749 : ::sqlite3_finalize(impl_->update_url_meta_stmt);
1240 : 749 : impl_->update_url_meta_stmt = nullptr;
1241 : : }
1242 : :
1243 [ + - ]: 749 : if VLIKELY (impl_->update_url_loss_stmt) {
1244 : 749 : ::sqlite3_finalize(impl_->update_url_loss_stmt);
1245 : 749 : impl_->update_url_loss_stmt = nullptr;
1246 : : }
1247 : :
1248 [ + - ]: 749 : if VLIKELY (impl_->update_header_stmt) {
1249 : 749 : ::sqlite3_finalize(impl_->update_header_stmt);
1250 : 749 : impl_->update_header_stmt = nullptr;
1251 : : }
1252 : :
1253 [ + - ]: 749 : if VLIKELY (impl_->update_complete_stmt) {
1254 : 749 : ::sqlite3_finalize(impl_->update_complete_stmt);
1255 : 749 : impl_->update_complete_stmt = nullptr;
1256 : : }
1257 : :
1258 [ + + ]: 749 : if (impl_->config.wal_mode) {
1259 : 3 : ::sqlite3_exec(impl_->db, "PRAGMA journal_mode = OFF;", nullptr, nullptr, nullptr);
1260 : : }
1261 : :
1262 [ + + ]: 749 : if (impl_->config.optimize_on_exit) {
1263 : 20 : ::sqlite3_exec(impl_->db, "PRAGMA optimize;", nullptr, nullptr, nullptr);
1264 : : }
1265 : :
1266 [ + - ]: 749 : if VLIKELY (impl_->db) {
1267 [ + - ]: 749 : const char* err_ptr = ::sqlite3_errmsg(impl_->db);
1268 [ + - + - : 749 : std::string close_err = err_ptr ? err_ptr : std::string{};
+ - - - ]
1269 [ + - ]: 749 : int ret = ::sqlite3_close_v2(impl_->db);
1270 : :
1271 [ - + ]: 749 : if VUNLIKELY (ret != SQLITE_OK) {
1272 : : CLOG_W("Failed to close database (rc=%d): %s.", ret, close_err.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1273 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1274 : : }
1275 : :
1276 : 749 : impl_->db = nullptr;
1277 : 749 : }
1278 : :
1279 [ + + + + : 749 : if (impl_->is_split_mode.load(std::memory_order_acquire) && impl_->config.max_split_count > 0) {
+ + ]
1280 [ + - ]: 486 : impl_->split_info_list.emplace_back(Impl::SplitInfo{impl_->current_row, std::move(impl_->url_map)});
1281 : : } else {
1282 : 263 : impl_->url_map.clear();
1283 : : }
1284 : 749 : impl_->ser_map.clear();
1285 : 749 : url_loss_map_ref().clear();
1286 : :
1287 : 749 : impl_->current_row = 0;
1288 : 749 : impl_->current_size = 0;
1289 : 749 : impl_->has_oversize = false;
1290 : :
1291 : 749 : impl_->in_cached.store(false, std::memory_order_relaxed);
1292 : 749 : impl_->cached_size.store(0, std::memory_order_relaxed);
1293 : 749 : impl_->cache_snapshot.reset();
1294 : : #endif
1295 : :
1296 : 749 : impl_->last_timestamp = 0;
1297 : : }
1298 : :
1299 : 963 : bool VDBWriter::write(const std::string& url, const std::string& ser_type, SchemaType schema_type,
1300 : : ActionType action_type, const Bytes& data, int64_t microseconds_timestamp) {
1301 : : #ifdef VLINK_ENABLE_SQLITE
1302 : :
1303 [ - + ]: 963 : if VUNLIKELY (!impl_->db) {
1304 : : VLOG_W("VDBWriter: Sqlite not open."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1305 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1306 : : }
1307 : :
1308 [ + + + + : 963 : if VUNLIKELY (impl_->is_split_mode.load(std::memory_order_acquire) && !impl_->split_first) {
+ + ]
1309 : 85 : impl_->split_first = true;
1310 : :
1311 [ + - ]: 85 : std::lock_guard split_lock(impl_->split_mtx);
1312 : :
1313 [ + + + + : 139 : if (!impl_->split_before && impl_->split_callback && impl_->split_index.load(std::memory_order_relaxed) == 0) {
+ - + + ]
1314 [ + - ]: 54 : impl_->split_callback(0, impl_->split_filename);
1315 : : }
1316 : 85 : }
1317 : :
1318 : 963 : int ret = 0;
1319 : :
1320 : 963 : bool do_compress = false;
1321 : 963 : Bytes compressed_data;
1322 : 963 : bool do_split = false;
1323 : :
1324 [ + - - + ]: 963 : if VUNLIKELY (!begin_cache()) {
1325 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1326 : : }
1327 : :
1328 : : // delete data, when limit
1329 [ + + + + : 971 : while (impl_->current_row > impl_->config.max_row_count || impl_->current_size > impl_->config.max_bytes_size) {
+ + ]
1330 [ + + ]: 10 : if VUNLIKELY (!impl_->has_oversize) {
1331 [ + + ]: 8 : if (impl_->config.enable_limit) {
1332 [ + - + - : 12 : VLOG_W("VDBWriter: The number of messages has reached the upper limit, the oldest data will be deleted.");
+ - ]
1333 : : } else {
1334 [ + - + - : 4 : VLOG_W("VDBWriter: The number of messages has reached the upper limit, data after that will be ignored.");
+ - ]
1335 : : }
1336 : :
1337 : 8 : impl_->has_oversize = true;
1338 : : }
1339 : :
1340 [ + + ]: 10 : if (!impl_->config.enable_limit) {
1341 : 2 : return false;
1342 : : }
1343 : :
1344 : 8 : ::sqlite3_stmt* select_stmt = nullptr;
1345 : 8 : ::sqlite3_stmt* delete_row_stmt = nullptr;
1346 [ + - ]: 8 : ret = ::sqlite3_prepare_v2(impl_->db, "SELECT rowid, url, length(data) FROM VLinkDatas ORDER BY rowid LIMIT 1;", -1,
1347 : : &select_stmt, nullptr);
1348 : :
1349 [ + - ]: 8 : if VLIKELY (ret == SQLITE_OK) {
1350 [ + - ]: 8 : ret = ::sqlite3_step(select_stmt);
1351 : :
1352 [ + - ]: 8 : if VLIKELY (ret == SQLITE_ROW) {
1353 [ + - ]: 8 : const auto erase_row_id = ::sqlite3_column_int64(select_stmt, 0);
1354 [ + - ]: 8 : const auto erase_url_id = ::sqlite3_column_int(select_stmt, 1);
1355 [ + - ]: 8 : const auto erase_blob_size = ::sqlite3_column_int64(select_stmt, 2);
1356 : 8 : auto erase_size = static_cast<size_t>(erase_blob_size);
1357 : :
1358 [ + + + - : 8 : if (impl_->enable_compressed && erase_blob_size >= 13) {
+ + ]
1359 : 6 : std::array<uint8_t, 13> erase_data{};
1360 : 6 : ::sqlite3_blob* erase_blob = nullptr;
1361 [ + - ]: 6 : ret = ::sqlite3_blob_open(impl_->db, "main", "VLinkDatas", "data", erase_row_id, 0, &erase_blob);
1362 : :
1363 [ + - ]: 6 : if VLIKELY (ret == SQLITE_OK) {
1364 [ + - ]: 6 : ret = ::sqlite3_blob_read(erase_blob, erase_data.data(), 9, 0);
1365 : : }
1366 : :
1367 [ + - ]: 6 : if VLIKELY (ret == SQLITE_OK) {
1368 [ + - ]: 6 : ret = ::sqlite3_blob_read(erase_blob, erase_data.data() + 9, 4, static_cast<int>(erase_blob_size) - 4);
1369 : : }
1370 : :
1371 [ + - ]: 6 : if (erase_blob) {
1372 [ + - ]: 6 : const auto close_ret = ::sqlite3_blob_close(erase_blob);
1373 [ + - ]: 6 : if (ret == SQLITE_OK) {
1374 : 6 : ret = close_ret;
1375 : : }
1376 : : }
1377 : :
1378 [ - + ]: 6 : if VUNLIKELY (ret != SQLITE_OK) {
1379 [ # # # # : 0 : CLOG_W("VDBWriter: Failed to read compressed data metadata while evicting: %s.",
# # # # ]
1380 : : ::sqlite3_errmsg(impl_->db));
1381 [ # # ]: 0 : ::sqlite3_finalize(select_stmt);
1382 [ # # ]: 0 : rollback_cache();
1383 : 0 : return false;
1384 : : }
1385 : :
1386 [ + - ]: 6 : if (Bytes::is_compress_data(erase_data.data(), erase_data.size())) {
1387 : 6 : erase_size = (static_cast<size_t>(erase_data[4]) << 24U) | (static_cast<size_t>(erase_data[5]) << 16U) |
1388 : 6 : (static_cast<size_t>(erase_data[6]) << 8U) | static_cast<size_t>(erase_data[7]);
1389 : : }
1390 : : }
1391 : :
1392 : 8 : auto total_url_iter = impl_->total_url_map.end();
1393 : :
1394 [ + - + - : 8 : if VLIKELY (erase_url_id >= 0 && static_cast<size_t>(erase_url_id) < impl_->total_url_list.size()) {
+ - ]
1395 [ + - ]: 8 : total_url_iter = impl_->total_url_map.find(impl_->total_url_list[static_cast<size_t>(erase_url_id)]);
1396 : : }
1397 : :
1398 [ - + ]: 8 : if VUNLIKELY (total_url_iter == impl_->total_url_map.end()) {
1399 [ # # # # : 0 : CLOG_W("VDBWriter: URL metadata not found while evicting the oldest data.");
# # ]
1400 [ # # ]: 0 : ::sqlite3_finalize(select_stmt);
1401 [ # # ]: 0 : rollback_cache();
1402 : 0 : return false;
1403 : : }
1404 : :
1405 [ + - ]: 8 : auto url_iter = impl_->url_map.find(total_url_iter->first);
1406 : :
1407 [ - + ]: 8 : if VUNLIKELY (url_iter == impl_->url_map.end()) {
1408 [ # # # # : 0 : CLOG_W("VDBWriter: Segment URL metadata not found while evicting the oldest data.");
# # ]
1409 [ # # ]: 0 : ::sqlite3_finalize(select_stmt);
1410 [ # # ]: 0 : rollback_cache();
1411 : 0 : return false;
1412 : : }
1413 : :
1414 [ + - ]: 8 : ::sqlite3_finalize(select_stmt);
1415 : 8 : select_stmt = nullptr;
1416 : :
1417 [ + - ]: 8 : ret = ::sqlite3_prepare_v2(impl_->db, "DELETE FROM VLinkDatas WHERE rowid = ?;", -1, &delete_row_stmt, nullptr);
1418 : :
1419 [ + - ]: 8 : if VLIKELY (ret == SQLITE_OK) {
1420 [ + - ]: 8 : ::sqlite3_bind_int64(delete_row_stmt, get_column(0), erase_row_id);
1421 [ + - ]: 8 : ret = ::sqlite3_step(delete_row_stmt);
1422 : : }
1423 : :
1424 [ + - ]: 8 : if VLIKELY (ret == SQLITE_DONE) {
1425 : 8 : --impl_->current_row;
1426 : 8 : impl_->current_size -= static_cast<int64_t>(erase_size);
1427 : 8 : --impl_->total_current_row;
1428 : 8 : impl_->total_current_size -= static_cast<int64_t>(erase_size);
1429 : :
1430 : 8 : auto& url_info = url_iter->second;
1431 : 8 : auto& total_url_info = total_url_iter->second;
1432 : 8 : --url_info.count;
1433 : 8 : url_info.size -= erase_size;
1434 : 8 : --total_url_info.count;
1435 : 8 : total_url_info.size -= erase_size;
1436 : :
1437 [ + + ]: 8 : if (url_info.count == 0) {
1438 : 6 : url_info.first_timestamp = -1;
1439 : 6 : url_info.last_timestamp = -1;
1440 : 6 : url_info.freq = 0;
1441 [ + - ]: 2 : } else if (url_info.first_timestamp >= 0) {
1442 [ - + ]: 2 : if (url_info.count == 1) {
1443 : 0 : url_info.first_timestamp = url_info.last_timestamp;
1444 : 0 : url_info.freq = 0;
1445 : : } else {
1446 : 2 : ::sqlite3_stmt* timestamp_stmt = nullptr;
1447 [ + - ]: 2 : ret = ::sqlite3_prepare_v2(
1448 : 2 : impl_->db, "SELECT elapsed FROM VLinkDatas WHERE rowid > ? AND url = ? ORDER BY rowid LIMIT 1;", -1,
1449 : : ×tamp_stmt, nullptr);
1450 : :
1451 [ + - ]: 2 : if VLIKELY (ret == SQLITE_OK) {
1452 [ + - ]: 2 : ::sqlite3_bind_int64(timestamp_stmt, get_column(0), erase_row_id);
1453 [ + - ]: 2 : ::sqlite3_bind_int(timestamp_stmt, get_column(1), erase_url_id);
1454 [ + - ]: 2 : ret = ::sqlite3_step(timestamp_stmt);
1455 : : }
1456 : :
1457 [ - + ]: 2 : if VUNLIKELY (ret != SQLITE_ROW) {
1458 [ # # # # : 0 : CLOG_W("VDBWriter: Failed to refresh URL metadata after eviction: %s.", ::sqlite3_errmsg(impl_->db));
# # # # ]
1459 [ # # ]: 0 : if (timestamp_stmt) {
1460 [ # # ]: 0 : ::sqlite3_finalize(timestamp_stmt);
1461 : : }
1462 [ # # ]: 0 : ::sqlite3_finalize(delete_row_stmt);
1463 [ # # ]: 0 : rollback_cache();
1464 : 0 : return false;
1465 : : }
1466 : :
1467 [ + - ]: 2 : url_info.first_timestamp = ::sqlite3_column_int64(timestamp_stmt, 0);
1468 [ + - ]: 2 : ::sqlite3_finalize(timestamp_stmt);
1469 : :
1470 : 2 : const auto duration = (url_info.last_timestamp - url_info.first_timestamp) / 1000'000.0;
1471 [ + - ]: 2 : url_info.freq = duration > 0 ? url_info.count / duration : 0;
1472 : : }
1473 : : }
1474 : :
1475 [ + + ]: 8 : if (total_url_info.count == 0) {
1476 : 6 : total_url_info.first_timestamp = -1;
1477 : 6 : total_url_info.last_timestamp = -1;
1478 : 6 : total_url_info.freq = 0;
1479 [ + - ]: 2 : } else if (total_url_info.count == url_info.count) {
1480 : 2 : total_url_info.first_timestamp = url_info.first_timestamp;
1481 : 2 : total_url_info.last_timestamp = url_info.last_timestamp;
1482 : 2 : total_url_info.freq = url_info.freq;
1483 [ # # ]: 0 : } else if (total_url_info.first_timestamp >= 0) {
1484 [ # # ]: 0 : if (url_info.count > 0) {
1485 : 0 : total_url_info.last_timestamp = url_info.last_timestamp;
1486 : : } else {
1487 : 0 : total_url_info.last_timestamp = total_url_info.previous_segment_last_timestamp;
1488 : : }
1489 : :
1490 : 0 : const auto duration = (total_url_info.last_timestamp - total_url_info.first_timestamp) / 1000'000.0;
1491 [ # # ]: 0 : total_url_info.freq = duration > 0 ? total_url_info.count / duration : 0;
1492 : : }
1493 : : } else {
1494 : : // LCOV_EXCL_START GCOVR_EXCL_START
1495 : : CLOG_W("Failed to erase datas table: %s.", ::sqlite3_errmsg(impl_->db));
1496 : : if VLIKELY (delete_row_stmt) {
1497 : : ::sqlite3_finalize(delete_row_stmt);
1498 : : delete_row_stmt = nullptr;
1499 : : }
1500 : : if VLIKELY (select_stmt) {
1501 : : ::sqlite3_finalize(select_stmt);
1502 : : select_stmt = nullptr;
1503 : : }
1504 : : rollback_cache();
1505 : : return false;
1506 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1507 : : }
1508 : : } else {
1509 [ # # ]: 0 : if (ret == SQLITE_DONE) {
1510 [ # # # # : 0 : CLOG_W("VDBWriter: No data available while enforcing the configured limit.");
# # ]
1511 : : } else {
1512 [ # # # # : 0 : CLOG_W("VDBWriter: Failed to select data while enforcing the configured limit: %s.",
# # # # ]
1513 : : ::sqlite3_errmsg(impl_->db));
1514 : : }
1515 [ # # ]: 0 : ::sqlite3_finalize(select_stmt);
1516 [ # # ]: 0 : rollback_cache();
1517 : 0 : return false;
1518 : : }
1519 : :
1520 [ + - ]: 8 : if VLIKELY (delete_row_stmt) {
1521 [ + - ]: 8 : ::sqlite3_finalize(delete_row_stmt);
1522 : 8 : delete_row_stmt = nullptr;
1523 : : }
1524 : :
1525 [ - + ]: 8 : if VLIKELY (select_stmt) {
1526 [ # # ]: 0 : ::sqlite3_finalize(select_stmt);
1527 : 0 : select_stmt = nullptr;
1528 : : }
1529 : : } else {
1530 : : CLOG_W("Failed to erase datas table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1531 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1532 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1533 : : }
1534 : : }
1535 : :
1536 : : // split
1537 : :
1538 [ + + + + : 961 : if (impl_->is_split_mode.load(std::memory_order_acquire) && !impl_->url_map.empty()) {
+ + ]
1539 [ + + + + ]: 846 : if (impl_->config.split_by_time > 0 &&
1540 : 273 : (microseconds_timestamp - impl_->config.begin_time * 1000) >
1541 : 273 : impl_->config.split_by_time * 1000 *
1542 [ + + ]: 546 : (static_cast<int64_t>(impl_->split_index.load(std::memory_order_relaxed)) + 1)) {
1543 : 272 : do_split = true;
1544 [ + + + - : 601 : } else if (impl_->config.split_by_time <= 0 && impl_->config.split_by_size > 0 &&
+ + ]
1545 [ + + ]: 300 : (impl_->current_size + static_cast<int64_t>(data.size())) > impl_->config.split_by_size) {
1546 : 295 : do_split = true;
1547 : : } else {
1548 : : do_split = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1549 : : }
1550 : :
1551 [ + + ]: 573 : if VUNLIKELY (do_split) {
1552 [ + - ]: 567 : std::lock_guard split_lock(impl_->split_mtx);
1553 : :
1554 : 567 : impl_->split_index.fetch_add(1, std::memory_order_relaxed);
1555 [ + - ]: 567 : const auto time_current = impl_->time_start + std::chrono::milliseconds(microseconds_timestamp / 1000U);
1556 [ + - ]: 567 : const bool same_millisecond = time_current == impl_->time_current;
1557 : 567 : impl_->time_current = time_current;
1558 : :
1559 [ + + ]: 567 : if (impl_->config.split_name_by_time) {
1560 [ + - + - ]: 268 : std::string filename = get_format_date(&impl_->time_current, true) + ".vdb";
1561 : :
1562 [ + + + - : 268 : if VUNLIKELY (same_millisecond || std::find(impl_->split_file_list.begin(), impl_->split_file_list.end(),
- + + + ]
1563 : : filename) != impl_->split_file_list.end()) {
1564 [ + - ]: 132 : filename.insert(filename.find_last_of('.'),
1565 [ + - + - ]: 396 : "." + std::to_string(impl_->split_index.load(std::memory_order_relaxed) + 1));
1566 : : }
1567 : :
1568 [ + + + - : 268 : impl_->split_filename = impl_->base_dir.empty() ? filename : impl_->base_dir + "/" + filename;
+ - + - +
+ - - ]
1569 : 268 : } else {
1570 : 299 : impl_->split_filename =
1571 [ + - + - : 897 : impl_->base_name + "." + std::to_string(impl_->split_index.load(std::memory_order_relaxed) + 1) + ".vdb";
+ - + - ]
1572 : : }
1573 : :
1574 [ + + + - : 567 : if (impl_->split_before && impl_->split_callback) {
+ + ]
1575 [ + - ]: 46 : impl_->split_callback(impl_->split_index.load(std::memory_order_relaxed), impl_->split_filename);
1576 : : }
1577 : :
1578 [ + - - + ]: 567 : if VUNLIKELY (!sync_cache()) {
1579 : : impl_->split_index.fetch_sub(1, std::memory_order_relaxed); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1580 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1581 : : }
1582 : :
1583 [ + - ]: 567 : close_segment();
1584 : :
1585 [ + - - + ]: 567 : if VUNLIKELY (!open_split(impl_->split_filename)) {
1586 : 0 : set_fail();
1587 : 0 : impl_->split_index.fetch_sub(1, std::memory_order_relaxed);
1588 : :
1589 [ # # # # ]: 0 : if VUNLIKELY (!write_filex(false)) {
1590 : 0 : set_fail();
1591 : : }
1592 : :
1593 : 0 : return false;
1594 : : }
1595 : :
1596 : 7 : auto discard_new_split = [this]() {
1597 [ + - ]: 1 : close_segment();
1598 : 1 : impl_->split_file_list.pop_back();
1599 [ + - ]: 1 : if (impl_->config.max_split_count > 0) {
1600 : 1 : impl_->split_info_list.pop_back();
1601 : : }
1602 : :
1603 : : #ifdef _WIN32
1604 : : const auto new_split_path = impl_->split_output_dir /
1605 : : std::filesystem::path(Helpers::string_to_wstring(impl_->split_filename)).filename();
1606 : : #else
1607 [ + - + - : 2 : const auto new_split_path = impl_->split_output_dir / std::filesystem::path(impl_->split_filename).filename();
+ - ]
1608 : : #endif
1609 : 1 : std::error_code remove_ec;
1610 : 1 : std::filesystem::remove(new_split_path, remove_ec);
1611 : 1 : impl_->split_index.fetch_sub(1, std::memory_order_relaxed);
1612 : 1 : };
1613 : :
1614 [ + - - + ]: 567 : if VUNLIKELY (!begin_cache()) {
1615 [ # # ]: 0 : discard_new_split();
1616 : :
1617 [ # # # # ]: 0 : if VUNLIKELY (!write_filex(false)) {
1618 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1619 : : }
1620 : :
1621 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1622 : : }
1623 : :
1624 : 567 : std::string oldest_split_file;
1625 : 567 : Impl::SplitInfo oldest_split_info;
1626 : :
1627 [ + + + + ]: 1011 : if (impl_->config.max_split_count > 0 &&
1628 [ + + ]: 444 : static_cast<int64_t>(impl_->split_file_list.size()) > impl_->config.max_split_count) {
1629 : 234 : oldest_split_file = std::move(impl_->split_file_list.front());
1630 [ + - ]: 234 : impl_->split_file_list.erase(impl_->split_file_list.begin());
1631 : 234 : oldest_split_info = std::move(impl_->split_info_list.front());
1632 [ + - ]: 234 : impl_->split_info_list.erase(impl_->split_info_list.begin());
1633 : : }
1634 : :
1635 [ + - - + ]: 567 : if VUNLIKELY (!write_filex(false)) {
1636 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1637 [ # # ]: 0 : discard_new_split();
1638 : :
1639 [ # # ]: 0 : if (!oldest_split_file.empty()) {
1640 [ # # ]: 0 : impl_->split_file_list.insert(impl_->split_file_list.begin(), std::move(oldest_split_file));
1641 [ # # ]: 0 : impl_->split_info_list.insert(impl_->split_info_list.begin(), std::move(oldest_split_info));
1642 : : }
1643 : :
1644 [ # # # # ]: 0 : if VUNLIKELY (!write_filex(false)) {
1645 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1646 : : }
1647 : :
1648 : 0 : return false;
1649 : : }
1650 : :
1651 [ + + ]: 567 : if (!oldest_split_file.empty()) {
1652 : : #ifdef _WIN32
1653 : : const auto oldest_path =
1654 : : impl_->split_output_dir / std::filesystem::path(Helpers::string_to_wstring(oldest_split_file));
1655 : : #else
1656 [ + - + - ]: 234 : const auto oldest_path = impl_->split_output_dir / std::filesystem::path(oldest_split_file);
1657 : : #endif
1658 : 234 : std::error_code remove_ec;
1659 : 234 : std::filesystem::remove(oldest_path, remove_ec);
1660 : :
1661 [ + + ]: 234 : if VUNLIKELY (remove_ec) {
1662 [ + - + - : 2 : CLOG_W("VDBWriter: Failed to remove oldest split file [%s]: %s.", oldest_split_file.c_str(),
+ - + - +
- - - ]
1663 : : remove_ec.message().c_str());
1664 : 1 : set_fail();
1665 [ + - ]: 1 : discard_new_split();
1666 [ + - ]: 1 : impl_->split_file_list.insert(impl_->split_file_list.begin(), std::move(oldest_split_file));
1667 [ + - ]: 1 : impl_->split_info_list.insert(impl_->split_info_list.begin(), std::move(oldest_split_info));
1668 : :
1669 [ + - - + ]: 1 : if VUNLIKELY (!write_filex(false)) {
1670 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1671 : : }
1672 : :
1673 : 1 : return false;
1674 : : }
1675 [ + + ]: 234 : }
1676 : :
1677 [ + + + + : 566 : if (!impl_->split_before && impl_->split_callback) {
+ + ]
1678 [ + - ]: 1070 : impl_->split_callback(impl_->split_index.load(std::memory_order_relaxed), impl_->split_filename);
1679 : : }
1680 [ + + + + : 569 : }
+ + ]
1681 : : }
1682 : :
1683 [ + - ]: 960 : auto total_url_iter_ret = impl_->total_url_map.try_emplace(url, Impl::UrlMsgInfo());
1684 [ + - ]: 960 : auto url_iter_ret = impl_->url_map.try_emplace(url, Impl::UrlMsgInfo());
1685 : :
1686 : 14 : auto discard_new_url_entries = [this, &url_iter_ret, &total_url_iter_ret]() {
1687 [ + + ]: 5 : if (url_iter_ret.second) {
1688 : 1 : impl_->url_map.erase(url_iter_ret.first);
1689 : : }
1690 : :
1691 [ + + ]: 5 : if (total_url_iter_ret.second) {
1692 : 1 : impl_->total_url_map.erase(total_url_iter_ret.first);
1693 : : }
1694 : 965 : };
1695 : :
1696 : 960 : Impl::UrlMsgInfo& total_url_msg_info = total_url_iter_ret.first->second;
1697 : 960 : Impl::UrlMsgInfo& url_msg_info = url_iter_ret.first->second;
1698 : 960 : auto resolved_schema_type = SchemaData::resolve_type(schema_type, ser_type);
1699 : :
1700 [ + + ]: 960 : const std::string& next_ser_type = total_url_msg_info.ser_type.empty() ? ser_type : total_url_msg_info.ser_type;
1701 : 960 : SchemaType next_schema_type = total_url_msg_info.schema_type;
1702 : :
1703 [ + + ]: 960 : if (total_url_iter_ret.second) {
1704 : 246 : next_schema_type = resolved_schema_type;
1705 [ + + + + : 714 : } else if VUNLIKELY (!ser_type.empty() && next_ser_type != ser_type) {
+ + ]
1706 [ + - + - : 6 : CLOG_E("VDBWriter: URL [%s] ser changed from [%s] to [%s].", url.c_str(), next_ser_type.c_str(), ser_type.c_str());
+ - ]
1707 [ + - ]: 3 : discard_new_url_entries();
1708 : 3 : return false;
1709 : : }
1710 : :
1711 [ + + ]: 957 : if (!next_ser_type.empty()) {
1712 : 845 : std::string split_ser_type;
1713 [ + + ]: 845 : const auto schema_ser_source = ser_type.empty() ? std::string_view{next_ser_type} : std::string_view{ser_type};
1714 : 845 : SchemaType schema_storage_type = SchemaData::resolve_type(schema_type, schema_ser_source);
1715 : 845 : bool has_split_method_schema = false;
1716 : :
1717 [ + + + + ]: 812 : if ((action_type == ActionType::kClientRequest || action_type == ActionType::kClientResponse ||
1718 [ + + + + : 1691 : action_type == ActionType::kServerRequest || action_type == ActionType::kServerResponse) &&
+ + ]
1719 [ + - ]: 37 : !schema_ser_source.empty()) {
1720 : 37 : const auto split_pos = schema_ser_source.find('|');
1721 : :
1722 [ + - ]: 37 : if (split_pos != std::string_view::npos) {
1723 [ + - ]: 37 : auto payload_ser_type = schema_ser_source.substr(0, split_pos);
1724 : :
1725 [ + + + + ]: 37 : if (action_type == ActionType::kClientResponse || action_type == ActionType::kServerResponse) {
1726 [ + - ]: 3 : payload_ser_type = schema_ser_source.substr(split_pos + 1);
1727 : : }
1728 : :
1729 [ + - ]: 37 : if (!payload_ser_type.empty()) {
1730 [ + - ]: 37 : split_ser_type.assign(payload_ser_type.begin(), payload_ser_type.end());
1731 : 37 : schema_storage_type = SchemaData::resolve_type(schema_type, payload_ser_type);
1732 : 37 : has_split_method_schema = true;
1733 : : }
1734 : : }
1735 : : }
1736 : :
1737 [ + + ]: 845 : const std::string& schema_ser_type = has_split_method_schema ? split_ser_type : next_ser_type;
1738 : 845 : SchemaData schema_data;
1739 : :
1740 [ + - + + ]: 845 : if VUNLIKELY (!load_schema(schema_ser_type, schema_storage_type, schema_data)) {
1741 [ + - ]: 2 : discard_new_url_entries();
1742 : 2 : return false;
1743 : : }
1744 : :
1745 : 843 : schema_storage_type = SchemaData::resolve_type(schema_storage_type, schema_ser_type, schema_data.encoding);
1746 : :
1747 [ + + ]: 843 : if (has_split_method_schema) {
1748 [ + + ]: 37 : if (schema_storage_type != SchemaType::kUnknown) {
1749 [ - + ]: 6 : if (next_schema_type == SchemaType::kUnknown) {
1750 : : next_schema_type = schema_storage_type; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1751 [ - + ]: 6 : } else if (next_schema_type != schema_storage_type) {
1752 : : next_schema_type = SchemaType::kUnknown; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1753 : : }
1754 : : }
1755 : : } else {
1756 [ + + ]: 806 : if (resolved_schema_type == SchemaType::kUnknown) {
1757 : : const auto inferred_schema_type =
1758 : 2 : SchemaData::resolve_type(schema_data.schema_type, schema_data.name, schema_data.encoding);
1759 : :
1760 [ + - ]: 2 : if (inferred_schema_type != SchemaType::kUnknown) {
1761 : 2 : resolved_schema_type = inferred_schema_type;
1762 : : }
1763 : : }
1764 : :
1765 [ + - ]: 806 : if (resolved_schema_type != SchemaType::kUnknown) {
1766 [ + + ]: 806 : if (next_schema_type == SchemaType::kUnknown) {
1767 : 2 : next_schema_type = resolved_schema_type;
1768 [ - + ]: 804 : } else if VUNLIKELY (next_schema_type != resolved_schema_type) {
1769 : : // LCOV_EXCL_START GCOVR_EXCL_START
1770 : : CLOG_E("VDBWriter: URL [%s] schema changed from [%d] to [%d].", url.c_str(),
1771 : : static_cast<int>(next_schema_type), static_cast<int>(resolved_schema_type));
1772 : : discard_new_url_entries();
1773 : : return false;
1774 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1775 : : }
1776 : : }
1777 : : }
1778 : :
1779 [ + - ]: 843 : std::string schema_key = schema_ser_type;
1780 [ + - ]: 843 : schema_key.push_back('\x1F');
1781 [ + - ]: 843 : schema_key.append(SchemaData::convert_type(schema_storage_type));
1782 : :
1783 [ + - + + ]: 843 : if (impl_->ser_map.find(schema_key) == impl_->ser_map.end()) {
1784 : : const bool has_schema_blob =
1785 [ + + + - : 803 : !schema_data.name.empty() && !schema_data.encoding.empty() && !schema_data.data.empty();
+ - ]
1786 : :
1787 [ + + ]: 803 : if (has_schema_blob) {
1788 [ + - - + ]: 8 : if VUNLIKELY (!insert_schema(schema_data)) {
1789 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1790 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1791 : : }
1792 : :
1793 [ + - ]: 8 : impl_->ser_map.emplace(schema_key);
1794 : : }
1795 : : }
1796 [ + - + + : 847 : }
+ + ]
1797 : :
1798 [ + + ]: 955 : if (total_url_iter_ret.second) {
1799 : 245 : total_url_msg_info.index = impl_->total_url_map.size() - 1;
1800 [ + - ]: 245 : impl_->total_url_list.emplace_back(url);
1801 : : }
1802 : :
1803 [ + + ]: 955 : if (url_iter_ret.second) {
1804 : : // insert url
1805 : 809 : url_msg_info.index = impl_->url_map.size() - 1;
1806 : :
1807 [ + + ]: 809 : if (!total_url_iter_ret.second) {
1808 : 564 : total_url_msg_info.previous_segment_last_timestamp = total_url_msg_info.last_timestamp;
1809 : : }
1810 : :
1811 [ + + + - : 809 : if (action_type == ActionType::kClientRequest || action_type == ActionType::kClientResponse ||
+ - ]
1812 [ - + ]: 776 : action_type == ActionType::kServerRequest ||
1813 : : action_type == ActionType::kServerResponse) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1814 [ + - ]: 33 : impl_->write_url_type = "Method";
1815 [ + + + + ]: 776 : } else if (action_type == ActionType::kPublish || action_type == ActionType::kSubscribe) {
1816 [ + - ]: 738 : impl_->write_url_type = "Event";
1817 [ + + + + ]: 38 : } else if (action_type == ActionType::kSet || action_type == ActionType::kGet) {
1818 [ + - ]: 37 : impl_->write_url_type = "Field";
1819 : : } else {
1820 [ + - ]: 1 : impl_->write_url_type = "Unknown";
1821 : : }
1822 : :
1823 [ + - ]: 809 : ::sqlite3_bind_int(impl_->urls_stmt, get_column(0), total_url_msg_info.index);
1824 [ + - ]: 809 : ::sqlite3_bind_text(impl_->urls_stmt, get_column(1), url.c_str(), url.size(), SQLITE_STATIC);
1825 [ + - ]: 809 : ::sqlite3_bind_text(impl_->urls_stmt, get_column(2), impl_->write_url_type.c_str(), impl_->write_url_type.size(),
1826 : : SQLITE_STATIC);
1827 [ + - ]: 809 : ::sqlite3_bind_text(impl_->urls_stmt, get_column(3), next_ser_type.c_str(), next_ser_type.size(), SQLITE_STATIC);
1828 [ + - ]: 809 : const std::string next_encoding(SchemaData::convert_type(next_schema_type));
1829 [ + - ]: 809 : ::sqlite3_bind_text(impl_->urls_stmt, get_column(4), next_encoding.c_str(), next_encoding.size(), SQLITE_STATIC);
1830 [ + - ]: 809 : ::sqlite3_bind_int64(impl_->urls_stmt, get_column(5), 0);
1831 [ + - ]: 809 : ::sqlite3_bind_double(impl_->urls_stmt, get_column(6), 0);
1832 [ + - ]: 809 : ::sqlite3_bind_int64(impl_->urls_stmt, get_column(7), 0);
1833 [ + - ]: 809 : ::sqlite3_bind_double(impl_->urls_stmt, get_column(8), 0);
1834 : :
1835 [ + - ]: 809 : ret = ::sqlite3_step(impl_->urls_stmt);
1836 : :
1837 [ - + ]: 809 : if VUNLIKELY (ret != SQLITE_DONE) {
1838 : : CLOG_W("Failed to insert urls table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1839 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1840 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1841 : : }
1842 : :
1843 [ + - ]: 809 : ret = ::sqlite3_reset(impl_->urls_stmt);
1844 : :
1845 [ - + ]: 809 : if VUNLIKELY (ret != SQLITE_OK) {
1846 : : CLOG_W("Failed to reset urls table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1847 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1848 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1849 : : }
1850 : :
1851 [ + - ]: 809 : url_msg_info.url_type = impl_->write_url_type;
1852 [ + - ]: 809 : total_url_msg_info.url_type = impl_->write_url_type;
1853 : :
1854 [ + + ]: 809 : if (!next_ser_type.empty()) {
1855 [ + - ]: 798 : url_msg_info.ser_type = next_ser_type;
1856 [ + - ]: 798 : total_url_msg_info.ser_type = next_ser_type;
1857 : : }
1858 : :
1859 : 809 : url_msg_info.schema_type = next_schema_type;
1860 : 809 : total_url_msg_info.schema_type = next_schema_type;
1861 [ + - + + : 955 : } else if (total_url_msg_info.ser_type != next_ser_type || total_url_msg_info.schema_type != next_schema_type) {
- + + + ]
1862 : 1 : const bool ser_changed = total_url_msg_info.ser_type != next_ser_type;
1863 : 1 : const bool schema_changed = total_url_msg_info.schema_type != next_schema_type;
1864 : :
1865 [ - + - - ]: 1 : if (ser_changed || schema_changed) {
1866 [ + - ]: 1 : ::sqlite3_bind_text(impl_->update_url_meta_stmt, get_column(0), next_ser_type.c_str(), next_ser_type.size(),
1867 : : SQLITE_STATIC);
1868 [ + - ]: 1 : const std::string next_encoding(SchemaData::convert_type(next_schema_type));
1869 [ + - ]: 1 : ::sqlite3_bind_text(impl_->update_url_meta_stmt, get_column(1), next_encoding.c_str(), next_encoding.size(),
1870 : : SQLITE_STATIC);
1871 [ + - ]: 1 : ::sqlite3_bind_text(impl_->update_url_meta_stmt, get_column(2), url.c_str(), url.size(), SQLITE_STATIC);
1872 : :
1873 [ + - ]: 1 : ret = ::sqlite3_step(impl_->update_url_meta_stmt);
1874 : :
1875 [ - + ]: 1 : if VUNLIKELY (ret != SQLITE_DONE) {
1876 : : // LCOV_EXCL_START GCOVR_EXCL_START
1877 : : CLOG_W("Failed to update urls metadata table: %s.", ::sqlite3_errmsg(impl_->db));
1878 : : rollback_cache();
1879 : : return false;
1880 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1881 : : }
1882 : :
1883 [ + - ]: 1 : ret = ::sqlite3_reset(impl_->update_url_meta_stmt);
1884 : :
1885 [ - + ]: 1 : if VUNLIKELY (ret != SQLITE_OK) {
1886 : : // LCOV_EXCL_START GCOVR_EXCL_START
1887 : : CLOG_W("Failed to reset urls metadata table: %s.", ::sqlite3_errmsg(impl_->db));
1888 : : rollback_cache();
1889 : : return false;
1890 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1891 : : }
1892 : :
1893 [ + - ]: 1 : if (ser_changed) {
1894 [ + - ]: 1 : total_url_msg_info.ser_type = next_ser_type;
1895 [ + - ]: 1 : url_msg_info.ser_type = next_ser_type;
1896 : : }
1897 : :
1898 : 1 : total_url_msg_info.schema_type = next_schema_type;
1899 : 1 : url_msg_info.schema_type = next_schema_type;
1900 [ + - ]: 1 : }
1901 : : }
1902 : :
1903 : : // update info
1904 : 955 : ++url_msg_info.count;
1905 : 955 : ++total_url_msg_info.count;
1906 : :
1907 : 955 : impl_->cached_size.fetch_add(data.size(), std::memory_order_relaxed);
1908 : :
1909 : 955 : ++impl_->current_row;
1910 : 955 : ++impl_->total_current_row;
1911 : 955 : impl_->total_timestamp = std::max(impl_->total_timestamp, microseconds_timestamp);
1912 : :
1913 [ + + + + : 955 : if (action_type == ActionType::kPublish || action_type == ActionType::kSubscribe || action_type == ActionType::kSet ||
+ + + + ]
1914 : : action_type == ActionType::kGet) {
1915 : 917 : double time_duration = 0;
1916 : :
1917 [ + + ]: 917 : if (total_url_msg_info.first_timestamp < 0) {
1918 : 217 : total_url_msg_info.first_timestamp = microseconds_timestamp;
1919 : : }
1920 : :
1921 : 917 : total_url_msg_info.last_timestamp = microseconds_timestamp;
1922 : 917 : time_duration = (total_url_msg_info.last_timestamp - total_url_msg_info.first_timestamp) / 1000'000.0;
1923 : :
1924 [ + + ]: 917 : if (time_duration > 0) {
1925 : 698 : total_url_msg_info.freq = total_url_msg_info.count / time_duration;
1926 : : } else {
1927 : 219 : total_url_msg_info.freq = 0;
1928 : : }
1929 : :
1930 [ + + ]: 917 : if (url_msg_info.first_timestamp < 0) {
1931 : 781 : url_msg_info.first_timestamp = microseconds_timestamp;
1932 : : }
1933 : :
1934 : 917 : url_msg_info.last_timestamp = microseconds_timestamp;
1935 : 917 : time_duration = (url_msg_info.last_timestamp - url_msg_info.first_timestamp) / 1000'000.0;
1936 : :
1937 [ + + ]: 917 : if (time_duration > 0) {
1938 : 135 : url_msg_info.freq = url_msg_info.count / time_duration;
1939 : : } else {
1940 : 782 : url_msg_info.freq = 0;
1941 : : }
1942 : : }
1943 : :
1944 : : // insert data
1945 [ + - ]: 955 : const auto write_action_type = convert_action(action_type);
1946 : :
1947 : 955 : impl_->last_timestamp = std::max(impl_->last_timestamp, microseconds_timestamp);
1948 : :
1949 [ + - ]: 955 : ::sqlite3_bind_int64(impl_->datas_stmt, get_column(0), microseconds_timestamp);
1950 [ + - ]: 955 : ::sqlite3_bind_int(impl_->datas_stmt, get_column(1), total_url_msg_info.index);
1951 [ + - ]: 955 : ::sqlite3_bind_text(impl_->datas_stmt, get_column(2), write_action_type.data(), write_action_type.size(),
1952 : : SQLITE_STATIC);
1953 : :
1954 : : // check compress
1955 : 955 : do_compress = false;
1956 : :
1957 [ + + + - : 982 : if (impl_->enable_compressed && static_cast<int64_t>(data.size()) >= impl_->config.compress_start_size &&
+ + ]
1958 [ + - + + ]: 27 : impl_->config.ignore_compress_urls.count(url) == 0) {
1959 [ + - ]: 18 : auto& compress_ignore_cnt = impl_->compress_ignore_map[url];
1960 : :
1961 [ + + ]: 18 : if (compress_ignore_cnt < kCompressMaxIgnoreCnt) {
1962 : 17 : compressed_data = Bytes::compress_data(data.data(), data.size(), impl_->config.compress_level > 3);
1963 : :
1964 [ + - + + : 17 : if (!compressed_data.empty() && compressed_data.size() < data.size() * kCompressMaxRatio) {
+ + ]
1965 : 12 : do_compress = true;
1966 : 12 : compress_ignore_cnt = 0;
1967 : : } else {
1968 : 5 : ++compress_ignore_cnt;
1969 : : }
1970 : : }
1971 : : }
1972 : :
1973 [ + + ]: 955 : if (do_compress) {
1974 [ + - ]: 12 : ::sqlite3_bind_blob(impl_->datas_stmt, get_column(3), compressed_data.data(), compressed_data.size(),
1975 : : SQLITE_STATIC);
1976 : : } else {
1977 [ + - ]: 943 : ::sqlite3_bind_blob(impl_->datas_stmt, get_column(3), data.data(), data.size(), SQLITE_STATIC);
1978 : : }
1979 : :
1980 : 955 : auto accounted_size = data.size();
1981 : :
1982 [ + + + + : 955 : if VUNLIKELY (!do_compress && impl_->enable_compressed && Bytes::is_compress_data(data.data(), data.size())) {
+ + + + +
+ ]
1983 : 6 : accounted_size = (static_cast<size_t>(data[4]) << 24U) | (static_cast<size_t>(data[5]) << 16U) |
1984 : 6 : (static_cast<size_t>(data[6]) << 8U) | static_cast<size_t>(data[7]);
1985 : : }
1986 : :
1987 : 955 : url_msg_info.size += accounted_size;
1988 : 955 : total_url_msg_info.size += accounted_size;
1989 : 955 : impl_->current_size += static_cast<int64_t>(accounted_size);
1990 : 955 : impl_->total_current_size += static_cast<int64_t>(accounted_size);
1991 : :
1992 [ + - ]: 955 : ret = ::sqlite3_step(impl_->datas_stmt);
1993 : :
1994 [ - + ]: 955 : if VUNLIKELY (ret != SQLITE_DONE) {
1995 : : CLOG_W("Failed to insert datas table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1996 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1997 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1998 : : }
1999 : :
2000 [ + - ]: 955 : ret = ::sqlite3_reset(impl_->datas_stmt);
2001 : :
2002 [ - + ]: 955 : if VUNLIKELY (ret != SQLITE_OK) {
2003 : : CLOG_W("Failed to reset datas table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2004 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2005 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2006 : : }
2007 : :
2008 : : // update on wal_mode
2009 : :
2010 [ + + ]: 955 : if (impl_->sync_timer.get() > kSyncWriteInterval) {
2011 : 1 : impl_->sync_timer.restart();
2012 : :
2013 [ + - ]: 1 : ::sqlite3_bind_int64(impl_->update_urls_stmt, get_column(0), url_msg_info.count);
2014 [ + - ]: 1 : ::sqlite3_bind_int64(impl_->update_urls_stmt, get_column(1), url_msg_info.size);
2015 [ + - ]: 1 : ::sqlite3_bind_double(impl_->update_urls_stmt, get_column(2), url_msg_info.freq);
2016 [ + - ]: 1 : ::sqlite3_bind_text(impl_->update_urls_stmt, get_column(3), url.c_str(), url.size(), SQLITE_STATIC);
2017 [ + - ]: 1 : ret = ::sqlite3_step(impl_->update_urls_stmt);
2018 : :
2019 [ - + ]: 1 : if VUNLIKELY (ret != SQLITE_DONE) {
2020 : : CLOG_W("Failed to update urls table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2021 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2022 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2023 : : }
2024 : :
2025 [ + - ]: 1 : ret = ::sqlite3_reset(impl_->update_urls_stmt);
2026 : :
2027 [ - + ]: 1 : if VUNLIKELY (ret != SQLITE_OK) {
2028 : : CLOG_W("Failed to reset urls table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2029 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2030 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2031 : : }
2032 : :
2033 [ + - ]: 1 : ::sqlite3_bind_int64(impl_->update_header_stmt, get_column(0), impl_->current_row);
2034 [ + - ]: 1 : ::sqlite3_bind_int64(impl_->update_header_stmt, get_column(1), impl_->last_timestamp);
2035 [ + - ]: 1 : ret = ::sqlite3_step(impl_->update_header_stmt);
2036 : :
2037 [ - + ]: 1 : if VUNLIKELY (ret != SQLITE_DONE) {
2038 : : CLOG_W("Failed to update header table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2039 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2040 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2041 : : }
2042 : :
2043 [ + - ]: 1 : ret = ::sqlite3_reset(impl_->update_header_stmt);
2044 : :
2045 [ - + ]: 1 : if VUNLIKELY (ret != SQLITE_OK) {
2046 : : CLOG_W("Failed to reset header table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2047 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2048 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2049 : : }
2050 : : }
2051 : :
2052 [ + + ]: 1910 : if VUNLIKELY (impl_->cached_size.load(std::memory_order_relaxed) > impl_->config.cache_size) {
2053 [ + - - + ]: 623 : if VUNLIKELY (!sync_cache()) {
2054 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2055 : : }
2056 : : }
2057 : :
2058 : 955 : return true;
2059 : : #else
2060 : : (void)url;
2061 : : (void)ser_type;
2062 : : (void)schema_type;
2063 : : (void)action_type;
2064 : : (void)data;
2065 : : (void)microseconds_timestamp;
2066 : : return false;
2067 : : #endif
2068 : 963 : }
2069 : :
2070 : 791 : bool VDBWriter::write_filex(bool complete) {
2071 : : #ifdef _WIN32
2072 : : std::filesystem::path file_path(Helpers::string_to_wstring(impl_->path));
2073 : : #else
2074 [ + - ]: 791 : std::filesystem::path file_path(impl_->path);
2075 : : #endif
2076 : :
2077 : : try {
2078 : 791 : nlohmann::ordered_json json;
2079 : :
2080 : 791 : auto message_count = impl_->total_current_row;
2081 [ + + ]: 791 : if (impl_->config.max_split_count > 0) {
2082 : 569 : message_count = impl_->current_row;
2083 : :
2084 [ + + ]: 2459 : for (const auto& split_info : impl_->split_info_list) {
2085 : 1890 : message_count += split_info.row;
2086 : : }
2087 : : }
2088 : :
2089 [ + - ]: 791 : json["VLinkHeader"] = {
2090 : : {"major", VLINK_VERSION_MAJOR}, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2091 : : {"minor", VLINK_VERSION_MINOR}, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2092 : : {"patch", VLINK_VERSION_PATCH}, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2093 : : {"count", message_count},
2094 : 791 : {"duration", impl_->total_timestamp},
2095 : : {"accuracy", "MicroSecond"},
2096 : 791 : {"compress", impl_->enable_compressed ? "lzav" : "None"},
2097 : 791 : {"process", impl_->app_name},
2098 [ + - ]: 791 : {"date", get_format_date(&impl_->time_start)},
2099 : 791 : {"tag", impl_->tag_name},
2100 [ + + ]: 791 : {"split_by_size", impl_->config.split_by_time > 0 ? 0 : impl_->config.split_by_size},
2101 : 791 : {"split_by_time", impl_->config.split_by_time},
2102 : : {"complete", complete},
2103 : 791 : {"timezone", impl_->timezone_diff},
2104 : 791 : {"start_timestamp", impl_->start_timestamp},
2105 [ + - + - : 44296 : };
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
+ + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ + + + +
+ + + + +
+ + + + +
+ + + + +
+ + + + +
+ + + + +
+ + - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - ]
2106 : :
2107 : 791 : nlohmann::ordered_json url_json;
2108 : :
2109 : : {
2110 [ + - + - ]: 791 : std::lock_guard lock(sample_mutex());
2111 : :
2112 [ + + ]: 1498 : for (const auto& url : impl_->total_url_list) {
2113 [ + - ]: 707 : const auto& ext_info = impl_->total_url_map[url];
2114 : 707 : auto count = ext_info.count;
2115 : 707 : auto size = ext_info.size;
2116 : 707 : auto freq = ext_info.freq;
2117 : :
2118 [ + + ]: 707 : if (impl_->config.max_split_count > 0) {
2119 : 527 : count = 0;
2120 : 527 : size = 0;
2121 : 527 : int64_t first_timestamp = -1;
2122 : 527 : int64_t last_timestamp = -1;
2123 : :
2124 : 2417 : auto merge_url_info = [&](const auto& url_map) {
2125 [ + - ]: 2417 : const auto iter = url_map.find(url);
2126 : :
2127 [ + + ]: 2417 : if (iter == url_map.end()) {
2128 : 527 : return;
2129 : : }
2130 : :
2131 : 1890 : const auto& url_info = iter->second;
2132 : 1890 : count += url_info.count;
2133 : 1890 : size += url_info.size;
2134 : :
2135 [ + - + + : 1890 : if (url_info.first_timestamp >= 0 && (first_timestamp < 0 || url_info.first_timestamp < first_timestamp)) {
- + ]
2136 : 439 : first_timestamp = url_info.first_timestamp;
2137 : : }
2138 : :
2139 : 1890 : last_timestamp = std::max(last_timestamp, url_info.last_timestamp);
2140 : 527 : };
2141 : :
2142 [ + + ]: 2417 : for (const auto& split_info : impl_->split_info_list) {
2143 [ + - ]: 1890 : merge_url_info(split_info.url_map);
2144 : : }
2145 : :
2146 [ + - ]: 527 : merge_url_info(impl_->url_map);
2147 : :
2148 [ + + ]: 527 : if (count == 0) {
2149 : 88 : continue;
2150 : : }
2151 : :
2152 : 439 : const auto duration = (last_timestamp - first_timestamp) / 1000'000.0;
2153 [ + + ]: 439 : freq = duration > 0 ? count / duration : 0;
2154 : : }
2155 : :
2156 [ + - + - : 619 : auto loss = total_url_loss_map_ref()[recover_recorded_url(url)];
+ - ]
2157 : :
2158 [ + - + - : 20427 : url_json.push_back({
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- + - + +
+ + + + +
+ + + + +
+ + + + +
+ + + - -
- - - - -
- - - - -
- - - - -
- - - ]
2159 : : // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2160 : 619 : {"index", ext_info.index},
2161 : : {"url", url},
2162 [ + - ]: 619 : {"type", ext_info.url_type},
2163 [ + - ]: 619 : {"ser", ext_info.ser_type},
2164 [ + - ]: 1238 : {"encoding", std::string(SchemaData::convert_type(ext_info.schema_type))},
2165 : : {"count", count},
2166 : : {"size", size},
2167 : : {"loss", loss},
2168 : : {"freq", freq},
2169 : : });
2170 : : }
2171 : 791 : }
2172 : :
2173 [ + - ]: 791 : json["VLinkUrls"] = std::move(url_json);
2174 : :
2175 : 791 : nlohmann::ordered_json files_json;
2176 : 791 : auto split_file_count = impl_->split_file_list.size();
2177 : :
2178 : : #ifdef VLINK_ENABLE_SQLITE
2179 [ + + + + : 791 : if (!complete && impl_->db && split_file_count > 0) {
+ - + + ]
2180 : 567 : --split_file_count;
2181 : : }
2182 : : #endif
2183 : :
2184 [ + + ]: 3517 : for (size_t i = 0; i < split_file_count; ++i) {
2185 [ + - + - ]: 2726 : files_json.push_back(impl_->split_file_list[i]);
2186 : : }
2187 : :
2188 [ + - ]: 791 : json["VLinkFiles"] = std::move(files_json);
2189 : :
2190 [ + - ]: 791 : std::ofstream filex(file_path);
2191 : :
2192 [ + - - + ]: 791 : if VUNLIKELY (!filex.is_open()) {
2193 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2194 : : }
2195 : :
2196 [ + - + - ]: 791 : filex << json.dump(4);
2197 [ + - ]: 791 : filex.close();
2198 : :
2199 [ + - - + ]: 791 : if VUNLIKELY (!filex) {
2200 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2201 : : }
2202 [ + - + - : 791 : } catch (const nlohmann::json::exception& e) {
+ - + - -
- ]
2203 : : CLOG_W("VDBWriter: JSON error during config export: %s.", e.what()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2204 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2205 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2206 : :
2207 : 791 : return true;
2208 : 791 : }
2209 : :
2210 : 3028 : bool VDBWriter::begin_cache() {
2211 : : #ifdef VLINK_ENABLE_SQLITE
2212 : :
2213 [ - + ]: 3028 : if VUNLIKELY (!impl_->db) {
2214 : : VLOG_W("VDBWriter: Sqlite not open."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2215 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2216 : : }
2217 : :
2218 [ + + ]: 3028 : if VLIKELY (impl_->in_cached.load(std::memory_order_relaxed)) {
2219 : 241 : return true;
2220 : : }
2221 : :
2222 : 2787 : int ret = ::sqlite3_exec(impl_->db, "BEGIN;", nullptr, nullptr, nullptr);
2223 : :
2224 [ - + ]: 2787 : if VUNLIKELY (ret != SQLITE_OK) {
2225 : : CLOG_W("Failed to begin transaction: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2226 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2227 : : }
2228 : :
2229 : 2787 : impl_->cache_snapshot = std::make_unique<Impl::WriteStateSnapshot>(*impl_);
2230 : 2787 : impl_->in_cached.store(true, std::memory_order_relaxed);
2231 : 2787 : impl_->cached_size.store(0, std::memory_order_relaxed);
2232 : :
2233 : 2787 : impl_->cache_timer.restart();
2234 : :
2235 : 2787 : return true;
2236 : : #else
2237 : : return false;
2238 : : #endif
2239 : : }
2240 : :
2241 : 3437 : bool VDBWriter::sync_cache() {
2242 : : #ifdef VLINK_ENABLE_SQLITE
2243 : :
2244 [ - + ]: 3437 : if VUNLIKELY (!impl_->db) {
2245 : : VLOG_W("VDBWriter: Sqlite not open."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2246 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2247 : : }
2248 : :
2249 [ + + ]: 3437 : if (!impl_->in_cached.load(std::memory_order_relaxed)) {
2250 : 650 : return true;
2251 : : }
2252 : :
2253 : 2787 : int ret = ::sqlite3_exec(impl_->db, "COMMIT;", nullptr, nullptr, nullptr);
2254 : :
2255 [ - + ]: 2787 : if VUNLIKELY (ret != SQLITE_OK) {
2256 : : CLOG_W("Failed to commit transaction: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2257 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2258 : : }
2259 : :
2260 : 2787 : impl_->in_cached.store(false, std::memory_order_relaxed);
2261 : 2787 : impl_->cached_size.store(0, std::memory_order_relaxed);
2262 : 2787 : impl_->cache_snapshot.reset();
2263 : :
2264 : 2787 : impl_->cache_timer.stop();
2265 : :
2266 : 2787 : return true;
2267 : : #else
2268 : : return false;
2269 : : #endif
2270 : : }
2271 : :
2272 : 0 : bool VDBWriter::rollback_cache() {
2273 : : #ifdef VLINK_ENABLE_SQLITE
2274 : :
2275 [ # # ]: 0 : if VUNLIKELY (!impl_->db) {
2276 : : VLOG_W("VDBWriter: Sqlite not open."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2277 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2278 : : }
2279 : :
2280 : 0 : ::sqlite3_exec(impl_->db, "ROLLBACK;", nullptr, nullptr, nullptr);
2281 : :
2282 [ # # ]: 0 : if (impl_->cache_snapshot) {
2283 : 0 : impl_->cache_snapshot->restore(*impl_);
2284 : 0 : impl_->cache_snapshot.reset();
2285 : : }
2286 : :
2287 : 0 : impl_->in_cached.store(false, std::memory_order_relaxed);
2288 : 0 : impl_->cached_size.store(0, std::memory_order_relaxed);
2289 : :
2290 : 0 : impl_->cache_timer.stop();
2291 : :
2292 : 0 : return true;
2293 : : #else
2294 : : return false;
2295 : : #endif
2296 : : }
2297 : :
2298 : 46 : bool VDBWriter::merge_schema(SchemaData& schema_data) {
2299 : : const auto resolved_schema_type =
2300 : 46 : SchemaData::resolve_type(schema_data.schema_type, schema_data.name, schema_data.encoding);
2301 : 46 : schema_data.schema_type = resolved_schema_type;
2302 : :
2303 [ + + ]: 46 : if VUNLIKELY (schema_data.name.empty()) {
2304 : 1 : return true;
2305 : : }
2306 : :
2307 [ + + + + : 45 : if (schema_data.encoding.empty() && SchemaData::is_real_type(resolved_schema_type)) {
+ + ]
2308 [ + - ]: 3 : schema_data.encoding = std::string(SchemaData::convert_type(resolved_schema_type));
2309 : : }
2310 : :
2311 [ + - ]: 45 : std::string schema_key = schema_data.name;
2312 [ + - ]: 45 : schema_key.push_back('\x1F');
2313 [ + - ]: 45 : schema_key.append(SchemaData::convert_type(resolved_schema_type));
2314 : :
2315 : 45 : std::string unknown_schema_key;
2316 [ + - ]: 45 : auto schema_iter = impl_->total_schema_map.find(schema_key);
2317 : :
2318 [ + + + + : 45 : if (schema_iter == impl_->total_schema_map.end() && SchemaData::is_real_type(resolved_schema_type)) {
+ + ]
2319 [ + - ]: 39 : unknown_schema_key = schema_data.name;
2320 [ + - ]: 39 : unknown_schema_key.push_back('\x1F');
2321 [ + - ]: 39 : schema_iter = impl_->total_schema_map.find(unknown_schema_key);
2322 : : }
2323 : :
2324 [ + + ]: 45 : if (schema_iter == impl_->total_schema_map.end()) {
2325 [ + + + - : 77 : if (!schema_data.encoding.empty() && !schema_data.data.empty() &&
+ - ]
2326 [ + - + + ]: 77 : impl_->ser_map.find(schema_key) == impl_->ser_map.end()) {
2327 [ + - - + ]: 38 : if VUNLIKELY (!insert_schema(schema_data)) {
2328 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2329 : : }
2330 : :
2331 [ + - ]: 38 : impl_->ser_map.emplace(schema_key);
2332 : : }
2333 : :
2334 [ + - ]: 39 : impl_->total_schema_map.emplace(schema_key, schema_data);
2335 : :
2336 : 39 : return true;
2337 : : }
2338 : :
2339 : 6 : const auto& current = schema_iter->second;
2340 : :
2341 [ + - + + : 6 : if VUNLIKELY ((!schema_data.encoding.empty() && !current.encoding.empty() &&
+ + - + +
- + - + +
+ + + + +
+ + + + -
+ + + + -
+ - + +
+ ]
2342 : : current.encoding != schema_data.encoding) ||
2343 : : (!schema_data.data.empty() && !current.data.empty() && current.data != schema_data.data) ||
2344 : : (SchemaData::is_real_type(resolved_schema_type) && SchemaData::is_real_type(current.schema_type) &&
2345 : : current.schema_type != resolved_schema_type)) {
2346 [ + - + - : 4 : CLOG_E("VDBWriter: Conflicting schema pushed for [%s].", schema_data.name.c_str());
+ - ]
2347 : 2 : return false;
2348 : : }
2349 : :
2350 [ + - ]: 4 : SchemaData merged_schema = current;
2351 : :
2352 [ + + + - : 4 : if (merged_schema.encoding.empty() && !schema_data.encoding.empty()) {
+ + ]
2353 [ + - ]: 1 : merged_schema.encoding = schema_data.encoding;
2354 : : }
2355 : :
2356 [ + + + - : 4 : if (merged_schema.data.empty() && !schema_data.data.empty()) {
+ + ]
2357 : 1 : merged_schema.data = schema_data.data;
2358 : : }
2359 : :
2360 [ + + + - : 4 : if (!SchemaData::is_real_type(merged_schema.schema_type) && SchemaData::is_real_type(resolved_schema_type)) {
+ + ]
2361 : 1 : merged_schema.schema_type = resolved_schema_type;
2362 : : }
2363 : :
2364 [ + - + - : 8 : if (!merged_schema.encoding.empty() && !merged_schema.data.empty() &&
+ + ]
2365 [ + - + + ]: 8 : impl_->ser_map.find(schema_key) == impl_->ser_map.end()) {
2366 [ + - - + ]: 1 : if VUNLIKELY (!insert_schema(merged_schema)) {
2367 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2368 : : }
2369 : :
2370 [ + - ]: 1 : impl_->ser_map.emplace(schema_key);
2371 : : }
2372 : :
2373 [ + - ]: 4 : schema_data = merged_schema;
2374 : :
2375 [ + + + - : 4 : if (schema_iter->first != schema_key && merged_schema.schema_type == resolved_schema_type) {
+ + ]
2376 [ + - ]: 1 : impl_->total_schema_map.erase(schema_iter);
2377 [ + - ]: 1 : impl_->total_schema_map.emplace(schema_key, schema_data);
2378 : : } else {
2379 [ + - ]: 3 : schema_iter->second = schema_data;
2380 : : }
2381 : :
2382 : 4 : return true;
2383 : 45 : }
2384 : :
2385 : 845 : bool VDBWriter::load_schema(const std::string& ser_type, SchemaType& schema_type, SchemaData& schema_data) {
2386 : : #ifdef VLINK_ENABLE_SQLITE
2387 : 845 : schema_data = SchemaData{};
2388 : :
2389 [ - + ]: 845 : if VUNLIKELY (ser_type.empty()) {
2390 : : return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2391 : : }
2392 : :
2393 [ + - ]: 845 : std::string schema_key = ser_type;
2394 [ + - ]: 845 : schema_key.push_back('\x1F');
2395 [ + - ]: 845 : schema_key.append(SchemaData::convert_type(schema_type));
2396 : :
2397 : 845 : std::string unknown_schema_key;
2398 : 845 : auto schema_iter = impl_->total_schema_map.end();
2399 : :
2400 [ + + ]: 845 : if (schema_type != SchemaType::kUnknown) {
2401 [ + - ]: 813 : schema_iter = impl_->total_schema_map.find(schema_key);
2402 : :
2403 [ + + ]: 813 : if (schema_iter == impl_->total_schema_map.end()) {
2404 [ + - ]: 773 : unknown_schema_key = ser_type;
2405 [ + - ]: 773 : unknown_schema_key.push_back('\x1F');
2406 [ + - ]: 773 : schema_iter = impl_->total_schema_map.find(unknown_schema_key);
2407 : : }
2408 : : } else {
2409 [ + - + - ]: 64 : const auto prefix = ser_type + std::string("\x1F");
2410 : :
2411 [ + + ]: 64 : for (auto iter = impl_->total_schema_map.begin(); iter != impl_->total_schema_map.end(); ++iter) {
2412 [ + + ]: 32 : if (!Helpers::has_startwith(iter->first, prefix)) {
2413 : 31 : continue;
2414 : : }
2415 : :
2416 [ - + ]: 1 : if (schema_iter != impl_->total_schema_map.end()) {
2417 : : schema_iter = impl_->total_schema_map.end(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2418 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2419 : : }
2420 : :
2421 : 1 : schema_iter = iter;
2422 : : }
2423 : 32 : }
2424 : :
2425 [ + + ]: 845 : if VLIKELY (schema_iter != impl_->total_schema_map.end()) {
2426 [ + - ]: 41 : schema_data.name = schema_iter->second.name;
2427 [ + - ]: 41 : schema_data.encoding = schema_iter->second.encoding;
2428 : 41 : schema_data.schema_type = schema_iter->second.schema_type;
2429 : 41 : schema_data.data.shallow_copy(schema_iter->second.data);
2430 [ - + ]: 804 : } else if (impl_->schema_plugin_interface) {
2431 : : schema_data =
2432 : : impl_->schema_plugin_interface->search_schema(ser_type, schema_type); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2433 [ + + ]: 804 : } else if (impl_->schema_callback) {
2434 [ + - ]: 8 : schema_data = impl_->schema_callback(ser_type, schema_type);
2435 : : }
2436 : :
2437 : 845 : schema_type = SchemaData::resolve_type(schema_type, ser_type, schema_data.encoding);
2438 : 845 : schema_data.schema_type = SchemaData::resolve_type(schema_data.schema_type, ser_type, schema_data.encoding);
2439 : :
2440 [ + + + + ]: 845 : if (schema_type != SchemaType::kUnknown && schema_data.schema_type != SchemaType::kUnknown &&
2441 [ + + ]: 812 : schema_type != schema_data.schema_type) {
2442 [ + - + - : 4 : CLOG_E("VDBWriter: Schema family mismatch for [%s], requested = %d, resolved = %d.", ser_type.c_str(),
+ - ]
2443 : : static_cast<int>(schema_type), static_cast<int>(schema_data.schema_type));
2444 : 2 : return false;
2445 : : }
2446 : :
2447 [ + + + + : 843 : if (schema_type != SchemaType::kUnknown && schema_data.encoding.empty()) {
+ + ]
2448 [ + - ]: 764 : schema_data.encoding = std::string(SchemaData::convert_type(schema_type));
2449 : : }
2450 : :
2451 [ + + + + ]: 843 : if (schema_data.schema_type == SchemaType::kUnknown && schema_type != SchemaType::kUnknown) {
2452 : : schema_data.schema_type = schema_type; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2453 : : }
2454 : :
2455 [ + + ]: 843 : if (!schema_data.name.empty()) {
2456 [ + - ]: 48 : std::string resolved_schema_key = ser_type;
2457 [ + - ]: 48 : resolved_schema_key.push_back('\x1F');
2458 [ + - ]: 48 : resolved_schema_key.append(SchemaData::convert_type(schema_data.schema_type));
2459 : :
2460 [ + + + - : 48 : if VLIKELY (schema_iter != impl_->total_schema_map.end() && schema_iter->first == resolved_schema_key) {
+ + ]
2461 [ - + - - : 41 : if (schema_iter->second.encoding.empty() && !schema_data.encoding.empty()) {
- + ]
2462 [ # # ]: 0 : schema_iter->second.encoding = schema_data.encoding;
2463 : : }
2464 : :
2465 : 41 : schema_iter->second.schema_type = schema_data.schema_type;
2466 : : } else {
2467 [ + - ]: 7 : SchemaData stored_schema = schema_data;
2468 : :
2469 [ - + - - : 7 : if (schema_iter != impl_->total_schema_map.end() && schema_iter->second.schema_type == SchemaType::kUnknown &&
- + ]
2470 [ # # ]: 0 : schema_data.schema_type != SchemaType::kUnknown) {
2471 : : impl_->total_schema_map.erase(schema_iter); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2472 : : }
2473 : :
2474 : : const auto stored_iter =
2475 [ + - ]: 7 : impl_->total_schema_map.insert_or_assign(resolved_schema_key, std::move(stored_schema)).first;
2476 : 7 : schema_data.data.shallow_copy(stored_iter->second.data);
2477 : 7 : }
2478 : 48 : }
2479 : :
2480 : 843 : return true;
2481 : : #else
2482 : : (void)ser_type;
2483 : : (void)schema_type;
2484 : : (void)schema_data;
2485 : : return false;
2486 : : #endif
2487 : 845 : }
2488 : :
2489 : 47 : bool VDBWriter::insert_schema(const SchemaData& schema_data) {
2490 : : #ifdef VLINK_ENABLE_SQLITE
2491 : :
2492 [ - + ]: 47 : if VUNLIKELY (!impl_->db) {
2493 : : VLOG_W("VDBWriter: Sqlite not open."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2494 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2495 : : }
2496 : :
2497 [ + - ]: 47 : std::string encoding = schema_data.encoding;
2498 : 47 : SchemaType schema_type = SchemaData::resolve_type(schema_data.schema_type, schema_data.name, encoding);
2499 : :
2500 [ - + - - : 47 : if (encoding.empty() && SchemaData::is_real_type(schema_type)) {
- + ]
2501 : : encoding = std::string(SchemaData::convert_type(schema_type)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2502 : : }
2503 : :
2504 [ + - + - : 47 : if (schema_data.name.empty() || encoding.empty() || schema_data.data.empty()) {
- + - + ]
2505 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2506 : : }
2507 : :
2508 : 47 : int ret = 0;
2509 : :
2510 [ + - ]: 47 : ::sqlite3_bind_text(impl_->schemas_stmt, get_column(0), schema_data.name.c_str(), schema_data.name.size(),
2511 : : SQLITE_STATIC);
2512 : :
2513 [ + - ]: 47 : ::sqlite3_bind_text(impl_->schemas_stmt, get_column(1), encoding.c_str(), encoding.size(), SQLITE_STATIC);
2514 : :
2515 [ + - ]: 47 : ::sqlite3_bind_blob(impl_->schemas_stmt, get_column(2), schema_data.data.data(), schema_data.data.size(),
2516 : : SQLITE_STATIC);
2517 : :
2518 [ + - ]: 47 : ret = ::sqlite3_step(impl_->schemas_stmt);
2519 : :
2520 [ - + ]: 47 : if VUNLIKELY (ret != SQLITE_DONE) {
2521 : : CLOG_W("Failed to insert schema table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2522 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2523 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2524 : : }
2525 : :
2526 [ + - ]: 47 : ret = ::sqlite3_reset(impl_->schemas_stmt);
2527 : :
2528 [ - + ]: 47 : if VUNLIKELY (ret != SQLITE_OK) {
2529 : : CLOG_W("Failed to reset schema table: %s.", ::sqlite3_errmsg(impl_->db)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2530 : : rollback_cache(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2531 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2532 : : }
2533 : :
2534 : 47 : return true;
2535 : : #else
2536 : : (void)schema_data;
2537 : : return false;
2538 : : #endif
2539 : 47 : }
2540 : :
2541 : : } // namespace vlink
|
