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