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