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/bag_reader.h"
25 : :
26 : : #include <algorithm>
27 : : #include <memory>
28 : : #include <shared_mutex>
29 : : #include <string>
30 : : #include <unordered_map>
31 : : #include <unordered_set>
32 : : #include <utility>
33 : :
34 : : #include "./base/helpers.h"
35 : : #include "./base/logger.h"
36 : : #include "./extension/bag_plugin_interface.h"
37 : : #include "./extension/vcap_reader.h"
38 : : #include "./extension/vdb_reader.h"
39 : : #include "./impl/url.h"
40 : :
41 : : namespace vlink {
42 : :
43 : : // UrlMeta
44 : 189 : bool BagReader::Info::UrlMeta::operator<(const BagReader::Info::UrlMeta& target) const noexcept {
45 : 189 : int lindex = Url::get_sort_index(url);
46 : 189 : int rindex = Url::get_sort_index(target.url);
47 : :
48 [ + + ]: 189 : if (lindex < rindex) {
49 : 1 : return true;
50 [ + + ]: 188 : } else if (lindex > rindex) {
51 : 3 : return false;
52 [ + + ]: 185 : } else if (url < target.url) {
53 : 6 : return true;
54 [ + + ]: 179 : } else if (url > target.url) {
55 : 166 : return false;
56 : : }
57 : :
58 : 13 : return index < target.index;
59 : : }
60 : :
61 : : // BagReader::Impl
62 : : struct BagReader::Impl final {
63 : : BagReader::OutputCallback output_callback;
64 : : std::shared_ptr<BagPluginInterface> plugin_interface;
65 : : std::unordered_map<std::string, std::string> playback_url_remap;
66 : : std::unordered_set<std::string> excluded_playback_urls;
67 : : std::unordered_map<std::string, std::string> url_to_ser_map;
68 : : std::unordered_map<std::string, SchemaType> url_to_schema_type_map;
69 : : mutable std::shared_mutex playback_state_mtx;
70 : : mutable std::shared_mutex output_callback_mtx;
71 : :
72 : : bool cursor_opened{false};
73 : : bool cursor_eof{false};
74 : : bool cursor_fail{false};
75 : : };
76 : :
77 : : // BagReader
78 : 158 : std::shared_ptr<BagReader> BagReader::create(const std::string& path, bool read_only, bool try_to_fix) {
79 [ + - ]: 158 : std::string suffix_check = path;
80 : :
81 : 158 : std::transform(suffix_check.begin(), suffix_check.end(), suffix_check.begin(),
82 : 12828 : [](unsigned char c) { return std::tolower(c); });
83 : :
84 [ + + + + : 158 : if (Helpers::has_endwith(suffix_check, ".vdb") || Helpers::has_endwith(suffix_check, ".vdbx")) {
+ + ]
85 [ + + ]: 90 : return std::make_shared<VDBReader>(path, read_only, try_to_fix);
86 [ + + + + : 68 : } else if (Helpers::has_endwith(suffix_check, ".vcap") || Helpers::has_endwith(suffix_check, ".vcapx")) {
+ + ]
87 [ + + ]: 67 : return std::make_shared<VCAPReader>(path, read_only, try_to_fix);
88 : : } else {
89 [ + - + - ]: 2 : CLOG_E("BagReader: Unknown bag suffix, path=%s", path.c_str());
90 : 1 : return nullptr;
91 : : }
92 : 158 : }
93 : :
94 [ + - ]: 173 : BagReader::BagReader(const std::string& path, bool read_only, bool try_to_fix) : impl_(std::make_unique<Impl>()) {
95 : : (void)path;
96 : : (void)read_only;
97 : : (void)try_to_fix;
98 : :
99 : 173 : Bytes::init_memory_pool();
100 : 173 : }
101 : :
102 : 173 : BagReader::~BagReader() {
103 : 173 : std::shared_ptr<BagPluginInterface> plugin_interface;
104 : :
105 : : {
106 : 173 : std::shared_lock state_lock(impl_->playback_state_mtx);
107 : 173 : plugin_interface = impl_->plugin_interface;
108 : 173 : }
109 : :
110 [ + + ]: 173 : if (plugin_interface) {
111 : 6 : plugin_interface->register_callback({});
112 : : }
113 : 173 : }
114 : :
115 : 140 : void BagReader::detach_plugin() {
116 : 140 : std::shared_ptr<BagPluginInterface> plugin_interface;
117 : :
118 : : {
119 [ + - ]: 140 : std::unique_lock state_lock(impl_->playback_state_mtx);
120 : 140 : plugin_interface = std::move(impl_->plugin_interface);
121 : 140 : }
122 : :
123 [ + + ]: 140 : if (plugin_interface) {
124 [ + - ]: 2 : plugin_interface->flush();
125 : 2 : plugin_interface->register_callback({});
126 : : }
127 : 140 : }
128 : :
129 : 17 : void BagReader::flush_plugin() {
130 : 17 : std::shared_ptr<BagPluginInterface> plugin_interface;
131 : :
132 : : {
133 [ + - ]: 17 : std::shared_lock state_lock(impl_->playback_state_mtx);
134 : 17 : plugin_interface = impl_->plugin_interface;
135 : 17 : }
136 : :
137 [ + + ]: 17 : if (plugin_interface) {
138 [ + - ]: 1 : plugin_interface->flush();
139 : : }
140 : 17 : }
141 : :
142 : 13 : void BagReader::bind_plugin_interface(const std::shared_ptr<BagPluginInterface>& plugin_interface) {
143 : 13 : std::shared_ptr<BagPluginInterface> old_plugin_interface;
144 : :
145 : : {
146 [ + - ]: 13 : std::shared_lock state_lock(impl_->playback_state_mtx);
147 : 13 : old_plugin_interface = impl_->plugin_interface;
148 : 13 : }
149 : :
150 [ + + + - : 13 : if (old_plugin_interface && old_plugin_interface != plugin_interface) {
+ + ]
151 [ + - ]: 3 : old_plugin_interface->flush();
152 : 3 : old_plugin_interface->register_callback({});
153 : : }
154 : :
155 [ + + ]: 13 : if VLIKELY (plugin_interface) {
156 : 11 : plugin_interface->bind_direction(BagPluginInterface::Direction::kRead);
157 : :
158 [ + - ]: 11 : plugin_interface->register_callback([this](const Frame& frame) {
159 : 5 : std::string output_url;
160 : :
161 [ + - - + ]: 5 : if VUNLIKELY (!convert_playback_url(frame.url, output_url)) {
162 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
163 : : }
164 : :
165 [ + - ]: 5 : std::shared_lock callback_lock(impl_->output_callback_mtx);
166 : :
167 [ + - ]: 5 : if VLIKELY (impl_->output_callback) {
168 : 5 : Frame out;
169 : 5 : out.timestamp = frame.timestamp;
170 : 5 : out.url = std::move(output_url);
171 [ + - ]: 5 : out.ser_type = frame.ser_type;
172 : 5 : out.schema_type = frame.schema_type;
173 : 5 : out.action_type = frame.action_type;
174 : 5 : out.data = Bytes::shallow_copy(frame.data.data(), frame.data.size());
175 [ + - ]: 5 : fill_frame_meta(out);
176 [ + - ]: 5 : impl_->output_callback(out);
177 : 5 : }
178 [ + - ]: 5 : });
179 : : }
180 : :
181 [ + - ]: 13 : std::unique_lock state_lock(impl_->playback_state_mtx);
182 : :
183 : 13 : impl_->plugin_interface = plugin_interface;
184 : 13 : impl_->playback_url_remap.clear();
185 : 13 : impl_->excluded_playback_urls.clear();
186 : 13 : }
187 : :
188 [ # # ]: 0 : void BagReader::clear_plugin_interface() { bind_plugin_interface(nullptr); }
189 : :
190 : 1 : void BagReader::register_status_callback(StatusCallback&& status_callback) { (void)status_callback; }
191 : :
192 : 1 : void BagReader::register_ready_callback(ReadyCallback&& ready_callback) { (void)ready_callback; }
193 : :
194 : 1 : void BagReader::register_finish_callback(FinishCallback&& finish_callback) { (void)finish_callback; }
195 : :
196 : 16 : void BagReader::register_output_callback(OutputCallback&& output_callback) {
197 [ + - ]: 16 : std::unique_lock lock(impl_->output_callback_mtx);
198 : :
199 : 16 : impl_->output_callback = std::move(output_callback);
200 : 16 : }
201 : :
202 : 96 : bool BagReader::open_cursor(const Config& config) {
203 : 96 : impl_->cursor_opened = false;
204 : 96 : impl_->cursor_eof = false;
205 : 96 : impl_->cursor_fail = false;
206 : :
207 [ + + ]: 96 : if VUNLIKELY (!do_open_cursor(config)) {
208 : 2 : impl_->cursor_fail = true;
209 : 2 : return false;
210 : : }
211 : :
212 : 94 : impl_->cursor_opened = true;
213 : 94 : return true;
214 : : }
215 : :
216 [ + - ]: 67 : bool BagReader::open_cursor() { return open_cursor(Config{}); }
217 : :
218 : 243 : bool BagReader::read_next(Frame& out) {
219 [ + + ]: 243 : if VUNLIKELY (impl_->cursor_fail) {
220 : 2 : return false;
221 : : }
222 : :
223 [ + + ]: 241 : if VUNLIKELY (impl_->cursor_eof) {
224 : 1 : return false;
225 : : }
226 : :
227 [ + + + - : 240 : if (!impl_->cursor_opened && !open_cursor(Config{})) {
- + + + -
+ - - ]
228 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
229 : : }
230 : :
231 : 240 : bool is_error = false;
232 : :
233 [ + - + + ]: 240 : if VLIKELY (do_read_next(out, is_error)) {
234 : 146 : return true;
235 : : }
236 : :
237 [ - + ]: 94 : if (is_error) {
238 : : impl_->cursor_fail = true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
239 : : } else {
240 : 94 : impl_->cursor_eof = true;
241 : : }
242 : :
243 : 94 : return false;
244 : : }
245 : :
246 : 12 : BagReader& BagReader::operator>>(Frame& out) {
247 : 12 : read_next(out);
248 : :
249 : 12 : return *this;
250 : : }
251 : :
252 : 21 : bool BagReader::eof() const noexcept { return impl_->cursor_eof; }
253 : :
254 : 13 : bool BagReader::fail() const noexcept { return impl_->cursor_fail; }
255 : :
256 [ + + + + ]: 14 : BagReader::operator bool() const noexcept { return !impl_->cursor_eof && !impl_->cursor_fail; }
257 : :
258 : 1 : bool BagReader::do_open_cursor(const Config& config) {
259 : : (void)config;
260 : :
261 : 1 : return false;
262 : : }
263 : :
264 : : bool BagReader::do_read_next(Frame& out, bool& is_error) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
265 : : (void)out;
266 : :
267 : : is_error = false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
268 : :
269 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
270 : : }
271 : :
272 : 78 : void BagReader::process_output(Frame& frame) {
273 : 78 : std::shared_ptr<BagPluginInterface> plugin_interface;
274 : :
275 : : {
276 [ + - ]: 78 : std::shared_lock state_lock(impl_->playback_state_mtx);
277 : :
278 : 78 : plugin_interface = impl_->plugin_interface;
279 : :
280 [ + + + - : 78 : if VUNLIKELY (plugin_interface && impl_->excluded_playback_urls.count(frame.url) != 0U) {
+ + + + ]
281 : 1 : return;
282 : : }
283 [ + + ]: 78 : }
284 : :
285 [ + + ]: 77 : if (plugin_interface) {
286 [ + - ]: 4 : plugin_interface->on_read(frame);
287 : : } else {
288 : 73 : std::string output_url;
289 : :
290 [ + - - + ]: 73 : if VUNLIKELY (!convert_playback_url(frame.url, output_url)) {
291 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
292 : : }
293 : :
294 [ + - ]: 73 : std::shared_lock callback_lock(impl_->output_callback_mtx);
295 : :
296 [ + + ]: 73 : if VLIKELY (impl_->output_callback) {
297 [ - + ]: 72 : if (output_url != frame.url) {
298 : : frame.url = std::move(output_url); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
299 : : }
300 : :
301 [ + - ]: 72 : fill_frame_meta(frame);
302 [ + - ]: 72 : impl_->output_callback(frame);
303 : : }
304 [ + - ]: 73 : }
305 [ + + ]: 78 : }
306 : :
307 : 215 : void BagReader::fill_frame_meta(Frame& frame) const {
308 [ + - ]: 215 : if (frame.ser_type.empty()) {
309 : 215 : frame.ser_type = get_ser_type(frame.url);
310 : : }
311 : :
312 [ + - ]: 215 : if (frame.schema_type == SchemaType::kUnknown) {
313 : 215 : frame.schema_type = get_schema_type(frame.url);
314 : : }
315 : 215 : }
316 : :
317 : 727 : std::unordered_map<std::string, std::string>& BagReader::url_ser_map() { return impl_->url_to_ser_map; }
318 : :
319 : 727 : std::unordered_map<std::string, SchemaType>& BagReader::url_schema_type_map() { return impl_->url_to_schema_type_map; }
320 : :
321 : 241 : std::string BagReader::get_ser_type(const std::string& url) const {
322 [ + - ]: 241 : std::shared_lock state_lock(impl_->playback_state_mtx);
323 : :
324 [ + - ]: 241 : auto iter = impl_->url_to_ser_map.find(url);
325 : :
326 [ + + ]: 241 : if VLIKELY (iter != impl_->url_to_ser_map.end()) {
327 [ + - ]: 226 : return iter->second;
328 : : }
329 : :
330 : 15 : return {};
331 : 241 : }
332 : :
333 : 229 : SchemaType BagReader::get_schema_type(const std::string& url) const {
334 [ + - ]: 229 : std::shared_lock state_lock(impl_->playback_state_mtx);
335 : :
336 [ + - ]: 229 : auto iter = impl_->url_to_schema_type_map.find(url);
337 : :
338 [ + + ]: 229 : if VLIKELY (iter != impl_->url_to_schema_type_map.end()) {
339 : 220 : return iter->second;
340 : : }
341 : :
342 : 9 : return SchemaType::kUnknown;
343 : 229 : }
344 : :
345 : 11 : void BagReader::process_url_metas(std::vector<Info::UrlMeta>& url_metas) {
346 : 11 : std::shared_ptr<BagPluginInterface> plugin_interface;
347 : :
348 : : {
349 [ + - ]: 11 : std::shared_lock state_lock(impl_->playback_state_mtx);
350 : 11 : plugin_interface = impl_->plugin_interface;
351 : 11 : }
352 : :
353 : 11 : std::unordered_map<std::string, std::string> playback_url_remap;
354 : 11 : std::unordered_set<std::string> excluded_playback_urls;
355 : :
356 [ + + ]: 11 : if (plugin_interface) {
357 [ + - ]: 16 : url_metas.erase(
358 [ + - ]: 8 : std::remove_if(url_metas.begin(), url_metas.end(),
359 : 41 : [&plugin_interface, &playback_url_remap, &excluded_playback_urls](Info::UrlMeta& meta) {
360 [ + - ]: 15 : const std::string input_url = meta.url;
361 : :
362 [ + - + + ]: 15 : if VUNLIKELY (!plugin_interface->convert_url_meta(meta.url, meta.ser_type, meta.schema_type)) {
363 [ + - ]: 4 : excluded_playback_urls.emplace(input_url);
364 : 4 : return true;
365 : : }
366 : :
367 [ + + ]: 11 : if (meta.url != input_url) {
368 [ + - + - ]: 7 : playback_url_remap[input_url] = meta.url;
369 : : }
370 : :
371 : 11 : return false;
372 : 15 : }),
373 : 16 : url_metas.end());
374 : : }
375 : :
376 : : {
377 [ + - ]: 11 : std::unique_lock state_lock(impl_->playback_state_mtx);
378 : :
379 [ + - ]: 11 : if (impl_->plugin_interface == plugin_interface) {
380 : 11 : impl_->playback_url_remap = std::move(playback_url_remap);
381 : 11 : impl_->excluded_playback_urls = std::move(excluded_playback_urls);
382 : : }
383 : 11 : }
384 : 11 : }
385 : :
386 : 498 : bool BagReader::convert_playback_url(const std::string& input_url, std::string& output_url) const {
387 [ + - ]: 498 : std::shared_lock state_lock(impl_->playback_state_mtx);
388 : :
389 [ + - + + ]: 498 : if VUNLIKELY (impl_->excluded_playback_urls.count(input_url) != 0U) {
390 : 2 : return false;
391 : : }
392 : :
393 [ + - ]: 496 : auto iter = impl_->playback_url_remap.find(input_url);
394 : :
395 [ + + ]: 496 : if (iter != impl_->playback_url_remap.end()) {
396 [ + - ]: 11 : output_url = iter->second;
397 : : } else {
398 [ + - ]: 485 : output_url = input_url;
399 : : }
400 : :
401 : 496 : return true;
402 : 498 : }
403 : :
404 : 283 : bool BagReader::match_playback_url_filter(std::string_view input_url,
405 : : const std::unordered_set<std::string>& filter_urls) const {
406 [ + + ]: 283 : if VUNLIKELY (!input_url.data()) {
407 : 1 : return false;
408 : : }
409 : :
410 : 282 : std::string output_url;
411 : :
412 [ + - + - : 282 : if VUNLIKELY (!convert_playback_url(std::string(input_url), output_url)) {
+ + ]
413 : 2 : return false;
414 : : }
415 : :
416 [ + + + - : 280 : return filter_urls.empty() || filter_urls.count(output_url) != 0U;
+ + ]
417 : 282 : }
418 : :
419 : 249 : void BagReader::rebuild_url_meta_lookup(const std::vector<Info::UrlMeta>& url_metas) {
420 [ + - ]: 249 : std::unique_lock state_lock(impl_->playback_state_mtx);
421 : :
422 [ + - ]: 249 : rebuild_url_meta_maps(url_metas, impl_->url_to_ser_map, impl_->url_to_schema_type_map);
423 : 249 : }
424 : :
425 : 251 : void BagReader::rebuild_url_meta_maps(const std::vector<Info::UrlMeta>& url_metas,
426 : : std::unordered_map<std::string, std::string>& ser_map,
427 : : std::unordered_map<std::string, SchemaType>& schema_type_map) {
428 : 251 : ser_map.clear();
429 : 251 : schema_type_map.clear();
430 [ + - ]: 251 : ser_map.reserve(url_metas.size());
431 [ + - ]: 251 : schema_type_map.reserve(url_metas.size());
432 : :
433 : 251 : std::unordered_set<std::string> ser_conflict_urls;
434 : 251 : std::unordered_set<std::string> schema_conflict_urls;
435 : :
436 [ + - ]: 251 : ser_conflict_urls.reserve(url_metas.size());
437 [ + - ]: 251 : schema_conflict_urls.reserve(url_metas.size());
438 : :
439 [ + + ]: 601 : for (const auto& meta : url_metas) {
440 [ + - ]: 350 : auto& merged_ser_type = ser_map[meta.url];
441 [ + - ]: 350 : auto& merged_schema_type = schema_type_map[meta.url];
442 : :
443 [ + - + + : 350 : if (ser_conflict_urls.count(meta.url) == 0U && !meta.ser_type.empty()) {
+ + + + ]
444 [ + + + + : 342 : if (merged_ser_type.empty() || merged_ser_type == "Bytes") {
+ + ]
445 [ + - ]: 337 : merged_ser_type = meta.ser_type;
446 [ + - + + : 5 : } else if (meta.ser_type != "Bytes" && meta.ser_type != merged_ser_type) {
+ + ]
447 [ + - + - ]: 2 : CLOG_E("BagReader: URL remap collision on %s, keeping ser_type unknown. ser [%s] vs [%s].", meta.url.c_str(),
448 : : merged_ser_type.c_str(), meta.ser_type.c_str());
449 : 1 : merged_ser_type.clear();
450 [ + - ]: 1 : ser_conflict_urls.emplace(meta.url);
451 : : }
452 : : }
453 : :
454 [ + - + + : 350 : if (schema_conflict_urls.count(meta.url) == 0U && meta.schema_type != SchemaType::kUnknown) {
+ + + + ]
455 [ + + ]: 301 : if (merged_schema_type == SchemaType::kUnknown) {
456 : 295 : merged_schema_type = meta.schema_type;
457 [ + + ]: 6 : } else if (merged_schema_type != meta.schema_type) {
458 : 1 : const auto current_label = SchemaData::convert_type(merged_schema_type);
459 : 1 : const auto new_label = SchemaData::convert_type(meta.schema_type);
460 [ + - + - ]: 2 : CLOG_E("BagReader: URL remap collision on %s, keeping schema_type unknown. schema [%.*s] vs [%.*s].",
461 : : meta.url.c_str(), static_cast<int>(current_label.size()), current_label.data(),
462 : : static_cast<int>(new_label.size()), new_label.data());
463 : 1 : merged_schema_type = SchemaType::kUnknown;
464 [ + - ]: 1 : schema_conflict_urls.emplace(meta.url);
465 : : }
466 : : }
467 : : }
468 : 251 : }
469 : :
470 : 265 : ActionType BagReader::convert_action(std::string_view str) {
471 [ + + ]: 265 : if (str == "C/Req") {
472 : 27 : return ActionType::kClientRequest;
473 [ + + ]: 238 : } else if (str == "C/Resp") {
474 : 4 : return ActionType::kClientResponse;
475 [ + + ]: 234 : } else if (str == "S/Req") {
476 : 3 : return ActionType::kServerRequest;
477 [ + + ]: 231 : } else if (str == "S/Resp") {
478 : 3 : return ActionType::kServerResponse;
479 [ + + ]: 228 : } else if (str == "Pub") {
480 : 184 : return ActionType::kPublish;
481 [ + + ]: 44 : } else if (str == "Sub") {
482 : 3 : return ActionType::kSubscribe;
483 [ + + ]: 41 : } else if (str == "Set") {
484 : 31 : return ActionType::kSet;
485 [ + + ]: 10 : } else if (str == "Get") {
486 : 2 : return ActionType::kGet;
487 : : } else {
488 : 8 : return ActionType::kUnknownAction;
489 : : }
490 : : }
491 : :
492 : : } // namespace vlink
|
