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/vcap_writer.h"
25 : :
26 : : #include <algorithm>
27 : : #include <cstdio>
28 : : #include <filesystem>
29 : : #include <fstream>
30 : : #include <memory>
31 : : #include <mutex>
32 : : #include <optional>
33 : : #include <string>
34 : : #include <string_view>
35 : : #include <unordered_map>
36 : : #include <utility>
37 : : #include <vector>
38 : :
39 : : #include "./base/elapsed_timer.h"
40 : : #include "./base/helpers.h"
41 : : #include "./base/logger.h"
42 : : #include "./base/utils.h"
43 : : #include "./version.h"
44 : :
45 : : // json
46 : : #include <nlohmann/json.hpp>
47 : :
48 : : // mcap
49 : : #include "./private/mcap_import.h"
50 : :
51 : : // schema_plugin
52 : : #include "./extension/schema_plugin_interface.h"
53 : :
54 : : namespace vlink {
55 : :
56 : : static constexpr std::string_view kMcapMagic{"\x89MCAP0\r\n", 8};
57 : :
58 : 769 : static const std::string& make_channel_key(std::string& buffer, const std::string& url, std::string_view action_name) {
59 : 769 : buffer.assign(url);
60 : 769 : buffer.push_back('\x1F');
61 : 769 : buffer.append(action_name);
62 : :
63 : 769 : return buffer;
64 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
65 : :
66 : : // VCAPWriter::Impl
67 : : struct VCAPWriter::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
68 : : // UrlMsgInfo
69 : : struct UrlMsgInfo final {
70 : : int index{0};
71 : : size_t count{0};
72 : : size_t size{0};
73 : : int64_t first_timestamp{-1};
74 : : int64_t last_timestamp{-1};
75 : : double freq{0};
76 : : double loss{0};
77 : : std::string url;
78 : : std::string url_type;
79 : : std::string ser_type;
80 : : SchemaType schema_type{SchemaType::kUnknown};
81 : : ActionType action_type{ActionType::kUnknownAction};
82 : :
83 : 37 : bool operator<(const UrlMsgInfo& target) const noexcept { return index < target.index; }
84 : : };
85 : :
86 : : struct SplitInfo final {
87 : : int64_t row{0};
88 : : std::unordered_map<std::string, UrlMsgInfo> url_map;
89 : : };
90 : :
91 : : struct MemoryCharge final {
92 : 298 : MemoryCharge(std::atomic<int64_t>& counter, int64_t bytes) : value(&counter), size(bytes) {}
93 : :
94 : 596 : MemoryCharge(MemoryCharge&& other) noexcept : value(std::exchange(other.value, nullptr)), size(other.size) {}
95 : :
96 : 894 : ~MemoryCharge() {
97 [ + + ]: 894 : if (value) {
98 : 298 : value->fetch_sub(size, std::memory_order_relaxed);
99 : : }
100 : 894 : }
101 : :
102 : : std::atomic<int64_t>* value;
103 : : int64_t size;
104 : :
105 : : VLINK_DISALLOW_COPY_AND_ASSIGN(MemoryCharge)
106 : : };
107 : :
108 : : std::atomic_bool is_dumping{false};
109 : : std::atomic_bool is_split_mode{false};
110 : : std::atomic<int> split_index{0};
111 : : std::atomic<int64_t> memory_size{0};
112 : : std::atomic_bool in_cached{false};
113 : : std::atomic<int64_t> cached_size{0};
114 : : std::atomic_bool quit_flag{false};
115 : :
116 : : std::string path;
117 : : std::filesystem::path active_path;
118 : : std::filesystem::path split_output_dir;
119 : : std::string base_dir;
120 : : std::string base_name;
121 : : BagWriter::Config config;
122 : : ElapsedTimer elapsed_timer{ElapsedTimer::kMicro};
123 : :
124 : : int64_t current_row{0};
125 : : int64_t current_size{0};
126 : : bool has_oversize{false};
127 : :
128 : : int64_t last_timestamp{0};
129 : :
130 : : BagWriter::SystemClock time_start;
131 : : BagWriter::SystemClock time_current;
132 : : int64_t start_timestamp{0};
133 : :
134 : : std::vector<std::string> split_file_list;
135 : : std::vector<SplitInfo> split_info_list;
136 : : bool split_before{false};
137 : : bool split_first{false};
138 : :
139 : : std::vector<std::string> total_url_list;
140 : : int64_t total_current_row{0};
141 : : int64_t total_current_size{0};
142 : : int64_t total_timestamp{0};
143 : :
144 : : std::unordered_map<std::string, UrlMsgInfo> url_map;
145 : : std::unordered_map<std::string, int> ser_map;
146 : : std::unordered_map<std::string, UrlMsgInfo> total_url_map;
147 : : std::unordered_map<std::string, SchemaData> total_schema_map;
148 : :
149 : : BagWriter::SplitCallback split_callback;
150 : : BagWriter::SchemaCallback schema_callback;
151 : : std::string split_filename;
152 : : std::mutex split_mtx;
153 : :
154 : : std::string app_name;
155 : : std::string tag_name;
156 : : int32_t timezone_diff{0};
157 : :
158 : : bool enable_compressed{false};
159 : : std::mutex write_mtx;
160 : :
161 : : std::string write_url_type;
162 : : std::string write_channel_key;
163 : :
164 : : // mcap
165 : : std::ofstream output;
166 : : std::optional<mcap::McapWriter> writer;
167 : : mcap::McapWriterOptions writer_options{"vlink"};
168 : :
169 : : // schema plugin interface
170 : : SchemaPluginInterface* schema_plugin_interface{nullptr};
171 : : };
172 : :
173 : : // VCAPWriter
174 : 137 : VCAPWriter::VCAPWriter(const std::string& path, const Config& config)
175 [ + - ]: 137 : : BagWriter(path, config), impl_{std::make_unique<Impl>()} {
176 [ + - + - ]: 137 : set_name("VCAPWriter");
177 : :
178 [ + - ]: 137 : impl_->url_map.reserve(128);
179 [ + - ]: 137 : impl_->ser_map.reserve(128);
180 [ + - + - ]: 137 : url_loss_map_ref().reserve(128);
181 [ + - ]: 137 : impl_->total_url_map.reserve(128);
182 [ + - + - ]: 137 : total_url_loss_map_ref().reserve(128);
183 [ + - ]: 137 : impl_->total_schema_map.reserve(128);
184 : :
185 [ + - ]: 137 : impl_->schema_plugin_interface = get_schema_interface();
186 : :
187 [ + - + - ]: 137 : impl_->app_name = get_default_app_name();
188 : :
189 [ + + ]: 137 : if (config.tag_name.empty()) {
190 [ + - + - ]: 69 : impl_->tag_name = get_default_tag_name();
191 : : } else {
192 [ + - ]: 68 : impl_->tag_name = config.tag_name;
193 : : }
194 : :
195 [ + - ]: 137 : impl_->timezone_diff = get_default_timezone_diff();
196 : :
197 [ + - ]: 137 : impl_->path = path;
198 [ + - ]: 137 : impl_->config = config;
199 : 137 : impl_->writer_options.noChunkCRC = !impl_->config.enable_chunk_crc;
200 : :
201 [ + + + + ]: 137 : impl_->enable_compressed = impl_->config.compress == kCompressAuto || impl_->config.compress == kCompressZstd;
202 : :
203 [ + + ]: 137 : if (impl_->enable_compressed) {
204 : : #ifdef VLINK_ENABLE_ZSTD
205 : 13 : impl_->writer_options.compression = mcap::Compression::Zstd;
206 : :
207 [ + + + + : 13 : switch (impl_->config.compress_level) {
+ + + ]
208 : 1 : case 0:
209 : 1 : impl_->writer_options.compressionLevel = mcap::CompressionLevel::Default;
210 : 1 : break;
211 : 1 : case 1:
212 : 1 : impl_->writer_options.compressionLevel = mcap::CompressionLevel::Fastest;
213 : 1 : break;
214 : 1 : case 2:
215 : 1 : impl_->writer_options.compressionLevel = mcap::CompressionLevel::Fast;
216 : 1 : break;
217 : 3 : case 3:
218 : 3 : impl_->writer_options.compressionLevel = mcap::CompressionLevel::Default;
219 : 3 : break;
220 : 3 : case 4:
221 : 3 : impl_->writer_options.compressionLevel = mcap::CompressionLevel::Slow;
222 : 3 : break;
223 : 3 : case 5:
224 : 3 : impl_->writer_options.compressionLevel = mcap::CompressionLevel::Slowest;
225 : 3 : break;
226 : 1 : default:
227 : 1 : impl_->writer_options.compressionLevel = mcap::CompressionLevel::Default;
228 : 1 : break;
229 : : }
230 : : #else
231 : : impl_->writer_options.compression = mcap::Compression::None;
232 : : impl_->config.compress = kCompressNone;
233 : :
234 : : impl_->enable_compressed = false;
235 : : VLOG_W("VCAPWriter: Compress is not supported.");
236 : : #endif
237 : : } else {
238 : 124 : impl_->writer_options.compression = mcap::Compression::None;
239 : : }
240 : :
241 [ + + ]: 137 : if (impl_->config.cache_size > 0) {
242 : 86 : impl_->writer_options.noChunking = false;
243 : 86 : impl_->writer_options.chunkSize = impl_->config.cache_size;
244 : : } else {
245 : 51 : impl_->writer_options.noChunking = true;
246 : 51 : impl_->writer_options.chunkSize = 0;
247 : :
248 : 51 : impl_->writer_options.compression = mcap::Compression::None;
249 : 51 : impl_->config.compress = kCompressNone;
250 : :
251 [ + + ]: 51 : if (impl_->enable_compressed) {
252 : 2 : impl_->enable_compressed = false;
253 [ + - + - : 4 : VLOG_W("VCAPWriter: Compress is not supported without cache_size > 0.");
+ - ]
254 : : }
255 : : }
256 : :
257 [ + + ]: 137 : if VUNLIKELY (impl_->config.max_task_depth <= 0) {
258 : 1 : impl_->config.max_task_depth = BagWriter::Config().max_task_depth;
259 : : }
260 : :
261 [ + - ]: 137 : reset_lockfree_capacity();
262 : :
263 [ + + ]: 137 : if VUNLIKELY (impl_->config.enable_limit) {
264 [ + - + - : 2 : VLOG_W("VCAPWriter: Enable limit is not supported.");
+ - ]
265 : : }
266 : :
267 : : try {
268 : : #ifdef _WIN32
269 : : std::filesystem::path file_path(Helpers::string_to_wstring(path));
270 : : std::string suffix = Helpers::path_to_string(file_path.extension());
271 : : #else
272 [ + - ]: 137 : std::filesystem::path file_path(path);
273 [ + - + - ]: 137 : std::string suffix = file_path.extension().string();
274 : : #endif
275 : :
276 [ + - ]: 137 : const auto parent_path = file_path.parent_path();
277 : :
278 : 901 : std::transform(suffix.begin(), suffix.end(), suffix.begin(), [](unsigned char c) { return std::tolower(c); });
279 : :
280 [ + + ]: 137 : if (suffix == ".vcapx") {
281 [ + - + + ]: 79 : if (std::filesystem::exists(file_path)) {
282 : : try {
283 : 2 : nlohmann::json root_json;
284 : :
285 : : {
286 [ + - ]: 2 : std::ifstream filex(file_path);
287 : :
288 [ + - ]: 2 : filex >> root_json;
289 : :
290 [ + - ]: 2 : filex.close();
291 : 2 : }
292 : :
293 [ + - + - ]: 2 : nlohmann::json files_json = root_json["VLinkFiles"];
294 : :
295 [ + - + - : 9 : for (const auto& file_info : files_json) {
+ - + + ]
296 [ + + ]: 7 : if (!file_info.is_string()) {
297 : 4 : continue;
298 : : }
299 : :
300 [ + - ]: 6 : const auto stale_file_name = file_info.get<std::string>();
301 : : #ifdef _WIN32
302 : : const std::filesystem::path stale_file_path(Helpers::string_to_wstring(stale_file_name));
303 : : #else
304 [ + - ]: 6 : const std::filesystem::path stale_file_path(stale_file_name);
305 : : #endif
306 : :
307 [ + - + - : 18 : if (stale_file_path.empty() || stale_file_path == "." || stale_file_path == ".." ||
+ - + - +
- + + + -
+ + - - -
- ]
308 [ + - + - : 12 : stale_file_path != stale_file_path.filename()) {
+ - - - ]
309 [ + - + - : 6 : CLOG_W("VCAPWriter: Ignore unsafe split file path [%s].", stale_file_name.c_str());
+ - ]
310 : 3 : continue;
311 : 3 : }
312 : :
313 [ + - ]: 3 : const auto stale_output_path = parent_path / stale_file_path;
314 : 3 : std::error_code remove_ec;
315 : 3 : std::filesystem::remove(stale_output_path, remove_ec);
316 : :
317 [ + + ]: 3 : if VUNLIKELY (remove_ec) {
318 [ + - + - : 2 : CLOG_W("VCAPWriter: Failed to remove stale split path [%s]: %s.", stale_file_name.c_str(),
+ - + - +
- - - ]
319 : : remove_ec.message().c_str());
320 : : }
321 [ + + + + ]: 9 : }
322 : :
323 [ + - ]: 2 : std::filesystem::remove(file_path);
324 [ - - ]: 2 : } catch (const nlohmann::json::exception& e) {
325 [ # # # # : 0 : CLOG_W("VCAPWriter: Failed to parse stale split manifest [%s]: %s.", path.c_str(), e.what());
# # ]
326 : 0 : }
327 : : }
328 : :
329 : 79 : impl_->is_split_mode.store(true, std::memory_order_relaxed);
330 : 79 : impl_->split_index.store(0, std::memory_order_relaxed);
331 : :
332 : : #ifdef _WIN32
333 : :
334 : : std::error_code absolute_ec;
335 : : auto absolute_path = std::filesystem::absolute(file_path, absolute_ec);
336 : :
337 : : if VUNLIKELY (absolute_ec) {
338 : : absolute_path = file_path;
339 : : }
340 : :
341 : : impl_->path = Helpers::path_to_string(absolute_path);
342 : : impl_->split_output_dir = absolute_path.parent_path();
343 : :
344 : : if (parent_path.empty()) {
345 : : impl_->base_dir.clear();
346 : : impl_->base_name = Helpers::path_to_string(file_path.stem());
347 : : } else {
348 : : impl_->base_dir = Helpers::path_to_string(parent_path);
349 : : impl_->base_name = Helpers::path_to_string(std::filesystem::path(parent_path / file_path.stem()));
350 : : }
351 : : #else
352 : :
353 : 79 : std::error_code absolute_ec;
354 [ + - ]: 79 : auto absolute_path = std::filesystem::absolute(file_path, absolute_ec);
355 : :
356 [ - + ]: 79 : if VUNLIKELY (absolute_ec) {
357 [ # # ]: 0 : absolute_path = file_path;
358 : : }
359 : :
360 [ + - ]: 79 : impl_->path = absolute_path.string();
361 [ + - ]: 79 : impl_->split_output_dir = absolute_path.parent_path();
362 : :
363 [ + + ]: 79 : if (parent_path.empty()) {
364 : 3 : impl_->base_dir.clear();
365 [ + - + - ]: 3 : impl_->base_name = file_path.stem().string();
366 : : } else {
367 [ + - ]: 76 : impl_->base_dir = parent_path.string();
368 [ + - + - : 76 : impl_->base_name = std::filesystem::path(parent_path / file_path.stem()).string();
+ - ]
369 : : }
370 : : #endif
371 : :
372 [ + - ]: 79 : impl_->time_start = std::chrono::time_point_cast<std::chrono::milliseconds>(std::chrono::system_clock::now());
373 : 79 : impl_->time_current = impl_->time_start;
374 : :
375 [ + + ]: 79 : if (impl_->config.start_timestamp > 0) {
376 : 25 : impl_->start_timestamp = impl_->config.start_timestamp;
377 : : } else {
378 : 54 : impl_->start_timestamp = impl_->time_start.time_since_epoch().count();
379 : : }
380 : :
381 [ + - ]: 79 : write_filex(false);
382 : :
383 [ + + ]: 79 : if (impl_->config.split_name_by_time) {
384 [ + + ]: 27 : if (impl_->base_dir.empty()) {
385 [ + - + - ]: 2 : impl_->split_filename = get_format_date(&impl_->time_current, true) + ".vcap";
386 : : } else {
387 [ + - + - : 25 : impl_->split_filename = impl_->base_dir + "/" + get_format_date(&impl_->time_current, true) + ".vcap";
+ - + - ]
388 : : }
389 : : } else {
390 : 52 : impl_->split_filename =
391 [ + - + - : 156 : impl_->base_name + "." + std::to_string(impl_->split_index.load(std::memory_order_relaxed) + 1) + ".vcap";
+ - + - ]
392 : : }
393 : :
394 [ + - ]: 79 : open_split(impl_->split_filename);
395 : 79 : } else {
396 [ + - ]: 58 : impl_->time_start = std::chrono::time_point_cast<std::chrono::milliseconds>(std::chrono::system_clock::now());
397 : 58 : impl_->time_current = impl_->time_start;
398 : :
399 [ + + ]: 58 : if (impl_->config.start_timestamp > 0) {
400 : 43 : impl_->start_timestamp = impl_->config.start_timestamp;
401 : : } else {
402 : 15 : impl_->start_timestamp = impl_->time_start.time_since_epoch().count();
403 : : }
404 : :
405 : 58 : impl_->is_split_mode.store(false, std::memory_order_relaxed);
406 : 58 : impl_->split_index.store(0, std::memory_order_relaxed);
407 : :
408 [ + - ]: 58 : open(path);
409 : : }
410 [ - - ]: 137 : } catch (std::filesystem::filesystem_error& e) {
411 : : VLOG_F("VCAPWriter: Filesystem error, ", e.what(), "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
412 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
413 : :
414 : 137 : impl_->elapsed_timer.start();
415 : 137 : }
416 : :
417 : 139 : VCAPWriter::~VCAPWriter() {
418 : : #ifdef _WIN32
419 : : if (Utils::is_terminating()) {
420 : : (void)impl_.release();
421 : : return;
422 : : }
423 : : #endif
424 : :
425 : 137 : detach_plugin();
426 : :
427 : 137 : impl_->quit_flag.store(true, std::memory_order_release);
428 : :
429 [ - + ]: 137 : if VUNLIKELY (!wait_for_idle(30000U)) {
430 : : VLOG_W("VCAPWriter: Force to quit."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
431 : : }
432 : :
433 : 137 : quit(true);
434 : :
435 : 137 : wait_for_quit();
436 : :
437 : 137 : close();
438 : 139 : }
439 : :
440 : 193 : void VCAPWriter::close() {
441 : 193 : close_segment();
442 : :
443 [ + + - + : 193 : if VUNLIKELY (impl_->is_split_mode.load(std::memory_order_relaxed) && !write_filex(true)) {
- + ]
444 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
445 : : }
446 : 193 : }
447 : :
448 : 72 : void VCAPWriter::register_split_callback(SplitCallback&& callback, bool before) {
449 [ + - ]: 72 : std::lock_guard lock(impl_->split_mtx);
450 : 72 : impl_->split_before = before;
451 : 72 : impl_->split_callback = std::move(callback);
452 : 72 : }
453 : :
454 : 6 : void VCAPWriter::register_schema_callback(SchemaCallback&& callback) {
455 [ + - ]: 6 : std::lock_guard lock(impl_->write_mtx);
456 : 6 : impl_->schema_callback = std::move(callback);
457 : 6 : }
458 : :
459 : 18 : bool VCAPWriter::push_schema(const SchemaData& schema_data) {
460 [ + - ]: 18 : SchemaData stored_schema = schema_data;
461 : :
462 [ + + ]: 18 : if (impl_->config.sync_mode) {
463 [ + - ]: 16 : std::lock_guard lock(impl_->write_mtx);
464 [ + - ]: 16 : return merge_schema(stored_schema);
465 : 16 : }
466 : :
467 [ + - + - ]: 2 : bool posted = post_persistent_task([this, stored_schema = std::move(stored_schema)]() mutable {
468 [ + - ]: 2 : std::lock_guard lock(impl_->write_mtx);
469 : :
470 [ + - - + ]: 2 : if VUNLIKELY (!merge_schema(stored_schema)) {
471 [ # # # # : 0 : CLOG_E("VCAPWriter: Deferred merge_schema failed for [%s] in async push_schema path.",
# # ]
472 : : stored_schema.name.c_str());
473 : 0 : set_fail();
474 : : }
475 : 2 : });
476 : :
477 : 2 : return posted;
478 : 18 : }
479 : :
480 : 8 : bool VCAPWriter::is_dumping() const { return impl_->is_dumping.load(std::memory_order_relaxed); }
481 : :
482 : 34 : bool VCAPWriter::is_split_mode() const { return impl_->is_split_mode.load(std::memory_order_relaxed); }
483 : :
484 : 88 : int VCAPWriter::get_split_index() const { return impl_->split_index.load(std::memory_order_relaxed); }
485 : :
486 : 772 : int64_t VCAPWriter::record(const Frame& frame, int64_t timestamp) {
487 : 772 : const std::string& url = frame.url;
488 : 772 : const std::string& ser_type = frame.ser_type;
489 : 772 : const SchemaType schema_type = frame.schema_type;
490 : 772 : const ActionType action_type = frame.action_type;
491 : 772 : const Bytes& data = frame.data;
492 : 772 : const int64_t microseconds_timestamp = timestamp;
493 : :
494 [ + + ]: 772 : if (impl_->config.sync_mode) {
495 [ + - ]: 473 : std::lock_guard lock(impl_->write_mtx);
496 : :
497 [ + - + + ]: 473 : if VUNLIKELY (!write(url, ser_type, schema_type, action_type, data, microseconds_timestamp)) {
498 : 6 : return -1;
499 : : }
500 [ + + ]: 473 : } else {
501 [ + + ]: 598 : if VUNLIKELY (impl_->memory_size.load(std::memory_order_relaxed) + static_cast<int64_t>(data.size()) >
502 : : impl_->config.max_memory_size) {
503 [ + - + - : 2 : CLOG_E("The memory data in the queue exceeds %.1fGB and the task is automatically discarded.",
+ - ]
504 : : impl_->config.max_memory_size / 1024.0 / 1024.0 / 1024.0);
505 : :
506 : 2 : return -1;
507 : : }
508 : :
509 : 298 : int url_index = -1;
510 : 298 : int ser_index = -1;
511 : :
512 [ + - ]: 298 : get_url_meta(url, ser_type, url_index, ser_index);
513 : :
514 : 298 : const auto queued_size = static_cast<int64_t>(data.size());
515 : :
516 : 298 : impl_->memory_size.fetch_add(queued_size, std::memory_order_relaxed);
517 : 298 : Impl::MemoryCharge memory_charge(impl_->memory_size, queued_size);
518 : :
519 [ + - + - ]: 596 : bool posted = post_persistent_task([this, url_index, ser_index, schema_type, action_type, data,
520 : 298 : memory_charge = std::move(memory_charge),
521 : : microseconds_timestamp]() { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
522 : : (void)memory_charge;
523 : :
524 [ + - ]: 297 : std::lock_guard lock(impl_->write_mtx);
525 : 297 : std::string url;
526 : 297 : std::string ser_type;
527 : :
528 [ + - ]: 297 : get_url_meta(url_index, ser_index, url, ser_type);
529 : :
530 [ + - + + ]: 297 : if VUNLIKELY (!write(url, ser_type, schema_type, action_type, data, microseconds_timestamp)) {
531 : 1 : set_fail();
532 : : }
533 : 297 : });
534 : :
535 [ + + ]: 298 : if VUNLIKELY (!posted) {
536 : : return -1; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
537 : : }
538 [ + + ]: 298 : }
539 : :
540 : 764 : return microseconds_timestamp;
541 : : }
542 : :
543 : 1 : int64_t VCAPWriter::get_record_timestamp() const { return impl_->elapsed_timer.get(); }
544 : :
545 : 301 : size_t VCAPWriter::get_max_task_count() const { return impl_->config.max_task_depth; }
546 : :
547 : 31 : void VCAPWriter::on_begin() {
548 : 31 : MessageLoop::on_begin();
549 : :
550 : 31 : impl_->elapsed_timer.restart();
551 : 31 : }
552 : :
553 : 31 : void VCAPWriter::on_end() { MessageLoop::on_end(); }
554 : :
555 : 699 : void VCAPWriter::open(const std::string& path) {
556 : : try {
557 : : #ifdef _WIN32
558 : : impl_->split_file_list.emplace_back(Helpers::path_to_string(std::filesystem::path(path).filename()));
559 : : std::filesystem::path file_path(Helpers::string_to_wstring(path));
560 : : #else
561 [ + - + - : 699 : impl_->split_file_list.emplace_back(std::filesystem::path(path).filename().string());
+ - + - ]
562 [ + - ]: 699 : std::filesystem::path file_path(path);
563 : : #endif
564 : :
565 : 699 : std::error_code absolute_ec;
566 [ + - ]: 699 : impl_->active_path = std::filesystem::absolute(file_path, absolute_ec);
567 : :
568 [ - + ]: 699 : if VUNLIKELY (absolute_ec) {
569 [ # # ]: 0 : impl_->active_path = file_path;
570 : : }
571 : :
572 [ + - + + ]: 699 : if (std::filesystem::exists(file_path)) {
573 [ + - ]: 1 : std::filesystem::remove(file_path);
574 : : } else {
575 [ + - ]: 698 : auto parent_path = file_path.parent_path();
576 : :
577 [ + + + - : 698 : if (!parent_path.empty() && !std::filesystem::exists(parent_path)) {
+ + + + ]
578 [ + - ]: 1 : std::filesystem::create_directories(parent_path);
579 : : }
580 : 698 : }
581 [ - - ]: 699 : } catch (std::filesystem::filesystem_error& e) {
582 : : VLOG_F("VCAPWriter: Filesystem error, ", e.what(), "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
583 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
584 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
585 : :
586 : 699 : mcap::Status status;
587 : :
588 : : #ifdef _WIN32
589 : : impl_->output.open(std::filesystem::path(Helpers::string_to_wstring(path)), std::ios::binary | std::ios::trunc);
590 : : #else
591 [ + - ]: 699 : impl_->output.open(path, std::ios::binary | std::ios::trunc);
592 : : #endif
593 : :
594 [ + - - + ]: 699 : if VUNLIKELY (!impl_->output) {
595 [ # # # # : 0 : VLOG_F("Failed to open output: ", path);
# # ]
596 : 0 : return;
597 : : }
598 : :
599 [ + - ]: 699 : impl_->writer.emplace();
600 [ + - ]: 699 : impl_->writer->open(impl_->output, impl_->writer_options);
601 : :
602 : 699 : mcap::Metadata header_meta_data;
603 [ + - ]: 699 : header_meta_data.name = "VLinkHeader";
604 [ + - + - : 699 : header_meta_data.metadata["tag"] = impl_->tag_name;
+ - ]
605 [ + - + - : 699 : header_meta_data.metadata["version"] = VLINK_VERSION;
+ - ]
606 [ + + + - : 699 : header_meta_data.metadata["compress"] = impl_->enable_compressed ? "zstd" : "None";
+ - + - ]
607 [ + - + - : 699 : header_meta_data.metadata["process"] = impl_->app_name;
+ - ]
608 [ + - + - : 699 : header_meta_data.metadata["date"] = get_format_date(&impl_->time_current);
+ - ]
609 [ + - + - : 699 : header_meta_data.metadata["timezone"] = std::to_string(impl_->timezone_diff);
+ - ]
610 [ + - + - : 699 : header_meta_data.metadata["start_timestamp"] = std::to_string(impl_->start_timestamp);
+ - ]
611 : :
612 [ + - ]: 699 : status = impl_->writer->write(header_meta_data);
613 : :
614 [ - + ]: 699 : if VUNLIKELY (!status.ok()) {
615 : : CLOG_F("VCAPWriter: Failed to write header meta data, error = %s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
616 : : status.message.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
617 : : impl_->writer->close(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
618 : : impl_->writer->terminate(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
619 : 0 : impl_->writer.reset();
620 : : impl_->output.close(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
621 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
622 : : }
623 : :
624 : 699 : impl_->last_timestamp = 0;
625 [ + - + - ]: 699 : }
626 : :
627 : 641 : bool VCAPWriter::open_split(const std::string& path) {
628 : 641 : const auto split_file_count = impl_->split_file_list.size();
629 : : #ifdef _WIN32
630 : : const auto file_name = std::filesystem::path(Helpers::string_to_wstring(path)).filename();
631 : : const auto split_path = impl_->split_output_dir / file_name;
632 : : open(Helpers::path_to_string(split_path));
633 : : #else
634 [ + - + - : 1282 : const auto split_path = impl_->split_output_dir / std::filesystem::path(path).filename();
+ - ]
635 [ + - + - ]: 641 : open(split_path.string());
636 : : #endif
637 : :
638 [ + - ]: 641 : if VLIKELY (impl_->writer) {
639 : 641 : return true;
640 : : }
641 : :
642 [ # # ]: 0 : impl_->split_file_list.resize(split_file_count);
643 : 0 : std::error_code remove_ec;
644 : 0 : std::filesystem::remove(split_path, remove_ec);
645 : 0 : return false;
646 : 641 : }
647 : :
648 : 756 : void VCAPWriter::close_segment() {
649 [ + + ]: 756 : if (!impl_->writer) {
650 : 57 : return;
651 : : }
652 : :
653 : 699 : mcap::Status status;
654 : :
655 : 699 : std::vector<Impl::UrlMsgInfo> msg_info_list;
656 [ + - ]: 699 : msg_info_list.reserve(impl_->url_map.size());
657 : :
658 : : {
659 [ + - + - ]: 699 : std::lock_guard lock(sample_mutex());
660 : :
661 [ + + ]: 1428 : for (const auto& entry : impl_->url_map) {
662 : 729 : const auto& msg_info = entry.second;
663 [ + - ]: 729 : msg_info_list.emplace_back(msg_info);
664 : :
665 : 729 : auto& last = msg_info_list.back();
666 [ + - + - : 729 : auto loss_iter = url_loss_map_ref().find(recover_recorded_url(last.url));
+ - ]
667 [ + - + + ]: 729 : last.loss = loss_iter == url_loss_map_ref().end() ? 0 : loss_iter->second;
668 : : }
669 : 699 : }
670 : :
671 [ + - ]: 699 : std::sort(msg_info_list.begin(), msg_info_list.end());
672 : :
673 [ + + ]: 1428 : for (const auto& msg_info : msg_info_list) {
674 : 729 : mcap::Metadata channel_meta_data;
675 [ + - + - ]: 729 : channel_meta_data.name = "VLinkChannel_" + std::to_string(msg_info.index + 1);
676 [ + - + - : 729 : channel_meta_data.metadata["index"] = std::to_string(msg_info.index);
+ - ]
677 [ + - + - : 729 : channel_meta_data.metadata["type"] = msg_info.url_type;
+ - ]
678 [ + - + - : 729 : channel_meta_data.metadata["action"] = std::string(convert_action(msg_info.action_type));
+ - + - ]
679 [ + - + - : 729 : channel_meta_data.metadata["encoding"] = std::string(SchemaData::convert_type(msg_info.schema_type));
+ - ]
680 [ + - + - : 729 : channel_meta_data.metadata["ser"] = msg_info.ser_type;
+ - ]
681 [ + - + - : 729 : channel_meta_data.metadata["count"] = std::to_string(msg_info.count);
+ - ]
682 [ + - + - : 729 : channel_meta_data.metadata["size"] = std::to_string(msg_info.size);
+ - ]
683 [ + - + - ]: 729 : channel_meta_data.metadata["loss"] = Helpers::double_to_string(msg_info.loss, 6);
684 [ + - + - : 729 : channel_meta_data.metadata["freq"] = std::to_string(msg_info.freq);
+ - ]
685 : :
686 [ + - ]: 729 : status = impl_->writer->write(channel_meta_data);
687 : :
688 [ - + ]: 729 : if VUNLIKELY (!status.ok()) {
689 : : CLOG_E("VCAPWriter: Failed to write channel meta data, error = %s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
690 : : status.message.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
691 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
692 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
693 : : }
694 [ + - ]: 729 : }
695 : :
696 [ + - ]: 699 : impl_->writer->close();
697 [ + - ]: 699 : impl_->writer->terminate();
698 : 699 : impl_->writer.reset();
699 [ + - ]: 699 : impl_->output.close();
700 : :
701 [ + - - + ]: 699 : if VUNLIKELY (impl_->output.fail()) {
702 : 0 : set_fail();
703 : : }
704 : :
705 : 699 : std::error_code footer_ec;
706 : 699 : const auto file_size = std::filesystem::file_size(impl_->active_path, footer_ec);
707 : :
708 [ + - - + : 699 : if VUNLIKELY (footer_ec || file_size < kMcapMagic.size()) {
- + ]
709 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
710 : : } else {
711 [ + - ]: 699 : std::ifstream tail_check(impl_->active_path, std::ios::binary);
712 : 699 : char magic[8] = {};
713 : :
714 [ + - ]: 699 : tail_check.seekg(-static_cast<std::streamoff>(kMcapMagic.size()), std::ios::end);
715 [ + - ]: 699 : tail_check.read(magic, static_cast<std::streamsize>(kMcapMagic.size()));
716 : :
717 [ + - + - : 699 : if VUNLIKELY (!tail_check || std::string_view(magic, kMcapMagic.size()) != kMcapMagic) {
- + - + ]
718 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
719 : : }
720 : 699 : }
721 : :
722 [ + + + + : 699 : if (impl_->is_split_mode.load(std::memory_order_relaxed) && impl_->config.max_split_count > 0) {
+ + ]
723 [ + - ]: 486 : impl_->split_info_list.emplace_back(Impl::SplitInfo{impl_->current_row, std::move(impl_->url_map)});
724 : : } else {
725 : 213 : impl_->url_map.clear();
726 : : }
727 : 699 : impl_->ser_map.clear();
728 [ + - ]: 699 : url_loss_map_ref().clear();
729 : :
730 : 699 : impl_->current_row = 0;
731 : 699 : impl_->current_size = 0;
732 : 699 : impl_->has_oversize = false;
733 : :
734 : 699 : impl_->in_cached.store(false, std::memory_order_relaxed);
735 : 699 : impl_->cached_size.store(0, std::memory_order_relaxed);
736 : 699 : }
737 : :
738 : 18 : bool VCAPWriter::merge_schema(SchemaData& schema_data) {
739 : : const auto resolved_schema_type =
740 : 18 : SchemaData::resolve_type(schema_data.schema_type, schema_data.name, schema_data.encoding);
741 : 18 : schema_data.schema_type = resolved_schema_type;
742 : :
743 [ + + ]: 18 : if VUNLIKELY (schema_data.name.empty()) {
744 : 1 : return true;
745 : : }
746 : :
747 [ + + + + : 17 : if (schema_data.encoding.empty() && SchemaData::is_real_type(resolved_schema_type)) {
+ + ]
748 [ + - ]: 3 : schema_data.encoding = std::string(SchemaData::convert_type(resolved_schema_type));
749 : : }
750 : :
751 [ + - ]: 17 : std::string schema_key = schema_data.name;
752 [ + - ]: 17 : schema_key.push_back('\x1F');
753 [ + - ]: 17 : schema_key.append(SchemaData::convert_type(resolved_schema_type));
754 : :
755 : 17 : std::string unknown_schema_key;
756 [ + - ]: 17 : auto schema_iter = impl_->total_schema_map.find(schema_key);
757 : :
758 [ + + + + : 17 : if (schema_iter == impl_->total_schema_map.end() && SchemaData::is_real_type(resolved_schema_type)) {
+ + ]
759 [ + - ]: 13 : unknown_schema_key = schema_data.name;
760 [ + - ]: 13 : unknown_schema_key.push_back('\x1F');
761 [ + - ]: 13 : schema_iter = impl_->total_schema_map.find(unknown_schema_key);
762 : : }
763 : :
764 [ + + ]: 17 : if (schema_iter == impl_->total_schema_map.end()) {
765 [ + - ]: 13 : impl_->total_schema_map.emplace(schema_key, schema_data);
766 : : } else {
767 : 4 : auto& current = schema_iter->second;
768 : :
769 [ + - + + : 4 : if VUNLIKELY ((!schema_data.encoding.empty() && !current.encoding.empty() &&
+ + - + +
- + - + +
+ + + + +
+ + + + -
+ + + + -
+ - + +
+ ]
770 : : current.encoding != schema_data.encoding) ||
771 : : (!schema_data.data.empty() && !current.data.empty() && current.data != schema_data.data) ||
772 : : (SchemaData::is_real_type(resolved_schema_type) && SchemaData::is_real_type(current.schema_type) &&
773 : : current.schema_type != resolved_schema_type)) {
774 [ + - + - : 4 : CLOG_E("VCAPWriter: Conflicting schema pushed for [%s].", schema_data.name.c_str());
+ - ]
775 : 2 : return false;
776 : : }
777 : :
778 [ + + + - : 2 : if (current.encoding.empty() && !schema_data.encoding.empty()) {
+ + ]
779 [ + - ]: 1 : current.encoding = schema_data.encoding;
780 : : }
781 : :
782 [ + + + - : 2 : if (current.data.empty() && !schema_data.data.empty()) {
+ + ]
783 : 1 : current.data = schema_data.data;
784 : : }
785 : :
786 [ + + + - : 2 : if (!SchemaData::is_real_type(current.schema_type) && SchemaData::is_real_type(resolved_schema_type)) {
+ + ]
787 : 1 : current.schema_type = resolved_schema_type;
788 : : }
789 : :
790 [ + - ]: 2 : schema_data = current;
791 : :
792 [ + + + - : 2 : if (schema_iter->first != schema_key && current.schema_type == resolved_schema_type) {
+ + ]
793 [ + - ]: 1 : impl_->total_schema_map.erase(schema_iter);
794 [ + - ]: 1 : schema_iter = impl_->total_schema_map.emplace(schema_key, schema_data).first;
795 : : }
796 : : }
797 : :
798 : 15 : return true;
799 : 17 : }
800 : :
801 : 752 : bool VCAPWriter::load_schema(const std::string& ser_type, SchemaType& schema_type, SchemaData& schema_data) {
802 : 752 : schema_data = SchemaData{};
803 : :
804 [ - + ]: 752 : if VUNLIKELY (ser_type.empty()) {
805 : : return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
806 : : }
807 : :
808 [ + - ]: 752 : std::string schema_key = ser_type;
809 [ + - ]: 752 : schema_key.push_back('\x1F');
810 [ + - ]: 752 : schema_key.append(SchemaData::convert_type(schema_type));
811 : :
812 : 752 : std::string unknown_schema_key;
813 : 752 : auto schema_iter = impl_->total_schema_map.end();
814 : :
815 [ + + ]: 752 : if (schema_type != SchemaType::kUnknown) {
816 [ + - ]: 743 : schema_iter = impl_->total_schema_map.find(schema_key);
817 : :
818 [ + + ]: 743 : if (schema_iter == impl_->total_schema_map.end()) {
819 [ + - ]: 727 : unknown_schema_key = ser_type;
820 [ + - ]: 727 : unknown_schema_key.push_back('\x1F');
821 [ + - ]: 727 : schema_iter = impl_->total_schema_map.find(unknown_schema_key);
822 : : }
823 : : } else {
824 [ + - + - ]: 18 : const auto prefix = ser_type + std::string("\x1F");
825 : :
826 [ + + ]: 18 : for (auto iter = impl_->total_schema_map.begin(); iter != impl_->total_schema_map.end(); ++iter) {
827 [ + + ]: 9 : if (!Helpers::has_startwith(iter->first, prefix)) {
828 : 8 : continue;
829 : : }
830 : :
831 [ - + ]: 1 : if (schema_iter != impl_->total_schema_map.end()) {
832 : : schema_iter = impl_->total_schema_map.end(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
833 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
834 : : }
835 : :
836 : 1 : schema_iter = iter;
837 : : }
838 : 9 : }
839 : :
840 [ + + ]: 752 : if VLIKELY (schema_iter != impl_->total_schema_map.end()) {
841 [ + - ]: 17 : schema_data.name = schema_iter->second.name;
842 [ + - ]: 17 : schema_data.encoding = schema_iter->second.encoding;
843 : 17 : schema_data.schema_type = schema_iter->second.schema_type;
844 : 17 : schema_data.data.shallow_copy(schema_iter->second.data);
845 [ - + ]: 735 : } else if (impl_->schema_plugin_interface) {
846 : : schema_data =
847 : : impl_->schema_plugin_interface->search_schema(ser_type, schema_type); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
848 [ + + ]: 735 : } else if (impl_->schema_callback) {
849 [ + - ]: 10 : schema_data = impl_->schema_callback(ser_type, schema_type);
850 : : }
851 : :
852 : 752 : schema_type = SchemaData::resolve_type(schema_type, ser_type, schema_data.encoding);
853 : 752 : schema_data.schema_type = SchemaData::resolve_type(schema_data.schema_type, ser_type, schema_data.encoding);
854 : :
855 [ + + + - ]: 752 : if (schema_type != SchemaType::kUnknown && schema_data.schema_type != SchemaType::kUnknown &&
856 [ + + ]: 744 : schema_type != schema_data.schema_type) {
857 [ + - + - : 6 : CLOG_E("VCAPWriter: Schema family mismatch for [%s], requested = %d, resolved = %d.", ser_type.c_str(),
+ - ]
858 : : static_cast<int>(schema_type), static_cast<int>(schema_data.schema_type));
859 : 3 : return false;
860 : : }
861 : :
862 [ + + + + : 749 : if (schema_type != SchemaType::kUnknown && schema_data.encoding.empty()) {
+ + ]
863 [ + - ]: 716 : schema_data.encoding = std::string(SchemaData::convert_type(schema_type));
864 : : }
865 : :
866 [ + + - + ]: 749 : if (schema_data.schema_type == SchemaType::kUnknown && schema_type != SchemaType::kUnknown) {
867 : : schema_data.schema_type = schema_type; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
868 : : }
869 : :
870 [ + + ]: 749 : if (!schema_data.name.empty()) {
871 [ + - ]: 25 : std::string resolved_schema_key = ser_type;
872 [ + - ]: 25 : resolved_schema_key.push_back('\x1F');
873 [ + - ]: 25 : resolved_schema_key.append(SchemaData::convert_type(schema_data.schema_type));
874 : :
875 [ + + + - : 25 : if VLIKELY (schema_iter != impl_->total_schema_map.end() && schema_iter->first == resolved_schema_key) {
+ + ]
876 [ - + - - : 17 : if (schema_iter->second.encoding.empty() && !schema_data.encoding.empty()) {
- + ]
877 [ # # ]: 0 : schema_iter->second.encoding = schema_data.encoding;
878 : : }
879 : :
880 : 17 : schema_iter->second.schema_type = schema_data.schema_type;
881 : : } else {
882 [ + - ]: 8 : SchemaData stored_schema = schema_data;
883 : :
884 [ - + - - : 8 : if (schema_iter != impl_->total_schema_map.end() && schema_iter->second.schema_type == SchemaType::kUnknown &&
- + ]
885 [ # # ]: 0 : schema_data.schema_type != SchemaType::kUnknown) {
886 : : impl_->total_schema_map.erase(schema_iter); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
887 : : }
888 : :
889 : : const auto stored_iter =
890 [ + - ]: 8 : impl_->total_schema_map.insert_or_assign(resolved_schema_key, std::move(stored_schema)).first;
891 : 8 : schema_data.data.shallow_copy(stored_iter->second.data);
892 : 8 : }
893 : 25 : }
894 : :
895 : 749 : return true;
896 : 752 : }
897 : :
898 : 770 : bool VCAPWriter::write(const std::string& url, const std::string& ser_type, SchemaType schema_type,
899 : : ActionType action_type, const Bytes& data, int64_t microseconds_timestamp) {
900 [ - + ]: 770 : if VUNLIKELY (!impl_->writer) {
901 : : VLOG_E("VCAPWriter: Writer is not open."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
902 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
903 : : }
904 : :
905 [ + + + + : 770 : if VUNLIKELY (impl_->is_split_mode.load(std::memory_order_relaxed) && !impl_->split_first) {
+ + ]
906 : 78 : impl_->split_first = true;
907 : :
908 [ + - ]: 78 : std::lock_guard split_lock(impl_->split_mtx);
909 : :
910 [ + + + + : 149 : if (!impl_->split_before && impl_->split_callback && impl_->split_index.load(std::memory_order_relaxed) == 0) {
+ - + + ]
911 [ + - ]: 71 : impl_->split_callback(0, impl_->split_filename);
912 : : }
913 : 78 : }
914 : :
915 : 770 : mcap::Status status;
916 : :
917 : 770 : bool do_split = false;
918 : :
919 : : // split
920 : :
921 [ + + + + : 770 : if (impl_->is_split_mode.load(std::memory_order_relaxed) && !impl_->url_map.empty()) {
+ + ]
922 [ + + + + ]: 836 : if (impl_->config.split_by_time > 0 &&
923 : 273 : (microseconds_timestamp - impl_->config.begin_time * 1000) >
924 : 273 : impl_->config.split_by_time * 1000 *
925 [ + + ]: 546 : (static_cast<int64_t>(impl_->split_index.load(std::memory_order_relaxed)) + 1)) {
926 : 272 : do_split = true;
927 [ + + + - : 581 : } else if (impl_->config.split_by_time <= 0 && impl_->config.split_by_size > 0 &&
+ + ]
928 [ + - ]: 290 : (impl_->current_size + static_cast<int64_t>(data.size())) > impl_->config.split_by_size) {
929 : 290 : do_split = true;
930 : : } else {
931 : 1 : do_split = false;
932 : : }
933 : :
934 [ + + ]: 563 : if VUNLIKELY (do_split) {
935 [ + - ]: 562 : std::lock_guard split_lock(impl_->split_mtx);
936 : :
937 : 562 : impl_->split_index.fetch_add(1, std::memory_order_relaxed);
938 [ + - ]: 562 : const auto time_current = impl_->time_start + std::chrono::milliseconds(microseconds_timestamp / 1000U);
939 [ + - ]: 562 : const bool same_millisecond = time_current == impl_->time_current;
940 : 562 : impl_->time_current = time_current;
941 : :
942 [ + + ]: 562 : if (impl_->config.split_name_by_time) {
943 [ + - + - ]: 268 : std::string filename = get_format_date(&impl_->time_current, true) + ".vcap";
944 : :
945 [ + + + - : 268 : if VUNLIKELY (same_millisecond || std::find(impl_->split_file_list.begin(), impl_->split_file_list.end(),
- + + + ]
946 : : filename) != impl_->split_file_list.end()) {
947 [ + - ]: 132 : filename.insert(filename.find_last_of('.'),
948 [ + - + - ]: 396 : "." + std::to_string(impl_->split_index.load(std::memory_order_relaxed) + 1));
949 : : }
950 : :
951 [ + + + - : 268 : impl_->split_filename = impl_->base_dir.empty() ? filename : impl_->base_dir + "/" + filename;
+ - + - +
+ - - ]
952 : 268 : } else {
953 : 294 : impl_->split_filename =
954 [ + - + - : 882 : impl_->base_name + "." + std::to_string(impl_->split_index.load(std::memory_order_relaxed) + 1) + ".vcap";
+ - + - ]
955 : : }
956 : :
957 [ + + + - : 562 : if (impl_->split_before && impl_->split_callback) {
+ + ]
958 [ + - ]: 2 : impl_->split_callback(impl_->split_index.load(std::memory_order_relaxed), impl_->split_filename);
959 : : }
960 : :
961 [ + - ]: 562 : close_segment();
962 : :
963 [ + - - + ]: 562 : if VUNLIKELY (!open_split(impl_->split_filename)) {
964 : 0 : set_fail();
965 : 0 : impl_->split_index.fetch_sub(1, std::memory_order_relaxed);
966 : :
967 [ # # # # ]: 0 : if VUNLIKELY (!write_filex(false)) {
968 : 0 : set_fail();
969 : : }
970 : :
971 : 0 : return false;
972 : : }
973 : :
974 : 7 : auto discard_new_split = [this]() {
975 [ + - ]: 1 : close_segment();
976 : 1 : impl_->split_file_list.pop_back();
977 [ + - ]: 1 : if (impl_->config.max_split_count > 0) {
978 : 1 : impl_->split_info_list.pop_back();
979 : : }
980 : :
981 : : #ifdef _WIN32
982 : : const auto new_split_path = impl_->split_output_dir /
983 : : std::filesystem::path(Helpers::string_to_wstring(impl_->split_filename)).filename();
984 : : #else
985 [ + - + - : 2 : const auto new_split_path = impl_->split_output_dir / std::filesystem::path(impl_->split_filename).filename();
+ - ]
986 : : #endif
987 : 1 : std::error_code remove_ec;
988 : 1 : std::filesystem::remove(new_split_path, remove_ec);
989 : 1 : impl_->split_index.fetch_sub(1, std::memory_order_relaxed);
990 : 1 : };
991 : :
992 : 562 : std::string oldest_split_file;
993 : 562 : Impl::SplitInfo oldest_split_info;
994 : :
995 [ + + + + ]: 1006 : if (impl_->config.max_split_count > 0 &&
996 [ + + ]: 444 : static_cast<int64_t>(impl_->split_file_list.size()) > impl_->config.max_split_count) {
997 : 234 : oldest_split_file = std::move(impl_->split_file_list.front());
998 [ + - ]: 234 : impl_->split_file_list.erase(impl_->split_file_list.begin());
999 : 234 : oldest_split_info = std::move(impl_->split_info_list.front());
1000 [ + - ]: 234 : impl_->split_info_list.erase(impl_->split_info_list.begin());
1001 : : }
1002 : :
1003 [ + - - + ]: 562 : if VUNLIKELY (!write_filex(false)) {
1004 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1005 [ # # ]: 0 : discard_new_split();
1006 : :
1007 [ # # ]: 0 : if (!oldest_split_file.empty()) {
1008 [ # # ]: 0 : impl_->split_file_list.insert(impl_->split_file_list.begin(), std::move(oldest_split_file));
1009 [ # # ]: 0 : impl_->split_info_list.insert(impl_->split_info_list.begin(), std::move(oldest_split_info));
1010 : : }
1011 : :
1012 [ # # # # ]: 0 : if VUNLIKELY (!write_filex(false)) {
1013 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1014 : : }
1015 : :
1016 : 0 : return false;
1017 : : }
1018 : :
1019 [ + + ]: 562 : if (!oldest_split_file.empty()) {
1020 : : #ifdef _WIN32
1021 : : const auto oldest_path =
1022 : : impl_->split_output_dir / std::filesystem::path(Helpers::string_to_wstring(oldest_split_file));
1023 : : #else
1024 [ + - + - ]: 234 : const auto oldest_path = impl_->split_output_dir / std::filesystem::path(oldest_split_file);
1025 : : #endif
1026 : 234 : std::error_code remove_ec;
1027 : 234 : std::filesystem::remove(oldest_path, remove_ec);
1028 : :
1029 [ + + ]: 234 : if VUNLIKELY (remove_ec) {
1030 [ + - + - : 2 : CLOG_W("VCAPWriter: Failed to remove oldest split file [%s]: %s.", oldest_split_file.c_str(),
+ - + - +
- - - ]
1031 : : remove_ec.message().c_str());
1032 : 1 : set_fail();
1033 [ + - ]: 1 : discard_new_split();
1034 [ + - ]: 1 : impl_->split_file_list.insert(impl_->split_file_list.begin(), std::move(oldest_split_file));
1035 [ + - ]: 1 : impl_->split_info_list.insert(impl_->split_info_list.begin(), std::move(oldest_split_info));
1036 : :
1037 [ + - - + ]: 1 : if VUNLIKELY (!write_filex(false)) {
1038 : : set_fail(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1039 : : }
1040 : :
1041 : 1 : return false;
1042 : : }
1043 [ + + ]: 234 : }
1044 : :
1045 [ + + + + : 561 : if (!impl_->split_before && impl_->split_callback) {
+ + ]
1046 [ + - ]: 1108 : impl_->split_callback(impl_->split_index.load(std::memory_order_relaxed), impl_->split_filename);
1047 : : }
1048 [ + + + + : 564 : }
+ + ]
1049 : : }
1050 : :
1051 : : // insert url
1052 [ + - + - ]: 769 : const std::string& channel_key = make_channel_key(impl_->write_channel_key, url, convert_action(action_type));
1053 [ + - ]: 769 : auto total_url_iter_ret = impl_->total_url_map.try_emplace(channel_key, Impl::UrlMsgInfo());
1054 [ + - ]: 769 : auto url_iter_ret = impl_->url_map.try_emplace(channel_key, Impl::UrlMsgInfo());
1055 : :
1056 : 20 : auto discard_new_url_entries = [this, &url_iter_ret, &total_url_iter_ret]() {
1057 [ + + ]: 6 : if (url_iter_ret.second) {
1058 : 2 : impl_->url_map.erase(url_iter_ret.first);
1059 : : }
1060 : :
1061 [ + + ]: 6 : if (total_url_iter_ret.second) {
1062 : 2 : impl_->total_url_map.erase(total_url_iter_ret.first);
1063 : : }
1064 : 775 : };
1065 : :
1066 : 769 : Impl::UrlMsgInfo& total_url_msg_info = total_url_iter_ret.first->second;
1067 : :
1068 : 769 : Impl::UrlMsgInfo& url_msg_info = url_iter_ret.first->second;
1069 : 769 : auto resolved_schema_type = SchemaData::resolve_type(schema_type, ser_type);
1070 [ + + ]: 769 : const std::string& next_ser_type = total_url_msg_info.ser_type.empty() ? ser_type : total_url_msg_info.ser_type;
1071 : 769 : SchemaType next_schema_type = total_url_msg_info.schema_type;
1072 : :
1073 [ + + ]: 769 : if (total_url_iter_ret.second) {
1074 : 171 : next_schema_type = resolved_schema_type;
1075 [ + + + + : 598 : } else if VUNLIKELY (!ser_type.empty() && next_ser_type != ser_type) {
+ + ]
1076 [ + - + - : 6 : CLOG_E("VCAPWriter: URL [%s] ser changed from [%s] to [%s].", url.c_str(), next_ser_type.c_str(), ser_type.c_str());
+ - ]
1077 [ + - ]: 3 : discard_new_url_entries();
1078 : 3 : return false;
1079 : : }
1080 : :
1081 : 766 : SchemaData schema_data;
1082 : 766 : std::string split_ser_type;
1083 [ + + ]: 766 : const auto schema_ser_source = ser_type.empty() ? std::string_view{next_ser_type} : std::string_view{ser_type};
1084 : 766 : SchemaType schema_storage_type = SchemaData::resolve_type(schema_type, schema_ser_source);
1085 : 766 : bool has_split_method_schema = false;
1086 : :
1087 [ + + + + ]: 756 : if ((action_type == ActionType::kClientRequest || action_type == ActionType::kClientResponse ||
1088 [ + + + + : 1533 : action_type == ActionType::kServerRequest || action_type == ActionType::kServerResponse) &&
+ + ]
1089 [ + - ]: 14 : !schema_ser_source.empty()) {
1090 : 14 : const auto split_pos = schema_ser_source.find('|');
1091 : :
1092 [ + - ]: 14 : if (split_pos != std::string_view::npos) {
1093 [ + - ]: 14 : auto payload_ser_type = schema_ser_source.substr(0, split_pos);
1094 : :
1095 [ + + + + ]: 14 : if (action_type == ActionType::kClientResponse || action_type == ActionType::kServerResponse) {
1096 [ + - ]: 3 : payload_ser_type = schema_ser_source.substr(split_pos + 1);
1097 : : }
1098 : :
1099 [ + - ]: 14 : if (!payload_ser_type.empty()) {
1100 [ + - ]: 14 : split_ser_type.assign(payload_ser_type.begin(), payload_ser_type.end());
1101 : 14 : schema_storage_type = SchemaData::resolve_type(schema_type, payload_ser_type);
1102 : 14 : has_split_method_schema = true;
1103 : : }
1104 : : }
1105 : : }
1106 : :
1107 [ + + ]: 766 : const std::string& schema_ser_type = has_split_method_schema ? split_ser_type : next_ser_type;
1108 : :
1109 [ + + ]: 766 : if (!next_ser_type.empty()) {
1110 [ + - + + ]: 752 : if VUNLIKELY (!load_schema(schema_ser_type, schema_storage_type, schema_data)) {
1111 [ + - ]: 3 : discard_new_url_entries();
1112 : 3 : return false;
1113 : : }
1114 : :
1115 : 749 : schema_storage_type = SchemaData::resolve_type(schema_storage_type, schema_ser_type, schema_data.encoding);
1116 : :
1117 [ + + ]: 749 : if (has_split_method_schema) {
1118 [ + + ]: 14 : if (schema_storage_type != SchemaType::kUnknown) {
1119 [ - + ]: 6 : if (next_schema_type == SchemaType::kUnknown) {
1120 : : next_schema_type = schema_storage_type; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1121 [ - + ]: 6 : } else if (next_schema_type != schema_storage_type) {
1122 : : next_schema_type = SchemaType::kUnknown; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1123 : : }
1124 : : }
1125 : : } else {
1126 [ + + ]: 735 : if (resolved_schema_type == SchemaType::kUnknown) {
1127 : : const auto inferred_schema_type =
1128 : 2 : SchemaData::resolve_type(schema_data.schema_type, schema_data.name, schema_data.encoding);
1129 : :
1130 [ + - ]: 2 : if (inferred_schema_type != SchemaType::kUnknown) {
1131 : 2 : resolved_schema_type = inferred_schema_type;
1132 : : }
1133 : : }
1134 : :
1135 [ + - ]: 735 : if (resolved_schema_type != SchemaType::kUnknown) {
1136 [ + + ]: 735 : if (next_schema_type == SchemaType::kUnknown) {
1137 : 2 : next_schema_type = resolved_schema_type;
1138 [ - + ]: 733 : } else if VUNLIKELY (next_schema_type != resolved_schema_type) {
1139 : : CLOG_E("VCAPWriter: URL [%s] schema changed from [%d] to [%d].", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1140 : : url.c_str(), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1141 : : static_cast<int>(next_schema_type), static_cast<int>(resolved_schema_type));
1142 : : discard_new_url_entries(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1143 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1144 : : }
1145 : : }
1146 : : }
1147 : : } else {
1148 : 14 : schema_storage_type = SchemaData::resolve_type(schema_storage_type, schema_ser_type, schema_data.encoding);
1149 : : }
1150 : :
1151 [ + - ]: 763 : std::string storage_schema_key = schema_ser_type;
1152 [ + - ]: 763 : storage_schema_key.push_back('\x1F');
1153 [ + - ]: 763 : storage_schema_key.append(SchemaData::convert_type(schema_storage_type));
1154 : :
1155 [ + + ]: 763 : if (total_url_iter_ret.second) {
1156 : 169 : total_url_msg_info.index = impl_->total_url_map.size() - 1;
1157 [ + - ]: 169 : total_url_msg_info.url = url;
1158 [ + - ]: 169 : impl_->total_url_list.emplace_back(channel_key);
1159 : : }
1160 : :
1161 [ + + + + ]: 1512 : if (!schema_ser_type.empty() &&
1162 [ + + - + ]: 749 : (schema_storage_type == SchemaType::kProtobuf || schema_storage_type == SchemaType::kFlatbuffers)) {
1163 [ + - + + ]: 23 : if (impl_->ser_map.find(storage_schema_key) == impl_->ser_map.end()) {
1164 [ + - + - : 21 : if (!schema_data.name.empty() && !schema_data.encoding.empty() && !schema_data.data.empty()) {
+ - + - ]
1165 : 21 : mcap::Schema schema;
1166 : 21 : schema.id = static_cast<mcap::SchemaId>(impl_->ser_map.size() + 1);
1167 [ + - ]: 21 : schema.name = schema_data.name;
1168 [ + - ]: 21 : schema.encoding = schema_data.encoding;
1169 [ + - ]: 21 : schema.data.assign(reinterpret_cast<const std::byte*>(schema_data.data.data()),
1170 : 21 : reinterpret_cast<const std::byte*>(schema_data.data.data()) + schema_data.data.size());
1171 : :
1172 [ + - ]: 21 : impl_->writer->addSchema(schema);
1173 : :
1174 [ + - ]: 21 : impl_->ser_map.emplace(storage_schema_key, schema.id);
1175 : 21 : }
1176 : : }
1177 : : }
1178 : :
1179 [ + + ]: 763 : if (url_iter_ret.second) {
1180 : 729 : url_msg_info.index = impl_->url_map.size() - 1;
1181 : :
1182 [ + + + + : 729 : if (action_type == ActionType::kClientRequest || action_type == ActionType::kClientResponse ||
+ + ]
1183 [ + + ]: 716 : action_type == ActionType::kServerRequest || action_type == ActionType::kServerResponse) {
1184 [ + - ]: 14 : impl_->write_url_type = "Method";
1185 [ + + + + ]: 715 : } else if (action_type == ActionType::kPublish || action_type == ActionType::kSubscribe) {
1186 [ + - ]: 700 : impl_->write_url_type = "Event";
1187 [ + + + + ]: 15 : } else if (action_type == ActionType::kSet || action_type == ActionType::kGet) {
1188 [ + - ]: 14 : impl_->write_url_type = "Field";
1189 : : } else {
1190 [ + - ]: 1 : impl_->write_url_type = "Unknown";
1191 : : }
1192 : :
1193 : 729 : mcap::SchemaId schema_id = 0;
1194 : :
1195 [ + + + + : 729 : if (impl_->write_url_type != "Method" && !next_ser_type.empty()) {
+ + ]
1196 [ + - ]: 709 : auto iter = impl_->ser_map.find(storage_schema_key);
1197 : :
1198 [ + + ]: 709 : if (iter != impl_->ser_map.end()) {
1199 : 17 : schema_id = iter->second;
1200 : : }
1201 : : }
1202 : :
1203 : 729 : mcap::Channel channel;
1204 : 729 : channel.id = url_msg_info.index + 1;
1205 : 729 : channel.schemaId = schema_id;
1206 [ + - ]: 729 : channel.topic = url;
1207 [ + - + - : 729 : channel.metadata["action"] = std::string(convert_action(action_type));
+ - + - ]
1208 : :
1209 [ + + ]: 729 : if (impl_->write_url_type != "Method") {
1210 [ + + ]: 715 : if (!schema_data.encoding.empty()) {
1211 [ + - ]: 709 : channel.messageEncoding = schema_data.encoding;
1212 [ + + ]: 6 : } else if (next_schema_type != SchemaType::kUnknown) {
1213 : : channel.messageEncoding = SchemaData::convert_type(next_schema_type); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1214 : : }
1215 : : }
1216 : :
1217 : 729 : url_msg_info.index = channel.id - 1;
1218 : :
1219 [ + - ]: 729 : impl_->writer->addChannel(channel);
1220 : :
1221 [ + - ]: 729 : url_msg_info.url = url;
1222 [ + - ]: 729 : url_msg_info.url_type = impl_->write_url_type;
1223 [ + - ]: 729 : total_url_msg_info.url_type = impl_->write_url_type;
1224 : :
1225 [ + + ]: 729 : if VLIKELY (!next_ser_type.empty()) {
1226 [ + - ]: 723 : url_msg_info.ser_type = next_ser_type;
1227 [ + - ]: 723 : total_url_msg_info.ser_type = next_ser_type;
1228 : : }
1229 : :
1230 : 729 : url_msg_info.schema_type = next_schema_type;
1231 : 729 : total_url_msg_info.schema_type = next_schema_type;
1232 : 729 : url_msg_info.action_type = action_type;
1233 : 729 : total_url_msg_info.action_type = action_type;
1234 : 729 : } else {
1235 [ + + + + : 34 : if VUNLIKELY (total_url_msg_info.ser_type.empty() && !next_ser_type.empty()) {
+ + ]
1236 [ + - ]: 1 : total_url_msg_info.ser_type = next_ser_type;
1237 [ + - ]: 1 : url_msg_info.ser_type = next_ser_type;
1238 : : }
1239 : :
1240 : 34 : total_url_msg_info.schema_type = next_schema_type;
1241 : 34 : url_msg_info.schema_type = next_schema_type;
1242 : : }
1243 : :
1244 : : // update count
1245 : 763 : ++url_msg_info.count;
1246 : 763 : ++total_url_msg_info.count;
1247 : :
1248 : : // update size
1249 : 763 : url_msg_info.size += data.size();
1250 : 763 : total_url_msg_info.size += data.size();
1251 : :
1252 [ + + + + : 763 : if (action_type == ActionType::kPublish || action_type == ActionType::kSubscribe || action_type == ActionType::kSet ||
+ + + + ]
1253 : : action_type == ActionType::kGet) {
1254 : 748 : double time_duration = 0;
1255 : :
1256 [ + + ]: 748 : if (total_url_msg_info.first_timestamp < 0) {
1257 : 154 : total_url_msg_info.first_timestamp = microseconds_timestamp;
1258 : : }
1259 : :
1260 : 748 : total_url_msg_info.last_timestamp = microseconds_timestamp;
1261 : 748 : time_duration = (total_url_msg_info.last_timestamp - total_url_msg_info.first_timestamp) / 1000'000.0;
1262 : :
1263 [ + + ]: 748 : if (time_duration > 0) {
1264 : 592 : total_url_msg_info.freq = total_url_msg_info.count / time_duration;
1265 : : } else {
1266 : 156 : total_url_msg_info.freq = 0;
1267 : : }
1268 : :
1269 [ + + ]: 748 : if (url_msg_info.first_timestamp < 0) {
1270 : 714 : url_msg_info.first_timestamp = microseconds_timestamp;
1271 : : }
1272 : :
1273 : 748 : url_msg_info.last_timestamp = microseconds_timestamp;
1274 : 748 : time_duration = (url_msg_info.last_timestamp - url_msg_info.first_timestamp) / 1000'000.0;
1275 : :
1276 [ + + ]: 748 : if (time_duration > 0) {
1277 : 33 : url_msg_info.freq = url_msg_info.count / time_duration;
1278 : : } else {
1279 : 715 : url_msg_info.freq = 0;
1280 : : }
1281 : : }
1282 : :
1283 : : // insert data
1284 : :
1285 : 763 : impl_->last_timestamp = std::max(impl_->last_timestamp, microseconds_timestamp);
1286 : :
1287 : 763 : mcap::Message message;
1288 : 763 : message.channelId = url_msg_info.index + 1;
1289 : 763 : message.sequence = url_msg_info.count;
1290 : 763 : message.logTime = microseconds_timestamp * 1000U + impl_->start_timestamp * 1000'000;
1291 : 763 : message.publishTime = message.logTime;
1292 : 763 : message.data = reinterpret_cast<const std::byte*>(data.data());
1293 : 763 : message.dataSize = data.size();
1294 : :
1295 [ + - ]: 763 : status = impl_->writer->write(message);
1296 : :
1297 [ + - + - : 763 : if VUNLIKELY (!status.ok() || !impl_->output) {
- + - + ]
1298 : : CLOG_W("VCAPWriter: Failed to write message data, error = %s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1299 : : status.message.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1300 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1301 : : }
1302 : :
1303 : 763 : impl_->cached_size.fetch_add(data.size(), std::memory_order_relaxed);
1304 : :
1305 : 763 : ++impl_->current_row;
1306 : 763 : impl_->current_size += data.size();
1307 : :
1308 : 763 : ++impl_->total_current_row;
1309 : 763 : impl_->total_current_size += data.size();
1310 : 763 : impl_->total_timestamp = std::max(impl_->total_timestamp, microseconds_timestamp);
1311 : :
1312 : 763 : return true;
1313 : 770 : }
1314 : :
1315 : 772 : bool VCAPWriter::write_filex(bool complete) {
1316 : : try {
1317 : : #ifdef _WIN32
1318 : : std::filesystem::path file_path(Helpers::string_to_wstring(impl_->path));
1319 : : #else
1320 [ + - ]: 772 : std::filesystem::path file_path(impl_->path);
1321 : : #endif
1322 : :
1323 : 772 : nlohmann::ordered_json json;
1324 : :
1325 : 772 : auto message_count = impl_->total_current_row;
1326 [ + + ]: 772 : if (impl_->config.max_split_count > 0) {
1327 : 569 : message_count = impl_->current_row;
1328 : :
1329 [ + + ]: 2459 : for (const auto& split_info : impl_->split_info_list) {
1330 : 1890 : message_count += split_info.row;
1331 : : }
1332 : : }
1333 : :
1334 [ + - ]: 772 : json["VLinkHeader"] = {
1335 : : {"major", VLINK_VERSION_MAJOR}, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1336 : : {"minor", VLINK_VERSION_MINOR}, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1337 : : {"patch", VLINK_VERSION_PATCH}, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1338 : : {"count", message_count},
1339 : 772 : {"duration", impl_->total_timestamp},
1340 : : {"accuracy", "MicroSecond"},
1341 : 772 : {"compress", impl_->enable_compressed ? "zstd" : "None"},
1342 : 772 : {"process", impl_->app_name},
1343 [ + - ]: 772 : {"date", get_format_date(&impl_->time_start)},
1344 : 772 : {"tag", impl_->tag_name},
1345 [ + + ]: 772 : {"split_by_size", impl_->config.split_by_time > 0 ? 0 : impl_->config.split_by_size},
1346 : 772 : {"split_by_time", impl_->config.split_by_time},
1347 : : {"complete", complete},
1348 : 772 : {"timezone", impl_->timezone_diff},
1349 : 772 : {"start_timestamp", impl_->start_timestamp},
1350 [ + - + - : 43232 : };
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - -
+ + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ + + + +
+ + + + +
+ + + + +
+ + + + +
+ + + + +
+ + + + +
+ + - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - ]
1351 : :
1352 : 772 : nlohmann::ordered_json url_json;
1353 : :
1354 : : {
1355 [ + - + - ]: 772 : std::lock_guard lock(sample_mutex());
1356 : :
1357 [ + + ]: 1466 : for (const auto& channel_key : impl_->total_url_list) {
1358 [ + - ]: 694 : const auto& ext_info = impl_->total_url_map[channel_key];
1359 : 694 : auto count = ext_info.count;
1360 : 694 : auto size = ext_info.size;
1361 : 694 : auto freq = ext_info.freq;
1362 : :
1363 [ + + ]: 694 : if (impl_->config.max_split_count > 0) {
1364 : 527 : count = 0;
1365 : 527 : size = 0;
1366 : 527 : int64_t first_timestamp = -1;
1367 : 527 : int64_t last_timestamp = -1;
1368 : :
1369 : 2417 : auto merge_url_info = [&](const auto& url_map) {
1370 [ + - ]: 2417 : const auto iter = url_map.find(channel_key);
1371 : :
1372 [ + + ]: 2417 : if (iter == url_map.end()) {
1373 : 527 : return;
1374 : : }
1375 : :
1376 : 1890 : const auto& url_info = iter->second;
1377 : 1890 : count += url_info.count;
1378 : 1890 : size += url_info.size;
1379 : :
1380 [ + - + + : 1890 : if (url_info.first_timestamp >= 0 && (first_timestamp < 0 || url_info.first_timestamp < first_timestamp)) {
- + ]
1381 : 439 : first_timestamp = url_info.first_timestamp;
1382 : : }
1383 : :
1384 : 1890 : last_timestamp = std::max(last_timestamp, url_info.last_timestamp);
1385 : 527 : };
1386 : :
1387 [ + + ]: 2417 : for (const auto& split_info : impl_->split_info_list) {
1388 [ + - ]: 1890 : merge_url_info(split_info.url_map);
1389 : : }
1390 : :
1391 [ + - ]: 527 : merge_url_info(impl_->url_map);
1392 : :
1393 [ + + ]: 527 : if (count == 0) {
1394 : 88 : continue;
1395 : : }
1396 : :
1397 : 439 : const auto duration = (last_timestamp - first_timestamp) / 1000'000.0;
1398 [ + + ]: 439 : freq = duration > 0 ? count / duration : 0;
1399 : : }
1400 : :
1401 [ + - + - : 606 : auto loss = total_url_loss_map_ref()[recover_recorded_url(ext_info.url)];
+ - ]
1402 : :
1403 [ + - + - : 23634 : url_json.push_back({
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + + +
+ + + + +
+ + + + +
+ + + + +
+ + + + -
- - - - -
- - - - -
- - - - -
- - - - -
- ]
1404 : : // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1405 : 606 : {"index", ext_info.index},
1406 [ + - ]: 606 : {"url", ext_info.url},
1407 [ + - ]: 606 : {"type", ext_info.url_type},
1408 [ + - + - ]: 1212 : {"action", std::string(convert_action(ext_info.action_type))},
1409 [ + - ]: 606 : {"ser", ext_info.ser_type},
1410 [ + - ]: 1212 : {"encoding", std::string(SchemaData::convert_type(ext_info.schema_type))},
1411 : : {"count", count},
1412 : : {"size", size},
1413 : : {"loss", loss},
1414 : : {"freq", freq},
1415 : : });
1416 : : }
1417 : 772 : }
1418 : :
1419 [ + - ]: 772 : json["VLinkUrls"] = std::move(url_json);
1420 : :
1421 : 772 : nlohmann::ordered_json files_json;
1422 : 772 : auto split_file_count = impl_->split_file_list.size();
1423 : :
1424 [ + + + + : 772 : if (!complete && impl_->writer && split_file_count > 0) {
+ - + + ]
1425 : 562 : --split_file_count;
1426 : : }
1427 : :
1428 [ + + ]: 3480 : for (size_t i = 0; i < split_file_count; ++i) {
1429 [ + - + - ]: 2708 : files_json.push_back(impl_->split_file_list[i]);
1430 : : }
1431 : :
1432 [ + - ]: 772 : json["VLinkFiles"] = std::move(files_json);
1433 : :
1434 [ + - ]: 772 : std::ofstream filex(file_path);
1435 : :
1436 [ + - - + ]: 772 : if VUNLIKELY (!filex.is_open()) {
1437 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1438 : : }
1439 : :
1440 [ + - + - ]: 772 : filex << json.dump(4);
1441 [ + - ]: 772 : filex.close();
1442 : :
1443 [ + - - + ]: 772 : if VUNLIKELY (!filex) {
1444 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1445 : : }
1446 [ + - + - : 772 : } catch (const nlohmann::json::exception& e) {
+ - + - +
- - - ]
1447 : : CLOG_W("VCAPWriter: JSON error during config export: %s.", e.what()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1448 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1449 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1450 : :
1451 : 772 : return true;
1452 : : }
1453 : :
1454 : : } // namespace vlink
|
