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 "./shm_factory.h"
25 : :
26 : : #include <charconv>
27 : : #include <filesystem>
28 : : #include <memory>
29 : : #include <new>
30 : : #include <string>
31 : : #include <utility>
32 : : #include <vector>
33 : :
34 : : #ifdef __QNX__
35 : : #include <filesystem>
36 : : #define SHM_QNX_LOCK_DIR "/var/lock"
37 : : #endif
38 : :
39 : : #include "./base/utils.h"
40 : : #include "./impl/server_impl.h"
41 : :
42 : : #define SHM_USE_RUNTIME_IMPL 1
43 : :
44 : : #include <iceoryx_posh/internal/roudi/roudi.hpp>
45 : : #include <iceoryx_posh/internal/runtime/ipc_runtime_interface.hpp>
46 : : #include <iceoryx_posh/internal/runtime/posh_runtime_impl.hpp>
47 : :
48 : : #ifdef _WIN32
49 : : #include <Windows.h>
50 : : #undef min
51 : : #undef max
52 : : #undef GetMessage
53 : : #endif
54 : :
55 : : namespace vlink {
56 : :
57 : : // ShmGlobal
58 : : struct ShmGlobal final {
59 : : std::atomic_bool has_roudi_inited{false};
60 : : std::atomic_bool has_runtime_inited{false};
61 : : std::atomic_bool has_factory_inited{false};
62 : : class ShmRuntime* runtime_instance{nullptr};
63 : : std::unique_ptr<shm::runtime::PoshRuntime> shm_runtime;
64 : : shm::RuntimeName_t shm_runtime_name;
65 : : #ifdef VLINK_SUPPORT_SHM_ROUDI
66 : : shm::cxx::optional<shm::roudi::IceOryxRouDiComponents> roudi_components;
67 : : shm::cxx::optional<shm::roudi::RouDi> roudi;
68 : : #endif
69 : :
70 : 829 : static ShmGlobal& get() {
71 : : #ifdef _WIN32
72 : : static ShmGlobal* instance = new ShmGlobal();
73 : : return *instance;
74 : : #else
75 [ + + + - ]: 829 : static ShmGlobal instance;
76 : 829 : return instance;
77 : : #endif
78 : : }
79 : :
80 : : private:
81 : 13 : ShmGlobal() = default;
82 : : };
83 : :
84 : : #if SHM_USE_RUNTIME_IMPL
85 : : // ShmRuntime
86 : : class ShmRuntime final : public shm::runtime::PoshRuntimeImpl {
87 : : public:
88 : 12 : explicit ShmRuntime(const shm::RuntimeName_t& name, bool same_process_from_roudi = false)
89 : : #if ICEORYX_VERSION_MAJOR == 2 && ICEORYX_VERSION_MINOR == 0
90 : 24 : : shm::runtime::PoshRuntimeImpl(shm::cxx::make_optional<const shm::RuntimeName_t*>(&name),
91 : : same_process_from_roudi
92 : : ? shm::runtime::RuntimeLocation::SAME_PROCESS_LIKE_ROUDI
93 [ + + ]: 24 : : shm::runtime::RuntimeLocation::SEPARATE_PROCESS_FROM_ROUDI)
94 : : #else
95 : : : shm::runtime::PoshRuntimeImpl(shm::cxx::make_optional<const shm::RuntimeName_t*>(&name), shm::DEFAULT_DOMAIN_ID,
96 : : same_process_from_roudi
97 : : ? shm::runtime::RuntimeLocation::SAME_PROCESS_LIKE_ROUDI
98 : : : shm::runtime::RuntimeLocation::SEPARATE_PROCESS_FROM_ROUDI)
99 : : #endif
100 : : {
101 : 12 : ShmGlobal::get().runtime_instance = this;
102 : :
103 : 12 : setRuntimeFactory(get_runtime_factory);
104 : 12 : }
105 : :
106 : 24 : ~ShmRuntime() override { ShmGlobal::get().runtime_instance = nullptr; }
107 : :
108 : 253 : static PoshRuntime& get_runtime_factory(shm::cxx::optional<const shm::RuntimeName_t*>) {
109 : 253 : return *ShmGlobal::get().runtime_instance;
110 : : }
111 : : };
112 : : #endif
113 : :
114 : : // ShmFactory
115 [ + - + - : 12 : ShmFactory::ShmFactory() {
+ - ]
116 [ + - + - ]: 12 : static auto& global_instance = ShmGlobal::get();
117 : :
118 : 12 : global_instance.has_factory_inited.store(true, std::memory_order_release);
119 : :
120 : 12 : Bytes::init_memory_pool();
121 : :
122 [ - + ]: 12 : if VUNLIKELY (ShmConf::get_thread_count() != 1) {
123 [ # # # # : 0 : VLOG_W("ShmFactory: Shm does not support setting thread count.");
# # ]
124 : : }
125 : :
126 [ + - + - ]: 12 : message_loop_.set_name("SHM-FACTORY");
127 : :
128 [ + - ]: 12 : detect_timer_.attach(&message_loop_);
129 [ + - ]: 12 : detect_timer_.set_interval(10);
130 [ + - ]: 12 : detect_timer_.set_loop_count(Timer::kInfinite);
131 [ + - + - ]: 12 : detect_timer_.set_callback([this]() {
132 [ - + ]: 408 : if VUNLIKELY (!global_instance.has_runtime_inited.load(std::memory_order_acquire)) {
133 : 0 : return;
134 : : }
135 : :
136 [ + - ]: 408 : detect_timer_.set_interval(50);
137 : :
138 [ + + ]: 408 : if (port_sub_) {
139 : 77 : port_sub_->take().and_then(
140 : 12 : [this](const shm::popo::Sample<const shm::roudi::PortIntrospectionFieldTopic>& sample) {
141 [ + - ]: 6 : std::lock_guard lock(topic_mtx_);
142 : 6 : topic_list_ = *sample;
143 : 6 : });
144 : : }
145 : :
146 : 408 : std::vector<DetectCallback> callbacks;
147 : :
148 : : {
149 [ + - ]: 408 : std::shared_lock lock(detect_mtx_);
150 [ + - ]: 408 : callbacks.reserve(detect_map_.size());
151 : :
152 [ + + ]: 840 : for (const auto& [publisher, callback] : detect_map_) {
153 : : (void)publisher;
154 [ + - ]: 432 : callbacks.emplace_back(callback);
155 : : }
156 : 408 : }
157 : :
158 [ + + ]: 840 : for (const auto& callback : callbacks) {
159 [ + - ]: 432 : callback();
160 : : }
161 : 408 : });
162 : :
163 [ + - ]: 12 : message_loop_.async_run();
164 : :
165 [ + - ]: 12 : init_log_level(true);
166 : :
167 [ + - + - ]: 12 : init_runtime();
168 : :
169 [ + - ]: 12 : init_log_level(false);
170 : :
171 : : {
172 [ + - + - ]: 24 : std::string depth_env_str = Utils::get_env("VLINK_SHM_DEPTH");
173 : :
174 [ - + ]: 12 : if (!depth_env_str.empty()) {
175 [ # # ]: 0 : auto [p, error] = std::from_chars(depth_env_str.data(), depth_env_str.data() + depth_env_str.size(), sub_depth_);
176 : :
177 [ # # # # : 0 : if VUNLIKELY (error != std::errc() || sub_depth_ <= 0) {
# # ]
178 : 0 : sub_depth_ = kDefaultSubDepth;
179 : : }
180 : : }
181 : 12 : }
182 : 12 : }
183 : :
184 : 12 : ShmFactory::~ShmFactory() {
185 : : #ifdef _WIN32
186 : : if (Utils::is_terminating()) {
187 : : (void)::new (std::nothrow) auto(std::move(listener_map_));
188 : : return;
189 : : }
190 : : #endif
191 : :
192 : 12 : detect_timer_.stop();
193 : 12 : detect_timer_.detach();
194 : :
195 : 12 : message_loop_.quit();
196 : 12 : message_loop_.wait_for_quit();
197 : :
198 : 12 : deinit_runtime();
199 : 12 : deinit_roudi();
200 : 12 : }
201 : :
202 : 3 : bool ShmFactory::has_roudi_inited() { return ShmGlobal::get().has_roudi_inited.load(std::memory_order_acquire); }
203 : :
204 : 496 : bool ShmFactory::has_runtime_inited() { return ShmGlobal::get().has_runtime_inited.load(std::memory_order_acquire); }
205 : :
206 : 13 : bool ShmFactory::has_roudi_running() {
207 : : #if ICEORYX_VERSION_MAJOR == 2 && ICEORYX_VERSION_MINOR == 0
208 : : shm::runtime::IpcInterfaceUser roudi_ipc(
209 : 13 : shm::RuntimeName_t(shm::cxx::TruncateToCapacity, shm::roudi::IPC_CHANNEL_ROUDI_NAME));
210 : :
211 : 13 : return roudi_ipc.isInitialized();
212 : : #else
213 : : shm::runtime::IpcInterfaceUser roudi_ipc(
214 : : shm::RuntimeName_t(shm::TruncateToCapacity, shm::roudi::IPC_CHANNEL_ROUDI_NAME), shm::DEFAULT_DOMAIN_ID,
215 : : shm::ResourceType::ICEORYX_DEFINED);
216 : :
217 : : return roudi_ipc.isInitialized();
218 : : #endif
219 : 13 : }
220 : :
221 : 84 : bool ShmFactory::auto_init_roudi(bool same_process_from_roudi, int memory_strategy) {
222 : 84 : Bytes::init_memory_pool();
223 : :
224 : : struct RoudiManager final {
225 : 11 : RoudiManager(bool same_process_from_roudi, int memory_strategy) {
226 : 11 : bool roudi_running = ShmFactory::has_roudi_running();
227 : :
228 [ - + ]: 11 : if (!roudi_running) {
229 : : #ifdef _WIN32
230 : : (void)memory_strategy;
231 : :
232 : : if (same_process_from_roudi) {
233 : : status_ = false;
234 : : return;
235 : : }
236 : : #else
237 [ # # ]: 0 : ShmFactory::init_roudi({}, memory_strategy);
238 : : #endif
239 : : }
240 : :
241 : 11 : ShmFactory::init_log_level(false);
242 [ - + - - : 11 : ShmFactory::init_runtime({}, !roudi_running && same_process_from_roudi);
+ - ]
243 : 11 : }
244 : :
245 : 11 : ~RoudiManager() {
246 : : #ifdef _WIN32
247 : : if (Utils::is_terminating()) {
248 : : return;
249 : : }
250 : : #endif
251 : :
252 [ + - ]: 11 : if (status_) {
253 : 11 : ShmFactory::deinit_runtime();
254 : 11 : ShmFactory::deinit_roudi();
255 : : }
256 : 11 : }
257 : :
258 : 84 : bool status() const { return status_; }
259 : :
260 : : private:
261 : : bool status_{true};
262 : : };
263 : :
264 [ + + + - : 84 : static RoudiManager manager(same_process_from_roudi, memory_strategy);
+ - - - ]
265 : :
266 : 84 : return manager.status();
267 : : }
268 : :
269 : 174 : shm::capro::ServiceDescription ShmFactory::get_description(const std::string& service, const std::string& instance,
270 : : const std::string& event) {
271 : : #if ICEORYX_VERSION_MAJOR == 2 && ICEORYX_VERSION_MINOR == 0
272 : 174 : return {shm::capro::IdString_t(shm::cxx::TruncateToCapacity, service.c_str()),
273 : 174 : shm::capro::IdString_t(shm::cxx::TruncateToCapacity, instance.c_str()),
274 : 174 : shm::capro::IdString_t(shm::cxx::TruncateToCapacity, event.c_str())};
275 : : #else
276 : : return {shm::capro::IdString_t(shm::TruncateToCapacity, service.c_str()),
277 : : shm::capro::IdString_t(shm::TruncateToCapacity, instance.c_str()),
278 : : shm::capro::IdString_t(shm::TruncateToCapacity, event.c_str())};
279 : : #endif
280 : : }
281 : :
282 : 36 : void ShmFactory::init_log_level(bool wait_roudi) {
283 : : #if ICEORYX_VERSION_MAJOR == 2 && ICEORYX_VERSION_MINOR == 0
284 : 36 : auto& log_manager = shm::log::LogManager::GetLogManager();
285 : :
286 [ + + ]: 36 : if (wait_roudi) {
287 : 12 : log_manager.SetDefaultLogLevel(shm::log::LogLevel::kWarn, shm::log::LogLevelOutput::kHideLogLevel);
288 : 12 : return;
289 : : }
290 : :
291 [ + + + - : 24 : static std::string shm_debug_str = Utils::get_env("VLINK_SHM_DEBUG");
+ - + - -
- ]
292 : :
293 [ - + ]: 24 : if (shm_debug_str == "1") {
294 : 0 : log_manager.SetDefaultLogLevel(shm::log::LogLevel::kInfo, shm::log::LogLevelOutput::kHideLogLevel);
295 : : } else {
296 : 24 : log_manager.SetDefaultLogLevel(shm::log::LogLevel::kFatal, shm::log::LogLevelOutput::kHideLogLevel);
297 : : }
298 : : #else
299 : :
300 : : if (wait_roudi) {
301 : : shm::log::Logger::setLogLevel(shm::log::LogLevel::Warn);
302 : : return;
303 : : }
304 : :
305 : : static std::string shm_debug_str = Utils::get_env("VLINK_SHM_DEBUG");
306 : :
307 : : if (shm_debug_str == "1") {
308 : : shm::log::Logger::setLogLevel(shm::log::LogLevel::Info);
309 : : } else {
310 : : shm::log::Logger::setLogLevel(shm::log::LogLevel::Fatal);
311 : : }
312 : : #endif
313 : : }
314 : :
315 : 1 : void ShmFactory::init_roudi(const std::string& config_path, int memory_strategy, bool monitoring_enable) {
316 : 1 : Bytes::init_memory_pool();
317 : :
318 : : #ifdef VLINK_SUPPORT_SHM_ROUDI
319 : :
320 : : #if __QNX__
321 : : try {
322 : : if (!std::filesystem::exists(SHM_QNX_LOCK_DIR)) {
323 : : std::filesystem::create_directories(SHM_QNX_LOCK_DIR);
324 : : }
325 : : } catch (std::filesystem::filesystem_error&) {
326 : : }
327 : : #endif
328 : :
329 : 1 : bool expected = false;
330 : :
331 [ - + ]: 1 : if VUNLIKELY (!ShmGlobal::get().has_roudi_inited.compare_exchange_strong(expected, true, std::memory_order_acq_rel,
332 : : std::memory_order_relaxed)) {
333 : 0 : return;
334 : : }
335 : :
336 [ + - ]: 1 : init_log_level(false);
337 : :
338 [ + - + - : 2 : VLOG_I("ShmFactory: Start shm roudi.");
+ - ]
339 : :
340 : 1 : shm::RouDiConfig_t config;
341 : :
342 [ - + ]: 1 : if (!config_path.empty()) {
343 : 0 : shm::config::CmdLineArgs_t cmd;
344 : :
345 : : #if ICEORYX_VERSION_MAJOR == 2 && ICEORYX_VERSION_MINOR == 0
346 : 0 : cmd.configFilePath = shm::roudi::ConfigFilePathString_t(shm::cxx::TruncateToCapacity, config_path.c_str());
347 : : #else
348 : : cmd.configFilePath = shm::roudi::ConfigFilePathString_t(shm::TruncateToCapacity, config_path.c_str());
349 : : #endif
350 : :
351 : 0 : shm::config::TomlRouDiConfigFileProvider provider(cmd);
352 : 0 : config = provider.parse().value();
353 : 0 : } else {
354 : 1 : shm::mepoo::MePooConfig poo_config;
355 : :
356 [ - + ]: 1 : if (memory_strategy == 1) {
357 : 0 : poo_config.addMemPool({1024, 1000});
358 : 0 : poo_config.addMemPool({16384, 500});
359 : 0 : poo_config.addMemPool({131072, 100});
360 : 0 : poo_config.addMemPool({1048576, 10});
361 [ - + ]: 1 : } else if (memory_strategy == 2) {
362 : 0 : poo_config.addMemPool({1024, 5000});
363 : 0 : poo_config.addMemPool({16384, 1000});
364 : 0 : poo_config.addMemPool({131072, 100});
365 : 0 : poo_config.addMemPool({1048576, 20});
366 : 0 : poo_config.addMemPool({4194304, 10});
367 : 0 : poo_config.addMemPool({8388608, 5});
368 [ - + ]: 1 : } else if (memory_strategy == 4) {
369 : 0 : poo_config.addMemPool({1024, 10000});
370 : 0 : poo_config.addMemPool({16384, 1000});
371 : 0 : poo_config.addMemPool({131072, 500});
372 : 0 : poo_config.addMemPool({1048576, 200});
373 : 0 : poo_config.addMemPool({3145728, 100});
374 : 0 : poo_config.addMemPool({6291456, 50});
375 : 0 : poo_config.addMemPool({13631488, 30});
376 : 0 : poo_config.addMemPool({25165824, 20});
377 : : } else { // default
378 : 1 : poo_config.addMemPool({1024, 10000});
379 : 1 : poo_config.addMemPool({16384, 1000});
380 : 1 : poo_config.addMemPool({131072, 500});
381 : 1 : poo_config.addMemPool({1048576, 100});
382 : 1 : poo_config.addMemPool({3145728, 50});
383 : 1 : poo_config.addMemPool({6291456, 30});
384 : 1 : poo_config.addMemPool({13631488, 20});
385 : : }
386 : :
387 : : #if ICEORYX_VERSION_MAJOR == 2 && ICEORYX_VERSION_MINOR == 0
388 : 1 : auto group_name = shm::posix::PosixGroup::getGroupOfCurrentProcess().getName();
389 : : #else
390 : : auto group_name = shm::PosixGroup::getGroupOfCurrentProcess().getName();
391 : : #endif
392 : :
393 : 1 : config.m_sharedMemorySegments.emplace_back(group_name, group_name, std::move(poo_config));
394 : 1 : }
395 : :
396 : : #if ICEORYX_VERSION_MAJOR == 2 && ICEORYX_VERSION_MINOR == 0
397 : : shm::roudi::RouDi::RoudiStartupParameters parameters{
398 : : monitoring_enable ? shm::roudi::MonitoringMode::ON : shm::roudi::MonitoringMode::OFF, true,
399 : : shm::roudi::RouDi::RuntimeMessagesThreadStart::IMMEDIATE, shm::version::CompatibilityCheckLevel::PATCH,
400 [ - + ]: 1 : shm::units::Duration::fromSeconds(10)};
401 : : #else
402 : : shm::config::RouDiConfig parameters;
403 : : parameters.sharesAddressSpaceWithApplications = true;
404 : : parameters.monitoringMode = monitoring_enable ? shm::roudi::MonitoringMode::ON : shm::roudi::MonitoringMode::OFF;
405 : : #endif
406 : :
407 : 1 : ShmGlobal::get().roudi_components.emplace(config);
408 : 2 : ShmGlobal::get().roudi.emplace(ShmGlobal::get().roudi_components.value().rouDiMemoryManager,
409 : 1 : ShmGlobal::get().roudi_components.value().portManager, parameters);
410 : : #else
411 : : (void)config_path;
412 : : (void)memory_strategy;
413 : : (void)monitoring_enable;
414 : :
415 : : VLOG_F("ShmFactory: Shm roudi is not supported.");
416 : : #endif
417 : 1 : }
418 : :
419 : 24 : void ShmFactory::init_runtime(std::string name, bool same_process_from_roudi) {
420 : 24 : Bytes::init_memory_pool();
421 : :
422 [ + + + - ]: 24 : static auto& global_instance = ShmGlobal::get();
423 : :
424 : 24 : bool expected = false;
425 : :
426 [ + + ]: 24 : if VLIKELY (global_instance.has_runtime_inited.compare_exchange_strong(expected, true, std::memory_order_acq_rel,
427 : : std::memory_order_relaxed)) {
428 [ + + ]: 12 : if (name.empty()) {
429 [ + - + - ]: 11 : name = Utils::get_app_name() + "_" + Utils::get_pid_str();
430 : : }
431 : :
432 : : #if ICEORYX_VERSION_MAJOR == 2 && ICEORYX_VERSION_MINOR == 0
433 : 12 : global_instance.shm_runtime_name = shm::RuntimeName_t(shm::cxx::TruncateToCapacity, name.c_str());
434 : : #else
435 : : global_instance.shm_runtime_name = shm::RuntimeName_t(shm::TruncateToCapacity, name.c_str());
436 : : #endif
437 : :
438 : : #if SHM_USE_RUNTIME_IMPL
439 : 12 : global_instance.shm_runtime =
440 [ + - ]: 12 : std::make_unique<ShmRuntime>(global_instance.shm_runtime_name, same_process_from_roudi);
441 : : #else
442 : : (void)same_process_from_roudi;
443 : : shm::runtime::PoshRuntime::initRuntime(global_instance.shm_runtime_name);
444 : : #endif
445 : : }
446 : 24 : }
447 : :
448 : 23 : void ShmFactory::deinit_runtime() {
449 [ + + + - ]: 23 : static auto& global_instance = ShmGlobal::get();
450 : :
451 : 23 : bool expected = true;
452 : :
453 [ + + ]: 23 : if VLIKELY (global_instance.has_runtime_inited.compare_exchange_strong(expected, false, std::memory_order_acq_rel,
454 : : std::memory_order_relaxed)) {
455 [ + - ]: 12 : if (global_instance.has_factory_inited.load(std::memory_order_acquire)) {
456 [ + - ]: 12 : auto& factory = ShmFactory::get();
457 : :
458 : : {
459 [ + - ]: 12 : std::lock_guard lock(factory.listener_mtx_);
460 : 12 : factory.listener_map_.clear();
461 : 12 : }
462 : :
463 : 12 : factory.port_sub_.reset();
464 : :
465 : 12 : Utils::yield_cpu();
466 : : }
467 : :
468 : : #if SHM_USE_RUNTIME_IMPL
469 : :
470 [ + - ]: 12 : if (global_instance.shm_runtime) {
471 : 12 : global_instance.shm_runtime.reset();
472 : : }
473 : : #endif
474 : : }
475 : 23 : }
476 : :
477 : 23 : void ShmFactory::deinit_roudi() {
478 : : #ifdef VLINK_SUPPORT_SHM_ROUDI
479 [ + + + - ]: 23 : static auto& global_instance = ShmGlobal::get();
480 : :
481 : 23 : bool expected = true;
482 : :
483 [ + + ]: 23 : if VLIKELY (global_instance.has_roudi_inited.compare_exchange_strong(expected, false, std::memory_order_acq_rel,
484 : : std::memory_order_relaxed)) {
485 : 1 : global_instance.roudi.reset();
486 : 1 : global_instance.roudi_components.reset();
487 : : }
488 : : #endif
489 : 23 : }
490 : :
491 : 111 : std::shared_ptr<shm::popo::Listener> ShmFactory::get_listener(int32_t domain) {
492 [ + - ]: 111 : std::lock_guard lock(listener_mtx_);
493 : :
494 [ + - ]: 111 : auto& entry = listener_map_[domain];
495 : 111 : auto listener = entry.lock();
496 : :
497 [ + + ]: 111 : if (!listener) {
498 [ + - ]: 78 : listener = std::make_shared<shm::popo::Listener>();
499 : 78 : entry = listener;
500 : : }
501 : :
502 : 222 : return listener;
503 : 111 : }
504 : :
505 : 62 : void ShmFactory::add_detect_callback(void* node, DetectCallback&& callback) {
506 [ + - ]: 62 : std::lock_guard lock(detect_mtx_);
507 : :
508 [ + + ]: 62 : if (detect_map_.empty()) {
509 [ + - ]: 58 : detect_timer_.restart();
510 : : }
511 : :
512 [ + - + - ]: 62 : detect_map_[node] = callback;
513 : :
514 [ + - + - ]: 124 : message_loop_.post_task([callback = std::move(callback)]() { callback(); });
515 : 62 : }
516 : :
517 : 62 : void ShmFactory::remove_detect_callback(void* node) {
518 [ + - ]: 62 : std::lock_guard lock(detect_mtx_);
519 : :
520 [ + - ]: 62 : detect_map_.erase(node);
521 : :
522 [ + + ]: 62 : if (detect_map_.empty()) {
523 [ + - ]: 58 : detect_timer_.stop();
524 : : }
525 : 62 : }
526 : :
527 : 1 : void ShmFactory::start_detect_node_count() {
528 [ + - ]: 1 : if (!port_sub_) {
529 : 1 : shm::popo::SubscriberOptions port_sub_opt;
530 : 1 : port_sub_opt.queueCapacity = 1U;
531 : 1 : port_sub_opt.historyRequest = 1U;
532 : 1 : port_sub_.emplace(shm::roudi::IntrospectionPortService, port_sub_opt);
533 : : }
534 : :
535 : 1 : detect_timer_.restart();
536 : 1 : }
537 : :
538 : 0 : uint64_t ShmFactory::get_publisher_count(const shm::capro::ServiceDescription& description) {
539 [ # # ]: 0 : std::shared_lock lock(topic_mtx_);
540 : :
541 : 0 : uint64_t count = 0;
542 : :
543 [ # # ]: 0 : for (const auto& info : topic_list_.m_publisherList) {
544 [ # # ]: 0 : if (info.m_caproServiceID == description.getServiceIDString() &&
545 [ # # # # : 0 : info.m_caproInstanceID == description.getInstanceIDString() &&
# # ]
546 : 0 : info.m_caproEventMethodID == description.getEventIDString()) {
547 : 0 : ++count;
548 : : }
549 : : }
550 : :
551 : 0 : return count;
552 : 0 : }
553 : :
554 : 1 : uint64_t ShmFactory::get_subscriber_count(const shm::capro::ServiceDescription& description) {
555 [ + - ]: 1 : std::shared_lock lock(topic_mtx_);
556 : :
557 : 1 : uint64_t count = 0;
558 : :
559 [ + + ]: 2 : for (const auto& info : topic_list_.m_subscriberList) {
560 [ - + ]: 2 : if (info.m_caproServiceID == description.getServiceIDString() &&
561 [ + - - - : 2 : info.m_caproInstanceID == description.getInstanceIDString() &&
- + ]
562 : 0 : info.m_caproEventMethodID == description.getEventIDString()) {
563 : 0 : ++count;
564 : : }
565 : : }
566 : :
567 : 1 : return count;
568 : 1 : }
569 : :
570 : 59 : int ShmFactory::get_sub_depth() const { return sub_depth_; }
571 : :
572 : : // ShmServer
573 : 24 : ShmServer::ShmServer(const ShmID& id) {
574 [ + + + - : 24 : static auto& factory = ShmFactory::get();
+ - - - ]
575 : :
576 : 24 : const auto& [impl_type, address, domain, depth, history, wait, method, owner] = id;
577 : :
578 : 24 : domain_ = domain;
579 : :
580 : 24 : shm::popo::ServerOptions options;
581 : 24 : options.offerOnCreate = false;
582 : :
583 [ + + ]: 24 : if (depth > 0) {
584 : 2 : options.requestQueueCapacity = depth;
585 : : } else {
586 : 22 : options.requestQueueCapacity = kDefaultReqDepth;
587 : : }
588 : :
589 [ + - + - ]: 24 : server_.emplace(ShmFactory::get_description("vlink.method." + std::to_string(domain), address, method), options);
590 : :
591 [ + - ]: 24 : listener_ = factory.get_listener(domain_);
592 : :
593 [ + - - + ]: 24 : if VUNLIKELY (listener_
594 : : ->attachEvent(server_.value(), shm::popo::ServerEvent::REQUEST_RECEIVED,
595 : : shm::popo::createNotificationCallback(ShmServer::on_request_received, *this))
596 : : .has_error()) {
597 [ # # # # : 0 : VLOG_F("ShmFactory: Failed to attach REQUEST_RECEIVED event to listener.");
# # ]
598 : : }
599 : 24 : }
600 : :
601 : 48 : ShmServer::~ShmServer() {
602 : 24 : quit_flag_.store(true, std::memory_order_release);
603 : :
604 : 24 : server_->stopOffer();
605 : :
606 : 24 : listener_->detachEvent(server_.value(), shm::popo::ServerEvent::REQUEST_RECEIVED);
607 : :
608 : : {
609 : 24 : std::lock_guard lock(callback_mtx_);
610 : 24 : }
611 : :
612 : 24 : server_->releaseQueuedRequests();
613 : 48 : }
614 : :
615 [ + - ]: 1 : std::any ShmServer::get_native_handle() const { return this; }
616 : :
617 : 1 : bool ShmServer::suspend() {
618 : 1 : is_suspend_.store(true, std::memory_order_relaxed);
619 : :
620 : 1 : return true;
621 : : }
622 : :
623 : 1 : bool ShmServer::resume() {
624 : 1 : is_suspend_.store(false, std::memory_order_relaxed);
625 : :
626 : 1 : return true;
627 : : }
628 : :
629 : 2 : bool ShmServer::is_suspend() const { return is_suspend_.load(std::memory_order_relaxed); }
630 : :
631 : 24 : void ShmServer::process_message() {
632 [ + + + - : 48 : while (server_->hasRequests() && !quit_flag_.load(std::memory_order_acquire)) {
+ + ]
633 : 48 : server_->take()
634 : 24 : .and_then([this](const void* buffer) {
635 : 24 : const auto* read_req = static_cast<const uint8_t*>(buffer);
636 : 24 : const auto* read_header = shm::popo::RequestHeader::fromPayload(read_req);
637 : :
638 : 24 : uint64_t channel = 0;
639 : 24 : uint64_t seq = 0;
640 : 24 : Bytes req_bytes;
641 : 24 : ShmFactory::read_data(read_req, read_header->getChunkHeader()->userPayloadSize(), channel, seq, req_bytes);
642 : :
643 : : (void)seq;
644 : :
645 : 24 : bool request_released = false;
646 : :
647 [ + - + - ]: 24 : traverse_req_resp_callback([this, channel, &req_bytes, &read_req, &read_header, &request_released](
648 : 210 : NodeImpl* impl, const auto& callback) {
649 [ + - ]: 24 : const auto* conf_ptr = impl->get_target_conf<ShmConf>();
650 : :
651 [ - + ]: 24 : if (static_cast<uint64_t>(conf_ptr->hash_code) != channel) {
652 : 0 : ignore_called();
653 : 0 : return;
654 : : }
655 : :
656 [ - + ]: 24 : if VUNLIKELY (has_called()) {
657 [ # # # # : 0 : VLOG_F(*conf_ptr, "Two identical service requests.");
# # ]
658 : 0 : return;
659 : : }
660 : :
661 [ + - ]: 24 : std::lock_guard lock(mtx_);
662 : :
663 [ + + ]: 24 : if (static_cast<ServerImpl*>(impl)->is_resp_type) {
664 : 21 : last_req_header_ = read_header;
665 : :
666 : 21 : Bytes resp_bytes;
667 : :
668 : 21 : auto seq = static_cast<uint64_t>(read_header->getSequenceId());
669 : :
670 [ + - ]: 21 : callback(seq, req_bytes, &resp_bytes);
671 : 21 : } else {
672 [ + - ]: 3 : callback(0, req_bytes, nullptr);
673 : : }
674 : :
675 : 24 : server_->releaseRequest(read_req);
676 : 24 : request_released = true;
677 : 24 : });
678 : :
679 [ - + ]: 24 : if VUNLIKELY (!request_released) {
680 : 0 : server_->releaseRequest(read_req);
681 : : }
682 : 24 : })
683 [ # # # # : 24 : .or_else([](auto& e) { VLOG_E("ShmFactory: Failed to take request, error: ", e, "."); });
# # ]
684 : : }
685 : 24 : }
686 : :
687 : 22 : void ShmServer::start() { server_->offer(); }
688 : :
689 : 0 : void ShmServer::stop() { server_->stopOffer(); }
690 : :
691 : 0 : bool ShmServer::has_clients() const { return server_->hasClients(); }
692 : :
693 : 1 : Bytes ShmServer::loan(uint64_t channel, int64_t size) {
694 [ - + ]: 1 : if VUNLIKELY (size <= 0) {
695 : 0 : return Bytes();
696 : : }
697 : :
698 : : auto write_req_result =
699 : 1 : server_->loan(last_req_header_, size + ShmFactory::get_loaned_offset(), ShmFactory::get_loaned_alignment());
700 : :
701 : 1 : seq_.fetch_add(1, std::memory_order_relaxed);
702 : :
703 [ - + ]: 1 : if VUNLIKELY (write_req_result.has_error()) {
704 [ # # # # : 0 : VLOG_E("ShmFactory: Failed to loan buffer, size: ", size + ShmFactory::get_loaned_offset(),
# # ]
705 : : ", error: ", write_req_result.get_error(), ".");
706 : 0 : return Bytes();
707 : : }
708 : :
709 : 1 : auto* write_req = static_cast<uint8_t*>(write_req_result.value());
710 : :
711 : 2 : ShmFactory::write_header(write_req, channel, seq_.load(std::memory_order_relaxed));
712 : :
713 : 1 : return Bytes::loan_internal(write_req + ShmFactory::get_loaned_offset(), size);
714 : 1 : }
715 : :
716 : 1 : bool ShmServer::release(const Bytes& bytes) {
717 [ + - ]: 1 : if VUNLIKELY (!bytes.is_loaned()) {
718 : 1 : return false;
719 : : }
720 : :
721 : 0 : server_->releaseResponse(const_cast<uint8_t*>(bytes.data()) - ShmFactory::get_loaned_offset());
722 : :
723 : 0 : return true;
724 : : }
725 : :
726 : 21 : bool ShmServer::reply(uint64_t channel, const Bytes& resp_data) {
727 [ - + ]: 21 : if (resp_data.is_loaned()) {
728 : 0 : auto send_result = server_->send(const_cast<uint8_t*>(resp_data.data() - ShmFactory::get_loaned_offset()));
729 : :
730 [ # # ]: 0 : if VUNLIKELY (send_result.has_error()) {
731 [ # # # # : 0 : VLOG_E("ShmFactory: Failed to send, error: ", send_result.get_error(), ".");
# # ]
732 : 0 : return false;
733 : : }
734 [ # # ]: 0 : } else {
735 : 21 : auto write_resp_result = server_->loan(last_req_header_, resp_data.size() + ShmFactory::get_loaned_offset(),
736 : 21 : ShmFactory::get_loaned_alignment());
737 : :
738 : 21 : seq_.fetch_add(1, std::memory_order_relaxed);
739 : :
740 [ - + ]: 21 : if VUNLIKELY (write_resp_result.has_error()) {
741 [ # # # # : 0 : VLOG_E("ShmFactory: Failed to loan buffer, size: ", resp_data.size() + ShmFactory::get_loaned_offset(),
# # ]
742 : : ", error: ", write_resp_result.get_error(), ".");
743 : 0 : return false;
744 : : }
745 : :
746 : 21 : auto* write_resp = static_cast<uint8_t*>(write_resp_result.value());
747 : :
748 : 42 : ShmFactory::write_data(write_resp, channel, seq_.load(std::memory_order_relaxed), resp_data);
749 : :
750 : 21 : auto send_result = server_->send(write_resp);
751 : :
752 [ - + ]: 21 : if VUNLIKELY (send_result.has_error()) {
753 [ # # # # : 0 : VLOG_E("ShmFactory: Failed to send, error: ", send_result.get_error(), ".");
# # ]
754 : 0 : return false;
755 : : }
756 [ + - + - ]: 21 : }
757 : :
758 : 21 : return true;
759 : : }
760 : :
761 : 25 : void ShmServer::on_request_received(shm::popo::UntypedServer*, ShmServer* target) {
762 [ + - ]: 25 : std::lock_guard callback_lock(target->callback_mtx_);
763 : :
764 [ - + ]: 25 : if VUNLIKELY (target->quit_flag_.load(std::memory_order_acquire)) {
765 : 0 : return;
766 : : }
767 : :
768 [ + + ]: 25 : if VUNLIKELY (target->is_suspend_.load(std::memory_order_relaxed)) {
769 : 1 : target->server_->releaseQueuedRequests();
770 : :
771 : 1 : return;
772 : : }
773 : :
774 [ + - ]: 24 : auto* impl = target->get_first_impl();
775 : :
776 [ - + ]: 24 : if VUNLIKELY (!impl) {
777 : 0 : return;
778 : : }
779 : :
780 : 24 : MessageLoop* message_loop = nullptr;
781 : :
782 [ + - + - ]: 48 : target->invoke_callback(impl, [&] { message_loop = impl->get_message_loop(); });
783 : :
784 [ + + ]: 24 : if (message_loop) {
785 : 9 : std::weak_ptr<ShmServer> weak_target = target->weak_from_this();
786 : :
787 [ + - + - ]: 9 : message_loop->post_task([weak_target, message_loop]() {
788 : 9 : auto target = weak_target.lock();
789 : :
790 [ - + ]: 9 : if VUNLIKELY (!target) {
791 : 0 : return;
792 : : }
793 : :
794 [ + - ]: 9 : auto* impl = target->get_first_impl();
795 : :
796 : 9 : bool attached = false;
797 [ + - + - ]: 18 : target->invoke_callback(impl, [&] { attached = impl->get_message_loop() == message_loop; });
798 : :
799 [ - + ]: 9 : if VUNLIKELY (!attached) {
800 : 0 : return;
801 : : }
802 : :
803 : 9 : target->process_message();
804 [ + - ]: 9 : });
805 : 9 : } else {
806 : 15 : target->process_message();
807 : : }
808 [ + + ]: 25 : }
809 : :
810 : : // ShmClient
811 : 22 : ShmClient::ShmClient(const ShmID& id) {
812 [ + + + - : 22 : static auto& factory = ShmFactory::get();
+ - - - ]
813 : :
814 : 22 : const auto& [impl_type, address, domain, depth, history, wait, method, owner] = id;
815 : :
816 : 22 : domain_ = domain;
817 : :
818 : 22 : shm::popo::ClientOptions options;
819 : 22 : options.connectOnCreate = true;
820 : :
821 [ + + ]: 22 : if (depth > 0) {
822 : 3 : options.responseQueueCapacity = depth;
823 : : } else {
824 : 19 : options.responseQueueCapacity = kDefaultRespDepth;
825 : : }
826 : :
827 [ + - + - ]: 22 : client_.emplace(ShmFactory::get_description("vlink.method." + std::to_string(domain), address, method), options);
828 : :
829 [ + - ]: 22 : listener_ = factory.get_listener(domain_);
830 : :
831 [ + - - + ]: 22 : if VUNLIKELY (listener_
832 : : ->attachEvent(client_.value(), shm::popo::ClientEvent::RESPONSE_RECEIVED,
833 : : shm::popo::createNotificationCallback(ShmClient::on_response_received, *this))
834 : : .has_error()) {
835 [ # # # # : 0 : VLOG_F("ShmFactory: Failed to attach RESPONSE_RECEIVED event to listener.");
# # ]
836 : : }
837 : 22 : }
838 : :
839 : 44 : ShmClient::~ShmClient() {
840 : 22 : quit_flag_.store(true, std::memory_order_release);
841 : :
842 : 22 : client_->disconnect();
843 : :
844 : 22 : disable_detect_timer();
845 : :
846 : 22 : listener_->detachEvent(client_.value(), shm::popo::ClientEvent::RESPONSE_RECEIVED);
847 : :
848 : : {
849 : 22 : std::lock_guard lock(callback_mtx_);
850 : 22 : }
851 : :
852 : 22 : client_->releaseQueuedResponses();
853 : 44 : }
854 : :
855 [ + - ]: 1 : std::any ShmClient::get_native_handle() const { return this; }
856 : :
857 : 21 : void ShmClient::process_message() {
858 [ + + + - : 42 : while (client_->hasResponses() && !quit_flag_.load(std::memory_order_acquire)) {
+ + ]
859 [ + - ]: 21 : std::unique_lock lock(mtx_);
860 : :
861 : 21 : auto received = client_->take();
862 : :
863 [ - + ]: 21 : if VUNLIKELY (received.has_error()) {
864 [ # # # # : 0 : VLOG_E("ShmFactory: Failed to take response, error: ", received.get_error(), ".");
# # ]
865 : 0 : break;
866 : : }
867 : :
868 : 21 : const auto* read_resp = static_cast<const uint8_t*>(received.value());
869 : 21 : const auto* read_header = shm::popo::ResponseHeader::fromPayload(read_resp);
870 [ + - ]: 21 : auto iter = callbacks_.find(read_header->getSequenceId());
871 : :
872 [ + + ]: 21 : if VUNLIKELY (iter == callbacks_.end()) {
873 : 1 : client_->releaseResponse(read_resp);
874 : 1 : continue;
875 : : }
876 : :
877 : 20 : auto callback = std::move(iter->second);
878 [ + - ]: 20 : callbacks_.erase(iter);
879 [ + - ]: 20 : lock.unlock();
880 : :
881 : 20 : uint64_t channel = 0;
882 : 20 : uint64_t seq = 0;
883 : 20 : Bytes resp_bytes;
884 : 20 : ShmFactory::read_data(read_resp, read_header->getChunkHeader()->userPayloadSize(), channel, seq, resp_bytes);
885 : :
886 : 20 : MessageLoop* message_loop = nullptr;
887 : :
888 [ + + ]: 20 : if (callback.dispatch) {
889 [ + - + - ]: 5 : invoke_callback(callback.owner, [&]() {
890 : 5 : message_loop = callback.owner->get_message_loop();
891 : :
892 [ + + ]: 5 : if (!message_loop) {
893 : 2 : callback.callback(resp_bytes);
894 : : }
895 : 5 : });
896 : : } else {
897 [ + - ]: 15 : callback.callback(resp_bytes);
898 : : }
899 : :
900 [ + + ]: 20 : if (message_loop) {
901 : 3 : auto release_response = [self = shared_from_this()](const uint8_t* data) {
902 [ + - ]: 3 : std::lock_guard lock(self->mtx_);
903 : :
904 : 3 : self->client_->releaseResponse(data);
905 [ + - ]: 6 : };
906 : 3 : std::unique_ptr<const uint8_t, decltype(release_response)> response(read_resp, std::move(release_response));
907 : :
908 [ + - + - ]: 6 : message_loop->post_task([this, message_loop, response = std::move(response), resp_bytes = std::move(resp_bytes),
909 : 3 : callback = std::move(callback)]() {
910 : 3 : bool attached = false;
911 [ + - + - ]: 6 : invoke_callback(callback.owner, [&]() { attached = callback.owner->get_message_loop() == message_loop; });
912 : :
913 [ + - ]: 3 : if (attached) {
914 [ + - ]: 3 : callback.callback(resp_bytes);
915 : : }
916 : 3 : });
917 : 3 : } else {
918 [ + - ]: 17 : lock.lock();
919 : :
920 : 17 : client_->releaseResponse(read_resp);
921 : : }
922 [ + - + + : 22 : }
- + ]
923 : 21 : }
924 : :
925 : 316 : bool ShmClient::is_connected() const { return client_->getConnectionState() == shm::ConnectionState::CONNECTED; }
926 : :
927 : 67 : void ShmClient::enable_detect_timer() {
928 [ + + ]: 67 : if (!has_detect_timer_.load(std::memory_order_relaxed)) {
929 : 20 : has_detect_timer_.store(true, std::memory_order_relaxed);
930 : :
931 [ + - + - ]: 20 : ShmFactory::get().add_detect_callback(this, [weak = weak_from_this()]() {
932 : 74 : auto self = weak.lock();
933 : :
934 [ + - ]: 74 : if VLIKELY (self) {
935 [ + - ]: 74 : self->detect_server();
936 : : }
937 : 74 : });
938 : : }
939 : 67 : }
940 : :
941 : 22 : void ShmClient::disable_detect_timer() {
942 [ + + ]: 22 : if (has_detect_timer_.load(std::memory_order_relaxed)) {
943 : 20 : has_detect_timer_.store(false, std::memory_order_relaxed);
944 : :
945 : 20 : ShmFactory::get().remove_detect_callback(this);
946 : : }
947 : 22 : }
948 : :
949 : 98 : Bytes ShmClient::loan(uint64_t channel, int64_t size) {
950 [ - + ]: 98 : if VUNLIKELY (size <= 0) {
951 : 0 : return Bytes();
952 : : }
953 : :
954 [ + - ]: 98 : std::lock_guard lock(mtx_);
955 : :
956 : 98 : auto write_req_result = client_->loan(size + ShmFactory::get_loaned_offset(), ShmFactory::get_loaned_alignment());
957 : :
958 [ - + ]: 98 : if VUNLIKELY (write_req_result.has_error()) {
959 [ # # # # : 0 : VLOG_E("ShmFactory: Failed to loan buffer, size: ", size + ShmFactory::get_loaned_offset(),
# # ]
960 : : ", error: ", write_req_result.get_error(), ".");
961 : 0 : return Bytes();
962 : : }
963 : :
964 : 98 : auto* write_req = static_cast<uint8_t*>(write_req_result.value());
965 : :
966 : 196 : ShmFactory::write_header(write_req, channel, seq_.load(std::memory_order_relaxed));
967 : :
968 : 98 : return Bytes::loan_internal(write_req + ShmFactory::get_loaned_offset(), size);
969 : 98 : }
970 : :
971 : 98 : bool ShmClient::release(const Bytes& bytes) {
972 [ + + ]: 98 : if VUNLIKELY (!bytes.is_loaned()) {
973 : 2 : return false;
974 : : }
975 : :
976 [ + - ]: 96 : std::lock_guard lock(mtx_);
977 : :
978 : 96 : client_->releaseRequest(const_cast<uint8_t*>(bytes.data()) - ShmFactory::get_loaned_offset());
979 : :
980 : 96 : return true;
981 : 96 : }
982 : :
983 : 57 : bool ShmClient::call(NodeImpl* owner, uint64_t channel, const Bytes& req_data, NodeImpl::MsgCallback&& callback,
984 : : uint64_t* seq_out, bool dispatch) {
985 [ + + ]: 57 : if VUNLIKELY (!is_connected()) {
986 [ + - ]: 32 : release(req_data);
987 : :
988 : 32 : return false;
989 : : }
990 : :
991 [ + - ]: 25 : std::lock_guard lock(mtx_);
992 : 25 : uint64_t response_seq = 0;
993 : 25 : bool has_response_callback = false;
994 : :
995 [ + + ]: 25 : if (req_data.is_loaned()) {
996 : : auto* write_header =
997 : 2 : shm::popo::RequestHeader::fromPayload(const_cast<uint8_t*>(req_data.data()) - ShmFactory::get_loaned_offset());
998 : :
999 [ + + ]: 2 : if (callback) {
1000 : 1 : response_seq = seq_.load(std::memory_order_relaxed);
1001 : 1 : has_response_callback = true;
1002 [ + - ]: 1 : callbacks_[response_seq] = ResponseCallback{owner, dispatch, std::move(callback)};
1003 : 1 : write_header->setSequenceId(response_seq);
1004 : :
1005 [ + - ]: 1 : if (seq_out) {
1006 : 1 : *seq_out = response_seq;
1007 : : }
1008 : :
1009 : 1 : seq_.fetch_add(1, std::memory_order_relaxed);
1010 : : }
1011 : :
1012 : 2 : auto send_result = client_->send(const_cast<uint8_t*>(req_data.data()) - ShmFactory::get_loaned_offset());
1013 : :
1014 [ - + ]: 2 : if VUNLIKELY (send_result.has_error()) {
1015 [ # # # # : 0 : VLOG_E("ShmFactory: Failed to send, error: ", send_result.get_error(), ".");
# # ]
1016 [ # # ]: 0 : if (has_response_callback) {
1017 [ # # ]: 0 : callbacks_.erase(response_seq);
1018 : : }
1019 : 0 : return false;
1020 : : }
1021 [ + - ]: 2 : } else {
1022 : : auto write_req_result =
1023 : 23 : client_->loan(req_data.size() + ShmFactory::get_loaned_offset(), ShmFactory::get_loaned_alignment());
1024 [ - + ]: 23 : if VUNLIKELY (write_req_result.has_error()) {
1025 [ # # # # : 0 : VLOG_E("ShmFactory: Failed to loan buffer, size: ", req_data.size() + ShmFactory::get_loaned_offset(),
# # ]
1026 : : ", error: ", write_req_result.get_error(), ".");
1027 : :
1028 : 0 : return false;
1029 : : }
1030 : :
1031 : 23 : auto* write_req = static_cast<uint8_t*>(write_req_result.value());
1032 : :
1033 : 23 : auto* write_header = shm::popo::RequestHeader::fromPayload(write_req);
1034 : :
1035 [ + + ]: 23 : if (callback) {
1036 : 21 : response_seq = seq_.load(std::memory_order_relaxed);
1037 : 21 : has_response_callback = true;
1038 [ + - ]: 21 : callbacks_[response_seq] = ResponseCallback{owner, dispatch, std::move(callback)};
1039 : 21 : write_header->setSequenceId(response_seq);
1040 : :
1041 [ + + ]: 21 : if (seq_out) {
1042 : 15 : *seq_out = response_seq;
1043 : : }
1044 : :
1045 : 21 : seq_.fetch_add(1, std::memory_order_relaxed);
1046 : : }
1047 : :
1048 : 46 : ShmFactory::write_data(write_req, channel, seq_.load(std::memory_order_relaxed), req_data);
1049 : :
1050 : 23 : auto send_result = client_->send(write_req);
1051 : :
1052 [ - + ]: 23 : if VUNLIKELY (send_result.has_error()) {
1053 [ # # # # : 0 : VLOG_E("ShmFactory: Failed to send, error: ", send_result.get_error(), ".");
# # ]
1054 : :
1055 [ # # ]: 0 : if (has_response_callback) {
1056 [ # # ]: 0 : callbacks_.erase(response_seq);
1057 : : }
1058 : :
1059 : 0 : return false;
1060 : : }
1061 [ + - + - ]: 23 : }
1062 : :
1063 : 25 : return true;
1064 : 25 : }
1065 : :
1066 : 1 : void ShmClient::remove_response_callback(uint64_t seq) {
1067 [ + - ]: 1 : std::lock_guard lock(mtx_);
1068 [ + - ]: 1 : callbacks_.erase(seq);
1069 : 1 : }
1070 : :
1071 : 48 : void ShmClient::cancel_calls(NodeImpl* owner) {
1072 [ + - ]: 48 : std::lock_guard lock(mtx_);
1073 : :
1074 [ + + ]: 49 : for (auto iter = callbacks_.begin(); iter != callbacks_.end();) {
1075 [ + - ]: 1 : if (iter->second.owner == owner) {
1076 [ + - ]: 1 : iter = callbacks_.erase(iter);
1077 : : } else {
1078 : 0 : ++iter;
1079 : : }
1080 : : }
1081 : 48 : }
1082 : :
1083 : 74 : void ShmClient::detect_server() {
1084 [ - + ]: 74 : if VUNLIKELY (quit_flag_.load(std::memory_order_acquire)) {
1085 : 0 : return;
1086 : : }
1087 : :
1088 [ - + ]: 74 : if VUNLIKELY (!ShmFactory::has_runtime_inited()) {
1089 : 0 : return;
1090 : : }
1091 : :
1092 : 74 : discovery_server(is_connected());
1093 : : }
1094 : :
1095 : 74 : void ShmClient::discovery_server(bool connect) {
1096 [ + + ]: 74 : if VLIKELY (last_connected_.load(std::memory_order_relaxed) == connect) {
1097 : 55 : return;
1098 : : }
1099 : :
1100 [ + - + - ]: 40 : traverse_server_connect_callback([connect](NodeImpl*, const auto& callback) { callback(connect); });
1101 : 19 : last_connected_.store(connect, std::memory_order_relaxed);
1102 : : }
1103 : :
1104 : 21 : void ShmClient::on_response_received(shm::popo::UntypedClient*, ShmClient* target) {
1105 [ + - ]: 21 : std::lock_guard callback_lock(target->callback_mtx_);
1106 : :
1107 [ - + ]: 21 : if VUNLIKELY (target->quit_flag_.load(std::memory_order_acquire)) {
1108 : 0 : return;
1109 : : }
1110 : :
1111 [ + - ]: 21 : target->process_message();
1112 [ + - ]: 21 : }
1113 : :
1114 : : // ShmPublisher
1115 : 63 : ShmPublisher::ShmPublisher(const ShmID& id) {
1116 [ + + + - : 63 : static auto& factory = ShmFactory::get();
+ - - - ]
1117 : :
1118 : 63 : const auto& [impl_type, address, domain, depth, history, wait, field, owner] = id;
1119 : :
1120 : 63 : domain_ = domain;
1121 : 63 : wait_ = wait;
1122 : :
1123 : 63 : shm::popo::PublisherOptions options;
1124 : 63 : options.offerOnCreate = true;
1125 : 63 : options.historyCapacity = history;
1126 : :
1127 [ + - ]: 63 : std::string event = "event";
1128 : :
1129 [ + + ]: 63 : if (domain != 0) {
1130 [ + - + - : 3 : event += ("_" + std::to_string(domain));
+ - ]
1131 : : }
1132 : :
1133 [ + - ]: 63 : pub_.emplace(impl_type == kSetter
1134 [ + + + - : 126 : ? ShmFactory::get_description("vlink.field." + std::to_string(domain), address, field)
+ - + - +
+ + + + +
+ + - - -
- - - -
- ]
1135 : : : ShmFactory::get_description("vlink", address, event),
1136 : : options);
1137 : :
1138 [ + + ]: 63 : if (wait > 0) {
1139 [ + - ]: 1 : factory.start_detect_node_count();
1140 : :
1141 [ + - ]: 1 : sem_.emplace();
1142 : :
1143 [ + - + - : 2 : std::string sem_address = std::to_string(domain) + "@" + address;
+ - ]
1144 : 1 : std::replace(sem_address.begin(), sem_address.end(), '/', '@');
1145 : :
1146 : : #ifdef __FreeBSD__
1147 : : sem_->attach("/vlink@shm@" + sem_address);
1148 : : #else
1149 [ + - + - ]: 1 : sem_->attach("vlink@shm@" + sem_address);
1150 : : #endif
1151 : 1 : }
1152 : 63 : }
1153 : :
1154 : 126 : ShmPublisher::~ShmPublisher() {
1155 : 63 : quit_flag_.store(true, std::memory_order_release);
1156 : :
1157 : 63 : pub_->stopOffer();
1158 : :
1159 : 63 : disable_detect_timer();
1160 : :
1161 [ + + ]: 63 : if (sem_) {
1162 : 1 : sem_->detach(true);
1163 : : }
1164 : 126 : }
1165 : :
1166 [ + - ]: 2 : std::any ShmPublisher::get_native_handle() const { return this; }
1167 : :
1168 : 4750 : bool ShmPublisher::has_subscribers() const { return pub_->hasSubscribers(); }
1169 : :
1170 : 72 : Bytes ShmPublisher::loan(uint64_t channel, int64_t size) {
1171 [ + + ]: 72 : if VUNLIKELY (size <= 0) {
1172 : 2 : return Bytes();
1173 : : }
1174 : :
1175 [ + - ]: 70 : std::lock_guard lock(mtx_);
1176 : :
1177 : 70 : auto write_msg_result = pub_->loan(size + ShmFactory::get_loaned_offset(), ShmFactory::get_loaned_alignment());
1178 : :
1179 : 70 : seq_.fetch_add(1, std::memory_order_relaxed);
1180 : :
1181 [ - + ]: 70 : if VUNLIKELY (write_msg_result.has_error()) {
1182 [ # # # # : 0 : VLOG_E("ShmFactory: Failed to loan buffer, size: ", size + ShmFactory::get_loaned_offset(),
# # ]
1183 : : ", error: ", write_msg_result.get_error(), ".");
1184 : 0 : return Bytes();
1185 : : }
1186 : :
1187 : 70 : auto* write_msg = static_cast<uint8_t*>(write_msg_result.value());
1188 : :
1189 : 140 : ShmFactory::write_header(write_msg, channel, seq_.load(std::memory_order_relaxed));
1190 : :
1191 : 70 : return Bytes::loan_internal(write_msg + ShmFactory::get_loaned_offset(), size);
1192 : 70 : }
1193 : :
1194 : 4 : bool ShmPublisher::release(const Bytes& bytes) {
1195 [ + + ]: 4 : if VUNLIKELY (!bytes.is_loaned()) {
1196 : 2 : return false;
1197 : : }
1198 : :
1199 [ + - ]: 2 : std::lock_guard lock(mtx_);
1200 : :
1201 : 2 : pub_->release(const_cast<uint8_t*>(bytes.data()) - ShmFactory::get_loaned_offset());
1202 : :
1203 : 2 : return true;
1204 : 2 : }
1205 : :
1206 : 4371 : bool ShmPublisher::publish(uint64_t channel, const Bytes& bytes) {
1207 [ + + ]: 4371 : if VUNLIKELY (wait_ > 0) {
1208 : 1 : uint64_t sem_count = sem_->get_count();
1209 : :
1210 [ - + ]: 1 : if (sem_count > 0) {
1211 : 0 : sem_->acquire(sem_count, wait_);
1212 : : }
1213 : : }
1214 : :
1215 : : {
1216 [ + - ]: 4371 : std::lock_guard lock(mtx_);
1217 : :
1218 [ + + ]: 4377 : if (bytes.is_loaned()) {
1219 : 68 : pub_->publish(const_cast<uint8_t*>(bytes.data()) - ShmFactory::get_loaned_offset());
1220 : : } else {
1221 : : auto write_msg_result =
1222 : 4308 : pub_->loan(bytes.size() + ShmFactory::get_loaned_offset(), ShmFactory::get_loaned_alignment());
1223 : :
1224 : 4312 : seq_.fetch_add(1, std::memory_order_relaxed);
1225 : :
1226 [ - + ]: 4312 : if VUNLIKELY (write_msg_result.has_error()) {
1227 [ # # # # : 0 : VLOG_E("ShmFactory: Failed to loan buffer, size: ", bytes.size() + ShmFactory::get_loaned_offset(),
# # ]
1228 : : ", error: ", write_msg_result.get_error(), ".");
1229 : 0 : return false;
1230 : : }
1231 : :
1232 : 4311 : auto* write_msg = static_cast<uint8_t*>(write_msg_result.value());
1233 : :
1234 : 8621 : ShmFactory::write_data(write_msg, channel, seq_.load(std::memory_order_relaxed), bytes);
1235 : :
1236 : 4308 : pub_->publish(write_msg);
1237 [ + # ]: 4315 : }
1238 [ + - ]: 4374 : }
1239 : :
1240 [ + + ]: 4380 : if VUNLIKELY (wait_ > 0) {
1241 [ + - ]: 1 : uint64_t sub_count = ShmFactory::get().get_subscriber_count(pub_->getServiceDescription());
1242 : 1 : sem_->acquire(sub_count, wait_);
1243 : : }
1244 : :
1245 : 4379 : return true;
1246 : : }
1247 : :
1248 : 48 : void ShmPublisher::enable_detect_timer() {
1249 [ + + ]: 48 : if (!has_detect_timer_.load(std::memory_order_relaxed)) {
1250 : 42 : has_detect_timer_.store(true, std::memory_order_relaxed);
1251 [ + - + - ]: 42 : ShmFactory::get().add_detect_callback(this, [weak = weak_from_this()]() {
1252 : 420 : auto self = weak.lock();
1253 : :
1254 [ + - ]: 420 : if VLIKELY (self) {
1255 [ + - ]: 420 : self->detect_subscribers();
1256 : : }
1257 : 420 : });
1258 : : }
1259 : 48 : }
1260 : :
1261 : 63 : void ShmPublisher::disable_detect_timer() {
1262 [ + + ]: 63 : if (has_detect_timer_.load(std::memory_order_relaxed)) {
1263 : 42 : has_detect_timer_.store(false, std::memory_order_relaxed);
1264 : 42 : ShmFactory::get().remove_detect_callback(this);
1265 : : }
1266 : 63 : }
1267 : :
1268 : 420 : void ShmPublisher::detect_subscribers() {
1269 [ - + ]: 420 : if VUNLIKELY (quit_flag_.load(std::memory_order_acquire)) {
1270 : 0 : return;
1271 : : }
1272 : :
1273 [ - + ]: 420 : if VUNLIKELY (!ShmFactory::has_runtime_inited()) {
1274 : 0 : return;
1275 : : }
1276 : :
1277 : 420 : discovery_subscribers(has_subscribers());
1278 : : }
1279 : :
1280 : 420 : void ShmPublisher::discovery_subscribers(bool has_subscribers) {
1281 [ + + ]: 420 : if VLIKELY (last_has_subscribers_.load(std::memory_order_relaxed) == has_subscribers) {
1282 : 377 : return;
1283 : : }
1284 : :
1285 [ + - + - ]: 87 : traverse_sub_connect_callback([has_subscribers](NodeImpl*, const auto& callback) { callback(has_subscribers); });
1286 : 43 : last_has_subscribers_.store(has_subscribers, std::memory_order_relaxed);
1287 : : }
1288 : :
1289 : : // ShmSubscriber
1290 : 65 : ShmSubscriber::ShmSubscriber(const ShmID& id) {
1291 [ + + + - : 65 : static auto& factory = ShmFactory::get();
+ - - - ]
1292 : :
1293 : 65 : const auto& [impl_type, address, domain, depth, history, wait, field, owner] = id;
1294 : :
1295 : 65 : domain_ = domain;
1296 : 65 : wait_ = wait;
1297 : :
1298 : 65 : shm::popo::SubscriberOptions options;
1299 : 65 : options.subscribeOnCreate = false;
1300 : 65 : options.historyRequest = history;
1301 : : // options.requiresPublisherHistorySupport = history > 0;
1302 : :
1303 [ + + ]: 65 : if (depth > 0) {
1304 : 6 : options.queueCapacity = depth;
1305 : : } else {
1306 : 59 : options.queueCapacity = factory.get_sub_depth();
1307 : : }
1308 : :
1309 [ + - ]: 65 : std::string event = "event";
1310 : :
1311 [ + + ]: 65 : if (domain != 0) {
1312 [ + - + - : 3 : event += ("_" + std::to_string(domain));
+ - ]
1313 : : }
1314 : :
1315 [ + - ]: 65 : sub_.emplace(impl_type == kGetter
1316 [ + + + - : 130 : ? ShmFactory::get_description("vlink.field." + std::to_string(domain), address, field)
+ - + - +
+ + + + +
+ + - - -
- - - -
- ]
1317 : : : ShmFactory::get_description("vlink", address, event),
1318 : : options);
1319 : :
1320 [ + - ]: 65 : listener_ = factory.get_listener(domain_);
1321 : :
1322 [ + - - + ]: 65 : if VUNLIKELY (listener_
1323 : : ->attachEvent(sub_.value(), shm::popo::SubscriberEvent::DATA_RECEIVED,
1324 : : shm::popo::createNotificationCallback(ShmSubscriber::on_msg_received, *this))
1325 : : .has_error()) {
1326 [ # # # # : 0 : VLOG_F("ShmFactory: Failed to attach DATA_RECEIVED event to listener.");
# # ]
1327 : : }
1328 : :
1329 [ + + ]: 65 : if (wait_ > 0) {
1330 [ + - ]: 1 : sem_.emplace();
1331 : :
1332 [ + - + - : 2 : std::string sem_address = std::to_string(domain) + "@" + address;
+ - ]
1333 : 1 : std::replace(sem_address.begin(), sem_address.end(), '/', '@');
1334 : :
1335 : : #ifdef __FreeBSD__
1336 : : sem_->attach("/vlink@shm@" + sem_address);
1337 : : #else
1338 [ + - + - ]: 1 : sem_->attach("vlink@shm@" + sem_address);
1339 : : #endif
1340 : 1 : }
1341 : 65 : }
1342 : :
1343 : 130 : ShmSubscriber::~ShmSubscriber() {
1344 : 65 : quit_flag_.store(true, std::memory_order_release);
1345 : :
1346 : 65 : sub_->unsubscribe();
1347 : :
1348 : 65 : listener_->detachEvent(sub_.value(), shm::popo::SubscriberEvent::DATA_RECEIVED);
1349 : :
1350 : : {
1351 : 65 : std::lock_guard lock(callback_mtx_);
1352 : 65 : }
1353 : :
1354 : 65 : sub_->releaseQueuedData();
1355 : :
1356 [ + + ]: 65 : if (sem_) {
1357 : 1 : sem_->detach(false);
1358 : : }
1359 : 130 : }
1360 : :
1361 [ + - ]: 2 : std::any ShmSubscriber::get_native_handle() const { return this; }
1362 : :
1363 : 1 : bool ShmSubscriber::suspend() {
1364 : 1 : is_suspend_.store(true, std::memory_order_relaxed);
1365 : :
1366 : 1 : return true;
1367 : : }
1368 : :
1369 : 1 : bool ShmSubscriber::resume() {
1370 : 1 : is_suspend_.store(false, std::memory_order_relaxed);
1371 : :
1372 : 1 : return true;
1373 : : }
1374 : :
1375 : 2 : bool ShmSubscriber::is_suspend() const { return is_suspend_.load(std::memory_order_relaxed); }
1376 : :
1377 : 3007 : void ShmSubscriber::process_message(MessageLoop* dispatched_loop) {
1378 [ + + + - : 7200 : while (sub_->hasData() && !quit_flag_.load(std::memory_order_acquire)) {
+ + ]
1379 [ + - ]: 4193 : std::unique_lock lock(receive_mtx_);
1380 : :
1381 : 4193 : auto received = sub_->take();
1382 : :
1383 [ - + ]: 4193 : if VUNLIKELY (received.has_error()) {
1384 [ # # # # : 0 : VLOG_E("ShmFactory: Failed to take sample, error: ", received.get_error(), ".");
# # ]
1385 : 0 : break;
1386 : : }
1387 : :
1388 : 4193 : const auto* read_msg = static_cast<const uint8_t*>(received.value());
1389 : 4193 : const auto* read_header = shm::mepoo::ChunkHeader::fromUserPayload(read_msg);
1390 : 4193 : uint64_t channel = 0;
1391 : 4193 : uint64_t seq = 0;
1392 : 4193 : Bytes msg_bytes;
1393 : 4193 : ShmFactory::read_data(read_msg, read_header->userPayloadSize(), channel, seq, msg_bytes);
1394 : :
1395 [ + + ]: 4193 : if VUNLIKELY (is_latency_and_lost_enabled_.load(std::memory_order_acquire)) {
1396 [ - + ]: 10 : calc_sample_.update(seq > 0 ? seq : read_header->sequenceNumber(),
1397 : 20 : static_cast<uint64_t>(read_header->originId()));
1398 : : }
1399 : :
1400 [ + - ]: 4193 : lock.unlock();
1401 : :
1402 : 4193 : std::vector<MessageLoop*> routes;
1403 : :
1404 [ + - + - ]: 4193 : traverse_msg_callback([channel, &msg_bytes, dispatched_loop, &routes](NodeImpl* impl, const auto& callback) {
1405 [ + - ]: 4325 : const auto* conf = impl->get_target_conf<ShmConf>();
1406 : :
1407 [ + + + + : 4325 : if VUNLIKELY (static_cast<uint64_t>(conf->hash_code) != channel ||
+ + ]
1408 : : impl->has_suspend.load(std::memory_order_acquire)) {
1409 : 6 : return;
1410 : : }
1411 : :
1412 [ + - ]: 4319 : auto* loop = impl->get_message_loop();
1413 : :
1414 [ + + + + : 4319 : if VLIKELY (!loop || loop == dispatched_loop) {
+ + ]
1415 [ + - ]: 4314 : callback(msg_bytes);
1416 [ + - + - ]: 5 : } else if (std::find(routes.begin(), routes.end(), loop) == routes.end()) {
1417 [ + - ]: 5 : routes.push_back(loop);
1418 : : }
1419 : : });
1420 : :
1421 [ + + ]: 4193 : if VLIKELY (routes.empty()) {
1422 [ + - ]: 4190 : lock.lock();
1423 : :
1424 : 4190 : sub_->release(read_msg);
1425 [ + - ]: 4190 : lock.unlock();
1426 : :
1427 [ + + ]: 4190 : if (sem_) {
1428 [ + - ]: 1 : sem_->release();
1429 : : }
1430 : :
1431 : 4190 : continue;
1432 : : }
1433 : :
1434 : 3 : std::shared_ptr<const uint8_t> retained(read_msg, [self = shared_from_this()](const uint8_t* data) {
1435 : : {
1436 [ + - ]: 3 : std::lock_guard lock(self->receive_mtx_);
1437 : :
1438 : 3 : self->sub_->release(data);
1439 : 3 : }
1440 : :
1441 [ - + ]: 3 : if (self->sem_) {
1442 : 0 : self->sem_->release();
1443 : : }
1444 [ + - + - ]: 6 : });
1445 : :
1446 [ + + ]: 8 : for (auto* loop : routes) {
1447 [ + - + - ]: 5 : loop->post_task([this, loop, channel, retained, size = msg_bytes.size()]() {
1448 : 5 : auto bytes = Bytes::shallow_copy(retained.get() + ShmFactory::get_loaned_offset(), size);
1449 : :
1450 [ + - + - ]: 5 : traverse_msg_callback([loop, channel, &bytes](NodeImpl* impl, const auto& callback) {
1451 : 10 : const auto* conf = impl->get_target_conf<ShmConf>();
1452 : :
1453 [ + + + - : 15 : if (impl->get_message_loop() == loop && !impl->has_suspend.load(std::memory_order_acquire) &&
+ + ]
1454 [ + - ]: 5 : static_cast<uint64_t>(conf->hash_code) == channel) {
1455 : 5 : callback(bytes);
1456 : : }
1457 : 10 : });
1458 : 5 : });
1459 : : }
1460 [ + + + + : 16763 : }
+ - + + -
+ ]
1461 : 3007 : }
1462 : :
1463 : 80 : void ShmSubscriber::subscribe() { sub_->subscribe(); }
1464 : :
1465 : 0 : void ShmSubscriber::unsubscribe() { sub_->unsubscribe(); }
1466 : :
1467 : 86 : void ShmSubscriber::set_latency_and_lost_enabled(bool enable) {
1468 : 86 : is_latency_and_lost_enabled_.store(enable, std::memory_order_release);
1469 : 86 : }
1470 : :
1471 : 6 : bool ShmSubscriber::is_latency_and_lost_enabled() const {
1472 : 6 : return is_latency_and_lost_enabled_.load(std::memory_order_acquire);
1473 : : }
1474 : :
1475 : 2 : const CalculateSample& ShmSubscriber::get_calculate_sample() const { return calc_sample_; }
1476 : :
1477 : 3008 : void ShmSubscriber::on_msg_received(shm::popo::UntypedSubscriber*, ShmSubscriber* target) {
1478 [ + - ]: 3008 : std::lock_guard callback_lock(target->callback_mtx_);
1479 : :
1480 [ - + ]: 3008 : if VUNLIKELY (target->quit_flag_.load(std::memory_order_acquire)) {
1481 : 0 : return;
1482 : : }
1483 : :
1484 [ + + ]: 3008 : if VUNLIKELY (target->is_suspend_.load(std::memory_order_relaxed)) {
1485 : 1 : target->sub_->releaseQueuedData();
1486 : 1 : return;
1487 : : }
1488 : :
1489 : 3007 : MessageLoop* message_loop = nullptr;
1490 : 3007 : bool first = true;
1491 : 3007 : bool common_loop = true;
1492 : :
1493 [ + - + - ]: 3007 : target->traverse_msg_callback([&](NodeImpl* impl, const auto&) {
1494 [ + + ]: 3138 : if (first) {
1495 : 3007 : message_loop = impl->get_message_loop();
1496 : 3007 : first = false;
1497 [ + + ]: 131 : } else if (message_loop != impl->get_message_loop()) {
1498 : 3 : common_loop = false;
1499 : : }
1500 : 3138 : });
1501 : :
1502 [ + + + + ]: 3007 : if (common_loop && message_loop) {
1503 : 4 : std::weak_ptr<ShmSubscriber> weak_target = target->weak_from_this();
1504 : :
1505 [ + - + - ]: 4 : message_loop->post_task([weak_target, message_loop]() {
1506 : 4 : auto target = weak_target.lock();
1507 : :
1508 [ - + ]: 4 : if VUNLIKELY (!target) {
1509 : 0 : return;
1510 : : }
1511 : :
1512 [ + - ]: 4 : target->process_message(message_loop);
1513 [ + - ]: 4 : });
1514 : 4 : } else {
1515 [ + - ]: 3003 : target->process_message();
1516 : : }
1517 [ + + ]: 3008 : }
1518 : :
1519 : : } // namespace vlink
|
