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_reader.h"
25 : :
26 : : #include <algorithm>
27 : : #include <atomic>
28 : : #include <cinttypes>
29 : : #include <cmath>
30 : : #include <filesystem>
31 : : #include <fstream>
32 : : #include <limits>
33 : : #include <memory>
34 : : #include <mutex>
35 : : #include <shared_mutex>
36 : : #include <string>
37 : : #include <string_view>
38 : : #include <unordered_map>
39 : : #include <utility>
40 : : #include <vector>
41 : :
42 : : #include "./base/condition_variable.h"
43 : : #include "./base/elapsed_timer.h"
44 : : #include "./base/helpers.h"
45 : : #include "./base/logger.h"
46 : : #include "./version.h"
47 : :
48 : : // json
49 : : #include <nlohmann/json.hpp>
50 : :
51 : : // mcap
52 : : #include "./private/mcap_import.h"
53 : :
54 : : namespace vlink {
55 : :
56 : : [[maybe_unused]] static constexpr size_t kMaxTaskSize = 50000U;
57 : :
58 : 466 : static mcap::Timestamp get_read_end_time(int64_t start_timestamp_ns, int64_t end_time_ms) {
59 [ + + + - : 498 : if (end_time_ms <= 0 || start_timestamp_ns < 0 ||
+ + ]
60 [ + + ]: 32 : end_time_ms > (std::numeric_limits<int64_t>::max() - start_timestamp_ns - 1000) / 1000'000) {
61 : 436 : return mcap::MaxTime;
62 : : }
63 : :
64 : 30 : return start_timestamp_ns + end_time_ms * 1000'000 + 1000;
65 : : }
66 : :
67 : : // VCAPReader::Impl
68 : : struct VCAPReader::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
69 : : std::atomic<BagReader::Status> status{VCAPReader::kStopped};
70 : : std::atomic_bool stop_flag{false};
71 : : std::atomic_bool pause_flag{false};
72 : : std::atomic_bool pause_next_flag{false};
73 : : std::atomic_bool jump_flag{false};
74 : : std::atomic<int64_t> pause_elapsed{0};
75 : : std::atomic<int64_t> offset_elapsed{0};
76 : : std::atomic<int64_t> real_elapsed{0};
77 : : std::atomic<int64_t> extra_elapsed{0};
78 : : std::atomic<int64_t> begin_time{0};
79 : : std::atomic<double> rate{1.0};
80 : : std::atomic<int> times{1};
81 : : std::atomic_bool is_pending{false};
82 : : std::atomic<int> split_index{0};
83 : :
84 : : bool read_only{false};
85 : : bool try_to_fix{false};
86 : : bool enable_compress{false};
87 : :
88 : : std::string path;
89 : : BagReader::Info info;
90 : : std::vector<BagReader::Info::UrlMeta> raw_url_metas;
91 : : std::mutex mtx;
92 : : ConditionVariable cv;
93 : :
94 : : BagReader::Config config;
95 : : std::mutex config_mtx;
96 : : std::shared_mutex time_mtx;
97 : :
98 : : ElapsedTimer elapsed_timer{ElapsedTimer::kMicro};
99 : : ElapsedTimer pause_elapsed_timer{ElapsedTimer::kMicro};
100 : : ElapsedTimer offset_timer{ElapsedTimer::kMicro};
101 : : ElapsedTimer real_timer{ElapsedTimer::kMicro};
102 : :
103 : : BagReader::StatusCallback status_callback;
104 : : BagReader::ReadyCallback ready_callback;
105 : : BagReader::FinishCallback finish_callback;
106 : :
107 : : int64_t total_start_timestamp_ns{-1};
108 : : bool total_has_completed{false};
109 : :
110 : : // mcap
111 : :
112 : : // WrapperFile
113 : : struct WrapperFile final {
114 : : std::string path;
115 : : std::unique_ptr<mcap::McapReader> reader;
116 : : std::unique_ptr<mcap::LinearMessageView> msg_view;
117 : : std::optional<mcap::LinearMessageView::Iterator> msg_view_begin;
118 : : std::optional<mcap::LinearMessageView::Iterator> msg_view_end;
119 : : int index{0};
120 : : int64_t start_timestamp_ns{0};
121 : : int64_t begin{0};
122 : : int64_t end{0};
123 : : std::unordered_map<std::string, int> url_to_id_map;
124 : : std::unordered_map<int, std::string> id_to_url_map;
125 : : std::unordered_map<int, ActionType> channel_action_map;
126 : : bool has_idx_elapsed{false};
127 : : bool has_idx_url{false};
128 : : bool has_schema{false};
129 : : bool has_completed{false};
130 : : bool is_channel_broken{false};
131 : :
132 : 805 : WrapperFile() {
133 [ + - ]: 805 : url_to_id_map.reserve(128);
134 [ + - ]: 805 : id_to_url_map.reserve(128);
135 [ + - ]: 805 : channel_action_map.reserve(128);
136 : 805 : }
137 : : };
138 : :
139 : : std::vector<WrapperFile> file_list;
140 : :
141 : : std::unique_ptr<mcap::LinearMessageView> cursor_msg_view;
142 : : std::optional<mcap::LinearMessageView::Iterator> cursor_iter;
143 : : std::optional<mcap::LinearMessageView::Iterator> cursor_iter_end;
144 : : int cursor_file_index{0};
145 : : int64_t cursor_begin_us{0};
146 : : int64_t cursor_end_us{0};
147 : : bool cursor_need_advance{false};
148 : : bool cursor_read_error{false};
149 : : BagReader::Config cursor_config;
150 : : };
151 : :
152 : : // VCAPReader
153 : 185 : VCAPReader::VCAPReader(const std::string& path, bool read_only, bool try_to_fix)
154 [ + - ]: 185 : : BagReader(path, read_only, try_to_fix), impl_{std::make_unique<Impl>()} {
155 [ + - + - ]: 185 : set_name("VCAPReader");
156 : :
157 [ + - + - ]: 185 : url_ser_map().reserve(128);
158 [ + - + - ]: 185 : url_schema_type_map().reserve(128);
159 : :
160 : 185 : impl_->read_only = read_only;
161 : 185 : impl_->try_to_fix = try_to_fix;
162 : :
163 [ + + ]: 185 : open(path);
164 : 192 : }
165 : :
166 : 178 : VCAPReader::~VCAPReader() {
167 [ + + ]: 178 : if (!impl_->stop_flag.load(std::memory_order_relaxed)) {
168 : 177 : do_stop();
169 : : }
170 : :
171 : 178 : quit(true);
172 : :
173 : 178 : impl_->cv.notify_one();
174 : :
175 : 178 : wait_for_quit();
176 : :
177 : 178 : detach_plugin();
178 : :
179 : 178 : close();
180 : 178 : }
181 : :
182 : 8 : void VCAPReader::bind_bag_interface(const std::shared_ptr<BagPluginInterface>& bag_interface) {
183 : 8 : BagReader::bind_bag_interface(bag_interface);
184 : 8 : impl_->info.url_metas = impl_->raw_url_metas;
185 : 8 : process_url_metas(impl_->info.url_metas);
186 : 8 : rebuild_url_meta_lookup(impl_->info.url_metas);
187 : 8 : }
188 : :
189 : 2 : void VCAPReader::register_status_callback(StatusCallback&& status_callback) {
190 : 2 : impl_->status_callback = std::move(status_callback);
191 : 2 : }
192 : :
193 : 4 : void VCAPReader::register_ready_callback(ReadyCallback&& ready_callback) {
194 : 4 : impl_->ready_callback = std::move(ready_callback);
195 : 4 : }
196 : :
197 : 15 : void VCAPReader::register_finish_callback(FinishCallback&& finish_callback) {
198 : 15 : impl_->finish_callback = std::move(finish_callback);
199 : 15 : }
200 : :
201 : 16 : void VCAPReader::register_output_callback(OutputCallback&& output_callback) {
202 : 16 : BagReader::register_output_callback(std::move(output_callback));
203 : 16 : }
204 : :
205 : 30 : void VCAPReader::play(const Config& config) {
206 [ + - + + ]: 30 : if VUNLIKELY (is_busy()) {
207 [ + - + - : 2 : VLOG_W("VCAPReader: Is busy.");
+ - ]
208 : : }
209 : :
210 [ + + ]: 30 : if (config.skip_blank) {
211 : 2 : impl_->begin_time.store(std::max(config.begin_time, impl_->info.blank_duration), std::memory_order_relaxed);
212 : : } else {
213 : 28 : impl_->begin_time.store(config.begin_time, std::memory_order_relaxed);
214 : : }
215 : :
216 [ + + ]: 30 : if (config.rate <= 0) {
217 : 1 : impl_->rate.store(1, std::memory_order_relaxed);
218 : : } else {
219 : 29 : impl_->rate.store(config.rate, std::memory_order_relaxed);
220 : : }
221 : :
222 : 30 : impl_->times.store(config.times, std::memory_order_relaxed);
223 : :
224 : 60 : impl_->real_elapsed.store(impl_->begin_time.load(std::memory_order_relaxed) * 1000U, std::memory_order_relaxed);
225 : 30 : impl_->is_pending.store(true, std::memory_order_relaxed);
226 : :
227 : : {
228 [ + - ]: 30 : std::unique_lock lock(impl_->mtx);
229 : 30 : impl_->stop_flag.store(false, std::memory_order_relaxed);
230 : 30 : impl_->pause_flag.store(false, std::memory_order_relaxed);
231 : 30 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
232 : 30 : impl_->jump_flag.store(false, std::memory_order_relaxed);
233 : 30 : }
234 : :
235 : 30 : Config config_snapshot;
236 : :
237 : : {
238 [ + - ]: 30 : std::unique_lock lock(impl_->config_mtx);
239 [ + - ]: 30 : impl_->config = config;
240 [ + - ]: 30 : config_snapshot = impl_->config;
241 : 30 : }
242 : :
243 [ + - + - : 60 : post_task([this, config_snapshot]() { read(config_snapshot); });
+ - ]
244 : 30 : }
245 : :
246 : 4 : void VCAPReader::stop() { do_stop(); }
247 : :
248 : 2 : void VCAPReader::pause() {
249 : : {
250 [ + - ]: 2 : std::unique_lock lock(impl_->mtx);
251 : 2 : impl_->pause_flag.store(true, std::memory_order_relaxed);
252 : 2 : }
253 : :
254 : 2 : impl_->cv.notify_one();
255 : 2 : }
256 : :
257 : 2 : void VCAPReader::resume() {
258 : : {
259 [ + - ]: 2 : std::unique_lock lock(impl_->mtx);
260 : 2 : impl_->pause_flag.store(false, std::memory_order_relaxed);
261 : 2 : }
262 : :
263 : 2 : impl_->cv.notify_one();
264 : 2 : }
265 : :
266 : 10 : void VCAPReader::pause_to_next() {
267 : : {
268 [ + - ]: 10 : std::unique_lock lock(impl_->mtx);
269 : :
270 [ + + ]: 10 : if (!impl_->pause_flag.load(std::memory_order_relaxed)) {
271 : 1 : return;
272 : : }
273 : :
274 : 9 : impl_->pause_next_flag.store(true, std::memory_order_relaxed);
275 [ + + ]: 10 : }
276 : :
277 : 9 : impl_->cv.notify_one();
278 : : }
279 : :
280 : 7 : void VCAPReader::jump(int64_t begin_time, double rate, int times, bool force_to_play) {
281 [ + + ]: 7 : if (begin_time < 0) {
282 : 1 : begin_time = 0;
283 [ + + ]: 6 : } else if (begin_time > impl_->info.total_duration) {
284 : 1 : begin_time = std::max<int64_t>(0, impl_->info.total_duration - 100);
285 : : }
286 : :
287 : 7 : impl_->real_elapsed.store(begin_time * 1000U, std::memory_order_relaxed);
288 : 7 : impl_->is_pending.store(true, std::memory_order_relaxed);
289 : :
290 : 7 : bool last_pause_flag = impl_->pause_flag.load(std::memory_order_relaxed);
291 : :
292 : : {
293 [ + - ]: 7 : std::unique_lock lock(impl_->mtx);
294 : 7 : impl_->stop_flag.store(false, std::memory_order_relaxed);
295 : 7 : impl_->pause_flag.store(false, std::memory_order_relaxed);
296 : 7 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
297 : 7 : impl_->jump_flag.store(true, std::memory_order_relaxed);
298 : 7 : }
299 : :
300 : 7 : impl_->cv.notify_one();
301 : :
302 [ + - ]: 7 : wait_for_idle();
303 : :
304 : 7 : impl_->begin_time.store(begin_time, std::memory_order_relaxed);
305 : :
306 [ + + ]: 7 : if (rate <= 0) {
307 : 3 : impl_->rate.store(1, std::memory_order_relaxed);
308 : : } else {
309 : 4 : impl_->rate.store(rate, std::memory_order_relaxed);
310 : : }
311 : :
312 : 7 : impl_->times.store(times, std::memory_order_relaxed);
313 : :
314 : : {
315 [ + - ]: 7 : std::unique_lock lock(impl_->mtx);
316 : 7 : impl_->stop_flag.store(false, std::memory_order_relaxed);
317 [ + + ]: 8 : impl_->pause_flag.store(force_to_play ? false : last_pause_flag, std::memory_order_relaxed);
318 : 7 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
319 : 7 : impl_->jump_flag.store(false, std::memory_order_relaxed);
320 : 7 : }
321 : :
322 : 7 : Config config_snapshot;
323 : :
324 : : {
325 [ + - ]: 7 : std::unique_lock lock(impl_->config_mtx);
326 [ + - ]: 7 : config_snapshot = impl_->config;
327 : 7 : }
328 : :
329 : 7 : config_snapshot.auto_pause = false;
330 [ + - + - : 14 : post_task([this, config_snapshot]() { read(config_snapshot); });
+ - ]
331 : 8 : }
332 : :
333 : 33 : std::future<bool> VCAPReader::check() {
334 [ + + ]: 33 : if VUNLIKELY (is_busy()) {
335 [ + - + - : 2 : VLOG_W("VCAPReader: Is busy.");
+ - ]
336 : : }
337 : :
338 : 390 : return invoke_task([this]() {
339 [ + + ]: 33 : if (!impl_->total_has_completed) {
340 [ + - + - : 2 : VLOG_W("VCAPReader: Incomplete data detected.");
+ - ]
341 : 1 : return false;
342 : : }
343 : :
344 : 32 : mcap::Status status;
345 : :
346 : : // LCOV_EXCL_START GCOVR_EXCL_START
347 : : auto status_function = [](const mcap::Status& status) {
348 : : if (!status.ok()) {
349 : : CLOG_W("VCAPReader: Failed to check summary, error = %s.", status.message.c_str());
350 : : }
351 : : };
352 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
353 : :
354 [ + + ]: 83 : for (auto& wrapper_file : impl_->file_list) {
355 [ - + ]: 51 : if VUNLIKELY (!wrapper_file.reader) {
356 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
357 : : }
358 : :
359 [ + - ]: 51 : status = wrapper_file.reader->readSummary(mcap::ReadSummaryMethod::ForceScan, status_function);
360 : :
361 [ - + ]: 51 : if VUNLIKELY (!status.ok()) {
362 : : CLOG_W("VCAPReader: Failed to check whole summary, error = %s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
363 : : status.message.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
364 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
365 : : }
366 : : }
367 : :
368 : 32 : bool is_ok = true;
369 : :
370 [ + + ]: 32 : if VUNLIKELY (impl_->info.total_duration < impl_->info.blank_duration) {
371 [ + - + - : 2 : CLOG_W("VCAPReader: Invalid duration, blank=%" PRId64 " total=%" PRId64 ".",
+ - ]
372 : : static_cast<int64_t>(impl_->info.blank_duration), static_cast<int64_t>(impl_->info.total_duration));
373 : 1 : is_ok = false;
374 : : }
375 : :
376 [ + + + + : 32 : if VUNLIKELY (impl_->info.message_count > 0 && impl_->info.url_metas.empty()) {
+ + ]
377 [ + - + - : 2 : CLOG_W("VCAPReader: Message count is %" PRId64 " but url meta list is empty.",
+ - ]
378 : : static_cast<int64_t>(impl_->info.message_count));
379 : 1 : is_ok = false;
380 : : }
381 : :
382 : 32 : size_t total_count = 0;
383 : 32 : size_t total_raw_size = 0;
384 : :
385 [ + + ]: 80 : for (const auto& url_meta : impl_->info.url_metas) {
386 : 48 : total_count += url_meta.count;
387 : 48 : total_raw_size += url_meta.size;
388 : :
389 [ - + ]: 48 : if VUNLIKELY (!url_meta.valid) {
390 : : CLOG_W("VCAPReader: Invalid url meta detected at index=%d.", url_meta.index); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
391 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
392 : : }
393 : :
394 [ + + ]: 48 : if VUNLIKELY (url_meta.url.empty()) {
395 [ + - + - : 2 : CLOG_W("VCAPReader: Empty url detected at index=%d.", url_meta.index);
+ - ]
396 : 1 : is_ok = false;
397 : : }
398 : :
399 [ + + ]: 48 : if VUNLIKELY (url_meta.url_type.empty()) {
400 [ + - + - : 2 : CLOG_W("VCAPReader: Empty url_type detected for url=%s.", url_meta.url.c_str());
+ - ]
401 : 1 : is_ok = false;
402 : : }
403 : :
404 [ + - + + : 48 : if VUNLIKELY (url_meta.count > 0 && url_meta.ser_type.empty()) {
+ + ]
405 [ + - + - : 4 : CLOG_W("VCAPReader: Empty ser_type detected for url=%s.", url_meta.url.c_str());
+ - ]
406 : 2 : is_ok = false;
407 : : }
408 : :
409 [ - + ]: 48 : if VUNLIKELY (!SchemaData::is_valid_type(url_meta.schema_type)) {
410 : : CLOG_W("VCAPReader: Invalid schema_type=%d detected for url=%s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
411 : : static_cast<int>(url_meta.schema_type), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
412 : : url_meta.url.c_str());
413 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
414 : : }
415 : :
416 : 48 : auto inferred_schema_type = SchemaData::infer_ser_type(url_meta.ser_type);
417 : :
418 [ + + + + : 48 : if VUNLIKELY (url_meta.schema_type == SchemaType::kUnknown && inferred_schema_type != SchemaType::kUnknown) {
+ + ]
419 : 1 : const auto schema_label = SchemaData::convert_type(inferred_schema_type);
420 [ + - + - : 2 : CLOG_W("VCAPReader: Missing schema_type for url=%s, inferred=%.*s.", url_meta.url.c_str(),
+ - ]
421 : : static_cast<int>(schema_label.size()), schema_label.data());
422 : 1 : is_ok = false;
423 : : }
424 : :
425 [ + - + - : 48 : if VUNLIKELY (!std::isfinite(url_meta.loss) ||
+ + + + +
+ + + +
+ ]
426 : : (url_meta.loss != -1.0 && (url_meta.loss < 0.0 || url_meta.loss > 1.0))) {
427 [ + - + - : 6 : CLOG_W("VCAPReader: Invalid loss=%f detected for url=%s.", url_meta.loss, url_meta.url.c_str());
+ - ]
428 : 3 : is_ok = false;
429 : : }
430 : :
431 [ + + ]: 48 : if VUNLIKELY (url_meta.freq < 0.0) {
432 [ + - + - : 6 : CLOG_W("VCAPReader: Invalid freq=%f detected for url=%s.", url_meta.freq, url_meta.url.c_str());
+ - ]
433 : 3 : is_ok = false;
434 : : }
435 : : }
436 : :
437 [ + + + + : 63 : if ((!impl_->info.url_metas.empty() || impl_->info.message_count != 0) &&
+ + ]
438 [ + + ]: 31 : total_count != static_cast<size_t>(impl_->info.message_count)) {
439 [ + - + - : 4 : CLOG_W("VCAPReader: Message count mismatch, header=%" PRId64 " metas=%zu.",
+ - ]
440 : : static_cast<int64_t>(impl_->info.message_count), total_count);
441 : 2 : is_ok = false;
442 : : }
443 : :
444 [ + + - + : 62 : if ((!impl_->info.url_metas.empty() || impl_->info.total_raw_size != 0) &&
- + ]
445 [ - + ]: 30 : total_raw_size != static_cast<size_t>(impl_->info.total_raw_size)) {
446 : : CLOG_W("VCAPReader: Raw size mismatch, header=%" PRId64 " metas=%zu.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
447 : : static_cast<int64_t>(impl_->info.total_raw_size), total_raw_size);
448 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
449 : : }
450 : :
451 [ + - + + ]: 39 : for (const auto& schema_data : detect_schema()) {
452 [ - + ]: 7 : if VUNLIKELY (schema_data.name.empty()) {
453 : : CLOG_W("VCAPReader: Empty schema name detected."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
454 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
455 : : }
456 : :
457 [ - + ]: 7 : if VUNLIKELY (schema_data.encoding.empty()) {
458 : : CLOG_W("VCAPReader: Empty schema encoding detected for name=%s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
459 : : schema_data.name.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
460 : : is_ok = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
461 : : }
462 : :
463 [ + - + + : 7 : if VUNLIKELY (!SchemaData::is_valid_type(schema_data.schema_type) ||
+ + ]
464 : : schema_data.schema_type == SchemaType::kUnknown) {
465 [ + - + - : 2 : CLOG_W("VCAPReader: Invalid schema_type=%d detected for schema=%s.", static_cast<int>(schema_data.schema_type),
+ - ]
466 : : schema_data.name.c_str());
467 : 1 : is_ok = false;
468 : : }
469 : 32 : }
470 : :
471 : 32 : return is_ok;
472 [ + - ]: 65 : });
473 : : }
474 : :
475 : 3 : std::future<bool> VCAPReader::reindex() {
476 [ + + ]: 3 : if VUNLIKELY (is_busy()) {
477 [ + - + - : 2 : VLOG_W("VCAPReader: Is busy.");
+ - ]
478 : : }
479 : :
480 : 3 : return invoke_task([]() {
481 [ + - + - : 6 : VLOG_W("VCAPReader: Reindex is not supported for vcap.");
+ - ]
482 : :
483 : 3 : return false;
484 [ + - ]: 3 : });
485 : : }
486 : :
487 : 3 : std::future<bool> VCAPReader::fix(bool rebuild) {
488 [ + + ]: 3 : if VUNLIKELY (is_busy()) {
489 [ + - + - : 2 : VLOG_W("VCAPReader: Is busy.");
+ - ]
490 : : }
491 : :
492 : 3 : return invoke_task([rebuild]() {
493 : : (void)rebuild;
494 : :
495 [ + - + - : 6 : VLOG_W("VCAPReader: Fix is not supported for vcap.");
+ - ]
496 : :
497 : 3 : return false;
498 [ + - ]: 3 : });
499 : : }
500 : :
501 : 8 : void VCAPReader::tag(const std::string& tag_name) {
502 [ + + ]: 8 : if VUNLIKELY (is_busy()) {
503 [ + - + - : 2 : VLOG_W("VCAPReader: Is busy.");
+ - ]
504 : : }
505 : :
506 [ + - + - ]: 8 : post_task([this, tag_name]() {
507 : : try {
508 : : #ifdef _WIN32
509 : : std::filesystem::path file_path(Helpers::string_to_wstring(impl_->path));
510 : : std::string suffix = Helpers::path_to_string(file_path.extension());
511 : : #else
512 [ + - ]: 8 : std::filesystem::path file_path(impl_->path);
513 [ + - + - ]: 8 : std::string suffix = file_path.extension().string();
514 : : #endif
515 : :
516 : 52 : std::transform(suffix.begin(), suffix.end(), suffix.begin(), [](unsigned char c) { return std::tolower(c); });
517 : :
518 [ + + ]: 8 : if (suffix == ".vcapx") {
519 : : try {
520 : 4 : nlohmann::ordered_json root_json;
521 : 4 : nlohmann::ordered_json header_json;
522 : :
523 : : {
524 [ + - ]: 4 : std::ifstream file(file_path);
525 : :
526 [ + + ]: 4 : file >> root_json;
527 : :
528 [ + - ]: 3 : file.close();
529 : 4 : }
530 : :
531 [ + - + - ]: 3 : header_json = root_json["VLinkHeader"];
532 : :
533 [ + - + - ]: 3 : header_json["tag"] = tag_name;
534 : :
535 [ + - ]: 3 : root_json["VLinkHeader"] = std::move(header_json);
536 : :
537 : : {
538 [ + - ]: 3 : std::ofstream filex(impl_->path, std::ios::out | std::ios::trunc);
539 : :
540 [ + - + - ]: 3 : if VLIKELY (filex.is_open()) {
541 [ + - + - ]: 3 : filex << root_json.dump(4);
542 [ + - ]: 3 : filex.close();
543 : : }
544 : 3 : }
545 [ - + ]: 6 : } catch (nlohmann::json::exception& e) {
546 [ + - + - : 2 : VLOG_W("VCAPReader: JSON parse error, ", e.what(), ".");
+ - ]
547 : 1 : }
548 : : } else {
549 [ + - + - : 8 : VLOG_W("VCAPReader: Tag is not supported for single vcap.");
+ - ]
550 : : }
551 [ - - ]: 8 : } catch (std::filesystem::filesystem_error& e) {
552 : : VLOG_F("VCAPReader: Filesystem error, ", e.what(), "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
553 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
554 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
555 : : });
556 : 8 : }
557 : :
558 : 5 : int64_t VCAPReader::get_timestamp() const {
559 [ + - ]: 5 : std::shared_lock time_lock(impl_->time_mtx);
560 : :
561 [ + + ]: 5 : if (impl_->status.load(std::memory_order_relaxed) == kPlaying) {
562 [ - + ]: 1 : if (impl_->is_pending.load(std::memory_order_relaxed)) {
563 : : return impl_->real_elapsed.load(std::memory_order_relaxed) / 1000U; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
564 : : } else {
565 : 1 : return (impl_->real_elapsed.load(std::memory_order_relaxed) +
566 : 1 : (impl_->real_timer.get() * impl_->rate.load(std::memory_order_relaxed))) /
567 : 1 : 1000U;
568 : : }
569 [ + + ]: 4 : } else if (impl_->status.load(std::memory_order_relaxed) == kPaused) {
570 : 2 : return (impl_->real_elapsed.load(std::memory_order_relaxed) +
571 : 2 : ((impl_->real_timer.get() - impl_->pause_elapsed_timer.get() -
572 : 2 : impl_->extra_elapsed.load(std::memory_order_relaxed)) *
573 : 2 : impl_->rate.load(std::memory_order_relaxed))) /
574 : 2 : 1000U;
575 : : } else {
576 : 2 : return 0;
577 : : }
578 : 5 : }
579 : :
580 : 5 : int64_t VCAPReader::get_real_timestamp() const {
581 [ + + + + : 9 : if (impl_->status.load(std::memory_order_relaxed) == kPlaying ||
+ + ]
582 : 4 : impl_->status.load(std::memory_order_relaxed) == kPaused) {
583 : 6 : return impl_->real_elapsed.load(std::memory_order_relaxed) / 1000U;
584 : : } else {
585 : 2 : return 0;
586 : : }
587 : : }
588 : :
589 : 20 : BagReader::Status VCAPReader::get_status() const { return impl_->status.load(std::memory_order_relaxed); }
590 : :
591 : 233 : const BagReader::Info& VCAPReader::get_info() const { return impl_->info; }
592 : :
593 : 36 : std::vector<SchemaData> VCAPReader::detect_schema() {
594 : 36 : std::vector<SchemaData> schema_list;
595 : 36 : std::unordered_map<std::string, size_t> schema_index_map;
596 : :
597 [ + + ]: 36 : if (!impl_->info.has_schema) {
598 : 25 : return schema_list;
599 : : }
600 : :
601 [ + - ]: 11 : schema_index_map.reserve(impl_->info.url_metas.size());
602 : :
603 [ + + ]: 24 : for (auto& wrapper_file : impl_->file_list) {
604 [ + - + + ]: 25 : for (const auto& [schema_id, schema_ptr] : wrapper_file.reader->schemas()) {
605 : : (void)schema_id;
606 : :
607 : 12 : SchemaData schema;
608 [ + - ]: 12 : schema.name = schema_ptr->name;
609 [ + - ]: 12 : schema.encoding = schema_ptr->encoding;
610 : 12 : schema.schema_type = SchemaData::resolve_type(SchemaType::kUnknown, schema.name, schema.encoding);
611 : :
612 [ + - + - : 12 : if (!schema.name.empty() && !schema_ptr->data.empty()) {
+ - ]
613 [ + - ]: 12 : std::string schema_key = schema.name;
614 [ + - ]: 12 : schema_key.push_back('\x1F');
615 [ + - ]: 12 : schema_key.append(SchemaData::convert_type(schema.schema_type));
616 [ + - ]: 12 : auto schema_index_iter = schema_index_map.find(schema_key);
617 : :
618 [ + + ]: 12 : if (schema_index_iter == schema_index_map.end()) {
619 : : schema.data =
620 : 10 : Bytes::deep_copy(reinterpret_cast<const uint8_t*>(schema_ptr->data.data()), schema_ptr->data.size());
621 [ + - ]: 10 : schema_index_map.emplace(schema_key, schema_list.size());
622 [ + - ]: 10 : schema_list.emplace_back(std::move(schema));
623 : : } else {
624 : 2 : auto& current_schema = schema_list[schema_index_iter->second];
625 : :
626 [ - + - - : 2 : if (current_schema.encoding.empty() && !schema.encoding.empty()) {
- + ]
627 : : current_schema.encoding = schema.encoding; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
628 : : }
629 : :
630 [ - + ]: 2 : if (current_schema.data.empty()) {
631 : 0 : current_schema.data = Bytes::deep_copy(
632 : : reinterpret_cast<const uint8_t*>(schema_ptr->data.data()), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
633 : : schema_ptr->data.size()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
634 : : }
635 : : }
636 : 12 : }
637 : 25 : }
638 : : }
639 : :
640 : 11 : return schema_list;
641 : 36 : }
642 : :
643 : 9 : bool VCAPReader::is_split_mode() const { return impl_->info.split_count > 0; }
644 : :
645 : 4 : int VCAPReader::get_split_index() const { return impl_->split_index.load(std::memory_order_relaxed); }
646 : :
647 : 6 : bool VCAPReader::is_jumping() const { return impl_->jump_flag.load(std::memory_order_relaxed); }
648 : :
649 : 84 : size_t VCAPReader::get_max_task_count() const { return kMaxTaskSize; }
650 : :
651 : 125 : void VCAPReader::on_begin() { MessageLoop::on_begin(); }
652 : :
653 : 125 : void VCAPReader::on_end() { MessageLoop::on_end(); }
654 : :
655 : 110 : bool VCAPReader::do_open_cursor(const Config& config) {
656 : 110 : impl_->cursor_config = config;
657 [ + + ]: 110 : impl_->cursor_begin_us = config.begin_time > 0 ? config.begin_time * 1000 : 0;
658 [ + + ]: 110 : impl_->cursor_end_us = config.end_time > 0 ? config.end_time * 1000 : 0;
659 : 110 : impl_->cursor_file_index = 0;
660 : 110 : impl_->cursor_read_error = false;
661 : :
662 [ - + ]: 110 : if VUNLIKELY (impl_->file_list.empty()) {
663 : : VLOG_W("VCAPReader: Cursor cannot find any data."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
664 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
665 : : }
666 : :
667 : 110 : return prepare_cursor_view(0);
668 : : }
669 : :
670 : 562 : bool VCAPReader::do_read_next(Frame& out, bool& is_error) {
671 : 562 : is_error = false;
672 : :
673 : : while (true) {
674 [ + - - + : 892 : if VUNLIKELY (!impl_->cursor_iter.has_value() || !impl_->cursor_iter_end.has_value()) {
- + ]
675 : 562 : return false;
676 : : }
677 : :
678 [ + - ]: 892 : auto& iter = impl_->cursor_iter.value();
679 [ + - ]: 892 : auto& iter_end = impl_->cursor_iter_end.value();
680 : :
681 [ + + ]: 892 : if (impl_->cursor_need_advance) {
682 [ + - + - ]: 452 : if (iter != iter_end) {
683 [ + - ]: 452 : iter++;
684 : : }
685 : :
686 : 452 : impl_->cursor_need_advance = false;
687 : : }
688 : :
689 [ + - + + ]: 892 : if (iter == iter_end) {
690 [ - + ]: 427 : if VUNLIKELY (impl_->cursor_read_error) {
691 : : is_error = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
692 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
693 : : }
694 : :
695 [ + + ]: 427 : if (impl_->cursor_file_index + 1 < static_cast<int>(impl_->file_list.size())) {
696 [ + - - + ]: 317 : if VUNLIKELY (!prepare_cursor_view(impl_->cursor_file_index + 1)) {
697 : : is_error = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
698 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
699 : : }
700 : :
701 : 317 : continue;
702 : : }
703 : :
704 : 110 : return false;
705 : : }
706 : :
707 [ + - ]: 465 : const int64_t timestamp = (static_cast<int64_t>(iter->message.logTime) - impl_->total_start_timestamp_ns) / 1000;
708 : :
709 [ + + - + : 465 : if (impl_->cursor_begin_us > 0 && timestamp < impl_->cursor_begin_us) {
- + ]
710 : : iter++; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
711 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
712 : : }
713 : :
714 [ + + + + : 465 : if (impl_->cursor_end_us > 0 && timestamp > impl_->cursor_end_us) {
+ + ]
715 [ + - ]: 13 : iter++;
716 : 13 : continue;
717 : : }
718 : :
719 [ + - - + ]: 452 : if VUNLIKELY (iter->message.dataSize > static_cast<uint64_t>(std::numeric_limits<size_t>::max())) {
720 : : // LCOV_EXCL_START GCOVR_EXCL_START
721 : : CLOG_W("VCAPReader: Cursor message data size is too large to address, size = %" PRIu64 ".",
722 : : static_cast<uint64_t>(iter->message.dataSize));
723 : : iter++;
724 : : continue;
725 : : }
726 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
727 : :
728 : 452 : std::string output_url;
729 : :
730 [ + - + - : 452 : if VUNLIKELY (!convert_playback_url(iter->channel->topic, output_url)) {
- + ]
731 : : iter++; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
732 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
733 : : }
734 : :
735 : 452 : ActionType action_type = ActionType::kUnknownAction;
736 [ + - ]: 452 : auto& wrapper_file = impl_->file_list.at(impl_->cursor_file_index);
737 : :
738 [ + - + - ]: 452 : if (auto action_iter = wrapper_file.channel_action_map.find(iter->message.channelId);
739 [ + - ]: 452 : action_iter != wrapper_file.channel_action_map.end()) {
740 : 452 : action_type = action_iter->second;
741 : : }
742 : :
743 [ + - ]: 452 : const auto* data = reinterpret_cast<const uint8_t*>(iter->message.data);
744 [ + - ]: 452 : const auto size = static_cast<size_t>(iter->message.dataSize);
745 : :
746 : 452 : out.timestamp = timestamp;
747 : 452 : out.url = std::move(output_url);
748 : 452 : out.ser_type.clear();
749 : 452 : out.schema_type = SchemaType::kUnknown;
750 : 452 : out.action_type = action_type;
751 : 452 : out.data = Bytes::shallow_copy(data, size);
752 : :
753 [ + - ]: 452 : fill_frame_meta(out);
754 : :
755 : 452 : impl_->cursor_need_advance = true;
756 : :
757 : 452 : return true;
758 [ + - ]: 782 : }
759 : : }
760 : :
761 : 427 : bool VCAPReader::prepare_cursor_view(int file_index) {
762 : 427 : impl_->cursor_iter.reset();
763 : 427 : impl_->cursor_iter_end.reset();
764 : 427 : impl_->cursor_msg_view.reset();
765 : 427 : impl_->cursor_need_advance = false;
766 : :
767 [ + - - + : 427 : if VUNLIKELY (file_index < 0 || file_index >= static_cast<int>(impl_->file_list.size())) {
- + ]
768 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
769 : : }
770 : :
771 [ + - ]: 427 : auto& wrapper_file = impl_->file_list.at(file_index);
772 : :
773 [ - + ]: 427 : if VUNLIKELY (!wrapper_file.reader) {
774 : : VLOG_W("VCAPReader: Cursor target vcap reader is empty."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
775 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
776 : : }
777 : :
778 : 0 : auto status_function = [this](const mcap::Status& status) {
779 : : // LCOV_EXCL_START GCOVR_EXCL_START
780 : : if (!status.ok()) {
781 : : CLOG_W("VCAPReader: Failed to read cursor message, error = %s.", status.message.c_str());
782 : : impl_->cursor_read_error = true;
783 : : }
784 : : };
785 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
786 : :
787 : 20 : auto filter_function = [this](std::string_view url) -> bool {
788 : 20 : return match_playback_url_filter(url, impl_->cursor_config.filter_urls);
789 : 427 : };
790 : :
791 : 427 : mcap::ReadMessageOptions read_options;
792 : 427 : read_options.startTime =
793 [ + + ]: 427 : impl_->cursor_begin_us > 0 ? impl_->cursor_begin_us * 1000 + impl_->total_start_timestamp_ns : 0;
794 : 427 : read_options.endTime = mcap::MaxTime;
795 [ + + + + : 427 : if (!impl_->cursor_config.filter_urls.empty() || has_playback_url_rules()) {
+ + ]
796 : 12 : read_options.topicFilter = filter_function;
797 : : }
798 : 427 : read_options.readOrder = mcap::ReadMessageOptions::ReadOrder::FileOrder;
799 : :
800 : 427 : const auto end_time = get_read_end_time(impl_->total_start_timestamp_ns, impl_->cursor_config.end_time);
801 [ + - ]: 427 : const auto [start_offset, end_offset] = wrapper_file.reader->byteRange(read_options.startTime, end_time);
802 : :
803 : : // NOLINTNEXTLINE(readability-redundant-smartptr-get)
804 [ + - ]: 854 : impl_->cursor_msg_view = std::make_unique<mcap::LinearMessageView>(*wrapper_file.reader.get(), read_options,
805 : 427 : start_offset, end_offset, status_function);
806 : :
807 [ + + ]: 427 : if (start_offset == end_offset) {
808 [ + - ]: 3 : impl_->cursor_iter.emplace(impl_->cursor_msg_view->end());
809 : : } else {
810 [ + - ]: 424 : impl_->cursor_iter.emplace(impl_->cursor_msg_view->begin());
811 : : }
812 : :
813 [ + - ]: 427 : impl_->cursor_iter_end.emplace(impl_->cursor_msg_view->end());
814 : 427 : impl_->cursor_file_index = file_index;
815 : :
816 : 427 : return true;
817 : 427 : }
818 : :
819 : 98 : void VCAPReader::update_status(Status status) {
820 : 98 : bool has_changed = false;
821 : :
822 [ + + ]: 98 : if (status == kStopped) {
823 [ + + ]: 34 : if (impl_->status.load(std::memory_order_relaxed) != kStopped) {
824 : 33 : impl_->status.store(kStopped, std::memory_order_relaxed);
825 : 33 : has_changed = true;
826 : : }
827 [ + + ]: 64 : } else if (status == kPaused) {
828 [ + + ]: 19 : if (impl_->status.load(std::memory_order_relaxed) != kPaused) {
829 : 14 : impl_->status.store(kPaused, std::memory_order_relaxed);
830 : 14 : has_changed = true;
831 : : }
832 [ + - ]: 45 : } else if (status == kPlaying) {
833 [ + + ]: 45 : if (impl_->status.load(std::memory_order_relaxed) != kPlaying) {
834 : 42 : impl_->status.store(kPlaying, std::memory_order_relaxed);
835 : 42 : has_changed = true;
836 : : }
837 : : }
838 : :
839 [ + + ]: 98 : if (has_changed) {
840 [ + + ]: 89 : if VLIKELY (impl_->status_callback) {
841 : 19 : impl_->status_callback(impl_->status.load(std::memory_order_relaxed));
842 : : }
843 : : }
844 : 98 : }
845 : :
846 : 181 : void VCAPReader::do_stop() {
847 : : {
848 [ + - ]: 181 : std::unique_lock lock(impl_->mtx);
849 : 181 : impl_->stop_flag.store(true, std::memory_order_relaxed);
850 : 181 : impl_->pause_flag.store(false, std::memory_order_relaxed);
851 : 181 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
852 : 181 : impl_->jump_flag.store(false, std::memory_order_relaxed);
853 : 181 : }
854 : :
855 : 181 : impl_->cv.notify_one();
856 : 181 : }
857 : :
858 : 14 : void VCAPReader::do_pause() {
859 [ + - ]: 14 : std::unique_lock lock(impl_->mtx);
860 : :
861 [ + - ]: 14 : while (impl_->pause_flag.load(std::memory_order_relaxed)) {
862 : 14 : impl_->pause_elapsed_timer.restart();
863 [ + - ]: 14 : update_status(kPaused);
864 : :
865 [ + - ]: 14 : impl_->cv.wait(lock, [this]() -> bool {
866 [ + + + + ]: 67 : return impl_->stop_flag.load(std::memory_order_relaxed) || !impl_->pause_flag.load(std::memory_order_relaxed) ||
867 [ + - ]: 30 : impl_->pause_next_flag.load(std::memory_order_relaxed) ||
868 [ + - - + ]: 55 : impl_->jump_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
869 : : });
870 : :
871 : 14 : impl_->pause_elapsed.fetch_add(impl_->pause_elapsed_timer.get(), std::memory_order_relaxed);
872 : :
873 : : {
874 [ + - ]: 14 : std::lock_guard time_lock(impl_->time_mtx);
875 : 14 : impl_->real_timer.restart();
876 : :
877 [ - + ]: 28 : if (impl_->offset_elapsed.load(std::memory_order_relaxed) > 0) {
878 : : impl_->real_elapsed.fetch_add( // LCOV_EXCL_LINE GCOVR_EXCL_LINE
879 : : (impl_->offset_timer.get() - impl_->pause_elapsed_timer.get()) * // LCOV_EXCL_LINE GCOVR_EXCL_LINE
880 : : impl_->rate.load(std::memory_order_relaxed), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
881 : : std::memory_order_relaxed);
882 : : }
883 : 14 : }
884 : :
885 [ + - ]: 14 : update_status(kPlaying);
886 : :
887 [ + - ]: 14 : if (impl_->pause_next_flag.load(std::memory_order_relaxed)) {
888 : 28 : impl_->pause_elapsed.fetch_sub(impl_->offset_elapsed.load(std::memory_order_relaxed), std::memory_order_relaxed);
889 : 14 : break;
890 : : // LCOV_EXCL_START GCOVR_EXCL_START
891 : : } else if (impl_->offset_elapsed.load(std::memory_order_relaxed) > 0) {
892 : : impl_->offset_timer.restart();
893 : :
894 : : impl_->cv.wait_for(lock, std::chrono::microseconds(impl_->offset_elapsed.load(std::memory_order_relaxed)),
895 : : [this]() -> bool {
896 : : return impl_->stop_flag.load(std::memory_order_relaxed) ||
897 : : impl_->pause_flag.load(std::memory_order_relaxed) ||
898 : : impl_->pause_next_flag.load(std::memory_order_relaxed) ||
899 : : impl_->jump_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
900 : : });
901 : :
902 : : if VUNLIKELY (impl_->pause_flag.load(std::memory_order_relaxed)) {
903 : : impl_->offset_elapsed.fetch_sub(impl_->offset_timer.get(), std::memory_order_relaxed);
904 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
905 : : } else {
906 : : impl_->offset_elapsed.store(0, std::memory_order_relaxed); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
907 : : }
908 : : }
909 : : }
910 : 14 : }
911 : :
912 : 804 : bool VCAPReader::prepare_file(void* file) {
913 : 804 : auto* wrapper_file = static_cast<Impl::WrapperFile*>(file);
914 : :
915 : 804 : wrapper_file->has_completed = true;
916 : :
917 : 804 : auto& reader = wrapper_file->reader;
918 : :
919 [ - + ]: 804 : if VUNLIKELY (!reader) {
920 : : wrapper_file->has_completed = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
921 : : CLOG_F("VCAPReader: Mcap [%s] reader is nullptr.", wrapper_file->path.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
922 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
923 : : }
924 : :
925 [ + - - + ]: 804 : if VUNLIKELY (!reader->header()) {
926 : : // LCOV_EXCL_START GCOVR_EXCL_START
927 : : wrapper_file->has_completed = false;
928 : : CLOG_F("VCAPReader: Mcap [%s] reader header is nullptr.", wrapper_file->path.c_str());
929 : : return false;
930 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
931 : : }
932 : :
933 [ + - - + ]: 804 : if VUNLIKELY (reader->header()->profile != "vlink") {
934 : : // LCOV_EXCL_START GCOVR_EXCL_START
935 : : wrapper_file->has_completed = false;
936 : : CLOG_F("VCAPReader: Mcap [%s] profile is %s, not valid.", wrapper_file->path.c_str(),
937 : : reader->header()->profile.c_str());
938 : : return false;
939 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
940 : : }
941 : :
942 : 804 : mcap::Status status;
943 : :
944 [ + - ]: 804 : status = reader->readSummary(mcap::ReadSummaryMethod::NoFallbackScan);
945 : :
946 [ - + ]: 804 : if VUNLIKELY (!status.ok()) {
947 : : // LCOV_EXCL_START GCOVR_EXCL_START
948 : : wrapper_file->has_completed = false;
949 : :
950 : : if (impl_->try_to_fix) {
951 : : CLOG_E("VCAPReader: Failed to read summary, error = %s. Trying to fix.", status.message.c_str());
952 : : status = reader->readSummary(mcap::ReadSummaryMethod::AllowFallbackScan);
953 : :
954 : : if VUNLIKELY (!status.ok()) {
955 : : CLOG_F("VCAPReader: Failed to read summary, error = %s.", status.message.c_str());
956 : : return false;
957 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
958 : : }
959 : :
960 : : } else {
961 : : CLOG_F("VCAPReader: Failed to read summary, error = %s.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
962 : : status.message.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
963 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
964 : : }
965 : : }
966 : :
967 [ + - ]: 804 : const auto& meta_index = reader->metadataIndexes();
968 [ + - ]: 804 : const auto& statistics = reader->statistics();
969 : :
970 [ - + ]: 804 : if (!statistics.has_value()) {
971 : : // LCOV_EXCL_START GCOVR_EXCL_START
972 : : wrapper_file->has_completed = false;
973 : : CLOG_F("VCAPReader: Mcap [%s] cannot find statistics.", wrapper_file->path.c_str());
974 : : return false;
975 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
976 : : }
977 : :
978 [ + - + + ]: 804 : for (const auto& [schema_id, schema_ptr] : reader->schemas()) {
979 : : (void)schema_id;
980 : :
981 [ + - + - : 19 : if (schema_ptr && !schema_ptr->data.empty()) {
+ - ]
982 : 19 : wrapper_file->has_schema = true;
983 : 19 : impl_->info.has_schema = true;
984 : 19 : break;
985 : : }
986 : 804 : }
987 : :
988 : : // read header
989 : : {
990 [ + - + - ]: 804 : auto header_iter = meta_index.find("VLinkHeader");
991 : :
992 [ - + ]: 804 : if VUNLIKELY (header_iter == meta_index.end()) {
993 : : // LCOV_EXCL_START GCOVR_EXCL_START
994 : : wrapper_file->has_completed = false;
995 : : CLOG_F("VCAPReader: Mcap [%s] cannot find header.", wrapper_file->path.c_str());
996 : : return false;
997 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
998 : : }
999 : :
1000 : 804 : const auto& header_index = header_iter->second;
1001 : :
1002 : : mcap::Record header_record;
1003 : :
1004 [ + - + - ]: 804 : status = mcap::McapReader::ReadRecord(*reader->dataSource(), header_index.offset, &header_record);
1005 : :
1006 [ - + ]: 804 : if VUNLIKELY (!status.ok()) {
1007 : : // LCOV_EXCL_START GCOVR_EXCL_START
1008 : : wrapper_file->has_completed = false;
1009 : : CLOG_F("VCAPReader: Failed to read header record for index, error = %s.", status.message.c_str());
1010 : : return false;
1011 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1012 : : }
1013 : :
1014 : 804 : mcap::Metadata header_meta_data;
1015 : :
1016 [ + - ]: 804 : status = mcap::McapReader::ParseMetadata(header_record, &header_meta_data);
1017 : :
1018 [ - + ]: 804 : if VUNLIKELY (!status.ok()) {
1019 : : // LCOV_EXCL_START GCOVR_EXCL_START
1020 : : wrapper_file->has_completed = false;
1021 : : CLOG_F("VCAPReader: Failed to parse header meta data, error = %s.", status.message.c_str());
1022 : : return false;
1023 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1024 : : }
1025 : :
1026 [ + - + - ]: 804 : auto& tag_str = header_meta_data.metadata["tag"];
1027 [ + - + - ]: 804 : auto& version_str = header_meta_data.metadata["version"];
1028 [ + - + - ]: 804 : auto& compress_str = header_meta_data.metadata["compress"];
1029 [ + - + - ]: 804 : auto& process_str = header_meta_data.metadata["process"];
1030 [ + - + - ]: 804 : auto& date_str = header_meta_data.metadata["date"];
1031 [ + - + - ]: 804 : auto& start_timestamp_str = header_meta_data.metadata["start_timestamp"];
1032 [ + - + - ]: 804 : auto& timezone_str = header_meta_data.metadata["timezone"];
1033 : :
1034 [ - + ]: 804 : if VUNLIKELY (version_str.empty()) {
1035 : : // LCOV_EXCL_START GCOVR_EXCL_START
1036 : : wrapper_file->has_completed = false;
1037 : :
1038 : : if (impl_->read_only) {
1039 : : CLOG_E("VCAPReader: Mcap [%s] cannot find version in header.", wrapper_file->path.c_str());
1040 : : }
1041 : :
1042 : : version_str = "0.0.0";
1043 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1044 : : } else {
1045 : 804 : auto version = Version::from_string(version_str);
1046 : :
1047 [ + + ]: 804 : if VUNLIKELY (!version.is_valid()) {
1048 : 1 : wrapper_file->has_completed = false;
1049 : :
1050 [ - + ]: 1 : if (impl_->read_only) {
1051 : : CLOG_E("VCAPReader: Mcap [%s] header version is invalid.", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1052 : : wrapper_file->path.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1053 : : }
1054 : : } else {
1055 [ + + ]: 803 : if VUNLIKELY (version.major != VLINK_VERSION_MAJOR) {
1056 : 1 : wrapper_file->has_completed = false;
1057 [ + - + - : 2 : VLOG_F("VCAPReader: Mcap version is incompatible.");
- + ]
1058 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1059 : : }
1060 : : }
1061 : : }
1062 : :
1063 [ - + ]: 803 : if VUNLIKELY (compress_str.empty()) {
1064 : : // LCOV_EXCL_START GCOVR_EXCL_START
1065 : : wrapper_file->has_completed = false;
1066 : :
1067 : : if (impl_->read_only) {
1068 : : CLOG_E("VCAPReader: Mcap [%s] cannot find compress in header.", wrapper_file->path.c_str());
1069 : : }
1070 : :
1071 : : compress_str = "Unknown";
1072 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1073 : : }
1074 : :
1075 [ - + ]: 803 : if VUNLIKELY (process_str.empty()) {
1076 : : // LCOV_EXCL_START GCOVR_EXCL_START
1077 : : wrapper_file->has_completed = false;
1078 : :
1079 : : if (impl_->read_only) {
1080 : : CLOG_E("VCAPReader: Mcap [%s] cannot find process in header.", wrapper_file->path.c_str());
1081 : : }
1082 : :
1083 : : process_str = "Unknown";
1084 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1085 : : }
1086 : :
1087 [ - + ]: 803 : if VUNLIKELY (date_str.empty()) {
1088 : : // LCOV_EXCL_START GCOVR_EXCL_START
1089 : : wrapper_file->has_completed = false;
1090 : :
1091 : : if (impl_->read_only) {
1092 : : CLOG_E("VCAPReader: Mcap [%s] cannot find date in header.", wrapper_file->path.c_str());
1093 : : }
1094 : :
1095 : : date_str = "Unknown";
1096 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1097 : : }
1098 : :
1099 [ - + ]: 803 : if VUNLIKELY (timezone_str.empty()) {
1100 : : // LCOV_EXCL_START GCOVR_EXCL_START
1101 : : wrapper_file->has_completed = false;
1102 : :
1103 : : if (impl_->read_only) {
1104 : : CLOG_E("VCAPReader: Mcap [%s] cannot find timezone in header.", wrapper_file->path.c_str());
1105 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1106 : : }
1107 : : }
1108 : :
1109 [ - + ]: 803 : if (tag_str.empty()) {
1110 : : tag_str = "Empty"; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1111 : : }
1112 : :
1113 [ + - ]: 803 : impl_->info.tag_name = tag_str;
1114 [ + - ]: 803 : impl_->info.version = version_str;
1115 [ + - ]: 803 : impl_->info.storage_type = "vcap";
1116 : 803 : impl_->info.message_count = statistics->messageCount;
1117 [ + - ]: 803 : impl_->info.time_accuracy = "MicroSecond";
1118 [ + - ]: 803 : impl_->info.compression_type = compress_str;
1119 [ + - ]: 803 : impl_->info.process_name = process_str;
1120 [ + - ]: 803 : impl_->info.date_time = date_str;
1121 : :
1122 : : try {
1123 [ + + ]: 803 : impl_->info.start_timestamp = std::stoll(start_timestamp_str);
1124 [ - + ]: 1 : } catch (std::exception&) {
1125 : 1 : impl_->info.start_timestamp = Helpers::convert_date_to_timestamp(impl_->info.date_time) / 1000'000;
1126 : 1 : }
1127 : :
1128 [ + + ]: 803 : if VUNLIKELY (impl_->info.start_timestamp < 0) {
1129 : 1 : impl_->info.start_timestamp = 0;
1130 : :
1131 [ - + ]: 1 : if (impl_->read_only) {
1132 : : VLOG_E("VCAPReader: Invalid start_timestamp_ns."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1133 : : }
1134 : : }
1135 : :
1136 : 803 : int64_t timestamp_diff = 0;
1137 : :
1138 : 803 : timestamp_diff = static_cast<int64_t>(statistics->messageStartTime / 1000'000) - impl_->info.start_timestamp;
1139 : :
1140 [ + + ]: 803 : if (timestamp_diff < 0) {
1141 : 3 : timestamp_diff = 0;
1142 : 3 : impl_->info.start_timestamp = static_cast<int64_t>(statistics->messageStartTime / 1000'000);
1143 : : }
1144 : :
1145 : 803 : impl_->info.blank_duration = timestamp_diff;
1146 : :
1147 : 803 : timestamp_diff = static_cast<int64_t>(statistics->messageEndTime / 1000'000) - impl_->info.start_timestamp;
1148 : :
1149 [ - + ]: 803 : if (timestamp_diff < 0) {
1150 : : timestamp_diff = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1151 : : }
1152 : :
1153 : 803 : impl_->info.total_duration = timestamp_diff;
1154 : :
1155 : : try {
1156 [ + - ]: 803 : impl_->info.timezone = std::stoi(timezone_str);
1157 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1158 : : impl_->info.timezone = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1159 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1160 [ + - ]: 804 : }
1161 : :
1162 : : // read channel
1163 : : {
1164 : 803 : int channel = 0;
1165 : 803 : std::string channel_str;
1166 : : mcap::Record channel_record;
1167 : :
1168 [ + - ]: 803 : url_ser_map().clear();
1169 [ + - ]: 803 : url_schema_type_map().clear();
1170 : 803 : impl_->info.url_metas.clear();
1171 : 803 : impl_->raw_url_metas.clear();
1172 : :
1173 : 803 : impl_->info.total_raw_size = 0;
1174 : :
1175 [ + + ]: 2446 : for (const auto& [name, index] : meta_index) {
1176 [ + + ]: 1643 : if (!Helpers::has_startwith(name, "VLinkChannel_")) {
1177 : 803 : continue;
1178 : : }
1179 : :
1180 [ + - ]: 840 : channel_str = name;
1181 [ + - + - ]: 840 : Helpers::replace_string(channel_str, "VLinkChannel_", "");
1182 : :
1183 : : try {
1184 [ + - ]: 840 : channel = std::stoi(channel_str);
1185 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1186 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1187 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1188 : :
1189 [ + - ]: 840 : auto channel_count_iter = statistics->channelMessageCounts.find(channel);
1190 : :
1191 [ - + ]: 840 : if VUNLIKELY (channel_count_iter == statistics->channelMessageCounts.end()) {
1192 : : // LCOV_EXCL_START GCOVR_EXCL_START
1193 : : wrapper_file->has_completed = false;
1194 : :
1195 : : if (impl_->read_only) {
1196 : : CLOG_E("VCAPReader: Mcap [%s] cannot read statistics in channel.", wrapper_file->path.c_str());
1197 : : }
1198 : :
1199 : : continue;
1200 : : }
1201 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1202 : :
1203 : 840 : auto channel_msg_count = channel_count_iter->second;
1204 : :
1205 [ + - ]: 840 : const auto& channel_ptr = reader->channel(channel);
1206 : :
1207 [ - + ]: 840 : if VUNLIKELY (!channel_ptr) {
1208 : : // LCOV_EXCL_START GCOVR_EXCL_START
1209 : : wrapper_file->has_completed = false;
1210 : :
1211 : : if (impl_->read_only) {
1212 : : CLOG_E("VCAPReader: Mcap [%s] cannot find ptr in channel.", wrapper_file->path.c_str());
1213 : : }
1214 : :
1215 : : continue;
1216 : : }
1217 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1218 : :
1219 [ + - + - ]: 840 : status = mcap::McapReader::ReadRecord(*reader->dataSource(), index.offset, &channel_record);
1220 : :
1221 [ - + ]: 840 : if VUNLIKELY (!status.ok()) {
1222 : : // LCOV_EXCL_START GCOVR_EXCL_START
1223 : : wrapper_file->has_completed = false;
1224 : :
1225 : : if (impl_->read_only) {
1226 : : CLOG_E("VCAPReader: Failed to read channel record for index, error = %s.", status.message.c_str());
1227 : : }
1228 : :
1229 : : continue;
1230 : : }
1231 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1232 : :
1233 : 840 : mcap::Metadata channel_meta_data;
1234 : :
1235 [ + - ]: 840 : status = mcap::McapReader::ParseMetadata(channel_record, &channel_meta_data);
1236 : :
1237 [ - + ]: 840 : if VUNLIKELY (!status.ok()) {
1238 : : // LCOV_EXCL_START GCOVR_EXCL_START
1239 : : wrapper_file->has_completed = false;
1240 : :
1241 : : if (impl_->read_only) {
1242 : : CLOG_E("VCAPReader: Failed to parse channel meta data, error = %s.", status.message.c_str());
1243 : : }
1244 : :
1245 : : continue;
1246 : : }
1247 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1248 : :
1249 : 840 : Info::UrlMeta url_meta;
1250 : :
1251 [ + - + - ]: 840 : auto& index_str = channel_meta_data.metadata["index"];
1252 [ + - + - ]: 840 : auto& type_str = channel_meta_data.metadata["type"];
1253 [ + - + - ]: 840 : auto& count_str = channel_meta_data.metadata["count"];
1254 [ + - + - ]: 840 : auto& size_str = channel_meta_data.metadata["size"];
1255 [ + - + - ]: 840 : auto& loss_str = channel_meta_data.metadata["loss"];
1256 [ + - + - ]: 840 : auto& freq_str = channel_meta_data.metadata["freq"];
1257 [ + - + - ]: 840 : auto ser_iter = channel_meta_data.metadata.find("ser");
1258 [ + - + - ]: 840 : auto encoding_iter = channel_meta_data.metadata.find("encoding");
1259 [ + - + - ]: 840 : auto action_iter = channel_meta_data.metadata.find("action");
1260 [ + - ]: 840 : const auto& schema_ptr = reader->schema(channel_ptr->schemaId);
1261 : :
1262 : 840 : int pindex = -1;
1263 : :
1264 : : try {
1265 [ + - ]: 840 : pindex = std::stoi(index_str);
1266 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1267 : : pindex = -1; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1268 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1269 : :
1270 [ - + ]: 840 : if VUNLIKELY (pindex != channel - 1) {
1271 : : // LCOV_EXCL_START GCOVR_EXCL_START
1272 : : wrapper_file->has_completed = false;
1273 : :
1274 : : if (impl_->read_only) {
1275 : : CLOG_E("VCAPReader: Mcap [%s] channel index error.", wrapper_file->path.c_str());
1276 : : }
1277 : :
1278 : : continue;
1279 : : }
1280 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1281 : :
1282 : 840 : url_meta.valid = true;
1283 : 840 : url_meta.index = pindex;
1284 [ + - ]: 840 : url_meta.url = channel_ptr->topic;
1285 [ + - ]: 840 : url_meta.url_type = type_str;
1286 : 840 : url_meta.schema_type = SchemaType::kUnknown;
1287 : :
1288 [ + - ]: 840 : if (ser_iter != channel_meta_data.metadata.end()) {
1289 [ + - ]: 840 : url_meta.ser_type = ser_iter->second;
1290 : : } else if (schema_ptr) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1291 : : url_meta.ser_type = schema_ptr->name; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1292 : : }
1293 : :
1294 [ + - ]: 840 : if (encoding_iter != channel_meta_data.metadata.end()) {
1295 : 840 : url_meta.schema_type =
1296 [ + + ]: 858 : SchemaData::resolve_type(SchemaData::convert_encoding(encoding_iter->second), url_meta.ser_type,
1297 : 858 : schema_ptr ? std::string_view(schema_ptr->encoding) : std::string_view{});
1298 : : // LCOV_EXCL_START GCOVR_EXCL_START
1299 : : } else if (schema_ptr) {
1300 : : url_meta.schema_type =
1301 : : SchemaData::resolve_type(SchemaData::convert_encoding(schema_ptr->encoding), url_meta.ser_type);
1302 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1303 : : }
1304 : :
1305 [ + + + - : 840 : if (schema_ptr && !schema_ptr->data.empty()) {
+ + ]
1306 : 18 : wrapper_file->has_schema = true;
1307 : 18 : impl_->info.has_schema = true;
1308 : : }
1309 : :
1310 : : try {
1311 [ + - ]: 840 : url_meta.count = static_cast<size_t>(std::stoull(count_str));
1312 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1313 : : url_meta.count = channel_msg_count; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1314 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1315 : :
1316 : : try {
1317 [ + - ]: 840 : url_meta.loss = std::stod(loss_str);
1318 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1319 : : url_meta.loss = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1320 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1321 : :
1322 : : try {
1323 [ + - ]: 840 : url_meta.size = static_cast<size_t>(std::stoull(size_str));
1324 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1325 : : url_meta.size = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1326 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1327 : :
1328 : : try {
1329 [ + - ]: 840 : url_meta.freq = std::stod(freq_str);
1330 : : } catch (std::exception&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1331 : : url_meta.freq = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1332 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1333 : :
1334 : 840 : impl_->info.total_raw_size += url_meta.size;
1335 : :
1336 [ + - ]: 840 : wrapper_file->id_to_url_map.emplace(url_meta.index, url_meta.url);
1337 [ + - ]: 840 : wrapper_file->url_to_id_map.emplace(url_meta.url, url_meta.index);
1338 : :
1339 [ + - + - ]: 840 : url_ser_map().emplace(url_meta.url, url_meta.ser_type);
1340 [ + - + - ]: 840 : url_schema_type_map().emplace(url_meta.url, url_meta.schema_type);
1341 : :
1342 [ + - ]: 840 : if (action_iter != channel_meta_data.metadata.end()) {
1343 [ + - ]: 840 : url_meta.action_type = convert_action(action_iter->second);
1344 [ + - ]: 840 : wrapper_file->channel_action_map.emplace(channel, url_meta.action_type);
1345 : : // LCOV_EXCL_START GCOVR_EXCL_START
1346 : : } else if (auto channel_action_iter = channel_ptr->metadata.find("action");
1347 : : channel_action_iter != channel_ptr->metadata.end()) {
1348 : : url_meta.action_type = convert_action(channel_action_iter->second);
1349 : : wrapper_file->channel_action_map.emplace(channel, url_meta.action_type);
1350 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1351 : : } else {
1352 : : wrapper_file->channel_action_map.emplace(channel, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1353 : : ActionType::kUnknownAction); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1354 : : }
1355 : :
1356 [ + - ]: 840 : impl_->info.url_metas.emplace_back(std::move(url_meta));
1357 [ + - + - : 840 : }
+ - + - ]
1358 : :
1359 : 803 : wrapper_file->is_channel_broken = false;
1360 : :
1361 [ + + ]: 803 : if (impl_->info.url_metas.empty()) {
1362 [ + - - + ]: 2 : for (const auto& [id, pchannel] : reader->channels()) {
1363 : : // LCOV_EXCL_START GCOVR_EXCL_START
1364 : : wrapper_file->is_channel_broken = true;
1365 : :
1366 : : auto pschema_ptr = reader->schema(pchannel->schemaId);
1367 : : auto channel_count_iter = statistics->channelMessageCounts.find(id);
1368 : :
1369 : : Info::UrlMeta url_meta;
1370 : :
1371 : : url_meta.valid = true;
1372 : : url_meta.index = id - 1;
1373 : : url_meta.url = pchannel->topic;
1374 : : url_meta.url_type = "Event";
1375 : :
1376 : : if (pschema_ptr) {
1377 : : url_meta.ser_type = pschema_ptr->name;
1378 : : url_meta.schema_type =
1379 : : SchemaData::resolve_type(SchemaData::convert_encoding(pschema_ptr->encoding), url_meta.ser_type);
1380 : :
1381 : : if (!pschema_ptr->data.empty()) {
1382 : : wrapper_file->has_schema = true;
1383 : : impl_->info.has_schema = true;
1384 : : }
1385 : : }
1386 : :
1387 : : if (channel_count_iter != statistics->channelMessageCounts.end()) {
1388 : : url_meta.count = channel_count_iter->second;
1389 : : }
1390 : :
1391 : : wrapper_file->id_to_url_map.emplace(url_meta.index, url_meta.url);
1392 : : wrapper_file->url_to_id_map.emplace(url_meta.url, url_meta.index);
1393 : :
1394 : : url_ser_map().emplace(url_meta.url, url_meta.ser_type);
1395 : : url_schema_type_map().emplace(url_meta.url, url_meta.schema_type);
1396 : :
1397 : : if (auto action_iter = pchannel->metadata.find("action"); action_iter != pchannel->metadata.end()) {
1398 : : url_meta.action_type = convert_action(action_iter->second);
1399 : : wrapper_file->channel_action_map.emplace(id, url_meta.action_type);
1400 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1401 : : } else {
1402 : : wrapper_file->channel_action_map.emplace(id, ActionType::kUnknownAction); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1403 : : }
1404 : :
1405 : : impl_->info.url_metas.emplace_back(std::move(url_meta)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1406 : 2 : }
1407 : : }
1408 : :
1409 [ + - ]: 803 : std::sort(impl_->info.url_metas.begin(), impl_->info.url_metas.end());
1410 [ + - ]: 803 : impl_->raw_url_metas = impl_->info.url_metas;
1411 [ + - ]: 803 : rebuild_url_meta_lookup(impl_->info.url_metas);
1412 : 803 : }
1413 : :
1414 : 803 : impl_->info.has_idx_elapsed = false;
1415 : 803 : impl_->info.has_idx_url = false;
1416 : :
1417 : 803 : impl_->info.has_completed = wrapper_file->has_completed;
1418 : :
1419 : 803 : return true;
1420 : 804 : }
1421 : :
1422 : 185 : void VCAPReader::open(const std::string& path) {
1423 [ + - ]: 185 : close();
1424 : :
1425 : 1609 : auto to_open = [this](Impl::WrapperFile& wrapper_file) -> bool {
1426 : 805 : mcap::Status status;
1427 : :
1428 [ + - ]: 805 : wrapper_file.reader = std::make_unique<mcap::McapReader>();
1429 : :
1430 [ + - ]: 805 : status = wrapper_file.reader->open(wrapper_file.path);
1431 : :
1432 [ + + ]: 805 : if VUNLIKELY (!status.ok()) {
1433 [ + - + - : 2 : CLOG_F("VCAPReader: Failed to open vcap, error = %s.", status.message.c_str());
- + ]
1434 : : wrapper_file.reader.reset(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1435 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1436 : : }
1437 : :
1438 [ + + - + ]: 804 : if VUNLIKELY (!prepare_file(&wrapper_file)) {
1439 : : wrapper_file.reader->close(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1440 : : wrapper_file.reader.reset(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1441 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1442 : : }
1443 : :
1444 : 803 : return true;
1445 : 805 : };
1446 : :
1447 [ + - ]: 185 : impl_->path = path;
1448 : :
1449 : 185 : impl_->total_start_timestamp_ns = -1;
1450 : :
1451 : 185 : impl_->total_has_completed = true;
1452 : :
1453 : : try {
1454 : : #ifdef _WIN32
1455 : : std::filesystem::path file_path(Helpers::string_to_wstring(path));
1456 : :
1457 : : impl_->info.file_name = Helpers::path_to_string(file_path.filename());
1458 : :
1459 : : impl_->info.file_size = 0;
1460 : :
1461 : : std::string suffix = Helpers::path_to_string(file_path.extension());
1462 : : #else
1463 [ + - ]: 185 : std::filesystem::path file_path(path);
1464 : :
1465 [ + - + - ]: 185 : impl_->info.file_name = file_path.filename().string();
1466 : :
1467 : 185 : impl_->info.file_size = 0;
1468 : :
1469 [ + - + - ]: 185 : std::string suffix = file_path.extension().string();
1470 : : #endif
1471 : :
1472 : 185 : std::error_code exists_ec;
1473 : :
1474 [ + + ]: 185 : if VUNLIKELY (!std::filesystem::exists(file_path, exists_ec)) {
1475 [ + - + - : 2 : CLOG_F("VCAPReader: Mcap [%s] does not exist.", path.c_str());
- + ]
1476 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1477 : : }
1478 : :
1479 : 184 : std::filesystem::path parent_path;
1480 : :
1481 : : try {
1482 [ + - ]: 184 : parent_path = file_path.parent_path();
1483 : : } catch (std::filesystem::filesystem_error&) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1484 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1485 : :
1486 : 1231 : std::transform(suffix.begin(), suffix.end(), suffix.begin(), [](unsigned char c) { return std::tolower(c); });
1487 : :
1488 [ + + ]: 184 : if (suffix == ".vcapx") {
1489 : : try {
1490 : 127 : int64_t blank_duration = -1;
1491 : :
1492 : 127 : nlohmann::json root_json;
1493 : :
1494 : : {
1495 [ + - ]: 127 : std::ifstream filex(file_path);
1496 : :
1497 [ + + ]: 127 : filex >> root_json;
1498 : :
1499 [ + - ]: 126 : filex.close();
1500 : 127 : }
1501 : :
1502 [ + - + - ]: 126 : nlohmann::json header_json = root_json["VLinkHeader"];
1503 [ + - + - ]: 126 : nlohmann::json urls_json = root_json["VLinkUrls"];
1504 [ + - + - ]: 126 : nlohmann::json files_json = root_json["VLinkFiles"];
1505 : :
1506 : 126 : impl_->info.file_size = 0;
1507 : :
1508 : 126 : int file_index = 0;
1509 : :
1510 [ + + ]: 126 : if (!files_json.empty()) {
1511 [ + - ]: 125 : impl_->file_list.reserve(files_json.size());
1512 : :
1513 : 125 : impl_->info.has_idx_elapsed = true;
1514 : 125 : impl_->info.has_idx_url = true;
1515 : 125 : impl_->info.has_schema = false;
1516 : : } else {
1517 : 1 : impl_->info.has_idx_elapsed = false;
1518 : 1 : impl_->info.has_idx_url = false;
1519 : 1 : impl_->info.has_schema = false;
1520 : : }
1521 : :
1522 : 126 : std::filesystem::path file_db;
1523 : 126 : std::string file_db_str;
1524 : :
1525 [ + - + - : 874 : for (const auto& file_info : files_json) {
+ - + + ]
1526 : : #ifdef _WIN32
1527 : :
1528 : : if (parent_path.empty()) {
1529 : : file_db = std::filesystem::path(Helpers::string_to_wstring(file_info));
1530 : : } else {
1531 : : file_db = parent_path / std::filesystem::path(Helpers::string_to_wstring(file_info));
1532 : : }
1533 : : #else
1534 : :
1535 [ + + ]: 749 : if (parent_path.empty()) {
1536 [ + - ]: 6 : file_db = std::filesystem::path(file_info);
1537 : : } else {
1538 [ + - + - ]: 743 : file_db = parent_path / std::filesystem::path(file_info);
1539 : : }
1540 : : #endif
1541 : :
1542 [ + - ]: 749 : file_db_str = file_db.string();
1543 : :
1544 : 749 : std::error_code db_exists_ec;
1545 : :
1546 [ + + ]: 749 : if VUNLIKELY (!std::filesystem::exists(file_db, db_exists_ec)) {
1547 [ + - + - : 2 : CLOG_F("VCAPReader: Mcap [%s] does not exist.", file_db_str.c_str());
- + ]
1548 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1549 : : }
1550 : :
1551 [ + - ]: 748 : Impl::WrapperFile wrapper_file;
1552 [ + - ]: 748 : wrapper_file.path = file_db_str;
1553 : 748 : wrapper_file.index = file_index;
1554 : :
1555 [ + - - + ]: 748 : if VUNLIKELY (!to_open(wrapper_file)) {
1556 : : CLOG_W("VCAPReader: Skipping invalid vcap [%s].", file_db_str.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1557 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1558 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1559 : :
1560 [ + - ]: 748 : if (!wrapper_file.has_idx_elapsed) {
1561 : 748 : impl_->info.has_idx_elapsed = false;
1562 : : }
1563 : :
1564 [ + - ]: 748 : if (!wrapper_file.has_idx_url) {
1565 : 748 : impl_->info.has_idx_url = false;
1566 : : }
1567 : :
1568 [ + + ]: 748 : if (wrapper_file.has_schema) {
1569 : 2 : impl_->info.has_schema = true;
1570 : : }
1571 : :
1572 : 748 : std::error_code db_size_ec;
1573 : 748 : std::uintmax_t file_size = std::filesystem::file_size(file_db, db_size_ec);
1574 : :
1575 [ - + ]: 748 : if VUNLIKELY (db_size_ec) {
1576 : : CLOG_W("VCAPReader: file_size failed for [%s]: %s.", file_db_str.c_str(), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1577 : : db_size_ec.message().c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1578 : : file_size = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1579 : : }
1580 : :
1581 : 748 : impl_->info.file_size += file_size;
1582 : :
1583 : 748 : wrapper_file.start_timestamp_ns = impl_->info.start_timestamp * 1000'000;
1584 : 748 : wrapper_file.begin = impl_->info.blank_duration;
1585 : 748 : wrapper_file.end = impl_->info.total_duration;
1586 : :
1587 [ + + ]: 748 : if (impl_->total_start_timestamp_ns < 0) {
1588 : 124 : impl_->total_start_timestamp_ns = wrapper_file.start_timestamp_ns;
1589 : : }
1590 : :
1591 [ - + ]: 748 : if (!wrapper_file.has_completed) {
1592 : : impl_->total_has_completed = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1593 : : }
1594 : :
1595 [ + + + + : 748 : if (blank_duration < 0 || impl_->info.blank_duration < blank_duration) {
+ + ]
1596 : 125 : blank_duration = impl_->info.blank_duration;
1597 : : }
1598 : :
1599 [ + - ]: 748 : impl_->file_list.emplace_back(std::move(wrapper_file));
1600 : 748 : ++file_index;
1601 [ + - ]: 748 : }
1602 : :
1603 : 125 : impl_->info.split_count = impl_->file_list.size();
1604 : :
1605 [ + + ]: 125 : if VUNLIKELY (impl_->file_list.empty()) {
1606 [ + - + - : 2 : VLOG_F("VCAPReader: DB list is empty.");
- + ]
1607 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1608 : : }
1609 : :
1610 [ + - + - ]: 124 : int version_major = header_json["major"];
1611 [ + - + - ]: 124 : int version_minor = header_json["minor"];
1612 [ + - + - ]: 124 : int version_patch = header_json["patch"];
1613 : :
1614 : 124 : impl_->info.version =
1615 [ + - + - : 248 : std::to_string(version_major) + "." + std::to_string(version_minor) + "." + std::to_string(version_patch);
+ - + - +
- + - +
- ]
1616 [ + - ]: 124 : impl_->info.storage_type = "vcap";
1617 [ + - + - ]: 124 : impl_->info.message_count = header_json["count"];
1618 [ + - + - ]: 124 : impl_->info.total_duration = header_json["duration"];
1619 : 124 : impl_->info.total_duration /= 1000U;
1620 [ + - + - ]: 124 : impl_->info.time_accuracy = header_json["accuracy"];
1621 [ + - + - ]: 124 : impl_->info.compression_type = header_json["compress"];
1622 [ + - + - ]: 124 : impl_->info.process_name = header_json["process"];
1623 [ + - + - ]: 124 : impl_->info.date_time = header_json["date"];
1624 : :
1625 [ + - + + ]: 124 : if (header_json.contains("start_timestamp")) {
1626 [ + - + - ]: 123 : impl_->info.start_timestamp = header_json["start_timestamp"];
1627 : : } else {
1628 : 1 : impl_->info.start_timestamp = Helpers::convert_date_to_timestamp(impl_->info.date_time) / 1000'000;
1629 : : }
1630 : :
1631 [ + + ]: 124 : if VUNLIKELY (impl_->info.start_timestamp < 0) {
1632 : 1 : impl_->info.start_timestamp = 0;
1633 : :
1634 [ + - ]: 1 : if (impl_->read_only) {
1635 [ + - + - : 2 : VLOG_E("VCAPReader: Invalid start_timestamp.");
+ - ]
1636 : : }
1637 : : }
1638 : :
1639 [ + - + + ]: 124 : if (header_json.contains("tag")) {
1640 [ + - + - ]: 123 : impl_->info.tag_name = header_json["tag"];
1641 : : } else {
1642 [ + - ]: 1 : impl_->info.tag_name = "Empty";
1643 : : }
1644 : :
1645 [ + - + + ]: 124 : if (header_json.contains("complete")) {
1646 [ + - + - ]: 123 : impl_->info.has_completed = header_json["complete"];
1647 : : } else {
1648 : 1 : impl_->info.has_completed = true;
1649 : : }
1650 : :
1651 [ + - + + ]: 124 : if (header_json.contains("timezone")) {
1652 [ + - + - ]: 123 : impl_->info.timezone = header_json["timezone"];
1653 : : } else {
1654 : 1 : impl_->info.timezone = 480;
1655 : : }
1656 : :
1657 [ + - + + ]: 124 : if (header_json.contains("split_by_size")) {
1658 [ + - + - ]: 123 : impl_->info.split_by_size = header_json["split_by_size"];
1659 : : }
1660 : :
1661 [ + - + + ]: 124 : if (header_json.contains("split_by_time")) {
1662 [ + - + - ]: 123 : impl_->info.split_by_time = header_json["split_by_time"];
1663 : : }
1664 : :
1665 : 124 : impl_->info.blank_duration = blank_duration;
1666 : :
1667 [ + + + + ]: 251 : if (impl_->info.compression_type.empty() || impl_->info.compression_type == "None" ||
1668 [ + - + + : 251 : impl_->info.compression_type == "NONE" || impl_->info.compression_type == "none") {
+ + ]
1669 : 123 : impl_->enable_compress = false;
1670 : : } else {
1671 : 1 : impl_->enable_compress = true;
1672 : : }
1673 : :
1674 [ + + ]: 124 : if VUNLIKELY (impl_->info.time_accuracy != "MicroSecond") {
1675 [ + - + - : 2 : VLOG_F("VCAPReader: MCAP accuracy is not supported.");
- + ]
1676 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1677 : : }
1678 : :
1679 [ + - ]: 123 : url_ser_map().clear();
1680 [ + - ]: 123 : url_schema_type_map().clear();
1681 : 123 : impl_->info.url_metas.clear();
1682 : 123 : impl_->raw_url_metas.clear();
1683 : :
1684 [ + - ]: 123 : impl_->info.url_metas.reserve(urls_json.size());
1685 : :
1686 : 123 : impl_->info.total_raw_size = 0;
1687 : :
1688 [ + - + - : 246 : for (const auto& url_info : urls_json) {
+ - + + ]
1689 : 123 : Info::UrlMeta url_meta;
1690 : :
1691 : 123 : url_meta.valid = true;
1692 [ + - + - ]: 123 : url_meta.index = url_info["index"];
1693 [ + - + - ]: 123 : url_meta.url = url_info["url"];
1694 [ + - + - ]: 123 : url_meta.url_type = url_info["type"];
1695 : :
1696 [ + - + + ]: 123 : if (url_info.contains("action")) {
1697 [ + - + - : 122 : url_meta.action_type = convert_action(url_info["action"].get<std::string>());
+ - ]
1698 : : }
1699 : :
1700 [ + - + - ]: 123 : url_meta.ser_type = url_info["ser"];
1701 : :
1702 [ + - + + ]: 123 : if (url_info.contains("encoding")) {
1703 : 244 : url_meta.schema_type = SchemaData::resolve_type(
1704 [ + - + - ]: 244 : SchemaData::convert_encoding(url_info["encoding"].get<std::string>()), url_meta.ser_type);
1705 : : } else {
1706 : 1 : url_meta.schema_type = SchemaType::kUnknown;
1707 : : }
1708 : :
1709 [ + - + - ]: 123 : url_meta.count = url_info["count"];
1710 [ + - + - ]: 123 : url_meta.loss = url_info["loss"];
1711 : :
1712 [ + - + + ]: 123 : if (url_info.contains("size")) {
1713 [ + - + - ]: 122 : url_meta.size = url_info["size"];
1714 : : }
1715 : :
1716 [ + - + + ]: 123 : if (url_info.contains("freq")) {
1717 [ + - + - ]: 122 : url_meta.freq = url_info["freq"];
1718 : : }
1719 : :
1720 : 123 : impl_->info.total_raw_size += url_meta.size;
1721 : :
1722 [ + - + - ]: 123 : url_ser_map().emplace(url_meta.url, url_meta.ser_type);
1723 [ + - + - ]: 123 : url_schema_type_map().emplace(url_meta.url, url_meta.schema_type);
1724 [ + - ]: 123 : impl_->info.url_metas.emplace_back(std::move(url_meta));
1725 : 123 : }
1726 : :
1727 [ + - ]: 123 : std::sort(impl_->info.url_metas.begin(), impl_->info.url_metas.end());
1728 [ + - ]: 123 : impl_->raw_url_metas = impl_->info.url_metas;
1729 [ + - ]: 123 : rebuild_url_meta_lookup(impl_->info.url_metas);
1730 [ + - + - : 146 : } catch (nlohmann::json::exception& e) {
+ - + - +
- + - +
+ ]
1731 [ + - + - : 2 : VLOG_F("VCAPReader: JSON parse error, ", e.what(), ".");
- + ]
1732 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1733 : 1 : }
1734 : : } else {
1735 [ + - ]: 57 : Impl::WrapperFile wrapper_file;
1736 [ + - ]: 57 : wrapper_file.path = impl_->path;
1737 : :
1738 [ + + - + ]: 57 : if VUNLIKELY (!to_open(wrapper_file)) {
1739 : : CLOG_F("VCAPReader: Failed to prepare vcap [%s].", path.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1740 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1741 : : }
1742 : :
1743 : 55 : impl_->info.file_size = 0;
1744 : :
1745 : 55 : std::error_code single_size_ec;
1746 : 55 : std::uintmax_t file_size = std::filesystem::file_size(file_path, single_size_ec);
1747 : :
1748 [ - + ]: 55 : if VUNLIKELY (single_size_ec) {
1749 : : CLOG_W("VCAPReader: file_size failed for [%s]: %s.", path.c_str(), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1750 : : single_size_ec.message().c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1751 : : file_size = 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1752 : : }
1753 : :
1754 : 55 : impl_->info.file_size += file_size;
1755 : :
1756 : 55 : wrapper_file.start_timestamp_ns = impl_->info.start_timestamp * 1000'000;
1757 : 55 : wrapper_file.begin = impl_->info.blank_duration;
1758 : 55 : wrapper_file.end = impl_->info.total_duration;
1759 : :
1760 [ + - ]: 55 : if (impl_->total_start_timestamp_ns < 0) {
1761 : 55 : impl_->total_start_timestamp_ns = wrapper_file.start_timestamp_ns;
1762 : : }
1763 : :
1764 [ + + ]: 55 : if (!wrapper_file.has_completed) {
1765 : 1 : impl_->total_has_completed = false;
1766 : : }
1767 : :
1768 [ + - ]: 55 : impl_->file_list.emplace_back(std::move(wrapper_file));
1769 [ + - ]: 57 : }
1770 [ + - + - : 205 : } catch (std::filesystem::filesystem_error& e) {
+ - + - ]
1771 : : VLOG_F("VCAPReader: Filesystem error, ", e.what(), "."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1772 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1773 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1774 : : }
1775 : :
1776 : 363 : void VCAPReader::close() {
1777 : 363 : impl_->cursor_iter.reset();
1778 : 363 : impl_->cursor_iter_end.reset();
1779 : 363 : impl_->cursor_msg_view.reset();
1780 : :
1781 [ + + ]: 1164 : for (auto& wrapper_file : impl_->file_list) {
1782 [ + - ]: 801 : if (wrapper_file.reader) {
1783 [ + - ]: 801 : wrapper_file.reader->close();
1784 : 801 : wrapper_file.reader.reset();
1785 : : }
1786 : : }
1787 : :
1788 : 363 : impl_->file_list.clear();
1789 : 363 : }
1790 : :
1791 : 39 : int VCAPReader::get_reset_index(const Config& config) {
1792 : 39 : impl_->is_pending.store(true, std::memory_order_relaxed);
1793 : :
1794 : : // LCOV_EXCL_START GCOVR_EXCL_START
1795 : : auto status_function = [](const mcap::Status& status) {
1796 : : if (!status.ok()) {
1797 : : CLOG_W("VCAPReader: Failed to read message, error = %s.", status.message.c_str());
1798 : : }
1799 : : };
1800 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1801 : :
1802 : 12 : auto filter_function = [this, &config](std::string_view url) -> bool {
1803 : 12 : return match_playback_url_filter(url, config.filter_urls);
1804 : 39 : };
1805 : :
1806 : 39 : int start_index = -1;
1807 : :
1808 : 39 : int64_t last_time = impl_->begin_time.load(std::memory_order_relaxed);
1809 : :
1810 : 39 : mcap::ReadMessageOptions read_options;
1811 : 39 : const auto end_time = get_read_end_time(impl_->total_start_timestamp_ns, config.end_time);
1812 : :
1813 [ + + ]: 91 : for (auto& wrapper_file : impl_->file_list) {
1814 [ + + + - : 134 : if (start_index < 0 && impl_->begin_time.load(std::memory_order_relaxed) >= last_time &&
+ + ]
1815 [ + + ]: 82 : impl_->begin_time.load(std::memory_order_relaxed) <= wrapper_file.end) {
1816 [ + + ]: 78 : if (impl_->begin_time.load(std::memory_order_relaxed) > 0) {
1817 : 7 : read_options.startTime =
1818 : 14 : impl_->begin_time.load(std::memory_order_relaxed) * 1000'000 + impl_->total_start_timestamp_ns;
1819 : 7 : read_options.endTime = mcap::MaxTime;
1820 [ + - - + : 7 : if (!config.filter_urls.empty() || has_playback_url_rules()) {
- + ]
1821 : 0 : read_options.topicFilter = filter_function;
1822 : : }
1823 : 7 : read_options.readOrder = mcap::ReadMessageOptions::ReadOrder::FileOrder;
1824 : : } else {
1825 : 32 : read_options.startTime = 0;
1826 : 32 : read_options.endTime = mcap::MaxTime;
1827 [ + + - + : 32 : if (!config.filter_urls.empty() || has_playback_url_rules()) {
+ + ]
1828 : 5 : read_options.topicFilter = filter_function;
1829 : : }
1830 : 32 : read_options.readOrder = mcap::ReadMessageOptions::ReadOrder::FileOrder;
1831 : : }
1832 : :
1833 : 39 : start_index = wrapper_file.index;
1834 : : } else {
1835 : : // LCOV_EXCL_START GCOVR_EXCL_START
1836 : : read_options.startTime = 0;
1837 : : read_options.endTime = mcap::MaxTime;
1838 : : if (!config.filter_urls.empty() || has_playback_url_rules()) {
1839 : : read_options.topicFilter = filter_function;
1840 : : }
1841 : : read_options.readOrder = mcap::ReadMessageOptions::ReadOrder::FileOrder;
1842 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1843 : : }
1844 : :
1845 [ + - ]: 52 : const auto [start_offset, end_offset] = wrapper_file.reader->byteRange(read_options.startTime, end_time);
1846 : :
1847 : : // NOLINTNEXTLINE(readability-redundant-smartptr-get)
1848 [ + - ]: 104 : wrapper_file.msg_view = std::make_unique<mcap::LinearMessageView>(*wrapper_file.reader.get(), read_options,
1849 : 52 : start_offset, end_offset, status_function);
1850 : :
1851 [ + + ]: 52 : if (start_offset == end_offset) {
1852 : : wrapper_file.msg_view_begin.emplace(wrapper_file.msg_view->end()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1853 : : wrapper_file.msg_view_end.emplace(wrapper_file.msg_view->end()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1854 : : } else {
1855 [ + - ]: 50 : wrapper_file.msg_view_begin.emplace(wrapper_file.msg_view->begin());
1856 [ + - ]: 50 : wrapper_file.msg_view_end.emplace(wrapper_file.msg_view->end());
1857 : : }
1858 : :
1859 : 52 : last_time = wrapper_file.end;
1860 : : }
1861 : :
1862 : 39 : impl_->is_pending.store(false, std::memory_order_relaxed);
1863 : :
1864 : 39 : return start_index;
1865 : 39 : }
1866 : :
1867 : 37 : void VCAPReader::read(const Config& config) {
1868 : 37 : int loop_times = 0;
1869 : :
1870 : 37 : reset_plugin();
1871 : :
1872 [ + + ]: 37 : if (config.auto_pause) {
1873 : 5 : impl_->pause_flag.store(true, std::memory_order_relaxed);
1874 : : }
1875 : :
1876 : 37 : bool is_interrupted = false;
1877 : :
1878 : : do {
1879 : : // prepare
1880 [ + - ]: 39 : int start_index = get_reset_index(config);
1881 : :
1882 [ + + ]: 39 : if (impl_->ready_callback) {
1883 [ + - ]: 13 : impl_->ready_callback();
1884 : : }
1885 : :
1886 [ + - - + : 39 : if VUNLIKELY (start_index < 0 || start_index > static_cast<int>(impl_->file_list.size()) - 1) {
- + ]
1887 : : // LCOV_EXCL_START GCOVR_EXCL_START
1888 : : VLOG_W("VCAPReader: Cannot find any data for play.");
1889 : :
1890 : : update_status(kStopped);
1891 : :
1892 : : if (config.auto_quit) {
1893 : : quit();
1894 : : }
1895 : :
1896 : : return;
1897 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1898 : : }
1899 : :
1900 : : {
1901 [ + - ]: 39 : std::lock_guard time_lock(impl_->time_mtx);
1902 : 39 : impl_->pause_elapsed.store(0, std::memory_order_relaxed);
1903 : 39 : impl_->offset_elapsed.store(0, std::memory_order_relaxed);
1904 : 78 : impl_->real_elapsed.store(impl_->begin_time.load(std::memory_order_relaxed) * 1000U, std::memory_order_relaxed);
1905 : :
1906 : 39 : impl_->elapsed_timer.restart();
1907 : 39 : impl_->pause_elapsed_timer.restart();
1908 : 39 : impl_->offset_timer.restart();
1909 : 39 : impl_->real_timer.restart();
1910 : 39 : }
1911 : :
1912 [ + + ]: 39 : if (impl_->stop_flag.load(std::memory_order_relaxed)) {
1913 : 1 : is_interrupted = true;
1914 [ + - ]: 1 : update_status(kStopped);
1915 : 9 : break;
1916 [ + + ]: 38 : } else if (impl_->jump_flag.load(std::memory_order_relaxed)) {
1917 : 2 : break;
1918 [ + + ]: 36 : } else if (impl_->pause_flag.load(std::memory_order_relaxed)) {
1919 : 5 : impl_->pause_elapsed_timer.restart();
1920 [ + - ]: 5 : update_status(kPaused);
1921 [ + - ]: 5 : do_pause();
1922 : : {
1923 [ + - ]: 5 : std::lock_guard time_lock(impl_->time_mtx);
1924 : 5 : impl_->pause_elapsed.store(0, std::memory_order_relaxed);
1925 : 5 : impl_->offset_elapsed.store(0, std::memory_order_relaxed);
1926 : 10 : impl_->real_elapsed.store(impl_->begin_time.load(std::memory_order_relaxed) * 1000U, std::memory_order_relaxed);
1927 : :
1928 : 5 : impl_->elapsed_timer.restart();
1929 : 5 : impl_->pause_elapsed_timer.restart();
1930 : 5 : impl_->offset_timer.restart();
1931 : 5 : impl_->real_timer.restart();
1932 : 5 : }
1933 : :
1934 : : } else {
1935 [ + - ]: 31 : update_status(kPlaying);
1936 : : }
1937 : :
1938 : 36 : int64_t elapsed = 0;
1939 : 36 : int64_t timestamp = 0;
1940 : 36 : int64_t last_timestamp = 0;
1941 : 36 : const uint8_t* data = nullptr;
1942 : 36 : size_t size = 0;
1943 : :
1944 : : // process files
1945 [ + + ]: 79 : for (int index = start_index; index < static_cast<int>(impl_->file_list.size()); ++index) {
1946 : 47 : impl_->split_index.store(index, std::memory_order_relaxed);
1947 : :
1948 [ + - ]: 94 : auto& wrapper_file = impl_->file_list.at(impl_->split_index.load(std::memory_order_relaxed));
1949 : :
1950 [ - + ]: 47 : if VUNLIKELY (!wrapper_file.reader) {
1951 : : VLOG_W("VCAPReader: Target vcap reader is empty."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1952 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1953 : : }
1954 : :
1955 : : // process datas
1956 [ + - ]: 47 : auto iter = std::move(wrapper_file.msg_view_begin).value(); // NOLINT
1957 [ + - ]: 47 : auto iter_end = std::move(wrapper_file.msg_view_end).value(); // NOLINT
1958 : :
1959 [ + - + - : 153 : for (; iter != iter_end; iter++) {
+ + ]
1960 [ + - ]: 110 : timestamp = (static_cast<int64_t>(iter->message.logTime) - impl_->total_start_timestamp_ns) / 1000;
1961 : :
1962 [ - + ]: 110 : if VUNLIKELY (last_timestamp > timestamp + 10'000U) {
1963 : : VLOG_W("VCAPReader: The vcap timestamp is incorrect."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1964 : : }
1965 : :
1966 : 110 : last_timestamp = timestamp;
1967 : :
1968 [ - + ]: 220 : if (timestamp < impl_->begin_time.load(std::memory_order_relaxed) * 1000U) {
1969 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1970 : : }
1971 : :
1972 [ + + + + ]: 110 : if (config.end_time > 0 && timestamp > config.end_time * 1000U) {
1973 [ + - + - : 24 : if (impl_->stop_flag.load(std::memory_order_relaxed) || impl_->jump_flag.load(std::memory_order_relaxed) ||
- + ]
1974 [ + - - + ]: 12 : is_ready_to_quit()) {
1975 : 0 : is_interrupted = true;
1976 : 4 : break;
1977 : : }
1978 : :
1979 : 12 : continue;
1980 : : }
1981 : :
1982 [ + - ]: 98 : data = reinterpret_cast<const uint8_t*>(iter->message.data);
1983 : :
1984 [ + - - + ]: 98 : if VUNLIKELY (iter->message.dataSize > static_cast<uint64_t>(std::numeric_limits<size_t>::max())) {
1985 : : // LCOV_EXCL_START GCOVR_EXCL_START
1986 : : CLOG_W("VCAPReader: Message data size is too large to address, size = %" PRIu64 ".",
1987 : : static_cast<uint64_t>(iter->message.dataSize));
1988 : : continue;
1989 : : }
1990 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1991 : :
1992 [ + - ]: 98 : size = static_cast<size_t>(iter->message.dataSize);
1993 : :
1994 : 98 : elapsed =
1995 : 98 : (timestamp / impl_->rate.load(std::memory_order_relaxed)) -
1996 : 98 : (impl_->elapsed_timer.get() - impl_->pause_elapsed.load(std::memory_order_relaxed)) -
1997 : 196 : (impl_->begin_time.load(std::memory_order_relaxed) * 1000U / impl_->rate.load(std::memory_order_relaxed));
1998 : :
1999 : : {
2000 [ + - ]: 98 : std::unique_lock lock(impl_->mtx);
2001 : :
2002 [ + + ]: 98 : if (config.force_delay > 0) {
2003 [ + - ]: 35 : impl_->cv.wait_for(lock, std::chrono::milliseconds(config.force_delay), [this]() -> bool {
2004 [ + - ]: 140 : return impl_->stop_flag.load(std::memory_order_relaxed) ||
2005 [ + + ]: 138 : impl_->pause_next_flag.load(std::memory_order_relaxed) ||
2006 [ + + - + ]: 209 : impl_->jump_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
2007 : : });
2008 [ + + + - ]: 63 : } else if (config.force_delay < 0 && elapsed > 0) {
2009 : 1 : impl_->offset_timer.restart();
2010 : :
2011 [ + - ]: 1 : impl_->cv.wait_for(lock, std::chrono::microseconds(elapsed), [this]() -> bool {
2012 [ + - ]: 4 : return impl_->stop_flag.load(std::memory_order_relaxed) ||
2013 [ + - ]: 4 : impl_->pause_next_flag.load(std::memory_order_relaxed) ||
2014 [ + - ]: 4 : impl_->jump_flag.load(std::memory_order_relaxed) ||
2015 [ + - - + ]: 6 : impl_->pause_flag.load(std::memory_order_relaxed) || is_ready_to_quit();
2016 : : });
2017 : :
2018 [ - + ]: 1 : if VUNLIKELY (impl_->pause_flag.load(std::memory_order_relaxed)) {
2019 : : impl_->offset_elapsed.store(elapsed - impl_->offset_timer.get(), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2020 : : std::memory_order_relaxed); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2021 : : }
2022 : : }
2023 : 98 : }
2024 : :
2025 [ + + + + : 192 : if (impl_->stop_flag.load(std::memory_order_relaxed) || impl_->jump_flag.load(std::memory_order_relaxed) ||
+ + ]
2026 [ + - - + ]: 94 : is_ready_to_quit()) {
2027 : 4 : is_interrupted = true;
2028 : 4 : break;
2029 [ + + ]: 94 : } else if (impl_->pause_flag.load(std::memory_order_relaxed)) {
2030 [ + - ]: 9 : do_pause();
2031 : 9 : impl_->pause_next_flag.store(false, std::memory_order_relaxed);
2032 : :
2033 [ + - + - : 18 : if (impl_->stop_flag.load(std::memory_order_relaxed) || impl_->jump_flag.load(std::memory_order_relaxed) ||
- + ]
2034 [ + - - + ]: 9 : is_ready_to_quit()) {
2035 : : is_interrupted = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2036 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
2037 : : }
2038 : : }
2039 : :
2040 : : {
2041 [ + - ]: 94 : std::lock_guard time_lock(impl_->time_mtx);
2042 : 94 : impl_->real_timer.restart();
2043 : 94 : impl_->real_elapsed.store(timestamp, std::memory_order_relaxed);
2044 : 94 : }
2045 : :
2046 : 94 : ActionType action_type = ActionType::kUnknownAction;
2047 : :
2048 [ + - + - ]: 94 : if (auto action_iter = wrapper_file.channel_action_map.find(iter->message.channelId);
2049 [ + - ]: 94 : action_iter != wrapper_file.channel_action_map.end()) {
2050 : 94 : action_type = action_iter->second;
2051 : : }
2052 : :
2053 : 94 : Frame frame;
2054 : 94 : frame.timestamp = timestamp;
2055 [ + - + - ]: 94 : frame.url = iter->channel->topic;
2056 : 94 : frame.action_type = action_type;
2057 : 94 : frame.data = Bytes::shallow_copy(data, size);
2058 : :
2059 [ + - ]: 94 : BagReader::process_output(frame);
2060 : 94 : }
2061 : :
2062 [ + + ]: 47 : if (is_interrupted) {
2063 : 4 : break;
2064 : : }
2065 [ + + + + ]: 51 : }
2066 : :
2067 [ + + ]: 36 : if (is_interrupted) {
2068 [ + + ]: 4 : if (impl_->stop_flag.load(std::memory_order_relaxed)) {
2069 [ + - ]: 2 : update_status(kStopped);
2070 : : }
2071 : :
2072 : 4 : break;
2073 : : }
2074 : :
2075 [ + + ]: 32 : if (!impl_->jump_flag.load(std::memory_order_relaxed)) {
2076 [ + - ]: 31 : update_status(kStopped);
2077 : :
2078 [ + + ]: 31 : if (config.skip_blank) {
2079 : 3 : impl_->begin_time.store(std::max(config.begin_time, impl_->info.blank_duration), std::memory_order_relaxed);
2080 : : } else {
2081 : 28 : impl_->begin_time.store(config.begin_time, std::memory_order_relaxed);
2082 : : }
2083 : : }
2084 : :
2085 [ + + + + : 62 : if (impl_->stop_flag.load(std::memory_order_relaxed) || impl_->jump_flag.load(std::memory_order_relaxed) ||
+ + ]
2086 [ + - - + ]: 30 : is_ready_to_quit()) {
2087 : 2 : is_interrupted = true;
2088 : 2 : break;
2089 : : }
2090 : :
2091 [ + - ]: 30 : flush_plugin();
2092 [ + - + + ]: 90 : } while (impl_->times.load(std::memory_order_relaxed) <= 0 ||
2093 [ + - ]: 60 : (impl_->times.load(std::memory_order_relaxed) > 0 &&
2094 [ + + ]: 60 : ++loop_times < impl_->times.load(std::memory_order_relaxed)));
2095 : :
2096 [ + + ]: 37 : if (impl_->stop_flag.load(std::memory_order_relaxed)) {
2097 : 4 : is_interrupted = true;
2098 : : }
2099 : :
2100 [ + + + + : 37 : if (!impl_->jump_flag.load(std::memory_order_relaxed) && impl_->finish_callback) {
+ + ]
2101 : 31 : impl_->finish_callback(is_interrupted);
2102 : : }
2103 : :
2104 [ + + + + : 37 : if (!impl_->jump_flag.load(std::memory_order_relaxed) && config.auto_quit) {
+ + ]
2105 : 1 : quit();
2106 : : }
2107 : :
2108 : : // clean msg_view
2109 [ + + ]: 87 : for (auto& wrapper_file : impl_->file_list) {
2110 : 50 : wrapper_file.msg_view.reset();
2111 : : }
2112 : : }
2113 : :
2114 : : } // namespace vlink
|
