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