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/trigger_recorder.h"
25 : :
26 : : #include <algorithm>
27 : : #include <atomic>
28 : : #include <cctype>
29 : : #include <chrono>
30 : : #include <cstdint>
31 : : #include <deque>
32 : : #include <exception>
33 : : #include <filesystem>
34 : : #include <limits>
35 : : #include <memory>
36 : : #include <mutex>
37 : : #include <shared_mutex>
38 : : #include <string>
39 : : #include <string_view>
40 : : #include <system_error>
41 : : #include <thread>
42 : : #include <unordered_map>
43 : : #include <unordered_set>
44 : : #include <utility>
45 : : #include <vector>
46 : :
47 : : #include "./base/elapsed_timer.h"
48 : : #include "./base/helpers.h"
49 : : #include "./base/logger.h"
50 : : #include "./base/timer.h"
51 : : #include "./base/utils.h"
52 : : #include "./extension/bag_writer.h"
53 : : #include "./extension/trigger_plugin_interface.h"
54 : :
55 : : namespace vlink {
56 : :
57 : : // TriggerRecorder::UrlBuffer
58 : : struct TriggerRecorder::UrlBuffer final {
59 : : struct Entry final {
60 : : int64_t capture_ts_us{0};
61 : : std::shared_ptr<const Bytes> payload;
62 : : };
63 : :
64 : : std::mutex mtx;
65 : : std::deque<Entry> ring;
66 : : int64_t bytes{0};
67 : : int64_t pre_us{0};
68 : : int64_t post_us{0};
69 : : std::atomic<int64_t> retention_us{0};
70 : : int64_t max_packet_size{0};
71 : : int64_t max_size{0};
72 : :
73 : : std::string url;
74 : : std::string ser_type;
75 : : SampleLostInfo last_lost;
76 : : SampleLostInfo final_lost;
77 : : SchemaType schema_type{SchemaType::kUnknown};
78 : : bool frozen{false};
79 : : bool disabled{false};
80 : : bool getter_semantics{false};
81 : :
82 : : std::shared_ptr<RawSub> sub;
83 : : };
84 : :
85 : : // TriggerRecorder::DumpJob
86 : : struct TriggerRecorder::DumpJob final {
87 : : int64_t trigger_ts{0};
88 : : TriggerParams params;
89 : : std::string path;
90 : : std::vector<std::shared_ptr<UrlBuffer>> frozen_buffers;
91 : : };
92 : :
93 : : // TriggerRecorder::Impl
94 : : struct TriggerRecorder::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
95 : : Config config;
96 : : RawSubFactory raw_sub_factory;
97 : : ElapsedTimer capture_timer{ElapsedTimer::kCpuTimestamp, ElapsedTimer::kMicro};
98 : : int64_t anchor_wall_us{0};
99 : :
100 : : std::atomic<int64_t> max_post_all_us{0};
101 : : std::atomic<int64_t> global_bytes{0};
102 : : std::atomic_bool cache_overflow_warned{false};
103 : : std::atomic_bool dumping{false};
104 : : std::atomic_bool writing{false};
105 : : std::atomic_bool running{false};
106 : :
107 : : std::mutex lifecycle_mtx;
108 : : std::shared_mutex url_buffer_mtx;
109 : :
110 : : std::unordered_map<std::string, std::shared_ptr<UrlBuffer>> url_buffer_map;
111 : : std::shared_ptr<DumpJob> active_dump_job;
112 : :
113 : : std::unique_ptr<DiscoveryViewer> viewer;
114 : : Timer sweep_timer;
115 : :
116 : : std::shared_ptr<TriggerPluginInterface> trigger_plugin;
117 : : std::shared_ptr<BagPluginInterface> bag_plugin;
118 : : };
119 : :
120 : : // TriggerRecorder
121 [ + - ]: 26 : TriggerRecorder::TriggerRecorder(const Config& config, RawSubFactory&& factory) : impl_(std::make_unique<Impl>()) {
122 [ + - + - ]: 26 : set_name("TriggerRecorder");
123 : :
124 [ + - ]: 26 : impl_->config = config;
125 : 26 : impl_->raw_sub_factory = std::move(factory);
126 : :
127 [ + + ]: 26 : if VUNLIKELY (!impl_->raw_sub_factory) {
128 [ + - - + ]: 2 : VLOG_F("TriggerRecorder: raw subscriber factory is not set");
129 : : }
130 : :
131 [ + + ]: 25 : if (impl_->config.dump_dir.empty()) {
132 [ + - ]: 6 : impl_->config.dump_dir = Utils::get_tmp_dir() + "/vlink-trigger";
133 : : }
134 : :
135 : 25 : impl_->config.default_pre_ms = std::max<int64_t>(0, impl_->config.default_pre_ms);
136 : 25 : impl_->config.default_post_ms = std::max<int64_t>(0, impl_->config.default_post_ms);
137 : 25 : impl_->config.retention_guard_ms = std::max<int64_t>(0, impl_->config.retention_guard_ms);
138 : :
139 [ + + - + : 25 : if VUNLIKELY (impl_->config.default_pre_ms > kMaxWindowMs || impl_->config.default_post_ms > kMaxWindowMs ||
+ + - + +
+ ]
140 : : impl_->config.retention_guard_ms > kMaxWindowMs) {
141 [ + - - + ]: 2 : VLOG_F("TriggerRecorder: pre/post/retention_guard exceeds the supported range");
142 : : }
143 : :
144 [ - + ]: 24 : if VUNLIKELY (impl_->config.default_post_ms + impl_->config.retention_guard_ms >
145 : : std::numeric_limits<uint32_t>::max()) {
146 [ # # # # ]: 0 : VLOG_F("TriggerRecorder: default post window plus retention guard exceeds the timer range");
147 : : }
148 : :
149 [ + - - + : 24 : if VUNLIKELY (impl_->config.default_max_packet_size < 0 || impl_->config.default_max_size < 0 ||
+ - + + +
+ - + + +
- + + + ]
150 : : impl_->config.max_cache_size < 0 || impl_->config.sleep_interval < 0 ||
151 : : impl_->config.sleep_time_ms < 0) {
152 [ + - - + ]: 2 : VLOG_F("TriggerRecorder: byte limits and sleep values must be non-negative");
153 : : }
154 : :
155 [ + + - + : 23 : if VUNLIKELY (impl_->config.file_type != kVdb && impl_->config.file_type != kVcap) {
- + ]
156 [ # # # # ]: 0 : VLOG_F("TriggerRecorder: unsupported file type");
157 : : }
158 : :
159 [ + - - + : 23 : if VUNLIKELY (impl_->config.overflow != kCoverOldest && impl_->config.overflow != kDropNewest) {
- + ]
160 [ # # # # ]: 0 : VLOG_F("TriggerRecorder: unsupported overflow policy");
161 : : }
162 : :
163 [ - + ]: 23 : for (const auto& [url, value] : impl_->config.url_overrides) {
164 [ # # # # : 0 : if VUNLIKELY (value.pre_ms > kMaxWindowMs || value.post_ms > kMaxWindowMs ||
# # # # #
# # # #
# ]
165 : : (value.post_ms >= 0 &&
166 : : value.post_ms + impl_->config.retention_guard_ms > std::numeric_limits<uint32_t>::max())) {
167 [ # # # # ]: 0 : VLOG_F("TriggerRecorder: URL window exceeds the supported range: ", url);
168 : : }
169 : : }
170 : :
171 : 23 : std::error_code ec;
172 [ + - + - ]: 23 : std::filesystem::create_directories(impl_->config.dump_dir, ec);
173 : :
174 : 23 : std::error_code query_ec;
175 : :
176 [ + + + - : 23 : if VUNLIKELY (ec && !std::filesystem::is_directory(impl_->config.dump_dir, query_ec)) {
+ - + + +
+ - - ]
177 : : VLOG_F("TriggerRecorder: cannot create dump_dir '", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
178 : : impl_->config.dump_dir, "': ", ec.message()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
179 : : }
180 : :
181 [ + - ]: 22 : impl_->viewer = std::make_unique<DiscoveryViewer>(impl_->config.discovery_filter);
182 [ + - + - ]: 42 : impl_->viewer->register_callback([this](const std::vector<DiscoveryViewer::Info>& list) { handle_discovery(list); });
183 : 30 : }
184 : :
185 : 22 : TriggerRecorder::~TriggerRecorder() {
186 : 22 : quit(true);
187 : 22 : wait_for_quit();
188 : 22 : }
189 : :
190 : 19 : bool TriggerRecorder::dump(const TriggerParams& params) {
191 : 19 : std::string out_file;
192 : :
193 [ + - ]: 38 : return dump(params, out_file);
194 : 19 : }
195 : :
196 : 23 : bool TriggerRecorder::dump(const TriggerParams& params, std::string& out_file) {
197 [ + + - + : 23 : if VUNLIKELY (params.pre_ms > kMaxWindowMs || params.post_ms > kMaxWindowMs) {
+ + ]
198 : 1 : out_file.clear();
199 [ + - + - ]: 2 : VLOG_E("TriggerRecorder: trigger window exceeds the supported range");
200 : 1 : return false;
201 : : }
202 : :
203 [ + - ]: 22 : std::lock_guard lifecycle_lock(impl_->lifecycle_mtx);
204 : :
205 [ + + + + : 22 : if VUNLIKELY (!impl_->running.load(std::memory_order_acquire) || impl_->dumping.load(std::memory_order_acquire)) {
+ + ]
206 : 3 : out_file.clear();
207 : 3 : return false;
208 : : }
209 : :
210 [ + - ]: 19 : auto job = std::make_shared<DumpJob>();
211 [ + - ]: 19 : job->params = params;
212 : 19 : const auto& request = job->params;
213 : 19 : out_file.clear();
214 : :
215 [ + + ]: 19 : if (!request.out_file.empty()) {
216 [ + - ]: 8 : job->path = request.out_file;
217 : : } else {
218 [ + + ]: 11 : const char* file_suffix = impl_->config.file_type == kVcap ? ".vcap" : ".vdb";
219 [ + + + - : 11 : std::string base = request.name_hint.empty() ? BagWriter::get_format_date(nullptr, true) : request.name_hint;
+ - ]
220 : 11 : std::replace(base.begin(), base.end(), '/', '_');
221 : 11 : std::replace(base.begin(), base.end(), '\\', '_');
222 [ + - + - : 11 : job->path = impl_->config.dump_dir + "/" + base + file_suffix;
+ - ]
223 : 11 : uint64_t suffix = 1;
224 : 11 : std::error_code ec;
225 : :
226 [ + - + + ]: 12 : while (std::filesystem::exists(job->path, ec)) {
227 [ + - + - : 1 : job->path = impl_->config.dump_dir + "/" + base + "_" + std::to_string(suffix++) + file_suffix;
+ - + - +
- + - ]
228 : : }
229 : 11 : }
230 : :
231 : 19 : int64_t max_post = 0;
232 : 19 : std::vector<std::string> filter_list = Helpers::split_any(request.filter_str);
233 : :
234 : : {
235 [ + - ]: 19 : std::shared_lock map_lock(impl_->url_buffer_mtx);
236 [ + - ]: 19 : job->frozen_buffers.reserve(impl_->url_buffer_map.size());
237 : 19 : job->trigger_ts = impl_->capture_timer.get();
238 [ + + ]: 30 : for (const auto& entry : impl_->url_buffer_map) {
239 [ - + ]: 11 : if (entry.second->disabled) {
240 : 2 : continue;
241 : : }
242 : :
243 [ + + + - : 21 : if ((!request.whitelist.empty() && request.whitelist.count(entry.second->url) == 0) ||
+ + + + ]
244 [ + - + + ]: 10 : request.blacklist.count(entry.second->url) != 0) {
245 : 2 : continue;
246 : : }
247 : :
248 [ - + ]: 9 : if (!filter_list.empty()) {
249 [ # # ]: 0 : bool skip = request.black_mode ? false : true;
250 : :
251 [ # # ]: 0 : std::string left_str = entry.second->url;
252 : 0 : std::transform(left_str.begin(), left_str.end(), left_str.begin(),
253 : 0 : [](unsigned char c) { return std::tolower(c); });
254 [ # # ]: 0 : for (const auto& filter : filter_list) {
255 [ # # ]: 0 : if (filter.empty()) {
256 : 0 : continue;
257 : : }
258 : :
259 [ # # ]: 0 : std::string right_str = filter;
260 : 0 : std::transform(right_str.begin(), right_str.end(), right_str.begin(),
261 : 0 : [](unsigned char c) { return std::tolower(c); });
262 : :
263 [ # # ]: 0 : if (left_str.find(right_str) != std::string::npos) {
264 [ # # ]: 0 : skip = request.black_mode ? true : false;
265 : 0 : break;
266 : : }
267 [ # # ]: 0 : }
268 : :
269 [ # # ]: 0 : if (skip) {
270 : 0 : continue;
271 : : }
272 [ # # ]: 0 : }
273 : :
274 : 9 : entry.second->frozen = true;
275 : : job->frozen_buffers.push_back(entry.second); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
276 : :
277 : : const int64_t post =
278 [ - + ]: 9 : request.post_ms >= 0 ? std::min(request.post_ms * 1000, entry.second->post_us) : entry.second->post_us;
279 : 9 : max_post = std::max(max_post, post);
280 : : }
281 : :
282 : 19 : impl_->active_dump_job = job;
283 : 19 : impl_->dumping.store(true, std::memory_order_release);
284 : 19 : }
285 : :
286 [ + + ]: 19 : const int64_t delay_ms = max_post > 0 ? max_post / 1000 + impl_->config.retention_guard_ms : 0;
287 : :
288 : 35 : auto task = [this, weak_job = std::weak_ptr<DumpJob>(job)]() {
289 [ + + ]: 18 : if (auto locked_job = weak_job.lock()) {
290 [ + - ]: 17 : do_dump(*locked_job);
291 : 18 : }
292 : 37 : };
293 : :
294 [ + - + - ]: 38 : VLOG_I("TriggerRecorder: dump scheduling -> ", job->path, " wait_ms=", delay_ms,
295 : : " urls=", job->frozen_buffers.size());
296 : :
297 [ + + + - : 22 : const bool ok = delay_ms > 0 ? Timer::call_once(this, static_cast<uint32_t>(delay_ms), std::move(task))
+ - - - ]
298 [ + - + - : 19 : : post_task(std::move(task));
+ + + + -
- ]
299 : :
300 [ + + ]: 19 : if VUNLIKELY (!ok) {
301 : : // LCOV_EXCL_START GCOVR_EXCL_START
302 : : finish_dump_locked(*job);
303 : :
304 : : return false;
305 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
306 : : }
307 : :
308 [ + - ]: 18 : out_file = job->path;
309 : :
310 : 18 : return true;
311 : 22 : }
312 : :
313 : 378 : bool TriggerRecorder::is_dumping() const noexcept { return impl_->dumping.load(std::memory_order_acquire); }
314 : :
315 : 10 : void TriggerRecorder::bind_trigger_interface(const std::shared_ptr<TriggerPluginInterface>& trigger_interface) {
316 [ + - ]: 10 : std::lock_guard lifecycle_lock(impl_->lifecycle_mtx);
317 : :
318 [ - + ]: 10 : if VUNLIKELY (impl_->running.load(std::memory_order_acquire)) {
319 [ # # # # ]: 0 : VLOG_E("TriggerRecorder: trigger plugin must be bound while the recorder is stopped");
320 : 0 : return;
321 : : }
322 : :
323 : 10 : impl_->trigger_plugin = trigger_interface;
324 [ + - ]: 10 : }
325 : :
326 [ # # ]: 0 : void TriggerRecorder::clear_trigger_interface() { bind_trigger_interface(nullptr); }
327 : :
328 : 6 : void TriggerRecorder::bind_bag_interface(const std::shared_ptr<BagPluginInterface>& bag_interface) {
329 [ + - ]: 6 : std::lock_guard lock(impl_->lifecycle_mtx);
330 : :
331 [ - + ]: 6 : if VUNLIKELY (impl_->running.load(std::memory_order_acquire)) {
332 [ # # # # ]: 0 : VLOG_E("TriggerRecorder: bag plugin must be bound while the recorder is stopped");
333 : 0 : return;
334 : : }
335 : :
336 : 6 : impl_->bag_plugin = bag_interface;
337 [ + - ]: 6 : }
338 : :
339 [ + - ]: 1 : void TriggerRecorder::clear_bag_interface() { bind_bag_interface(nullptr); }
340 : :
341 : 20 : void TriggerRecorder::on_begin() {
342 [ + - ]: 20 : MessageLoop::on_begin();
343 : :
344 [ + - ]: 20 : std::lock_guard lifecycle_lock(impl_->lifecycle_mtx);
345 : :
346 [ + + ]: 20 : if (!impl_->viewer) {
347 : : try {
348 [ + - ]: 2 : impl_->viewer = std::make_unique<DiscoveryViewer>(impl_->config.discovery_filter);
349 [ + - + - ]: 2 : impl_->viewer->register_callback(
350 : 2 : [this](const std::vector<DiscoveryViewer::Info>& list) { handle_discovery(list); });
351 [ - - ]: 0 : } catch (const std::exception& e) {
352 [ # # # # ]: 0 : VLOG_E("TriggerRecorder: discovery viewer setup failed: ", e.what());
353 : 0 : return;
354 : 0 : }
355 : : }
356 : :
357 : 20 : impl_->capture_timer.start();
358 : 20 : impl_->cache_overflow_warned.store(false, std::memory_order_relaxed);
359 : 40 : impl_->anchor_wall_us =
360 : 20 : static_cast<int64_t>(ElapsedTimer::get_sys_timestamp(ElapsedTimer::kMicro)) - impl_->capture_timer.get();
361 : :
362 [ + - - + ]: 20 : if VUNLIKELY (!impl_->viewer->async_run()) {
363 : : VLOG_E("TriggerRecorder: discovery viewer failed to start"); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
364 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
365 : : }
366 : :
367 [ + - - + ]: 20 : if VUNLIKELY (!impl_->sweep_timer.attach(this)) {
368 : 0 : return;
369 : : }
370 : :
371 [ + - + - ]: 22 : impl_->sweep_timer.start([this]() { sweep_evict(); });
372 : :
373 : 20 : impl_->running.store(true, std::memory_order_release);
374 : :
375 [ + + ]: 20 : if (impl_->trigger_plugin) {
376 [ + - ]: 11 : impl_->trigger_plugin->on_start();
377 : : }
378 [ + - ]: 20 : }
379 : :
380 : 20 : void TriggerRecorder::on_end() {
381 [ + - ]: 20 : std::unique_lock lifecycle_lock(impl_->lifecycle_mtx);
382 : :
383 : 20 : const bool recorder_started = impl_->running.exchange(false, std::memory_order_acq_rel);
384 [ + - ]: 20 : impl_->sweep_timer.stop();
385 : :
386 [ + - ]: 20 : if (impl_->viewer) {
387 [ + - ]: 20 : impl_->viewer->quit(true);
388 [ + - ]: 20 : impl_->viewer->wait_for_quit();
389 : 20 : impl_->viewer.reset();
390 : : }
391 : :
392 : 20 : impl_->dumping.store(false, std::memory_order_release);
393 : 20 : auto abandoned_job = std::move(impl_->active_dump_job);
394 : :
395 : 20 : std::vector<std::shared_ptr<RawSub>> retired_subscribers;
396 : :
397 : : {
398 [ + - ]: 20 : std::unique_lock lock(impl_->url_buffer_mtx);
399 [ + - ]: 20 : retired_subscribers.reserve(impl_->url_buffer_map.size());
400 : :
401 [ + + ]: 32 : for (auto& entry : impl_->url_buffer_map) {
402 [ + - + - ]: 12 : retired_subscribers.push_back(deactivate_url_buffer(*entry.second));
403 : : }
404 : :
405 : 20 : impl_->url_buffer_map.clear();
406 : 20 : impl_->global_bytes.store(0, std::memory_order_relaxed);
407 : 20 : impl_->max_post_all_us.store(0, std::memory_order_relaxed);
408 : 20 : }
409 : :
410 : 20 : retired_subscribers.clear();
411 : :
412 [ + - ]: 20 : if (recorder_started) {
413 [ + + ]: 20 : if VUNLIKELY (abandoned_job) {
414 [ + - ]: 1 : notify_dump_failed(*abandoned_job, "dump abandoned at shutdown");
415 : : }
416 : :
417 [ + + ]: 20 : if (impl_->trigger_plugin) {
418 [ + - ]: 11 : impl_->trigger_plugin->flush();
419 [ + - ]: 11 : impl_->trigger_plugin->on_stop();
420 : : }
421 : : }
422 : :
423 [ + - ]: 20 : lifecycle_lock.unlock();
424 : :
425 [ + - ]: 20 : MessageLoop::on_end();
426 : 20 : }
427 : :
428 : : // LCOV_EXCL_START GCOVR_EXCL_START
429 : : void TriggerRecorder::handle_data(UrlBuffer& url_buffer, const Bytes& data) {
430 : : if VUNLIKELY (!impl_->running.load(std::memory_order_relaxed) || url_buffer.disabled) {
431 : : return;
432 : : }
433 : :
434 : : if VUNLIKELY (impl_->config.busy_skip_data && impl_->writing.load(std::memory_order_relaxed)) {
435 : : return;
436 : : }
437 : :
438 : : const size_t data_size = data.size();
439 : :
440 : : if VUNLIKELY (data_size > static_cast<size_t>(std::numeric_limits<int64_t>::max())) {
441 : : return;
442 : : }
443 : :
444 : : const auto incoming = static_cast<int64_t>(data_size);
445 : :
446 : : if VUNLIKELY (url_buffer.max_packet_size > 0 && incoming > url_buffer.max_packet_size) {
447 : : return;
448 : : }
449 : :
450 : : if VUNLIKELY (url_buffer.max_size > 0 && incoming > url_buffer.max_size) {
451 : : return;
452 : : }
453 : :
454 : : if VUNLIKELY (impl_->config.max_cache_size > 0 && incoming > impl_->config.max_cache_size) {
455 : : if (!impl_->cache_overflow_warned.exchange(true, std::memory_order_relaxed)) {
456 : : VLOG_W("TriggerRecorder: max_cache_size exceeded, current=", impl_->global_bytes.load(std::memory_order_relaxed),
457 : : " incoming=", incoming, " limit=", impl_->config.max_cache_size, " action=drop");
458 : : }
459 : :
460 : : return;
461 : : }
462 : :
463 : : auto payload = std::make_shared<const Bytes>(Bytes::deep_copy(data.data(), data_size));
464 : :
465 : : if VUNLIKELY (payload->size() != data_size) {
466 : : return;
467 : : }
468 : :
469 : : std::lock_guard lock(url_buffer.mtx);
470 : :
471 : : if VUNLIKELY (!url_buffer.sub) {
472 : : return;
473 : : }
474 : :
475 : : const int64_t capture_ts = impl_->capture_timer.get();
476 : : const int64_t horizon = capture_ts - url_buffer.retention_us.load(std::memory_order_relaxed);
477 : :
478 : : while (!url_buffer.ring.empty() && url_buffer.ring.front().capture_ts_us < horizon) {
479 : : const auto size = static_cast<int64_t>(url_buffer.ring.front().payload->size());
480 : : url_buffer.bytes -= size;
481 : : impl_->global_bytes.fetch_sub(size, std::memory_order_relaxed);
482 : : url_buffer.ring.pop_front();
483 : : }
484 : :
485 : : const size_t evictable_count = url_buffer.ring.size();
486 : : size_t evict_count = 0;
487 : : int64_t evict_bytes = 0;
488 : : int64_t local_need = 0;
489 : :
490 : : if (url_buffer.max_size > 0 && url_buffer.bytes > url_buffer.max_size - incoming) {
491 : : local_need = url_buffer.bytes - (url_buffer.max_size - incoming);
492 : : }
493 : :
494 : : if (local_need > 0 && impl_->config.overflow == kDropNewest) {
495 : : return;
496 : : }
497 : :
498 : : url_buffer.ring.push_back(UrlBuffer::Entry{capture_ts, std::move(payload)});
499 : :
500 : : int64_t global = impl_->global_bytes.load(std::memory_order_relaxed);
501 : :
502 : : while (true) {
503 : : int64_t global_need = 0;
504 : :
505 : : if (impl_->config.max_cache_size > 0 && global > impl_->config.max_cache_size - incoming) {
506 : : global_need = global - (impl_->config.max_cache_size - incoming);
507 : :
508 : : if (!impl_->cache_overflow_warned.exchange(true, std::memory_order_relaxed)) {
509 : : VLOG_W("TriggerRecorder: max_cache_size exceeded, current=", global, " incoming=", incoming,
510 : : " limit=", impl_->config.max_cache_size,
511 : : " policy=", impl_->config.overflow == kDropNewest ? "drop" : "cover");
512 : : }
513 : :
514 : : if (impl_->config.overflow == kDropNewest) {
515 : : url_buffer.ring.pop_back();
516 : : return;
517 : : }
518 : : }
519 : :
520 : : const int64_t need = std::max(local_need, global_need);
521 : :
522 : : while (evict_count < evictable_count && evict_bytes < need) {
523 : : evict_bytes += static_cast<int64_t>(url_buffer.ring[evict_count].payload->size());
524 : : ++evict_count;
525 : : }
526 : :
527 : : while (evict_count > 0 &&
528 : : evict_bytes - static_cast<int64_t>(url_buffer.ring[evict_count - 1].payload->size()) >= need) {
529 : : evict_bytes -= static_cast<int64_t>(url_buffer.ring[evict_count - 1].payload->size());
530 : : --evict_count;
531 : : }
532 : :
533 : : if (evict_bytes < need || evict_bytes > global) {
534 : : url_buffer.ring.pop_back();
535 : : return;
536 : : }
537 : :
538 : : if VUNLIKELY (global - evict_bytes > std::numeric_limits<int64_t>::max() - incoming) {
539 : : url_buffer.ring.pop_back();
540 : : return;
541 : : }
542 : :
543 : : const int64_t desired = global - evict_bytes + incoming;
544 : :
545 : : if (impl_->global_bytes.compare_exchange_weak(global, desired, std::memory_order_relaxed,
546 : : std::memory_order_relaxed)) {
547 : : break;
548 : : }
549 : : }
550 : :
551 : : for (size_t index = 0; index < evict_count; ++index) {
552 : : url_buffer.ring.pop_front();
553 : : }
554 : :
555 : : url_buffer.bytes = url_buffer.bytes - evict_bytes + incoming;
556 : : }
557 : :
558 : : std::shared_ptr<TriggerRecorder::UrlBuffer> TriggerRecorder::build_url_buffer(const DiscoveryViewer::Info& info) {
559 : : auto url_buffer = std::make_shared<UrlBuffer>();
560 : : url_buffer->url = info.url;
561 : : url_buffer->ser_type = info.ser_type;
562 : : url_buffer->schema_type = info.schema_type;
563 : :
564 : : UrlConfig url_config;
565 : :
566 : : auto override_iter = impl_->config.url_overrides.find(info.url);
567 : :
568 : : if (override_iter != impl_->config.url_overrides.end()) {
569 : : url_config = override_iter->second;
570 : : }
571 : :
572 : : const int64_t pre_ms = url_config.pre_ms >= 0 ? url_config.pre_ms : impl_->config.default_pre_ms;
573 : : const int64_t post_ms = url_config.post_ms >= 0 ? url_config.post_ms : impl_->config.default_post_ms;
574 : :
575 : : url_buffer->pre_us = url_config.only_back ? 0 : pre_ms * 1000;
576 : : url_buffer->post_us = url_config.only_front ? 0 : post_ms * 1000;
577 : : url_buffer->disabled = url_config.only_front && url_config.only_back;
578 : : url_buffer->getter_semantics = (info.type & (kGetter | kSetter)) != 0;
579 : : url_buffer->max_packet_size =
580 : : url_config.max_packet_size >= 0 ? url_config.max_packet_size : impl_->config.default_max_packet_size;
581 : : url_buffer->max_size = url_config.max_size >= 0 ? url_config.max_size : impl_->config.default_max_size;
582 : :
583 : : const int64_t guard = impl_->config.retention_guard_ms * 1000;
584 : : const int64_t initial_max_post =
585 : : std::max(impl_->max_post_all_us.load(std::memory_order_relaxed), url_buffer->post_us);
586 : : url_buffer->retention_us.store(url_buffer->pre_us + initial_max_post + 2 * guard, std::memory_order_relaxed);
587 : :
588 : : if (url_buffer->disabled) {
589 : : VLOG_W("TriggerRecorder: URL has both only_front and only_back, window is empty: ", info.url);
590 : : }
591 : :
592 : : try {
593 : : auto sub = impl_->raw_sub_factory(info.url, InitType::kWithoutInit);
594 : :
595 : : if VUNLIKELY (!sub) {
596 : : VLOG_W("TriggerRecorder: raw subscriber factory returned null, URL skipped: ", info.url);
597 : : return nullptr;
598 : : }
599 : :
600 : : if (url_buffer->getter_semantics) {
601 : : sub->mark_as_getter();
602 : : }
603 : :
604 : : sub->set_latency_and_lost_enabled(true);
605 : : sub->set_ser_type(info.ser_type, info.schema_type);
606 : : sub->set_discovery_enabled(false);
607 : : sub->set_safety_quit(true);
608 : :
609 : : if VUNLIKELY (!sub->init()) {
610 : : VLOG_W("TriggerRecorder: subscriber init failed, URL skipped: ", info.url);
611 : : return nullptr;
612 : : }
613 : :
614 : : url_buffer->sub = std::move(sub);
615 : :
616 : : if VUNLIKELY (!url_buffer->sub->listen([this, weak = std::weak_ptr<UrlBuffer>(url_buffer)](const Bytes& data) {
617 : : if (auto locked = weak.lock()) {
618 : : handle_data(*locked, data);
619 : : }
620 : : })) {
621 : : deactivate_url_buffer(*url_buffer);
622 : : VLOG_W("TriggerRecorder: subscriber listen failed, URL skipped: ", info.url);
623 : : return nullptr;
624 : : }
625 : : } catch (const std::exception& e) {
626 : : if (url_buffer->sub) {
627 : : deactivate_url_buffer(*url_buffer);
628 : : }
629 : :
630 : : VLOG_W("TriggerRecorder: subscriber setup failed, URL skipped: ", info.url, " (", e.what(), ")");
631 : : return nullptr;
632 : : }
633 : :
634 : : return url_buffer;
635 : : }
636 : :
637 : : std::shared_ptr<TriggerRecorder::RawSub> TriggerRecorder::deactivate_url_buffer(UrlBuffer& url_buffer) {
638 : : std::lock_guard ring_lock(url_buffer.mtx);
639 : :
640 : : if (url_buffer.frozen) {
641 : : url_buffer.final_lost = url_buffer.sub->get_lost();
642 : : } else if (url_buffer.bytes > 0) {
643 : : impl_->global_bytes.fetch_sub(url_buffer.bytes, std::memory_order_relaxed);
644 : : }
645 : :
646 : : return std::move(url_buffer.sub);
647 : : }
648 : :
649 : : void TriggerRecorder::recompute_retention() {
650 : : int64_t max_post = 0;
651 : :
652 : : for (const auto& entry : impl_->url_buffer_map) {
653 : : max_post = std::max(max_post, entry.second->post_us);
654 : : }
655 : :
656 : : impl_->max_post_all_us.store(max_post, std::memory_order_relaxed);
657 : :
658 : : const int64_t guard = impl_->config.retention_guard_ms * 1000;
659 : :
660 : : for (const auto& entry : impl_->url_buffer_map) {
661 : : UrlBuffer* url_buffer = entry.second.get();
662 : : const int64_t updated = url_buffer->pre_us + max_post + 2 * guard;
663 : :
664 : : if (url_buffer->frozen && updated < url_buffer->retention_us.load(std::memory_order_relaxed)) {
665 : : continue;
666 : : }
667 : :
668 : : url_buffer->retention_us.store(updated, std::memory_order_relaxed);
669 : : }
670 : : }
671 : :
672 : : void TriggerRecorder::handle_discovery(const std::vector<DiscoveryViewer::Info>& list) {
673 : : if VUNLIKELY (!impl_->running.load(std::memory_order_acquire)) {
674 : : return;
675 : : }
676 : :
677 : : std::vector<std::shared_ptr<UrlBuffer>> new_buffers;
678 : : std::unordered_set<std::string_view> current;
679 : :
680 : : if (impl_->config.destroy_on_offline) {
681 : : current.reserve(list.size());
682 : : }
683 : :
684 : : for (const auto& info : list) {
685 : : if (impl_->config.destroy_on_offline) {
686 : : current.insert(info.url);
687 : : }
688 : :
689 : : if ((info.type & (kPublisher | kSetter)) == 0) {
690 : : continue;
691 : : }
692 : :
693 : : if ((std::find(impl_->config.blacklist.begin(), impl_->config.blacklist.end(), info.url) !=
694 : : impl_->config.blacklist.end()) ||
695 : : (!impl_->config.whitelist.empty() && std::find(impl_->config.whitelist.begin(), impl_->config.whitelist.end(),
696 : : info.url) == impl_->config.whitelist.end())) {
697 : : continue;
698 : : }
699 : :
700 : : auto existing = impl_->url_buffer_map.find(info.url);
701 : :
702 : : if VLIKELY (existing != impl_->url_buffer_map.end()) {
703 : : if VLIKELY (existing->second->ser_type == info.ser_type && existing->second->schema_type == info.schema_type &&
704 : : existing->second->getter_semantics == ((info.type & (kGetter | kSetter)) != 0)) {
705 : : continue;
706 : : }
707 : : }
708 : :
709 : : auto url_buffer = build_url_buffer(info);
710 : :
711 : : if (url_buffer) {
712 : : new_buffers.push_back(std::move(url_buffer));
713 : : }
714 : : }
715 : :
716 : : if (new_buffers.empty() && !impl_->config.destroy_on_offline) {
717 : : return;
718 : : }
719 : :
720 : : std::vector<std::shared_ptr<RawSub>> retired_subscribers;
721 : :
722 : : {
723 : : std::unique_lock lock(impl_->url_buffer_mtx);
724 : : bool retention_changed = false;
725 : :
726 : : for (auto& url_buffer : new_buffers) {
727 : : std::string url = url_buffer->url;
728 : : auto existing = impl_->url_buffer_map.find(url);
729 : :
730 : : if (existing == impl_->url_buffer_map.end()) {
731 : : impl_->url_buffer_map.emplace(std::move(url), std::move(url_buffer));
732 : : retention_changed = true;
733 : : continue;
734 : : }
735 : :
736 : : retired_subscribers.push_back(deactivate_url_buffer(*existing->second));
737 : : existing->second = std::move(url_buffer);
738 : : }
739 : :
740 : : if (impl_->config.destroy_on_offline) {
741 : : for (auto iter = impl_->url_buffer_map.begin(); iter != impl_->url_buffer_map.end();) {
742 : : if VUNLIKELY (current.count(std::string_view(iter->first)) == 0) {
743 : : retired_subscribers.push_back(deactivate_url_buffer(*iter->second));
744 : : iter = impl_->url_buffer_map.erase(iter);
745 : : retention_changed = true;
746 : : } else {
747 : : ++iter;
748 : : }
749 : : }
750 : : }
751 : :
752 : : if (retention_changed) {
753 : : recompute_retention();
754 : : }
755 : : }
756 : : }
757 : :
758 : : void TriggerRecorder::sweep_evict() {
759 : : const int64_t now = impl_->capture_timer.get();
760 : :
761 : : std::shared_lock lock(impl_->url_buffer_mtx);
762 : :
763 : : for (const auto& entry : impl_->url_buffer_map) {
764 : : UrlBuffer* url_buffer = entry.second.get();
765 : :
766 : : std::lock_guard ring_lock(url_buffer->mtx);
767 : :
768 : : const int64_t horizon = now - url_buffer->retention_us.load(std::memory_order_relaxed);
769 : :
770 : : while (!url_buffer->ring.empty() && url_buffer->ring.front().capture_ts_us < horizon) {
771 : : const auto size = static_cast<int64_t>(url_buffer->ring.front().payload->size());
772 : : url_buffer->bytes -= size;
773 : : impl_->global_bytes.fetch_sub(size, std::memory_order_relaxed);
774 : : url_buffer->ring.pop_front();
775 : : }
776 : : }
777 : : }
778 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
779 : :
780 : 17 : void TriggerRecorder::finish_dump(DumpJob& job) {
781 [ + - ]: 17 : std::lock_guard lifecycle_lock(impl_->lifecycle_mtx);
782 [ + - ]: 17 : finish_dump_locked(job);
783 : 17 : }
784 : :
785 : 18 : void TriggerRecorder::finish_dump_locked(DumpJob& job) {
786 [ - + ]: 18 : if VUNLIKELY (impl_->active_dump_job.get() != &job) {
787 : 0 : return;
788 : : }
789 : :
790 [ + - ]: 18 : std::shared_lock map_lock(impl_->url_buffer_mtx);
791 : :
792 [ + + ]: 26 : for (const auto& url_buffer : job.frozen_buffers) {
793 [ + - ]: 8 : std::lock_guard ring_lock(url_buffer->mtx);
794 : 8 : url_buffer->frozen = false;
795 : :
796 [ - + - - : 8 : if (!url_buffer->sub && url_buffer->bytes > 0) {
- + ]
797 : 0 : impl_->global_bytes.fetch_sub(url_buffer->bytes, std::memory_order_relaxed);
798 : : }
799 : 8 : }
800 : :
801 : 18 : impl_->writing.store(false, std::memory_order_release);
802 [ + - ]: 18 : recompute_retention();
803 : 18 : impl_->dumping.store(false, std::memory_order_release);
804 : 18 : impl_->active_dump_job.reset();
805 : 18 : }
806 : :
807 : : // LCOV_EXCL_START GCOVR_EXCL_START
808 : : void TriggerRecorder::notify_dump_failed(const DumpJob& job, std::string_view error) {
809 : : VLOG_E("TriggerRecorder: dump failed -> ", job.path, " error=", error);
810 : :
811 : : if (!impl_->trigger_plugin) {
812 : : return;
813 : : }
814 : :
815 : : TriggerPluginInterface::DumpResult result;
816 : : result.reason = job.params.reason;
817 : : result.path = job.path;
818 : : result.error = error;
819 : : impl_->trigger_plugin->on_dump_failed(result);
820 : : }
821 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
822 : :
823 : 17 : void TriggerRecorder::do_dump(DumpJob& job) {
824 : : struct FinishGuard final {
825 : : TriggerRecorder& recorder;
826 : : DumpJob& job;
827 : :
828 : 17 : ~FinishGuard() { recorder.finish_dump(job); }
829 : : };
830 : :
831 : : struct SnapFrame final {
832 : : int64_t capture_ts_us;
833 : : std::shared_ptr<const Bytes> payload;
834 : : const UrlBuffer* source;
835 : : };
836 : :
837 : : struct LossInfo final {
838 : : const UrlBuffer* source;
839 : : double loss;
840 : : };
841 : :
842 : 17 : FinishGuard guard{*this, job};
843 : :
844 : 17 : const auto& params = job.params;
845 : 17 : const auto& path = job.path;
846 : 17 : auto* trigger_plugin = impl_->trigger_plugin.get();
847 : 17 : const int64_t trigger_ts = job.trigger_ts;
848 : 17 : const int64_t dump_start = impl_->capture_timer.get();
849 : :
850 [ - + ]: 17 : const int64_t trigger_pre_us = params.pre_ms >= 0 ? params.pre_ms * 1000 : -1;
851 [ - + ]: 17 : const int64_t trigger_post_us = params.post_ms >= 0 ? params.post_ms * 1000 : -1;
852 : :
853 : 17 : std::vector<SnapFrame> snapshot;
854 : 17 : std::vector<LossInfo> losses;
855 [ + - ]: 17 : losses.reserve(job.frozen_buffers.size());
856 : :
857 : : // LCOV_EXCL_START GCOVR_EXCL_START
858 : : for (const auto& url_buffer : job.frozen_buffers) {
859 : : std::lock_guard ring_lock(url_buffer->mtx);
860 : : SampleLostInfo current_lost = url_buffer->sub ? url_buffer->sub->get_lost() : url_buffer->final_lost;
861 : : const uint64_t delta_total = current_lost.total - url_buffer->last_lost.total;
862 : : const uint64_t delta_lost = current_lost.lost - url_buffer->last_lost.lost;
863 : : const double loss = delta_total > 0 ? static_cast<double>(delta_lost) / static_cast<double>(delta_total) : 0.0;
864 : : url_buffer->last_lost = current_lost;
865 : :
866 : : const int64_t pre = trigger_pre_us >= 0 ? std::min(trigger_pre_us, url_buffer->pre_us) : url_buffer->pre_us;
867 : : const int64_t post = trigger_post_us >= 0 ? std::min(trigger_post_us, url_buffer->post_us) : url_buffer->post_us;
868 : :
869 : : const int64_t window_begin = trigger_ts - pre;
870 : : const int64_t window_end = trigger_ts + post;
871 : :
872 : : auto first = std::lower_bound(
873 : : url_buffer->ring.begin(), url_buffer->ring.end(), window_begin,
874 : : [](const UrlBuffer::Entry& entry, int64_t timestamp) { return entry.capture_ts_us < timestamp; });
875 : :
876 : : for (; first != url_buffer->ring.end() && first->capture_ts_us <= window_end; ++first) {
877 : : snapshot.push_back(SnapFrame{first->capture_ts_us, first->payload, url_buffer.get()});
878 : : }
879 : :
880 : : losses.push_back(LossInfo{url_buffer.get(), loss});
881 : : }
882 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
883 : :
884 [ + + ]: 17 : if (trigger_plugin) {
885 : 14 : TriggerPluginInterface::TriggerContext context;
886 [ + - ]: 14 : context.reason = params.reason;
887 [ + - ]: 14 : context.name_hint = params.name_hint;
888 [ + - ]: 14 : context.out_file = params.out_file;
889 : 14 : context.pre_ms = params.pre_ms;
890 : 14 : context.post_ms = params.post_ms;
891 : 14 : context.trigger_timestamp = impl_->anchor_wall_us + trigger_ts;
892 : :
893 [ + - ]: 14 : trigger_plugin->on_trigger(context);
894 : 14 : }
895 : :
896 [ + + + - : 17 : if (params.out_file.empty() && impl_->config.max_dump_file_count > 0) {
+ + ]
897 [ + - + - ]: 10 : const auto file_extension = std::filesystem::path(path).extension();
898 : 10 : std::vector<std::string> removed;
899 : 10 : std::error_code ec;
900 : 10 : std::vector<std::filesystem::directory_entry> files;
901 [ + - + - ]: 10 : std::filesystem::directory_iterator iter(impl_->config.dump_dir, ec);
902 : :
903 [ + - ]: 10 : if (!ec) {
904 [ + - + + ]: 16 : for (const std::filesystem::directory_iterator end; iter != end; iter.increment(ec)) {
905 : 6 : std::error_code type_ec;
906 : :
907 [ + - + - : 6 : if (iter->is_regular_file(type_ec) && iter->path().extension() == file_extension) {
+ - + - +
- - - ]
908 [ + - ]: 6 : files.push_back(*iter);
909 : : }
910 : 10 : }
911 : :
912 [ + - ]: 10 : std::sort(files.begin(), files.end(),
913 : 2 : [](const std::filesystem::directory_entry& left, const std::filesystem::directory_entry& right) {
914 : 2 : std::error_code left_error;
915 : 2 : std::error_code right_error;
916 [ + - ]: 2 : return std::filesystem::last_write_time(left, left_error) <
917 : 4 : std::filesystem::last_write_time(right, right_error);
918 : : });
919 : :
920 : 10 : const auto max_count = static_cast<size_t>(impl_->config.max_dump_file_count);
921 [ + + ]: 10 : const size_t remove_count = files.size() >= max_count ? files.size() - max_count + 1 : 0;
922 : :
923 [ + + ]: 12 : for (size_t index = 0; index < remove_count; ++index) {
924 : 2 : std::error_code remove_error;
925 : :
926 [ + - + - : 2 : if (std::filesystem::remove(files[index].path(), remove_error) && trigger_plugin) {
+ - ]
927 [ + - + - ]: 2 : removed.push_back(files[index].path().string());
928 : : }
929 : : }
930 : : }
931 : :
932 [ + + ]: 10 : if (trigger_plugin) {
933 [ + + ]: 11 : for (const auto& removed_path : removed) {
934 [ + - ]: 2 : trigger_plugin->on_file_rotated(removed_path);
935 : : }
936 : : }
937 : 10 : }
938 : :
939 [ + - ]: 17 : std::stable_sort(snapshot.begin(), snapshot.end(), [](const SnapFrame& left, const SnapFrame& right) {
940 : : return left.capture_ts_us < right.capture_ts_us; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
941 : : });
942 : :
943 [ + + ]: 17 : const int64_t min_capture = snapshot.empty() ? trigger_ts : std::min(trigger_ts, snapshot.front().capture_ts_us);
944 : :
945 : 17 : BagWriter::Config writer_config;
946 [ - + ]: 17 : writer_config.compress = impl_->config.enable_compress ? BagWriter::kCompressAuto : BagWriter::kCompressNone;
947 [ + - ]: 17 : writer_config.tag_name = params.reason;
948 : 17 : writer_config.sync_mode = true;
949 : 17 : writer_config.optimize_on_exit = true;
950 : 17 : writer_config.start_timestamp = (impl_->anchor_wall_us + min_capture) / 1000;
951 : :
952 : 17 : std::shared_ptr<BagWriter> writer;
953 : :
954 : : try {
955 [ + - ]: 17 : writer = BagWriter::create(path, writer_config);
956 [ - - ]: 0 : } catch (const std::exception& e) {
957 : : // LCOV_EXCL_START GCOVR_EXCL_START
958 : : notify_dump_failed(job, e.what());
959 : : return;
960 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
961 : 0 : }
962 : :
963 [ - + ]: 17 : if VUNLIKELY (!writer) {
964 : : // LCOV_EXCL_START GCOVR_EXCL_START
965 : : notify_dump_failed(job, "unsupported bag suffix");
966 : : return;
967 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
968 : : }
969 : :
970 [ + + ]: 17 : if (impl_->bag_plugin) {
971 [ + - ]: 3 : writer->bind_bag_interface(impl_->bag_plugin);
972 : : }
973 : :
974 [ + + ]: 24 : for (const auto& info : losses) {
975 : : writer->set_url_loss(info.source->url, info.loss); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
976 : : }
977 : :
978 : 17 : TriggerPluginInterface::DumpContext dump_context;
979 : :
980 : 17 : impl_->writing.store(true, std::memory_order_release);
981 [ + - + - ]: 34 : VLOG_I("TriggerRecorder: dump writing -> ", path, " frames=", snapshot.size(), " urls=", losses.size());
982 : :
983 [ + + ]: 17 : if (trigger_plugin) {
984 [ + - ]: 14 : dump_context.reason = params.reason;
985 [ + - ]: 14 : dump_context.path = path;
986 : 14 : dump_context.start_timestamp = writer_config.start_timestamp;
987 : 14 : dump_context.url_count = static_cast<int64_t>(losses.size());
988 [ + - ]: 14 : trigger_plugin->on_dump_started(dump_context);
989 : : }
990 : :
991 : 17 : int64_t throttle_bytes = 0;
992 : 17 : int64_t byte_count = 0;
993 : 17 : bool persistence_failed = false;
994 : 17 : const size_t snapshot_frame_count = snapshot.size();
995 : 17 : Frame frame;
996 : 17 : frame.action_type = ActionType::kSubscribe;
997 : :
998 [ + + ]: 115 : for (auto& item : snapshot) {
999 : : // LCOV_EXCL_START GCOVR_EXCL_START
1000 : : const auto item_size = static_cast<int64_t>(item.payload->size());
1001 : : frame.timestamp = item.capture_ts_us - min_capture;
1002 : : frame.url = item.source->url;
1003 : : frame.ser_type = item.source->ser_type;
1004 : : frame.schema_type = item.source->schema_type;
1005 : : frame.data = Bytes::shallow_copy(item.payload->data(), item.payload->size());
1006 : :
1007 : : if VUNLIKELY (writer->push(frame) < 0) {
1008 : : frame.data.clear();
1009 : : item.payload.reset();
1010 : : persistence_failed = true;
1011 : : break;
1012 : : }
1013 : :
1014 : : byte_count += item_size;
1015 : :
1016 : : if (trigger_plugin) {
1017 : : trigger_plugin->on_frame(frame, dump_context);
1018 : : }
1019 : :
1020 : : frame.data.clear();
1021 : : item.payload.reset();
1022 : :
1023 : : if (impl_->config.sleep_time_ms > 0 && impl_->config.sleep_interval > 0) {
1024 : : throttle_bytes += item_size;
1025 : :
1026 : : if (throttle_bytes >= impl_->config.sleep_interval) {
1027 : : std::this_thread::sleep_for(std::chrono::milliseconds(impl_->config.sleep_time_ms));
1028 : : throttle_bytes = 0;
1029 : : }
1030 : : }
1031 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1032 : : }
1033 : :
1034 : 17 : frame.data.clear();
1035 : :
1036 : 17 : std::vector<SnapFrame>().swap(snapshot);
1037 : :
1038 [ + + ]: 17 : if (impl_->bag_plugin) {
1039 [ + - ]: 3 : writer->clear_bag_interface();
1040 : : }
1041 : :
1042 [ + - ]: 17 : writer->close();
1043 : 17 : impl_->writing.store(false, std::memory_order_release);
1044 : :
1045 : 17 : const bool writer_failed = writer->fail();
1046 : :
1047 : 17 : writer.reset();
1048 : :
1049 [ + - - + : 17 : if VUNLIKELY (persistence_failed || writer_failed) {
- + ]
1050 [ # # ]: 0 : notify_dump_failed(job, "writer persistence failure");
1051 : 0 : return;
1052 : : }
1053 : :
1054 [ + + ]: 17 : if (trigger_plugin) {
1055 : 14 : TriggerPluginInterface::DumpResult result;
1056 [ + - ]: 14 : result.reason = params.reason;
1057 [ + - ]: 14 : result.path = path;
1058 : 14 : result.frame_count = static_cast<int64_t>(snapshot_frame_count);
1059 : 14 : result.byte_count = byte_count;
1060 : 14 : result.url_count = static_cast<int64_t>(losses.size());
1061 : 14 : result.start_timestamp = writer_config.start_timestamp;
1062 : 14 : result.duration_us = impl_->capture_timer.get() - dump_start;
1063 : 14 : result.success = true;
1064 : :
1065 [ + - ]: 14 : trigger_plugin->on_dump_finished(result);
1066 : 14 : }
1067 : :
1068 [ + - + - ]: 34 : VLOG_I("TriggerRecorder: dump finished -> ", path, " frames=", snapshot_frame_count, " bytes=", byte_count);
1069 [ + - + - : 17 : }
+ - + - +
- + - +
- ]
1070 : :
1071 : : } // namespace vlink
|
