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/discovery_viewer.h"
25 : :
26 : : #include <algorithm>
27 : : #include <charconv>
28 : : #include <cstring>
29 : : #include <ctime>
30 : : #include <exception>
31 : : #include <iterator>
32 : : #include <map>
33 : : #include <memory>
34 : : #include <shared_mutex>
35 : : #include <string>
36 : : #include <thread>
37 : : #include <unordered_map>
38 : : #include <unordered_set>
39 : : #include <utility>
40 : : #include <vector>
41 : :
42 : : #include "./base/bytes.h"
43 : : #include "./base/elapsed_timer.h"
44 : : #include "./base/helpers.h"
45 : : #include "./base/logger.h"
46 : : #include "./base/utils.h"
47 : : #include "./impl/url.h"
48 : :
49 : : #if __has_include(<unistd.h>)
50 : : #include <unistd.h>
51 : : #endif
52 : :
53 : : #ifdef _WIN32
54 : : #include <Winsock2.h>
55 : : #include <ws2tcpip.h>
56 : : #else
57 : : #ifdef __QNX__
58 : : #include <sys/time.h>
59 : : #endif
60 : : #include <arpa/inet.h>
61 : : #include <netinet/in.h>
62 : : #include <poll.h>
63 : : #include <sys/socket.h>
64 : : #endif
65 : :
66 : : #define VLINK_DISCOVERY_MULTICAST 1
67 : : #define VLINK_DISCOVERY_OFFLINE 0
68 : :
69 : : namespace vlink {
70 : :
71 : : #ifdef _WIN32
72 : : using SocketHandle = SOCKET;
73 : : static constexpr SocketHandle kInvalidSocket = INVALID_SOCKET;
74 : : #else
75 : : using SocketHandle = int;
76 : : static constexpr SocketHandle kInvalidSocket = -1;
77 : : #endif
78 : :
79 : : [[maybe_unused]] static constexpr int kCollectInterval = 500;
80 : : [[maybe_unused]] static constexpr int kReportInterval = 500;
81 : : [[maybe_unused]] static constexpr int kReportTimeout = kReportInterval * 4;
82 : : [[maybe_unused]] static constexpr size_t kMaxTaskSize = 50000U;
83 : : [[maybe_unused]] static constexpr uint32_t kMaxElapsedTime = 1000;
84 : : [[maybe_unused]] static constexpr int kBroadcastBasePort = 51600;
85 : : [[maybe_unused]] static constexpr size_t kBufferSize = 1024 * 1024U;
86 : : [[maybe_unused]] static constexpr size_t kMaxBatchSize = 64U;
87 : : [[maybe_unused]] static constexpr uint32_t kMaxDiscoveryDomain = 255U;
88 : : [[maybe_unused]] static constexpr int kTimeoutWarnInterval = 10000;
89 : : [[maybe_unused]] static constexpr int64_t kConflictWarnInterval = 60000;
90 : : [[maybe_unused]] static constexpr size_t kMaxRecvPerBatch = 256U;
91 : :
92 : : #if VLINK_DISCOVERY_MULTICAST
93 : : [[maybe_unused]] static constexpr const char* kBroadcastAddress = "239.255.0.100";
94 : : #else
95 : : [[maybe_unused]] static constexpr const char* kBroadcastAddress = "255.255.255.255";
96 : : #endif
97 : :
98 : 184 : [[maybe_unused]] static bool has_pending_data(SocketHandle sock) {
99 : : #ifdef _WIN32
100 : : u_long available = 0;
101 : :
102 : : return ::ioctlsocket(sock, FIONREAD, &available) == 0 && available > 0;
103 : : #else
104 : : pollfd target_fd;
105 : 184 : std::memset(&target_fd, 0, sizeof(target_fd));
106 : :
107 : 184 : target_fd.fd = sock;
108 : 184 : target_fd.events = POLLIN;
109 : :
110 [ + - - + : 184 : return ::poll(&target_fd, 1, 0) > 0 && (target_fd.revents & POLLIN) != 0;
- - ]
111 : : #endif
112 : : }
113 : :
114 : 42 : [[maybe_unused]] static std::string& get_invalid_listen_domain() {
115 [ + - + - ]: 42 : static std::string invalid_domain;
116 : 42 : return invalid_domain;
117 : : }
118 : :
119 : 29 : [[maybe_unused]] static std::string node_count_to_string(size_t node_count) {
120 [ + + ]: 29 : if (node_count > 9) {
121 [ + - ]: 1 : return std::string("~");
122 : : }
123 : :
124 : 28 : return std::to_string(node_count);
125 : : }
126 : :
127 : : // DiscoveryMessage
128 : : struct DiscoveryMessage final {
129 : : std::string message;
130 : : std::string target_ip;
131 : : };
132 : :
133 : : // DiscoveryViewer::Impl
134 : : struct DiscoveryViewer::Impl final {
135 : : struct Comparator final {
136 : 1000 : bool operator()(const DiscoveryViewer::Info& lhs, const DiscoveryViewer::Info& rhs) const {
137 [ + + ]: 1000 : if (lhs.sort_index < rhs.sort_index) {
138 : 100 : return true;
139 [ + + ]: 900 : } else if (lhs.sort_index > rhs.sort_index) {
140 : 73 : return false;
141 : : }
142 : :
143 [ + + ]: 827 : if (lhs.url < rhs.url) {
144 : 478 : return true;
145 [ + + ]: 349 : } else if (lhs.url > rhs.url) {
146 : 236 : return false;
147 : : }
148 : :
149 [ + + ]: 113 : if (lhs.type < rhs.type) {
150 : 17 : return true;
151 [ + + ]: 96 : } else if (lhs.type > rhs.type) {
152 : 10 : return false;
153 : : }
154 : :
155 [ - + ]: 86 : if (lhs.schema_type < rhs.schema_type) {
156 : 0 : return true;
157 [ - + ]: 86 : } else if (lhs.schema_type > rhs.schema_type) {
158 : 0 : return false;
159 : : }
160 : :
161 [ - + ]: 86 : if (lhs.ser_type < rhs.ser_type) {
162 : : return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
163 [ - + ]: 86 : } else if (lhs.ser_type > rhs.ser_type) {
164 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
165 : : }
166 : :
167 : 86 : return lhs.process_list < rhs.process_list;
168 : : }
169 : : };
170 : :
171 : : DiscoveryViewer::FilterType filter_type{DiscoveryViewer::kFilterNone};
172 : : std::string native_hostname;
173 : :
174 : : std::map<DiscoveryViewer::Info, ElapsedTimer, Comparator> info_map;
175 : : std::unordered_map<std::string, std::string> ser_map;
176 : : std::unordered_map<std::string, SchemaType> schema_type_map;
177 : : std::vector<DiscoveryViewer::Info> info_list;
178 : :
179 : : std::recursive_mutex mtx;
180 : : std::shared_mutex ser_mtx;
181 : : bool list_dirty{false};
182 : : bool callback_dirty{false};
183 : : std::unordered_map<std::string, ElapsedTimer> ser_warned_urls;
184 : :
185 : : DiscoveryViewer::Callback callback;
186 : : std::thread thread;
187 : : std::vector<uint8_t> buffer;
188 : : Timer timer;
189 : : ElapsedTimer schedule_elapsed;
190 : :
191 : : SocketHandle sock{kInvalidSocket};
192 : : sockaddr_in address;
193 : : bool enable_native_discovery{false};
194 : : #ifdef _WIN32
195 : : bool winsock_initialized{false};
196 : : #endif
197 : : };
198 : :
199 : : // DiscoveryViewer::Process
200 : 193 : bool DiscoveryViewer::Process::operator<(const DiscoveryViewer::Process& target) const noexcept {
201 [ + + ]: 193 : if (type < target.type) {
202 : 1 : return true;
203 [ + + ]: 192 : } else if (type > target.type) {
204 : 8 : return false;
205 [ + + ]: 184 : } else if (host < target.host) {
206 : 1 : return true;
207 [ + + ]: 183 : } else if (host > target.host) {
208 : 1 : return false;
209 [ + + ]: 182 : } else if (name < target.name) {
210 : 1 : return true;
211 [ - + ]: 181 : } else if (name > target.name) {
212 : 0 : return false;
213 [ + + ]: 181 : } else if (pid < target.pid) {
214 : 3 : return true;
215 [ + + ]: 178 : } else if (pid > target.pid) {
216 : 1 : return false;
217 [ - + ]: 177 : } else if (bridge != target.bridge) {
218 : 0 : return bridge < target.bridge;
219 : : } else {
220 : 177 : return std::lexicographical_compare(ip_list.begin(), ip_list.end(), target.ip_list.begin(), target.ip_list.end(),
221 : 177 : Utils::ip_less);
222 : : }
223 : : }
224 : :
225 : : // DiscoveryViewer::Info
226 : 9 : bool DiscoveryViewer::Info::operator<(const DiscoveryViewer::Info& target) const noexcept {
227 [ + + ]: 9 : if (type < target.type) {
228 : 1 : return true;
229 [ - + ]: 8 : } else if (type > target.type) {
230 : 0 : return false;
231 [ + + ]: 8 : } else if (sort_index < target.sort_index) {
232 : 1 : return true;
233 [ + + ]: 7 : } else if (sort_index > target.sort_index) {
234 : 1 : return false;
235 [ + + ]: 6 : } else if (url < target.url) {
236 : 1 : return true;
237 [ + + ]: 5 : } else if (url > target.url) {
238 : 1 : return false;
239 [ + + ]: 4 : } else if (schema_type < target.schema_type) {
240 : 1 : return true;
241 [ - + ]: 3 : } else if (schema_type > target.schema_type) {
242 : 0 : return false;
243 [ + + ]: 3 : } else if (ser_type < target.ser_type) {
244 : 1 : return true;
245 [ - + ]: 2 : } else if (ser_type > target.ser_type) {
246 : 0 : return false;
247 : : } else {
248 : 2 : return process_list < target.process_list;
249 : : }
250 : : }
251 : :
252 : : // DiscoveryViewer
253 : 226 : ImplType DiscoveryViewer::convert_type(std::string_view str) {
254 [ + + ]: 226 : if (str == "Ser") {
255 : 2 : return kServer;
256 [ + + ]: 224 : } else if (str == "Cli") {
257 : 26 : return kClient;
258 [ + + ]: 198 : } else if (str == "Pub") {
259 : 52 : return kPublisher;
260 [ + + ]: 146 : } else if (str == "Sub") {
261 : 103 : return kSubscriber;
262 [ + + ]: 43 : } else if (str == "Set") {
263 : 13 : return kSetter;
264 [ + + ]: 30 : } else if (str == "Get") {
265 : 26 : return kGetter;
266 : : } else {
267 : 4 : return kUnknownImplType;
268 : : }
269 : : }
270 : :
271 : 20 : std::string DiscoveryViewer::convert_type_to_view(uint32_t type) {
272 [ + + + + : 20 : switch (type) {
+ + + + +
+ + + ]
273 : 1 : case kPublisher | kSubscriber:
274 [ + - ]: 1 : return "Pub|Sub";
275 : 1 : case kSetter | kGetter:
276 [ + - ]: 1 : return "Set|Get";
277 : 1 : case kServer | kClient:
278 [ + - ]: 1 : return "Ser|Cli";
279 : 1 : case kPublisher:
280 [ + - ]: 1 : return "Pub|---";
281 : 1 : case kSubscriber:
282 [ + - ]: 1 : return "---|Sub";
283 : 1 : case kSetter:
284 [ + - ]: 1 : return "Set|---";
285 : 1 : case kGetter:
286 [ + - ]: 1 : return "---|Get";
287 : 1 : case kServer:
288 [ + - ]: 1 : return "Ser|---";
289 : 1 : case kClient:
290 [ + - ]: 1 : return "---|Cli";
291 : 1 : case kPublisher | kGetter:
292 [ + - ]: 1 : return "Pub|Get";
293 : 1 : case kSetter | kSubscriber:
294 [ + - ]: 1 : return "Set|Sub";
295 : 9 : default:
296 [ + + ]: 9 : if (type == (kPublisher | kSetter)) {
297 [ + - ]: 1 : return "Pub|---";
298 [ + + ]: 8 : } else if (type == (kPublisher | kSetter | kSubscriber)) {
299 [ + - ]: 1 : return "Pub|Sub";
300 [ + + ]: 7 : } else if (type == (kPublisher | kSetter | kGetter)) {
301 [ + - ]: 1 : return "Pub|Get";
302 [ + + ]: 6 : } else if (type == (kPublisher | kSetter | kSubscriber | kGetter)) {
303 [ + - ]: 1 : return "Pub|Sub";
304 [ + + ]: 5 : } else if (type == (kSubscriber | kGetter)) {
305 [ + - ]: 1 : return "---|Sub";
306 [ + + ]: 4 : } else if (type == (kSubscriber | kGetter | kPublisher)) {
307 [ + - ]: 1 : return "Pub|Sub";
308 [ + + ]: 3 : } else if (type == (kSubscriber | kGetter | kSetter)) {
309 [ + - ]: 1 : return "Set|Sub";
310 : : } else {
311 [ + - ]: 2 : return "???|???";
312 : : }
313 : : }
314 : : }
315 : :
316 : 21 : std::string DiscoveryViewer::convert_type_to_view(uint32_t type, const std::vector<Process>& process_list) {
317 : 21 : int left_cnt = 0;
318 : 21 : int right_cnt = 0;
319 : :
320 [ + + + + : 21 : switch (type) {
+ + + + +
+ + + ]
321 : 1 : case kPublisher | kSubscriber:
322 [ + + ]: 3 : for (const auto& process : process_list) {
323 [ + + ]: 2 : if (process.type == kPublisher) {
324 : 1 : ++left_cnt;
325 [ + - ]: 1 : } else if (process.type == kSubscriber) {
326 : 1 : ++right_cnt;
327 : : }
328 : : }
329 : :
330 [ + - + - : 2 : return "Pub*" + node_count_to_string(left_cnt) + "|Sub*" + node_count_to_string(right_cnt);
+ - + - ]
331 : 1 : case kSetter | kGetter:
332 [ + + ]: 4 : for (const auto& process : process_list) {
333 [ + + ]: 3 : if (process.type == kSetter) {
334 : 1 : ++left_cnt;
335 [ + - ]: 2 : } else if (process.type == kGetter) {
336 : 2 : ++right_cnt;
337 : : }
338 : : }
339 : :
340 [ + - + - : 2 : return "Set*" + node_count_to_string(left_cnt) + "|Get*" + node_count_to_string(right_cnt);
+ - + - ]
341 : 1 : case kServer | kClient:
342 [ + + ]: 4 : for (const auto& process : process_list) {
343 [ + + ]: 3 : if (process.type == kServer) {
344 : 2 : ++left_cnt;
345 [ + - ]: 1 : } else if (process.type == kClient) {
346 : 1 : ++right_cnt;
347 : : }
348 : : }
349 : :
350 [ + - + - : 2 : return "Ser*" + node_count_to_string(left_cnt) + "|Cli*" + node_count_to_string(right_cnt);
+ - + - ]
351 : 2 : case kPublisher:
352 [ + + ]: 23 : for (const auto& process : process_list) {
353 [ + + ]: 21 : if (process.type == kPublisher) {
354 : 16 : ++left_cnt;
355 : : }
356 : : }
357 : :
358 [ + - + - ]: 4 : return "Pub*" + node_count_to_string(left_cnt) + "|-----";
359 : 1 : case kSubscriber:
360 [ + + ]: 7 : for (const auto& process : process_list) {
361 [ + + ]: 6 : if (process.type == kSubscriber) {
362 : 1 : ++right_cnt;
363 : : }
364 : : }
365 : :
366 [ + - ]: 2 : return "-----|Sub*" + node_count_to_string(right_cnt);
367 : 1 : case kSetter:
368 [ + + ]: 7 : for (const auto& process : process_list) {
369 [ + + ]: 6 : if (process.type == kSetter) {
370 : 1 : ++left_cnt;
371 : : }
372 : : }
373 : :
374 [ + - + - ]: 2 : return "Set*" + node_count_to_string(left_cnt) + "|-----";
375 : 1 : case kGetter:
376 [ + + ]: 7 : for (const auto& process : process_list) {
377 [ + + ]: 6 : if (process.type == kGetter) {
378 : 1 : ++right_cnt;
379 : : }
380 : : }
381 : :
382 [ + - ]: 2 : return "-----|Get*" + node_count_to_string(right_cnt);
383 : 1 : case kServer:
384 [ + + ]: 7 : for (const auto& process : process_list) {
385 [ + + ]: 6 : if (process.type == kServer) {
386 : 1 : ++left_cnt;
387 : : }
388 : : }
389 : :
390 [ + - + - ]: 2 : return "Ser*" + node_count_to_string(left_cnt) + "|-----";
391 : 1 : case kClient:
392 [ + + ]: 7 : for (const auto& process : process_list) {
393 [ + + ]: 6 : if (process.type == kClient) {
394 : 1 : ++right_cnt;
395 : : }
396 : : }
397 : :
398 [ + - ]: 2 : return "-----|Cli*" + node_count_to_string(right_cnt);
399 : :
400 : 1 : case kPublisher | kGetter:
401 [ + + ]: 7 : for (const auto& process : process_list) {
402 [ + + ]: 6 : if (process.type == kPublisher) {
403 : 1 : ++left_cnt;
404 [ + + ]: 5 : } else if (process.type == kGetter) {
405 : 1 : ++right_cnt;
406 : : }
407 : : }
408 : :
409 [ + - + - : 2 : return "Pub*" + node_count_to_string(left_cnt) + "|Get*" + node_count_to_string(right_cnt);
+ - + - ]
410 : :
411 : 1 : case kSetter | kSubscriber:
412 [ + + ]: 7 : for (const auto& process : process_list) {
413 [ + + ]: 6 : if (process.type == kSetter) {
414 : 1 : ++left_cnt;
415 [ + + ]: 5 : } else if (process.type == kSubscriber) {
416 : 1 : ++right_cnt;
417 : : }
418 : : }
419 : :
420 [ + - + - : 2 : return "Set*" + node_count_to_string(left_cnt) + "|Sub*" + node_count_to_string(right_cnt);
+ - + - ]
421 : 9 : default:
422 [ + + ]: 9 : if (type == (kPublisher | kSetter)) {
423 [ + + ]: 7 : for (const auto& process : process_list) {
424 [ + + + + ]: 6 : if (process.type == kPublisher || process.type == kSetter) {
425 : 2 : ++left_cnt;
426 : : }
427 : : }
428 : :
429 [ + - + - ]: 2 : return "Pub*" + node_count_to_string(left_cnt) + "|-----";
430 [ + + ]: 8 : } else if (type == (kPublisher | kSetter | kSubscriber)) {
431 [ + + ]: 7 : for (const auto& process : process_list) {
432 [ + + + + ]: 6 : if (process.type == kPublisher || process.type == kSetter) {
433 : 2 : ++left_cnt;
434 [ + + ]: 4 : } else if (process.type == kSubscriber) {
435 : 1 : ++right_cnt;
436 : : }
437 : : }
438 : :
439 [ + - + - : 2 : return "Pub*" + node_count_to_string(left_cnt) + "|Sub*" + node_count_to_string(right_cnt);
+ - + - ]
440 [ + + ]: 7 : } else if (type == (kPublisher | kSetter | kGetter)) {
441 [ + + ]: 7 : for (const auto& process : process_list) {
442 [ + + + + ]: 6 : if (process.type == kPublisher || process.type == kSetter) {
443 : 2 : ++left_cnt;
444 [ + + ]: 4 : } else if (process.type == kGetter) {
445 : 1 : ++right_cnt;
446 : : }
447 : : }
448 : :
449 [ + - + - : 2 : return "Pub*" + node_count_to_string(left_cnt) + "|Get*" + node_count_to_string(right_cnt);
+ - + - ]
450 [ + + ]: 6 : } else if (type == (kPublisher | kSetter | kSubscriber | kGetter)) {
451 [ + + ]: 7 : for (const auto& process : process_list) {
452 [ + + + + ]: 6 : if (process.type == kPublisher || process.type == kSetter) {
453 : 2 : ++left_cnt;
454 [ + + + + ]: 4 : } else if (process.type == kSubscriber || process.type == kGetter) {
455 : 2 : ++right_cnt;
456 : : }
457 : : }
458 : :
459 [ + - + - : 2 : return "Pub*" + node_count_to_string(left_cnt) + "|Sub*" + node_count_to_string(right_cnt);
+ - + - ]
460 [ + + ]: 5 : } else if (type == (kSubscriber | kGetter)) {
461 [ + + ]: 7 : for (const auto& process : process_list) {
462 [ + + + + ]: 6 : if (process.type == kSubscriber || process.type == kGetter) {
463 : 2 : ++right_cnt;
464 : : }
465 : : }
466 : :
467 [ + - ]: 2 : return "-----|Sub*" + node_count_to_string(right_cnt);
468 [ + + ]: 4 : } else if (type == (kSubscriber | kGetter | kPublisher)) {
469 [ + + ]: 7 : for (const auto& process : process_list) {
470 [ + + ]: 6 : if (process.type == kPublisher) {
471 : 1 : ++left_cnt;
472 [ + + + + ]: 5 : } else if (process.type == kSubscriber || process.type == kGetter) {
473 : 2 : ++right_cnt;
474 : : }
475 : : }
476 : :
477 [ + - + - : 2 : return "Pub*" + node_count_to_string(left_cnt) + "|Sub*" + node_count_to_string(right_cnt);
+ - + - ]
478 [ + + ]: 3 : } else if (type == (kSubscriber | kGetter | kSetter)) {
479 [ + + ]: 7 : for (const auto& process : process_list) {
480 [ + + ]: 6 : if (process.type == kSetter) {
481 : 1 : ++left_cnt;
482 [ + + + + ]: 5 : } else if (process.type == kSubscriber || process.type == kGetter) {
483 : 2 : ++right_cnt;
484 : : }
485 : : }
486 : :
487 [ + - + - : 2 : return "Set*" + node_count_to_string(left_cnt) + "|Sub*" + node_count_to_string(right_cnt);
+ - + - ]
488 : : } else {
489 [ + - ]: 2 : return "?????|?????";
490 : : }
491 : : }
492 : : }
493 : :
494 : 122 : uint32_t DiscoveryViewer::get_listen_domain() {
495 : 42 : static uint32_t domain = []() -> uint32_t {
496 [ + - + - ]: 84 : const std::string domain_str = Utils::get_env("VLINK_DISCOVER_DOMAIN");
497 : :
498 [ + - ]: 42 : if (domain_str.empty()) {
499 : 42 : return 0U;
500 : : }
501 : :
502 : 0 : uint32_t value = 0U;
503 : :
504 [ # # ]: 0 : auto [ptr, error] = std::from_chars(domain_str.data(), domain_str.data() + domain_str.size(), value);
505 : :
506 [ # # # # : 0 : if VUNLIKELY (error != std::errc() || ptr != domain_str.data() + domain_str.size() || value > kMaxDiscoveryDomain) {
# # # # #
# ]
507 [ # # ]: 0 : get_invalid_listen_domain() = domain_str;
508 : 0 : return 0U;
509 : : }
510 : :
511 : 0 : return value;
512 [ + + + - : 164 : }();
+ - - - ]
513 : :
514 : 122 : return domain;
515 : : }
516 : :
517 [ + - ]: 2 : std::string DiscoveryViewer::get_listen_address() { return kBroadcastAddress; }
518 : :
519 : 78 : uint16_t DiscoveryViewer::get_listen_port() { return static_cast<uint16_t>(kBroadcastBasePort + get_listen_domain()); }
520 : :
521 [ + - ]: 36 : DiscoveryViewer::DiscoveryViewer(FilterType type) : impl_(std::make_unique<Impl>()) {
522 [ + - + - ]: 36 : set_name("DiscoveryViewer");
523 : :
524 [ + - + - ]: 72 : const std::string native_discovery = Utils::get_env("VLINK_DISCOVER_NATIVE");
525 : :
526 [ - + ]: 36 : if (native_discovery == "1") {
527 : 0 : impl_->enable_native_discovery = true;
528 : : }
529 : :
530 [ + - ]: 36 : impl_->ser_map.reserve(128);
531 [ + - ]: 36 : impl_->schema_type_map.reserve(128);
532 : :
533 : 36 : impl_->filter_type = type;
534 : 36 : impl_->native_hostname = Utils::get_host_name();
535 : :
536 : : #ifdef _WIN32
537 : : ::WSADATA wsa_data;
538 : :
539 : : if VUNLIKELY (::WSAStartup(MAKEWORD(2, 2), &wsa_data) != 0) {
540 : : VLOG_F("DiscoveryViewer: Failed to initialize winsock.");
541 : : return;
542 : : }
543 : :
544 : : impl_->winsock_initialized = true;
545 : : #endif
546 : :
547 : : // timer
548 [ + - ]: 36 : impl_->timer.set_interval(kCollectInterval);
549 [ + - ]: 36 : impl_->timer.set_loop_count(Timer::kInfinite);
550 [ + - ]: 36 : impl_->timer.attach(this);
551 : :
552 [ + - + - ]: 214 : impl_->timer.start([this]() { process_timeout(); });
553 : :
554 : : // socket init
555 : 36 : impl_->sock = ::socket(AF_INET, SOCK_DGRAM, 0);
556 : :
557 [ - + ]: 36 : if VUNLIKELY (impl_->sock == kInvalidSocket) {
558 : : VLOG_F("DiscoveryViewer: Failed to create socket."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
559 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
560 : : }
561 : :
562 : : #ifdef _WIN32
563 : : DWORD timeout = 100U;
564 : :
565 : : if VUNLIKELY (::setsockopt(impl_->sock, SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast<const char*>(&timeout),
566 : : sizeof(timeout)) < 0) {
567 : : VLOG_F("DiscoveryViewer: Failed to set receive timeout.");
568 : : return;
569 : : }
570 : : #else
571 : : timeval timeout;
572 : 36 : std::memset(&timeout, 0, sizeof(timeout));
573 : 36 : timeout.tv_sec = 0U;
574 : 36 : timeout.tv_usec = 1000 * 100U;
575 : :
576 [ - + ]: 36 : if VUNLIKELY (::setsockopt(impl_->sock, SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast<const char*>(&timeout),
577 : : sizeof(timeout)) < 0) {
578 : : VLOG_F("DiscoveryViewer: Failed to set receive timeout."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
579 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
580 : : }
581 : : #endif
582 : :
583 : 36 : int enable_reuse_addr = 1;
584 : :
585 [ - + ]: 36 : if VUNLIKELY (::setsockopt(impl_->sock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<const char*>(&enable_reuse_addr),
586 : : sizeof(enable_reuse_addr)) < 0) {
587 : : VLOG_F("DiscoveryViewer: Failed to set reuse address option."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
588 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
589 : : }
590 : :
591 : : #ifdef SO_REUSEPORT
592 : 36 : int enable_reuse_port = 1;
593 : :
594 [ - + ]: 36 : if VUNLIKELY (::setsockopt(impl_->sock, SOL_SOCKET, SO_REUSEPORT, reinterpret_cast<const char*>(&enable_reuse_port),
595 : : sizeof(enable_reuse_port)) < 0) {
596 : : VLOG_F("DiscoveryViewer: Failed to set reuse port option."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
597 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
598 : : }
599 : : #endif
600 : :
601 [ + - ]: 36 : warn_listen_domain();
602 : :
603 : 36 : std::memset(&impl_->address, 0, sizeof(impl_->address));
604 : 36 : impl_->address.sin_family = AF_INET;
605 [ + - ]: 36 : impl_->address.sin_port = htons(get_listen_port());
606 : 36 : impl_->address.sin_addr.s_addr = htonl(INADDR_ANY);
607 : :
608 [ - + ]: 36 : if VUNLIKELY (::bind(impl_->sock, reinterpret_cast<sockaddr*>(&impl_->address), sizeof(impl_->address)) < 0) {
609 : : VLOG_F("DiscoveryViewer: Failed to bind socket."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
610 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
611 : : }
612 : :
613 : : #if VLINK_DISCOVERY_MULTICAST
614 : 36 : std::vector<std::string> ip_list;
615 : :
616 [ - + ]: 36 : if (impl_->enable_native_discovery) {
617 [ # # ]: 0 : ip_list.emplace_back("127.0.0.1");
618 : : } else {
619 [ + - + - ]: 36 : ip_list = Helpers::split_any(Utils::get_env("VLINK_DISCOVER_IP"));
620 : : }
621 : :
622 [ + - ]: 36 : if (ip_list.empty()) {
623 : : ip_mreq mreq;
624 : 36 : std::memset(&mreq, 0, sizeof(mreq));
625 : 36 : mreq.imr_multiaddr.s_addr = inet_addr(kBroadcastAddress);
626 : 36 : mreq.imr_interface.s_addr = htonl(INADDR_ANY);
627 : :
628 [ - + ]: 36 : if VUNLIKELY (::setsockopt(impl_->sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, reinterpret_cast<const char*>(&mreq),
629 : : sizeof(mreq)) < 0) {
630 [ # # # # : 0 : VLOG_F("DiscoveryViewer: Failed to join multicast group [", kBroadcastAddress,
# # ]
631 : : "], please add the route entry to the target device.");
632 : 0 : return;
633 : : }
634 : : } else {
635 : 0 : size_t joined_count = 0;
636 : :
637 [ # # ]: 0 : for (const auto& ip : ip_list) {
638 : : ip_mreq mreq;
639 : 0 : std::memset(&mreq, 0, sizeof(mreq));
640 : :
641 : 0 : mreq.imr_multiaddr.s_addr = inet_addr(kBroadcastAddress);
642 : 0 : mreq.imr_interface.s_addr = inet_addr(ip.c_str());
643 : :
644 [ # # ]: 0 : if VUNLIKELY (::setsockopt(impl_->sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, reinterpret_cast<const char*>(&mreq),
645 : : sizeof(mreq)) < 0) {
646 [ # # # # : 0 : VLOG_W("DiscoveryViewer: Failed to join multicast group on [", ip, "].");
# # ]
647 : 0 : continue;
648 : 0 : }
649 : :
650 : 0 : ++joined_count;
651 : : }
652 : :
653 [ # # ]: 0 : if VUNLIKELY (joined_count == 0) {
654 [ # # # # : 0 : VLOG_F("DiscoveryViewer: Failed to join multicast group [", kBroadcastAddress, "] on any interface.");
# # ]
655 : 0 : return;
656 : : }
657 : : }
658 : :
659 : : #else
660 : : int enable_broadcast = 1;
661 : :
662 : : if VUNLIKELY (::setsockopt(impl_->sock, SOL_SOCKET, SO_BROADCAST, reinterpret_cast<const char*>(&enable_broadcast),
663 : : sizeof(enable_broadcast)) < 0) {
664 : : VLOG_F("DiscoveryViewer: Failed to enable broadcast.");
665 : : return;
666 : : }
667 : : #endif
668 : :
669 [ + - ]: 36 : impl_->buffer.resize(kBufferSize);
670 : 36 : std::memset(impl_->buffer.data(), 0, impl_->buffer.size());
671 : :
672 : 3 : static auto global_ip_set = []() {
673 : 3 : auto list = Utils::get_all_ipv4_address();
674 [ + - ]: 6 : return std::unordered_set<std::string>(list.begin(), list.end());
675 [ + + + - : 39 : }();
+ - - - ]
676 : :
677 : : // socket listen
678 [ + - ]: 72 : impl_->thread = std::thread([this]() {
679 : : sockaddr_in target_address;
680 : 36 : socklen_t target_address_len = sizeof(target_address);
681 : 36 : std::memset(&target_address, 0, target_address_len);
682 : :
683 : 36 : char target_ip[INET_ADDRSTRLEN] = {0};
684 : :
685 : 36 : std::vector<DiscoveryMessage> batch;
686 : :
687 : : for (;;) {
688 : 969 : batch.clear();
689 : :
690 : 969 : size_t recv_count = 0;
691 : :
692 : : for (;;) {
693 : 969 : target_address_len = sizeof(target_address);
694 : :
695 [ + - ]: 969 : int size = ::recvfrom(impl_->sock, reinterpret_cast<char*>(impl_->buffer.data()), impl_->buffer.size(), 0,
696 : 969 : reinterpret_cast<sockaddr*>(&target_address), &target_address_len);
697 : :
698 [ + - + + ]: 969 : if VUNLIKELY (is_ready_to_quit()) {
699 : 36 : return;
700 : : }
701 : :
702 [ + + - + : 933 : if VUNLIKELY (size <= 0 || static_cast<size_t>(size) > impl_->buffer.size()) {
+ + ]
703 : 749 : break;
704 : : }
705 : :
706 [ - + ]: 184 : if VUNLIKELY (::inet_ntop(AF_INET, &target_address.sin_addr, target_ip, INET_ADDRSTRLEN) == nullptr) {
707 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
708 : : }
709 : :
710 [ + + + - : 184 : if (impl_->filter_type == kFilterNative && global_ip_set.count(target_ip) == 0) {
+ - - + +
+ + + - +
- - - - ]
711 [ # # # # : 0 : if (++recv_count < kMaxRecvPerBatch && has_pending_data(impl_->sock)) {
# # # # ]
712 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
713 : : }
714 : :
715 : : break; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
716 : : }
717 : :
718 [ + - ]: 184 : batch.emplace_back(
719 [ + - ]: 368 : DiscoveryMessage{Bytes::shallow_copy(impl_->buffer.data(), size).to_string(), std::string(target_ip)});
720 : :
721 [ + - + - : 184 : if (batch.size() >= kMaxBatchSize || ++recv_count >= kMaxRecvPerBatch || !has_pending_data(impl_->sock)) {
+ - + - +
- ]
722 : 184 : break;
723 : : }
724 : 0 : }
725 : :
726 [ + + ]: 933 : if (batch.empty()) {
727 : 749 : continue;
728 : : }
729 : :
730 [ + - - + ]: 184 : if VUNLIKELY (!is_running()) {
731 : 0 : continue;
732 : : }
733 : :
734 [ + - + - ]: 184 : post_task([this, batch = std::move(batch)]() {
735 [ + - ]: 184 : std::lock_guard lock(impl_->mtx);
736 : :
737 [ + + ]: 368 : for (const auto& entry : batch) {
738 : 184 : const std::string& message = entry.message;
739 : 184 : const std::string& target_ip_str = entry.target_ip;
740 : :
741 : : #if VLINK_DISCOVERY_OFFLINE
742 : : if VUNLIKELY (Helpers::has_startwith(message, "offline")) {
743 : : auto offline_list_view = Helpers::split_view(message, '\n');
744 : :
745 : : if VUNLIKELY (offline_list_view.size() < 2) {
746 : : continue;
747 : : }
748 : :
749 : : if VUNLIKELY (offline_list_view.at(0) != "offline") {
750 : : continue;
751 : : }
752 : :
753 : : auto process_view = offline_list_view.at(1);
754 : :
755 : : auto process_list_view = Helpers::split_view(process_view, ':');
756 : :
757 : : if VUNLIKELY (process_list_view.size() < 3) {
758 : : continue;
759 : : }
760 : :
761 : : uint32_t process_pid = 0;
762 : :
763 : : const std::string hostname = Helpers::unescape_field(process_list_view.at(0));
764 : :
765 : : auto process_pid_view = process_list_view.at(1);
766 : :
767 : : auto [ptr, error] = std::from_chars(process_pid_view.data(),
768 : : process_pid_view.data() + process_pid_view.size(), process_pid);
769 : :
770 : : if VUNLIKELY (error != std::errc{}) {
771 : : process_pid = 0;
772 : : }
773 : :
774 : : const std::string process_name = Helpers::unescape_field(process_list_view.at(2));
775 : :
776 : : process_offline(hostname, process_pid, process_name);
777 : :
778 : : continue;
779 : : }
780 : : #endif
781 : :
782 : 184 : auto message_list_view = Helpers::split_view(message, '\n');
783 : :
784 [ - + ]: 184 : if VUNLIKELY (message_list_view.empty()) {
785 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
786 : : }
787 : :
788 [ + + ]: 400 : for (auto str : message_list_view) {
789 : 216 : auto list = Helpers::split_view(str, ' ');
790 : :
791 [ - + ]: 216 : if VUNLIKELY (list.size() < 5) {
792 : 0 : continue;
793 : : }
794 : :
795 [ + - + - ]: 216 : ImplType type = convert_type(list.at(0));
796 [ + - ]: 216 : auto url_view = list.at(1);
797 [ + - ]: 216 : auto ser_type_view = list.at(2);
798 [ + - ]: 216 : auto schema_type_view = list.at(3);
799 [ + - ]: 216 : auto process_view = list.at(4);
800 : :
801 : 216 : uint32_t schema_type_num = 0;
802 : 216 : SchemaType schema_type = SchemaType::kUnknown;
803 : 216 : uint32_t process_pid = 0;
804 : 216 : double profiler = -1;
805 : :
806 : : {
807 [ + - ]: 216 : auto [ptr, error] = std::from_chars(schema_type_view.data(),
808 : 216 : schema_type_view.data() + schema_type_view.size(), schema_type_num);
809 : :
810 [ - + ]: 216 : if VUNLIKELY (error != std::errc{}) {
811 : 0 : schema_type_num = 0;
812 : : }
813 : :
814 [ + - ]: 216 : if (error == std::errc{}) {
815 : 216 : auto parsed_schema_type = static_cast<SchemaType>(schema_type_num);
816 : :
817 [ + - ]: 216 : if (SchemaData::is_valid_type(parsed_schema_type)) {
818 : 216 : schema_type = parsed_schema_type;
819 : : }
820 : : }
821 : : }
822 : :
823 : 216 : auto process_list_view = Helpers::split_view(process_view, ':');
824 : :
825 [ - + ]: 216 : if VUNLIKELY (process_list_view.size() < 3) {
826 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
827 : : }
828 : :
829 : 216 : std::string url = Helpers::unescape_field(url_view);
830 : 216 : std::string ser_type = Helpers::unescape_field(ser_type_view);
831 [ + - ]: 216 : std::string hostname = Helpers::unescape_field(process_list_view.at(0));
832 : :
833 [ + + ]: 216 : if (impl_->filter_type == kFilterAvailable) {
834 [ - + ]: 48 : if (hostname != impl_->native_hostname) {
835 [ # # # # ]: 0 : if (Url::is_local_type(url)) {
836 : 0 : continue;
837 : : }
838 : : }
839 [ + - ]: 168 : } else if (impl_->filter_type == kFilterNative) {
840 [ - + ]: 168 : if (hostname != impl_->native_hostname) {
841 : 0 : continue;
842 : : }
843 : : }
844 : :
845 [ + - ]: 216 : auto process_pid_view = process_list_view.at(1);
846 : :
847 : : {
848 [ + - ]: 216 : auto [ptr, error] = std::from_chars(process_pid_view.data(),
849 : 216 : process_pid_view.data() + process_pid_view.size(), process_pid);
850 : :
851 [ - + ]: 216 : if VUNLIKELY (error != std::errc{}) {
852 : 0 : process_pid = 0;
853 : : }
854 : : }
855 : :
856 [ + - ]: 216 : std::string process_name = Helpers::unescape_field(process_list_view.at(2));
857 : :
858 [ - + ]: 216 : if (process_list_view.size() >= 4) {
859 [ # # ]: 0 : auto profiler_view = process_list_view.at(3);
860 : :
861 : : try {
862 [ # # # # ]: 0 : profiler = std::stod(std::string(profiler_view));
863 [ - - ]: 0 : } catch (std::exception&) {
864 : 0 : profiler = -1;
865 : 0 : }
866 : : }
867 : :
868 [ + - - + : 216 : if VUNLIKELY (type == kUnknownImplType || url.empty()) {
- + ]
869 : 0 : continue;
870 : : }
871 : :
872 [ + - ]: 216 : int sort_index = Url::get_sort_index(url);
873 : :
874 [ - + ]: 216 : if VUNLIKELY (sort_index < 0) {
875 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
876 : : }
877 : :
878 [ + + ]: 216 : if (ser_type == "{}") {
879 : 82 : ser_type.clear();
880 : : }
881 : :
882 [ + - ]: 216 : const bool bridge = std::find(list.begin() + 5, list.end(), "bridge=1") != list.end();
883 : :
884 : : Info info{
885 : 216 : sort_index, type,
886 : : url, ser_type,
887 [ + - + - : 648 : schema_type, {Process{type, hostname, process_pid, process_name, {target_ip_str}, profiler, bridge}}};
+ - + - +
- + - + -
+ + + + -
- - - ]
888 : :
889 [ + - ]: 216 : auto [iter, inserted] = impl_->info_map.try_emplace(std::move(info), ElapsedTimer{});
890 : 216 : iter->second.restart();
891 : :
892 [ + + ]: 216 : if VUNLIKELY (inserted) {
893 : 173 : impl_->list_dirty = true;
894 : 173 : impl_->callback_dirty = true;
895 : : } else {
896 : 43 : auto& existing_info = const_cast<Info&>(iter->first);
897 : :
898 [ + - - + : 43 : if (!existing_info.process_list.empty() && existing_info.process_list[0].profiler != profiler) {
- + ]
899 : 0 : existing_info.process_list[0].profiler = profiler;
900 : :
901 : 0 : impl_->list_dirty = true;
902 : 0 : impl_->callback_dirty = true;
903 : : }
904 : : }
905 [ + - + - : 216 : }
+ - + - +
- + - ]
906 [ + - ]: 184 : }
907 : 184 : });
908 : 933 : }
909 : 72 : });
910 [ + - + - ]: 36 : }
911 : :
912 : 64 : DiscoveryViewer::~DiscoveryViewer() {
913 : 36 : quit(true);
914 : :
915 [ + - ]: 36 : if VLIKELY (impl_->sock != kInvalidSocket) {
916 : : #ifdef _WIN32
917 : : ::shutdown(impl_->sock, SD_BOTH);
918 : : #else
919 : 36 : ::shutdown(impl_->sock, SHUT_RDWR);
920 : : #endif
921 : : }
922 : :
923 : 36 : wait_for_quit();
924 : :
925 [ + - ]: 36 : if VLIKELY (impl_->thread.joinable()) {
926 : 36 : impl_->thread.join();
927 : : }
928 : :
929 [ + - ]: 36 : if VLIKELY (impl_->sock != kInvalidSocket) {
930 : : #ifdef _WIN32
931 : : ::closesocket(impl_->sock);
932 : : #else
933 : 36 : ::close(impl_->sock);
934 : : #endif
935 : 36 : impl_->sock = kInvalidSocket;
936 : : }
937 : :
938 : : #ifdef _WIN32
939 : : if VLIKELY (impl_->winsock_initialized) {
940 : : ::WSACleanup();
941 : : impl_->winsock_initialized = false;
942 : : }
943 : : #endif
944 : :
945 : 36 : impl_->buffer.clear();
946 : 64 : }
947 : :
948 : 29 : void DiscoveryViewer::register_callback(Callback&& callback) {
949 [ + - ]: 29 : std::lock_guard lock(impl_->mtx);
950 : 29 : impl_->callback = std::move(callback);
951 : 29 : impl_->callback_dirty = true;
952 : 29 : }
953 : :
954 : 1 : std::vector<DiscoveryViewer::Info> DiscoveryViewer::get_info_list() {
955 : 1 : std::vector<std::string> warnings;
956 : 1 : std::vector<Info> info_list;
957 : :
958 : : {
959 [ + - ]: 1 : std::lock_guard lock(impl_->mtx);
960 [ + - ]: 1 : refresh_list(warnings);
961 [ + - ]: 1 : info_list = impl_->info_list;
962 : 1 : }
963 : :
964 [ - + ]: 1 : for (const auto& warning : warnings) {
965 [ # # # # : 0 : VLOG_W(warning);
# # ]
966 : : }
967 : :
968 : 2 : return info_list;
969 : 1 : }
970 : :
971 : 1 : std::string DiscoveryViewer::get_ser_type(const std::string& url) const {
972 : 1 : std::vector<std::string> warnings;
973 : 1 : std::string ser_type;
974 : :
975 : : {
976 [ + - ]: 1 : std::lock_guard lock(impl_->mtx);
977 [ + - ]: 1 : refresh_list(warnings);
978 : 1 : }
979 : :
980 : : {
981 [ + - ]: 1 : std::shared_lock lock(impl_->ser_mtx);
982 [ + - ]: 1 : auto iter = impl_->ser_map.find(url);
983 : :
984 [ - + ]: 1 : if VLIKELY (iter != impl_->ser_map.end()) {
985 [ # # ]: 0 : ser_type = iter->second;
986 : : }
987 : 1 : }
988 : :
989 [ - + ]: 1 : for (const auto& warning : warnings) {
990 [ # # # # : 0 : VLOG_W(warning);
# # ]
991 : : }
992 : :
993 : 2 : return ser_type;
994 : 1 : }
995 : :
996 : 1 : SchemaType DiscoveryViewer::get_schema_type(const std::string& url) const {
997 : 1 : std::vector<std::string> warnings;
998 : 1 : SchemaType schema_type = SchemaType::kUnknown;
999 : :
1000 : : {
1001 [ + - ]: 1 : std::lock_guard lock(impl_->mtx);
1002 [ + - ]: 1 : refresh_list(warnings);
1003 : 1 : }
1004 : :
1005 : : {
1006 [ + - ]: 1 : std::shared_lock lock(impl_->ser_mtx);
1007 [ + - ]: 1 : auto iter = impl_->schema_type_map.find(url);
1008 : :
1009 [ - + ]: 1 : if VLIKELY (iter != impl_->schema_type_map.end()) {
1010 : 0 : schema_type = iter->second;
1011 : : }
1012 : 1 : }
1013 : :
1014 [ - + ]: 1 : for (const auto& warning : warnings) {
1015 [ # # # # : 0 : VLOG_W(warning);
# # ]
1016 : : }
1017 : :
1018 : 1 : return schema_type;
1019 : 1 : }
1020 : :
1021 : 362 : size_t DiscoveryViewer::get_max_task_count() const { return kMaxTaskSize; }
1022 : :
1023 : 1086 : uint32_t DiscoveryViewer::get_max_elapsed_time() const { return kMaxElapsedTime; }
1024 : :
1025 : 25 : void DiscoveryViewer::on_begin() { MessageLoop::on_begin(); }
1026 : :
1027 : 25 : void DiscoveryViewer::on_end() { MessageLoop::on_end(); }
1028 : :
1029 : 0 : void DiscoveryViewer::on_task_timeout(MessageLoop::Callback&& callback, uint32_t elapsed_time) {
1030 [ # # # # : 0 : VLOG_W_EVERY_MS(kTimeoutWarnInterval, "DiscoveryViewer: Task was delayed for ", elapsed_time,
# # ]
1031 : : "ms, the message loop is starved.");
1032 : :
1033 [ # # ]: 0 : if VLIKELY (callback) {
1034 : 0 : callback();
1035 : : }
1036 : 0 : }
1037 : :
1038 : 76 : void DiscoveryViewer::warn_listen_domain() {
1039 : 42 : static const bool kWarned = []() {
1040 : 42 : (void)get_listen_domain();
1041 : :
1042 : 42 : const std::string& invalid_domain = get_invalid_listen_domain();
1043 : :
1044 [ - + ]: 42 : if VUNLIKELY (!invalid_domain.empty()) {
1045 [ # # # # : 0 : VLOG_W("DiscoveryViewer: Invalid VLINK_DISCOVER_DOMAIN [", invalid_domain, "], fallback to 0.");
# # ]
1046 : : }
1047 : :
1048 : 42 : return true;
1049 [ + + + - : 76 : }();
+ - - - ]
1050 : :
1051 : : (void)kWarned;
1052 : 76 : }
1053 : :
1054 : 178 : void DiscoveryViewer::process_timeout() {
1055 : 178 : const int64_t schedule_elapsed = impl_->schedule_elapsed.restart();
1056 : 178 : const int64_t expired_time = std::max<int64_t>(kReportTimeout, schedule_elapsed * 2);
1057 : :
1058 : : {
1059 [ + - ]: 178 : std::lock_guard lock(impl_->mtx);
1060 : :
1061 [ + + ]: 903 : for (auto iter = impl_->info_map.begin(); iter != impl_->info_map.end();) {
1062 [ + + - + : 725 : if (iter->second.get() > expired_time || !iter->second.is_active()) {
+ + ]
1063 [ + - ]: 129 : iter = impl_->info_map.erase(iter);
1064 : 129 : impl_->list_dirty = true;
1065 : 129 : impl_->callback_dirty = true;
1066 : : } else {
1067 : 596 : ++iter;
1068 : : }
1069 : : }
1070 : 178 : }
1071 : :
1072 : 178 : report_list();
1073 : 178 : }
1074 : :
1075 : : // LCOV_EXCL_START GCOVR_EXCL_START
1076 : : void DiscoveryViewer::process_offline(std::string_view hostname, uint32_t pid, std::string_view process_name) {
1077 : : std::vector<std::pair<DiscoveryViewer::Info, DiscoveryViewer::Info>> updated_list;
1078 : :
1079 : : {
1080 : : std::lock_guard lock(impl_->mtx);
1081 : :
1082 : : for (const auto& [info, timer] : impl_->info_map) {
1083 : : auto it = std::find_if(info.process_list.begin(), info.process_list.end(),
1084 : : [pid, process_name, hostname](const auto& process) {
1085 : : return process.pid == pid && process.name == process_name && process.host == hostname;
1086 : : });
1087 : :
1088 : : if (it != info.process_list.end()) {
1089 : : DiscoveryViewer::Info new_info = info;
1090 : : auto offset = std::distance(info.process_list.begin(), it);
1091 : : auto it_copy = new_info.process_list.begin() + offset;
1092 : :
1093 : : new_info.process_list.erase(it_copy);
1094 : : updated_list.emplace_back(info, std::move(new_info));
1095 : : }
1096 : : }
1097 : :
1098 : : if (updated_list.empty()) {
1099 : : return;
1100 : : }
1101 : :
1102 : : for (const auto& [old_info, new_info] : updated_list) {
1103 : : impl_->info_map.erase(old_info);
1104 : :
1105 : : if (!new_info.process_list.empty()) {
1106 : : auto [it, inserted] = impl_->info_map.try_emplace(new_info, ElapsedTimer{});
1107 : : it->second.restart();
1108 : : }
1109 : : }
1110 : :
1111 : : impl_->list_dirty = true;
1112 : : impl_->callback_dirty = true;
1113 : : }
1114 : : }
1115 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1116 : :
1117 : 18 : void DiscoveryViewer::sort_url(std::vector<std::string>& warnings) const {
1118 : 18 : impl_->info_list.clear();
1119 : :
1120 : 18 : std::unordered_map<std::string, std::string> next_ser_map;
1121 : 18 : std::unordered_map<std::string, SchemaType> next_schema_type_map;
1122 : 18 : std::unordered_set<std::string> ser_conflict_urls;
1123 : 18 : std::unordered_set<std::string> schema_conflict_urls;
1124 : :
1125 [ + - ]: 18 : next_ser_map.reserve(impl_->info_map.size());
1126 [ + - ]: 18 : next_schema_type_map.reserve(impl_->info_map.size());
1127 [ + - ]: 18 : ser_conflict_urls.reserve(impl_->info_map.size());
1128 [ + - ]: 18 : schema_conflict_urls.reserve(impl_->info_map.size());
1129 : :
1130 [ + + ]: 69 : for (const auto& [info, timer] : impl_->info_map) {
1131 [ + + ]: 51 : if VLIKELY (!impl_->info_list.empty()) {
1132 : 33 : auto& merged = impl_->info_list.back();
1133 : :
1134 [ + + ]: 33 : if (merged.url == info.url) {
1135 : 4 : merged.type |= info.type;
1136 [ + - ]: 4 : merged.process_list.insert(merged.process_list.end(), info.process_list.begin(), info.process_list.end());
1137 : :
1138 [ - + - - : 4 : if (!info.ser_type.empty() && ser_conflict_urls.count(info.url) == 0) {
- - - + ]
1139 [ # # # # : 0 : if (merged.ser_type.empty() || (merged.ser_type == "Bytes" && info.ser_type != "Bytes")) {
# # # # ]
1140 [ # # ]: 0 : merged.ser_type = info.ser_type;
1141 [ # # # # : 0 : } else if (merged.ser_type != "Bytes" && info.ser_type != "Bytes" && info.ser_type != merged.ser_type) {
# # # # ]
1142 [ # # ]: 0 : auto [warn_iter, warn_inserted] = impl_->ser_warned_urls.try_emplace(info.url, ElapsedTimer{});
1143 : :
1144 [ # # # # : 0 : if (warn_inserted || warn_iter->second.get() > kConflictWarnInterval) {
# # ]
1145 [ # # # # ]: 0 : const std::string process_name = info.process_list.empty() ? std::string() : info.process_list[0].name;
1146 [ # # ]: 0 : const uint32_t process_pid = info.process_list.empty() ? 0U : info.process_list[0].pid;
1147 : :
1148 : 0 : warn_iter->second.restart();
1149 : :
1150 [ # # # # ]: 0 : warnings.emplace_back("DiscoveryViewer: Different ser: url = " + info.url +
1151 [ # # # # : 0 : ", current_ser = " + merged.ser_type + ", new_ser = " + info.ser_type +
# # # # #
# ]
1152 [ # # # # ]: 0 : ", process_name = " + process_name +
1153 [ # # # # : 0 : ", process_pid = " + std::to_string(process_pid) + ".");
# # ]
1154 : 0 : }
1155 : :
1156 : 0 : merged.ser_type.clear();
1157 [ # # ]: 0 : ser_conflict_urls.emplace(info.url);
1158 : : }
1159 : : }
1160 : :
1161 [ + - + - : 4 : if (info.schema_type != SchemaType::kUnknown && schema_conflict_urls.count(info.url) == 0) {
+ - + - ]
1162 [ - + ]: 4 : if (merged.schema_type == SchemaType::kUnknown) {
1163 : 0 : merged.schema_type = info.schema_type;
1164 [ - + ]: 4 : } else if (merged.schema_type != info.schema_type) {
1165 : 0 : merged.schema_type = SchemaType::kUnknown;
1166 [ # # ]: 0 : schema_conflict_urls.emplace(info.url);
1167 : : }
1168 : : }
1169 : :
1170 : 4 : continue;
1171 : 4 : }
1172 : : }
1173 : :
1174 [ + - ]: 47 : impl_->info_list.emplace_back(info);
1175 : : }
1176 : :
1177 [ + + ]: 65 : for (auto& info : impl_->info_list) {
1178 [ + + ]: 47 : if (info.process_list.size() > 1) {
1179 [ + - ]: 4 : std::sort(info.process_list.begin(), info.process_list.end());
1180 : :
1181 : 4 : size_t process_count = 0;
1182 : :
1183 [ + + ]: 12 : for (auto& process : info.process_list) {
1184 [ + + ]: 8 : if (process_count > 0) {
1185 : 4 : auto& previous = info.process_list[process_count - 1];
1186 : :
1187 [ # # # # : 0 : if (previous.type == process.type && previous.host == process.host && previous.pid == process.pid &&
# # ]
1188 [ - + - - : 4 : previous.name == process.name && previous.bridge == process.bridge) {
- + ]
1189 [ # # # # : 0 : previous.ip_list.insert(previous.ip_list.end(), std::make_move_iterator(process.ip_list.begin()),
# # ]
1190 : : std::make_move_iterator(process.ip_list.end()));
1191 : 0 : continue;
1192 : : }
1193 : : }
1194 : :
1195 [ - + ]: 8 : if (&info.process_list[process_count] != &process) {
1196 : 0 : info.process_list[process_count] = std::move(process);
1197 : : }
1198 : :
1199 : 8 : ++process_count;
1200 : : }
1201 : :
1202 [ + - ]: 4 : info.process_list.resize(process_count);
1203 : :
1204 [ + + ]: 12 : for (auto& process : info.process_list) {
1205 [ + - ]: 8 : std::sort(process.ip_list.begin(), process.ip_list.end(), Utils::ip_less);
1206 [ + - + - ]: 8 : process.ip_list.erase(std::unique(process.ip_list.begin(), process.ip_list.end()), process.ip_list.end());
1207 : : }
1208 : : }
1209 : :
1210 [ + - ]: 47 : auto& ser_type = next_ser_map[info.url];
1211 [ + - ]: 47 : auto& schema_type = next_schema_type_map[info.url];
1212 : :
1213 [ + - - + ]: 47 : if (ser_conflict_urls.count(info.url) != 0) {
1214 : 0 : ser_type.clear();
1215 [ - + - - : 47 : } else if ((ser_type.empty() || ser_type == "Bytes") && !info.ser_type.empty()) {
+ + + + ]
1216 [ + - ]: 19 : ser_type = info.ser_type;
1217 : : }
1218 : :
1219 [ + - - + ]: 47 : if (schema_conflict_urls.count(info.url) != 0) {
1220 : 0 : schema_type = SchemaType::kUnknown;
1221 [ + - ]: 47 : } else if (info.schema_type != SchemaType::kUnknown) {
1222 [ + - ]: 47 : if (schema_type == SchemaType::kUnknown) {
1223 : 47 : schema_type = info.schema_type;
1224 : : } else if (schema_type != info.schema_type) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1225 : : schema_type = SchemaType::kUnknown; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1226 : : schema_conflict_urls.emplace(info.url); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1227 : : }
1228 : : }
1229 : : }
1230 : :
1231 [ - + ]: 18 : for (auto iter = impl_->ser_warned_urls.begin(); iter != impl_->ser_warned_urls.end();) {
1232 [ # # # # : 0 : if (ser_conflict_urls.count(iter->first) == 0 && iter->second.get() > kConflictWarnInterval) {
# # # # ]
1233 [ # # ]: 0 : iter = impl_->ser_warned_urls.erase(iter);
1234 : : } else {
1235 : 0 : ++iter;
1236 : : }
1237 : : }
1238 : :
1239 : : {
1240 [ + - ]: 18 : std::unique_lock lock(impl_->ser_mtx);
1241 : 18 : impl_->ser_map.swap(next_ser_map);
1242 : 18 : impl_->schema_type_map.swap(next_schema_type_map);
1243 : 18 : }
1244 : 18 : }
1245 : :
1246 : 21 : void DiscoveryViewer::refresh_list(std::vector<std::string>& warnings) const {
1247 [ + + ]: 21 : if (impl_->list_dirty) {
1248 : 18 : sort_url(warnings);
1249 : 18 : impl_->list_dirty = false;
1250 : : }
1251 : 21 : }
1252 : :
1253 : 178 : void DiscoveryViewer::report_list() {
1254 : 178 : Callback callback;
1255 : 178 : std::vector<Info> info_list;
1256 : 178 : std::vector<std::string> warnings;
1257 : :
1258 : : {
1259 [ + - ]: 178 : std::lock_guard lock(impl_->mtx);
1260 : :
1261 [ + + ]: 178 : if (!impl_->callback_dirty) {
1262 : 56 : return;
1263 : : }
1264 : :
1265 : 122 : impl_->callback_dirty = false;
1266 : :
1267 [ + + ]: 122 : if (!impl_->callback) {
1268 : 104 : return;
1269 : : }
1270 : :
1271 [ + - ]: 18 : refresh_list(warnings);
1272 : :
1273 [ + - ]: 18 : callback = impl_->callback;
1274 [ + - ]: 18 : info_list = impl_->info_list;
1275 [ + + ]: 178 : }
1276 : :
1277 [ - + ]: 18 : for (const auto& warning : warnings) {
1278 [ # # # # : 0 : VLOG_W(warning);
# # ]
1279 : : }
1280 : :
1281 [ + - ]: 18 : if (callback) {
1282 [ + - ]: 18 : callback(info_list);
1283 : : }
1284 [ + + + + : 498 : }
+ + ]
1285 : :
1286 : : } // namespace vlink
|
