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