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 : 818 : static ShmGlobal& get() {
69 [ + + + - ]: 818 : static ShmGlobal instance;
70 : 818 : 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 : 214 : static PoshRuntime& get_runtime_factory(shm::cxx::optional<const shm::RuntimeName_t*>) {
102 : 214 : 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 [ - + ]: 433 : if VUNLIKELY (!global_instance.has_runtime_inited.load(std::memory_order_acquire)) {
126 : 0 : return;
127 : : }
128 : :
129 [ + - ]: 433 : detect_timer_.set_interval(50);
130 : :
131 [ + + ]: 433 : if (port_sub_) {
132 : 69 : port_sub_->take().and_then(
133 : 10 : [this](const shm::popo::Sample<const shm::roudi::PortIntrospectionFieldTopic>& sample) {
134 [ + - ]: 5 : std::lock_guard lock(topic_mtx_);
135 : 5 : topic_list_ = *sample;
136 : 5 : });
137 : : }
138 : :
139 : 433 : std::vector<DetectCallback> callbacks;
140 : :
141 : : {
142 [ + - ]: 433 : std::shared_lock lock(detect_mtx_);
143 [ + - ]: 433 : callbacks.reserve(detect_map_.size());
144 : :
145 [ + + ]: 891 : for (const auto& [publisher, callback] : detect_map_) {
146 : : (void)publisher;
147 [ + - ]: 458 : callbacks.emplace_back(callback);
148 : : }
149 : 433 : }
150 : :
151 [ + + ]: 891 : for (const auto& callback : callbacks) {
152 [ + - ]: 458 : callback();
153 : : }
154 : 433 : });
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 : 524 : 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 : 72 : bool ShmFactory::auto_init_roudi(bool same_process_from_roudi) {
208 : 72 : 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 : 72 : bool status() const { return status_; }
238 : :
239 : : private:
240 : : bool status_{true};
241 : : };
242 : :
243 [ + + + - : 72 : static RoudiManager manager(same_process_from_roudi);
+ - - - ]
244 : :
245 : 72 : return manager.status();
246 : : }
247 : :
248 : 146 : 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 : 146 : return {shm::capro::IdString_t(shm::cxx::TruncateToCapacity, service.c_str()),
252 : 146 : shm::capro::IdString_t(shm::cxx::TruncateToCapacity, instance.c_str()),
253 : 146 : 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->shutdown();
446 : 12 : global_instance.shm_runtime.reset();
447 : : }
448 : : #endif
449 : : }
450 : 23 : }
451 : :
452 : 23 : void ShmFactory::deinit_roudi() {
453 : : #ifdef VLINK_SUPPORT_SHM_ROUDI
454 [ + + + - ]: 23 : static auto& global_instance = ShmGlobal::get();
455 : :
456 : 23 : bool expected = true;
457 : :
458 [ + + ]: 23 : if VLIKELY (global_instance.has_roudi_inited.compare_exchange_strong(expected, false, std::memory_order_acq_rel,
459 : : std::memory_order_relaxed)) {
460 : 1 : global_instance.roudi.reset();
461 : 1 : global_instance.roudi_components.reset();
462 : : }
463 : : #endif
464 : 23 : }
465 : :
466 : 92 : shm::popo::Listener* ShmFactory::get_listener(int32_t domain) {
467 [ + - ]: 92 : std::lock_guard lock(listener_mtx_);
468 : :
469 [ + - ]: 92 : auto iter = listener_map_.find(domain);
470 : :
471 [ + + ]: 92 : if (iter == listener_map_.end()) {
472 [ + - ]: 67 : auto listener = std::make_shared<shm::popo::Listener>();
473 : :
474 [ + - ]: 67 : return listener_map_.emplace(domain, std::move(listener)).first->second.get();
475 : 67 : }
476 : :
477 : 25 : return iter->second.get();
478 : 92 : }
479 : :
480 : 92 : void ShmFactory::try_to_destroy_listener(int32_t domain, shm::popo::Listener* listener) {
481 [ + - ]: 92 : std::lock_guard lock(listener_mtx_);
482 : :
483 [ + - ]: 92 : if (listener) {
484 [ + + ]: 92 : if (listener->size() == 0) {
485 [ + - ]: 67 : listener_map_.erase(domain);
486 : : }
487 : : } else {
488 [ # # ]: 0 : auto iter = listener_map_.find(domain);
489 [ # # ]: 0 : if (iter != listener_map_.end()) {
490 [ # # ]: 0 : if (iter->second->size() == 0) {
491 [ # # ]: 0 : listener_map_.erase(iter);
492 : : }
493 : : }
494 : : }
495 : 92 : }
496 : :
497 : 64 : void ShmFactory::add_detect_callback(void* node, DetectCallback&& callback) {
498 [ + - ]: 64 : std::lock_guard lock(detect_mtx_);
499 : :
500 [ + + ]: 64 : if (detect_map_.empty()) {
501 [ + - ]: 60 : detect_timer_.restart();
502 : : }
503 : :
504 [ + - + - ]: 64 : detect_map_[node] = callback;
505 : :
506 [ + - + - ]: 128 : message_loop_.post_task([callback = std::move(callback)]() { callback(); });
507 : 64 : }
508 : :
509 : 64 : void ShmFactory::remove_detect_callback(void* node) {
510 [ + - ]: 64 : std::lock_guard lock(detect_mtx_);
511 : :
512 [ + - ]: 64 : detect_map_.erase(node);
513 : :
514 [ + + ]: 64 : if (detect_map_.empty()) {
515 [ + - ]: 60 : detect_timer_.stop();
516 : : }
517 : 64 : }
518 : :
519 : 1 : void ShmFactory::start_detect_node_count() {
520 [ + - ]: 1 : if (!port_sub_) {
521 : 1 : shm::popo::SubscriberOptions port_sub_opt;
522 : 1 : port_sub_opt.queueCapacity = 1U;
523 : 1 : port_sub_opt.historyRequest = 1U;
524 : 1 : port_sub_.emplace(shm::roudi::IntrospectionPortService, port_sub_opt);
525 : : }
526 : :
527 : 1 : detect_timer_.restart();
528 : 1 : }
529 : :
530 : 0 : uint64_t ShmFactory::get_publisher_count(const shm::capro::ServiceDescription& description) {
531 [ # # ]: 0 : std::shared_lock lock(topic_mtx_);
532 : :
533 : 0 : uint64_t count = 0;
534 : :
535 [ # # ]: 0 : for (const auto& info : topic_list_.m_publisherList) {
536 [ # # ]: 0 : if (info.m_caproServiceID == description.getServiceIDString() &&
537 [ # # # # : 0 : info.m_caproInstanceID == description.getInstanceIDString() &&
# # ]
538 : 0 : info.m_caproEventMethodID == description.getEventIDString()) {
539 : 0 : ++count;
540 : : }
541 : : }
542 : :
543 : 0 : return count;
544 : 0 : }
545 : :
546 : 1 : uint64_t ShmFactory::get_subscriber_count(const shm::capro::ServiceDescription& description) {
547 [ + - ]: 1 : std::shared_lock lock(topic_mtx_);
548 : :
549 : 1 : uint64_t count = 0;
550 : :
551 [ + + ]: 2 : for (const auto& info : topic_list_.m_subscriberList) {
552 [ - + ]: 2 : if (info.m_caproServiceID == description.getServiceIDString() &&
553 [ + - - - : 2 : info.m_caproInstanceID == description.getInstanceIDString() &&
- + ]
554 : 0 : info.m_caproEventMethodID == description.getEventIDString()) {
555 : 0 : ++count;
556 : : }
557 : : }
558 : :
559 : 1 : return count;
560 : 1 : }
561 : :
562 : 53 : int ShmFactory::get_sub_depth() const { return sub_depth_; }
563 : :
564 : : // ShmServer
565 : 20 : ShmServer::ShmServer(const ShmID& id) {
566 [ + + + - : 20 : static auto& factory = ShmFactory::get();
+ - - - ]
567 : :
568 : 20 : const auto& [impl_type, address, domain, depth, history, wait] = id;
569 : :
570 : 20 : domain_ = domain;
571 : :
572 : 20 : shm::popo::ServerOptions options;
573 : 20 : options.offerOnCreate = false;
574 : :
575 [ + + ]: 20 : if (depth > 0) {
576 : 1 : options.requestQueueCapacity = depth;
577 : : } else {
578 : 19 : options.requestQueueCapacity = kDefaultReqDepth;
579 : : }
580 : :
581 [ + - ]: 20 : std::string event = "method";
582 : :
583 [ + + ]: 20 : if (domain != 0) {
584 [ + - + - : 1 : event += ("_" + std::to_string(domain));
+ - ]
585 : : }
586 : :
587 [ + - ]: 20 : server_.emplace(ShmFactory::get_description("vlink", address, event), options);
588 : :
589 [ + - ]: 20 : listener_ = factory.get_listener(domain_);
590 : :
591 : 20 : listener_
592 [ + - ]: 20 : ->attachEvent(server_.value(), shm::popo::ServerEvent::REQUEST_RECEIVED,
593 : 20 : shm::popo::createNotificationCallback(ShmServer::on_request_received, *this))
594 [ # # # # ]: 20 : .or_else([](auto&) { VLOG_F("ShmFactory: Failed to attach REQUEST_RECEIVED event to listener."); });
595 : 20 : }
596 : :
597 : 40 : ShmServer::~ShmServer() {
598 [ + + + - ]: 20 : static auto& factory = ShmFactory::get();
599 : :
600 : 20 : quit_flag_.store(true, std::memory_order_release);
601 : :
602 : 20 : server_->stopOffer();
603 : :
604 : 20 : listener_->detachEvent(server_.value(), shm::popo::ServerEvent::REQUEST_RECEIVED);
605 : :
606 : : {
607 : 20 : std::lock_guard lock(callback_mtx_);
608 : 20 : }
609 : :
610 : 20 : factory.try_to_destroy_listener(domain_, listener_);
611 : :
612 : 20 : server_->releaseQueuedRequests();
613 : 40 : }
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 : 17 : void ShmServer::process_message() {
632 [ + + + - : 34 : while (server_->hasRequests() && !quit_flag_.load(std::memory_order_acquire)) {
+ + ]
633 : 34 : server_->take()
634 : 17 : .and_then([this](const void* buffer) {
635 : 17 : const auto* read_req = static_cast<const uint8_t*>(buffer);
636 : 17 : const auto* read_header = shm::popo::RequestHeader::fromPayload(read_req);
637 : :
638 : 17 : uint64_t channel = 0;
639 : 17 : uint64_t seq = 0;
640 : 17 : Bytes req_bytes;
641 : 17 : ShmFactory::read_data(read_req, read_header->getChunkHeader()->userPayloadSize(), channel, seq, req_bytes);
642 : :
643 : : (void)seq;
644 : :
645 : 17 : bool request_released = false;
646 : :
647 [ + - + - ]: 17 : traverse_req_resp_callback([this, channel, &req_bytes, &read_req, &read_header, &request_released](
648 : 147 : NodeImpl* impl, const auto& callback) {
649 [ + - ]: 17 : const auto* conf_ptr = impl->get_target_conf<ShmConf>();
650 : :
651 [ - + ]: 17 : if (static_cast<uint64_t>(conf_ptr->hash_code) != channel) {
652 : 0 : ignore_called();
653 : 0 : return;
654 : : }
655 : :
656 [ - + ]: 17 : if VUNLIKELY (has_called()) {
657 [ # # # # ]: 0 : VLOG_F(*conf_ptr, "Two identical service requests.");
658 : 0 : return;
659 : : }
660 : :
661 [ + - ]: 17 : std::lock_guard lock(mtx_);
662 : :
663 [ + + ]: 17 : if (static_cast<ServerImpl*>(impl)->is_resp_type) {
664 : 14 : last_req_header_ = read_header;
665 : :
666 : 14 : Bytes resp_bytes;
667 : :
668 : 14 : auto seq = static_cast<uint64_t>(read_header->getSequenceId());
669 : :
670 [ + - ]: 14 : callback(seq, req_bytes, &resp_bytes);
671 : 14 : } else {
672 [ + - ]: 3 : callback(0, req_bytes, nullptr);
673 : : }
674 : :
675 : 17 : server_->releaseRequest(read_req);
676 : 17 : request_released = true;
677 : 17 : });
678 : :
679 [ - + ]: 17 : if VUNLIKELY (!request_released) {
680 : 0 : server_->releaseRequest(read_req);
681 : : }
682 : 17 : })
683 [ # # # # ]: 17 : .or_else([](auto& e) { VLOG_E("ShmFactory: Failed to take request, error: ", e, "."); });
684 : : }
685 : 17 : }
686 : :
687 : 18 : 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 : 14 : Bytes ShmServer::loan(uint64_t channel, int64_t size) {
694 [ + + ]: 14 : if VUNLIKELY (size <= 0) {
695 : 13 : 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 : 14 : bool ShmServer::reply(uint64_t channel, const Bytes& resp_data) {
727 [ - + ]: 14 : 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 : 14 : auto write_resp_result = server_->loan(last_req_header_, resp_data.size() + ShmFactory::get_loaned_offset(),
736 : 14 : ShmFactory::get_loaned_alignment());
737 : :
738 : 14 : seq_.fetch_add(1, std::memory_order_relaxed);
739 : :
740 [ - + ]: 14 : 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 : 14 : auto* write_resp = static_cast<uint8_t*>(write_resp_result.value());
747 : :
748 : 28 : ShmFactory::write_data(write_resp, channel, seq_.load(std::memory_order_relaxed), resp_data);
749 : :
750 : 14 : auto send_result = server_->send(write_resp);
751 : :
752 [ - + ]: 14 : 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 [ + - + - ]: 14 : }
757 : :
758 : 14 : return true;
759 : : }
760 : :
761 : 18 : void ShmServer::on_request_received(shm::popo::UntypedServer*, ShmServer* target) {
762 [ + - ]: 18 : std::lock_guard callback_lock(target->callback_mtx_);
763 : :
764 [ - + ]: 18 : if VUNLIKELY (target->quit_flag_.load(std::memory_order_acquire)) {
765 : 0 : return;
766 : : }
767 : :
768 [ + + ]: 18 : if VUNLIKELY (target->is_suspend_.load(std::memory_order_relaxed)) {
769 : 1 : target->server_->releaseQueuedRequests();
770 : :
771 : 1 : return;
772 : : }
773 : :
774 [ + - ]: 17 : auto* impl = target->get_first_impl();
775 : :
776 [ - + ]: 17 : if VUNLIKELY (!impl) {
777 : 0 : return;
778 : : }
779 : :
780 [ + - ]: 17 : auto* message_loop = impl->get_message_loop();
781 : :
782 [ + + ]: 17 : if (message_loop) {
783 : 2 : std::weak_ptr<ShmServer> weak_target = target->weak_from_this();
784 [ + - + - ]: 2 : message_loop->post_task([weak_target]() {
785 : 2 : auto target = weak_target.lock();
786 : :
787 [ - + ]: 2 : if VUNLIKELY (!target) {
788 : 0 : return;
789 : : }
790 : :
791 [ + - ]: 2 : auto* impl = target->get_first_impl();
792 : :
793 [ + - + - : 2 : if VUNLIKELY (!impl || !impl->get_message_loop()) {
- + - + ]
794 : 0 : return;
795 : : }
796 : :
797 : 2 : target->process_message();
798 [ + - ]: 2 : });
799 : 2 : } else {
800 : 15 : target->process_message();
801 : : }
802 [ + + ]: 18 : }
803 : :
804 : : // ShmClient
805 : 17 : ShmClient::ShmClient(const ShmID& id) {
806 [ + + + - : 17 : static auto& factory = ShmFactory::get();
+ - - - ]
807 : :
808 : 17 : const auto& [impl_type, address, domain, depth, history, wait] = id;
809 : :
810 : 17 : domain_ = domain;
811 : :
812 : 17 : shm::popo::ClientOptions options;
813 : 17 : options.connectOnCreate = true;
814 : :
815 [ + + ]: 17 : if (depth > 0) {
816 : 1 : options.responseQueueCapacity = depth;
817 : : } else {
818 : 16 : options.responseQueueCapacity = kDefaultRespDepth;
819 : : }
820 : :
821 [ + - ]: 17 : std::string event = "method";
822 : :
823 [ + + ]: 17 : if (domain != 0) {
824 [ + - + - : 1 : event += ("_" + std::to_string(domain));
+ - ]
825 : : }
826 : :
827 [ + - ]: 17 : client_.emplace(ShmFactory::get_description("vlink", address, event), options);
828 : :
829 [ + - ]: 17 : listener_ = factory.get_listener(domain_);
830 : :
831 : 17 : listener_
832 [ + - ]: 17 : ->attachEvent(client_.value(), shm::popo::ClientEvent::RESPONSE_RECEIVED,
833 : 17 : shm::popo::createNotificationCallback(ShmClient::on_response_received, *this))
834 [ # # # # ]: 17 : .or_else([](auto&) { VLOG_F("ShmFactory: Failed to attach RESPONSE_RECEIVED event to listener."); });
835 : 17 : }
836 : :
837 : 34 : ShmClient::~ShmClient() {
838 [ + + + - ]: 17 : static auto& factory = ShmFactory::get();
839 : :
840 : 17 : quit_flag_.store(true, std::memory_order_release);
841 : :
842 : 17 : client_->disconnect();
843 : :
844 : 17 : disable_detect_timer();
845 : :
846 : 17 : listener_->detachEvent(client_.value(), shm::popo::ClientEvent::RESPONSE_RECEIVED);
847 : :
848 : : {
849 : 17 : std::lock_guard lock(callback_mtx_);
850 : 17 : }
851 : :
852 : 17 : factory.try_to_destroy_listener(domain_, listener_);
853 : :
854 : 17 : client_->releaseQueuedResponses();
855 : 34 : }
856 : :
857 [ + - ]: 1 : std::any ShmClient::get_native_handle() const { return this; }
858 : :
859 : 14 : void ShmClient::process_message() {
860 [ + + + - : 28 : while (client_->hasResponses() && !quit_flag_.load(std::memory_order_acquire)) {
+ + ]
861 : 28 : client_->take()
862 : 14 : .and_then([this](const void* buffer) {
863 [ + - ]: 14 : std::unique_lock lock(mtx_);
864 : :
865 : 14 : const auto* read_resp = static_cast<const uint8_t*>(buffer);
866 : 14 : const auto* read_header = shm::popo::ResponseHeader::fromPayload(read_resp);
867 [ + - ]: 14 : auto iter = callbacks_.find(read_header->getSequenceId());
868 : :
869 [ - + ]: 14 : if VUNLIKELY (iter == callbacks_.end()) {
870 : 0 : client_->releaseResponse(read_resp);
871 : 0 : return;
872 : : }
873 : :
874 : 14 : auto callback = std::move(iter->second);
875 : 14 : auto seq_id = iter->first;
876 : :
877 [ + - ]: 14 : lock.unlock();
878 : :
879 : 14 : uint64_t channel = 0;
880 : 14 : uint64_t seq = 0;
881 : 14 : Bytes resp_bytes;
882 : 14 : ShmFactory::read_data(read_resp, read_header->getChunkHeader()->userPayloadSize(), channel, seq, resp_bytes);
883 : :
884 : : (void)seq;
885 : :
886 [ + - ]: 14 : callback(channel, resp_bytes);
887 : :
888 : 14 : client_->releaseResponse(read_resp);
889 : :
890 [ + - ]: 14 : lock.lock();
891 : :
892 [ + - ]: 14 : callbacks_.erase(seq_id);
893 [ + - ]: 14 : })
894 [ # # # # ]: 14 : .or_else([](auto& e) { VLOG_E("ShmFactory: Failed to take response, error: ", e, "."); });
895 : : }
896 : 14 : }
897 : :
898 : 220 : bool ShmClient::is_connected() const { return client_->getConnectionState() == shm::ConnectionState::CONNECTED; }
899 : :
900 : 28 : void ShmClient::enable_detect_timer() {
901 [ + + ]: 28 : if (!has_detect_timer_.load(std::memory_order_relaxed)) {
902 : 15 : has_detect_timer_.store(true, std::memory_order_relaxed);
903 : :
904 [ + - + - ]: 15 : ShmFactory::get().add_detect_callback(this, [weak = weak_from_this()]() {
905 : 62 : auto self = weak.lock();
906 : :
907 [ + - ]: 62 : if VLIKELY (self) {
908 [ + - ]: 62 : self->detect_server();
909 : : }
910 : 62 : });
911 : : }
912 : 28 : }
913 : :
914 : 17 : void ShmClient::disable_detect_timer() {
915 [ + + ]: 17 : if (has_detect_timer_.load(std::memory_order_relaxed)) {
916 : 15 : has_detect_timer_.store(false, std::memory_order_relaxed);
917 : :
918 : 15 : ShmFactory::get().remove_detect_callback(this);
919 : : }
920 : 17 : }
921 : :
922 : 19 : Bytes ShmClient::loan(uint64_t channel, int64_t size) {
923 [ + + ]: 19 : if VUNLIKELY (size <= 0) {
924 : 17 : return Bytes();
925 : : }
926 : :
927 : 2 : auto write_req_result = client_->loan(size + ShmFactory::get_loaned_offset(), ShmFactory::get_loaned_alignment());
928 : :
929 [ - + ]: 2 : if VUNLIKELY (write_req_result.has_error()) {
930 [ # # # # ]: 0 : VLOG_E("ShmFactory: Failed to loan buffer, size: ", size + ShmFactory::get_loaned_offset(),
931 : : ", error: ", write_req_result.get_error(), ".");
932 : 0 : return Bytes();
933 : : }
934 : :
935 : 2 : auto* write_req = static_cast<uint8_t*>(write_req_result.value());
936 : :
937 : 4 : ShmFactory::write_header(write_req, channel, seq_.load(std::memory_order_relaxed));
938 : :
939 : 2 : return Bytes::loan_internal(write_req + ShmFactory::get_loaned_offset(), size);
940 : 2 : }
941 : :
942 : 1 : bool ShmClient::release(const Bytes& bytes) {
943 [ + - ]: 1 : if VUNLIKELY (!bytes.is_loaned()) {
944 : 1 : return false;
945 : : }
946 : :
947 : 0 : client_->releaseRequest(const_cast<uint8_t*>(bytes.data()) - ShmFactory::get_loaned_offset());
948 : :
949 : 0 : return true;
950 : : }
951 : :
952 : 18 : bool ShmClient::call(uint64_t channel, const Bytes& req_data, NodeImpl::MsgCallback&& callback, uint64_t* seq_out) {
953 [ - + ]: 18 : if VUNLIKELY (!is_connected()) {
954 : 0 : return false;
955 : : }
956 : :
957 [ + - ]: 18 : std::lock_guard lock(mtx_);
958 : 18 : uint64_t response_seq = 0;
959 : 18 : bool has_response_callback = false;
960 : :
961 [ + + ]: 18 : if (req_data.is_loaned()) {
962 : : auto* write_header =
963 : 2 : shm::popo::RequestHeader::fromPayload(const_cast<uint8_t*>(req_data.data()) - ShmFactory::get_loaned_offset());
964 : :
965 [ + + ]: 2 : if (callback) {
966 : 1 : response_seq = seq_.load(std::memory_order_relaxed);
967 : 1 : has_response_callback = true;
968 : 1 : callbacks_[response_seq] = [callback = std::move(callback), channel](uint64_t target_channel,
969 : 1 : const Bytes& bytes) {
970 [ - + ]: 1 : if (channel != target_channel) {
971 : 0 : return;
972 : : }
973 : :
974 : 1 : callback(bytes);
975 [ + - + - ]: 1 : };
976 : 1 : write_header->setSequenceId(response_seq);
977 : :
978 [ + - ]: 1 : if (seq_out) {
979 : 1 : *seq_out = response_seq;
980 : : }
981 : :
982 : 1 : seq_.fetch_add(1, std::memory_order_relaxed);
983 : : }
984 : :
985 : 2 : auto send_result = client_->send(const_cast<uint8_t*>(req_data.data()) - ShmFactory::get_loaned_offset());
986 : :
987 [ - + ]: 2 : if VUNLIKELY (send_result.has_error()) {
988 [ # # # # ]: 0 : VLOG_E("ShmFactory: Failed to send, error: ", send_result.get_error(), ".");
989 [ # # ]: 0 : if (has_response_callback) {
990 [ # # ]: 0 : callbacks_.erase(response_seq);
991 : : }
992 : 0 : return false;
993 : : }
994 [ + - ]: 2 : } else {
995 : : auto write_req_result =
996 : 16 : client_->loan(req_data.size() + ShmFactory::get_loaned_offset(), ShmFactory::get_loaned_alignment());
997 [ - + ]: 16 : if VUNLIKELY (write_req_result.has_error()) {
998 [ # # # # ]: 0 : VLOG_E("ShmFactory: Failed to loan buffer, size: ", req_data.size() + ShmFactory::get_loaned_offset(),
999 : : ", error: ", write_req_result.get_error(), ".");
1000 : :
1001 : 0 : return false;
1002 : : }
1003 : :
1004 : 16 : auto* write_req = static_cast<uint8_t*>(write_req_result.value());
1005 : :
1006 : 16 : auto* write_header = shm::popo::RequestHeader::fromPayload(write_req);
1007 : :
1008 [ + + ]: 16 : if (callback) {
1009 : 14 : response_seq = seq_.load(std::memory_order_relaxed);
1010 : 14 : has_response_callback = true;
1011 : 14 : callbacks_[response_seq] = [callback = std::move(callback), channel](uint64_t target_channel,
1012 : 13 : const Bytes& bytes) {
1013 [ - + ]: 13 : if (channel != target_channel) {
1014 : 0 : return;
1015 : : }
1016 : :
1017 : 13 : callback(bytes);
1018 [ + - + - ]: 14 : };
1019 : 14 : write_header->setSequenceId(response_seq);
1020 : :
1021 [ + + ]: 14 : if (seq_out) {
1022 : 11 : *seq_out = response_seq;
1023 : : }
1024 : :
1025 : 14 : seq_.fetch_add(1, std::memory_order_relaxed);
1026 : : }
1027 : :
1028 : 32 : ShmFactory::write_data(write_req, channel, seq_.load(std::memory_order_relaxed), req_data);
1029 : :
1030 : 16 : auto send_result = client_->send(write_req);
1031 : :
1032 [ - + ]: 16 : if VUNLIKELY (send_result.has_error()) {
1033 [ # # # # ]: 0 : VLOG_E("ShmFactory: Failed to send, error: ", send_result.get_error(), ".");
1034 : :
1035 [ # # ]: 0 : if (has_response_callback) {
1036 [ # # ]: 0 : callbacks_.erase(response_seq);
1037 : : }
1038 : :
1039 : 0 : return false;
1040 : : }
1041 [ + - + - ]: 16 : }
1042 : :
1043 : 18 : return true;
1044 : 18 : }
1045 : :
1046 : 1 : void ShmClient::remove_response_callback(uint64_t seq) {
1047 [ + - ]: 1 : std::lock_guard lock(mtx_);
1048 [ + - ]: 1 : callbacks_.erase(seq);
1049 : 1 : }
1050 : :
1051 : 62 : void ShmClient::detect_server() {
1052 [ - + ]: 62 : if VUNLIKELY (quit_flag_.load(std::memory_order_acquire)) {
1053 : 0 : return;
1054 : : }
1055 : :
1056 [ - + ]: 62 : if VUNLIKELY (!ShmFactory::has_runtime_inited()) {
1057 : 0 : return;
1058 : : }
1059 : :
1060 : 62 : discovery_server(is_connected());
1061 : : }
1062 : :
1063 : 62 : void ShmClient::discovery_server(bool connect) {
1064 [ + + ]: 62 : if VLIKELY (last_connected_.load(std::memory_order_relaxed) == connect) {
1065 : 47 : return;
1066 : : }
1067 : :
1068 [ + - + - ]: 30 : traverse_server_connect_callback([connect](NodeImpl*, const auto& callback) { callback(connect); });
1069 : 15 : last_connected_.store(connect, std::memory_order_relaxed);
1070 : : }
1071 : :
1072 : 14 : void ShmClient::on_response_received(shm::popo::UntypedClient*, ShmClient* target) {
1073 [ + - ]: 14 : std::lock_guard callback_lock(target->callback_mtx_);
1074 : :
1075 [ - + ]: 14 : if VUNLIKELY (target->quit_flag_.load(std::memory_order_acquire)) {
1076 : 0 : return;
1077 : : }
1078 : :
1079 [ + - ]: 14 : auto* impl = target->get_first_impl();
1080 : :
1081 [ - + ]: 14 : if VUNLIKELY (!impl) {
1082 : 0 : return;
1083 : : }
1084 : :
1085 [ + - ]: 14 : auto* message_loop = impl->get_message_loop();
1086 : :
1087 [ + + ]: 14 : if (message_loop) {
1088 : 2 : std::weak_ptr<ShmClient> weak_target = target->weak_from_this();
1089 [ + - + - ]: 2 : message_loop->post_task([weak_target]() {
1090 : 2 : auto target = weak_target.lock();
1091 : :
1092 [ - + ]: 2 : if VUNLIKELY (!target) {
1093 : 0 : return;
1094 : : }
1095 : :
1096 [ + - ]: 2 : auto* impl = target->get_first_impl();
1097 : :
1098 [ + - + - : 2 : if VUNLIKELY (!impl || !impl->get_message_loop()) {
- + - + ]
1099 : 0 : return;
1100 : : }
1101 : :
1102 : 2 : target->process_message();
1103 [ + - ]: 2 : });
1104 : 2 : } else {
1105 : 12 : target->process_message();
1106 : : }
1107 [ + - ]: 14 : }
1108 : :
1109 : : // ShmPublisher
1110 : 54 : ShmPublisher::ShmPublisher(const ShmID& id) {
1111 [ + + + - : 54 : static auto& factory = ShmFactory::get();
+ - - - ]
1112 : :
1113 : 54 : const auto& [impl_type, address, domain, depth, history, wait] = id;
1114 : :
1115 : 54 : domain_ = domain;
1116 : 54 : wait_ = wait;
1117 : :
1118 : 54 : shm::popo::PublisherOptions options;
1119 : 54 : options.offerOnCreate = true;
1120 : 54 : options.historyCapacity = history;
1121 : :
1122 [ + - ]: 54 : std::string event = "event";
1123 : :
1124 [ + + ]: 54 : if (domain != 0) {
1125 [ + - + - : 3 : event += ("_" + std::to_string(domain));
+ - ]
1126 : : }
1127 : :
1128 [ + - + - ]: 54 : pub_.emplace(ShmFactory::get_description("vlink", address, event), options);
1129 : :
1130 [ + + ]: 54 : if (wait > 0) {
1131 [ + - ]: 1 : factory.start_detect_node_count();
1132 : :
1133 [ + - ]: 1 : sem_.emplace();
1134 : :
1135 [ + - ]: 1 : std::string sem_address = address;
1136 : 1 : std::replace(sem_address.begin(), sem_address.end(), '/', '@');
1137 : :
1138 : : #ifdef __FreeBSD__
1139 : : sem_->attach("/vlink@shm@" + sem_address);
1140 : : #else
1141 [ + - + - ]: 1 : sem_->attach("vlink@shm@" + sem_address);
1142 : : #endif
1143 : 1 : }
1144 : 54 : }
1145 : :
1146 : 108 : ShmPublisher::~ShmPublisher() {
1147 : 54 : quit_flag_.store(true, std::memory_order_release);
1148 : :
1149 : 54 : pub_->stopOffer();
1150 : :
1151 : 54 : disable_detect_timer();
1152 : :
1153 [ + + ]: 54 : if (sem_) {
1154 : 1 : sem_->detach(true);
1155 : : }
1156 : 108 : }
1157 : :
1158 [ + - ]: 2 : std::any ShmPublisher::get_native_handle() const { return this; }
1159 : :
1160 : 851 : bool ShmPublisher::has_subscribers() const { return pub_->hasSubscribers(); }
1161 : :
1162 : 189 : Bytes ShmPublisher::loan(uint64_t channel, int64_t size) {
1163 [ + + ]: 189 : if VUNLIKELY (size <= 0) {
1164 : 184 : return Bytes();
1165 : : }
1166 : :
1167 : 5 : auto write_msg_result = pub_->loan(size + ShmFactory::get_loaned_offset(), ShmFactory::get_loaned_alignment());
1168 : :
1169 : 5 : seq_.fetch_add(1, std::memory_order_relaxed);
1170 : :
1171 [ - + ]: 5 : if VUNLIKELY (write_msg_result.has_error()) {
1172 [ # # # # ]: 0 : VLOG_E("ShmFactory: Failed to loan buffer, size: ", size + ShmFactory::get_loaned_offset(),
1173 : : ", error: ", write_msg_result.get_error(), ".");
1174 : 0 : return Bytes();
1175 : : }
1176 : :
1177 : 5 : auto* write_msg = static_cast<uint8_t*>(write_msg_result.value());
1178 : :
1179 : 10 : ShmFactory::write_header(write_msg, channel, seq_.load(std::memory_order_relaxed));
1180 : :
1181 : 5 : return Bytes::loan_internal(write_msg + ShmFactory::get_loaned_offset(), size);
1182 : 5 : }
1183 : :
1184 : 3 : bool ShmPublisher::release(const Bytes& bytes) {
1185 [ + + ]: 3 : if VUNLIKELY (!bytes.is_loaned()) {
1186 : 2 : return false;
1187 : : }
1188 : :
1189 : 1 : pub_->release(const_cast<uint8_t*>(bytes.data()) - ShmFactory::get_loaned_offset());
1190 : :
1191 : 1 : return true;
1192 : : }
1193 : :
1194 : 212 : bool ShmPublisher::publish(uint64_t channel, const Bytes& bytes) {
1195 [ + + ]: 212 : if VUNLIKELY (wait_ > 0) {
1196 [ + - ]: 1 : uint64_t sem_count = sem_->get_count();
1197 : :
1198 [ + - + - ]: 2 : VLOG_I("ShmFactory: Wait sem_count: ", sem_count, ".");
1199 : :
1200 [ - + ]: 1 : if (sem_count > 0) {
1201 [ # # ]: 0 : sem_->acquire(sem_count, wait_);
1202 : : }
1203 : : }
1204 : :
1205 [ + + ]: 212 : if (bytes.is_loaned()) {
1206 : 4 : pub_->publish(const_cast<uint8_t*>(bytes.data()) - ShmFactory::get_loaned_offset());
1207 : : } else {
1208 : : auto write_msg_result =
1209 : 205 : pub_->loan(bytes.size() + ShmFactory::get_loaned_offset(), ShmFactory::get_loaned_alignment());
1210 : :
1211 : 209 : seq_.fetch_add(1, std::memory_order_relaxed);
1212 : :
1213 [ - + ]: 209 : if VUNLIKELY (write_msg_result.has_error()) {
1214 [ # # # # ]: 0 : VLOG_E("ShmFactory: Failed to loan buffer, size: ", bytes.size() + ShmFactory::get_loaned_offset(),
1215 : : ", error: ", write_msg_result.get_error(), ".");
1216 : 0 : return false;
1217 : : }
1218 : :
1219 : 204 : auto* write_msg = static_cast<uint8_t*>(write_msg_result.value());
1220 : :
1221 : 408 : ShmFactory::write_data(write_msg, channel, seq_.load(std::memory_order_relaxed), bytes);
1222 : :
1223 : 204 : pub_->publish(write_msg);
1224 [ + - ]: 212 : }
1225 : :
1226 [ + + ]: 213 : if VUNLIKELY (wait_ > 0) {
1227 [ + - + - ]: 1 : uint64_t sub_count = ShmFactory::get().get_subscriber_count(pub_->getServiceDescription());
1228 [ + - + - ]: 2 : VLOG_I("ShmFactory: Wait sub_count: ", sub_count, ".");
1229 [ + - ]: 1 : sem_->acquire(sub_count, wait_);
1230 : : }
1231 : :
1232 : 213 : return true;
1233 : : }
1234 : :
1235 : 52 : void ShmPublisher::enable_detect_timer() {
1236 [ + + ]: 52 : if (!has_detect_timer_.load(std::memory_order_relaxed)) {
1237 : 49 : has_detect_timer_.store(true, std::memory_order_relaxed);
1238 [ + - + - ]: 49 : ShmFactory::get().add_detect_callback(this, [weak = weak_from_this()]() {
1239 : 460 : auto self = weak.lock();
1240 : :
1241 [ + - ]: 460 : if VLIKELY (self) {
1242 [ + - ]: 460 : self->detect_subscribers();
1243 : : }
1244 : 460 : });
1245 : : }
1246 : 52 : }
1247 : :
1248 : 54 : void ShmPublisher::disable_detect_timer() {
1249 [ + + ]: 54 : if (has_detect_timer_.load(std::memory_order_relaxed)) {
1250 : 49 : has_detect_timer_.store(false, std::memory_order_relaxed);
1251 : 49 : ShmFactory::get().remove_detect_callback(this);
1252 : : }
1253 : 54 : }
1254 : :
1255 : 460 : void ShmPublisher::detect_subscribers() {
1256 [ - + ]: 460 : if VUNLIKELY (quit_flag_.load(std::memory_order_acquire)) {
1257 : 0 : return;
1258 : : }
1259 : :
1260 [ - + ]: 460 : if VUNLIKELY (!ShmFactory::has_runtime_inited()) {
1261 : 0 : return;
1262 : : }
1263 : :
1264 : 460 : discovery_subscribers(has_subscribers());
1265 : : }
1266 : :
1267 : 460 : void ShmPublisher::discovery_subscribers(bool has_subscribers) {
1268 [ + + ]: 460 : if VLIKELY (last_has_subscribers_.load(std::memory_order_relaxed) == has_subscribers) {
1269 : 414 : return;
1270 : : }
1271 : :
1272 [ + - + - ]: 85 : traverse_sub_connect_callback([has_subscribers](NodeImpl*, const auto& callback) { callback(has_subscribers); });
1273 : 46 : last_has_subscribers_.store(has_subscribers, std::memory_order_relaxed);
1274 : : }
1275 : :
1276 : : // ShmSubscriber
1277 : 55 : ShmSubscriber::ShmSubscriber(const ShmID& id) {
1278 [ + + + - : 55 : static auto& factory = ShmFactory::get();
+ - - - ]
1279 : :
1280 : 55 : const auto& [impl_type, address, domain, depth, history, wait] = id;
1281 : :
1282 : 55 : domain_ = domain;
1283 : 55 : wait_ = wait;
1284 : :
1285 : 55 : shm::popo::SubscriberOptions options;
1286 : 55 : options.subscribeOnCreate = false;
1287 : 55 : options.historyRequest = history;
1288 : : // options.requiresPublisherHistorySupport = history > 0;
1289 : :
1290 [ + + ]: 55 : if (depth > 0) {
1291 : 2 : options.queueCapacity = depth;
1292 : : } else {
1293 : 53 : options.queueCapacity = factory.get_sub_depth();
1294 : : }
1295 : :
1296 [ + - ]: 55 : std::string event = "event";
1297 : :
1298 [ + + ]: 55 : if (domain != 0) {
1299 [ + - + - : 3 : event += ("_" + std::to_string(domain));
+ - ]
1300 : : }
1301 : :
1302 [ + - + - ]: 55 : sub_.emplace(ShmFactory::get_description("vlink", address, event), options);
1303 : :
1304 [ + - ]: 55 : listener_ = factory.get_listener(domain_);
1305 : :
1306 : 55 : listener_
1307 [ + - ]: 55 : ->attachEvent(sub_.value(), shm::popo::SubscriberEvent::DATA_RECEIVED,
1308 : 55 : shm::popo::createNotificationCallback(ShmSubscriber::on_msg_received, *this))
1309 [ # # # # ]: 55 : .or_else([](auto&) { VLOG_F("ShmFactory: Failed to attach DATA_RECEIVED event to listener."); });
1310 : :
1311 [ + + ]: 55 : if (wait_ > 0) {
1312 [ + - ]: 1 : sem_.emplace();
1313 : :
1314 [ + - ]: 1 : std::string sem_address = address;
1315 : 1 : std::replace(sem_address.begin(), sem_address.end(), '/', '@');
1316 : :
1317 : : #ifdef __FreeBSD__
1318 : : sem_->attach("/vlink@shm@" + sem_address);
1319 : : #else
1320 [ + - + - ]: 1 : sem_->attach("vlink@shm@" + sem_address);
1321 : : #endif
1322 : 1 : }
1323 : 55 : }
1324 : :
1325 : 110 : ShmSubscriber::~ShmSubscriber() {
1326 [ + + + - ]: 55 : static auto& factory = ShmFactory::get();
1327 : :
1328 : 55 : quit_flag_.store(true, std::memory_order_release);
1329 : :
1330 : 55 : sub_->unsubscribe();
1331 : :
1332 : 55 : listener_->detachEvent(sub_.value(), shm::popo::SubscriberEvent::DATA_RECEIVED);
1333 : :
1334 : : {
1335 : 55 : std::lock_guard lock(callback_mtx_);
1336 : 55 : }
1337 : :
1338 : 55 : factory.try_to_destroy_listener(domain_, listener_);
1339 : :
1340 : 55 : sub_->releaseQueuedData();
1341 : :
1342 [ + + ]: 55 : if (sem_) {
1343 : 1 : sem_->detach(false);
1344 : : }
1345 : 110 : }
1346 : :
1347 [ + - ]: 2 : std::any ShmSubscriber::get_native_handle() const { return this; }
1348 : :
1349 : 2 : bool ShmSubscriber::suspend() {
1350 : 2 : is_suspend_.store(true, std::memory_order_relaxed);
1351 : :
1352 : 2 : return true;
1353 : : }
1354 : :
1355 : 2 : bool ShmSubscriber::resume() {
1356 : 2 : is_suspend_.store(false, std::memory_order_relaxed);
1357 : :
1358 : 2 : return true;
1359 : : }
1360 : :
1361 : 5 : bool ShmSubscriber::is_suspend() const { return is_suspend_.load(std::memory_order_relaxed); }
1362 : :
1363 : 171 : void ShmSubscriber::process_message() {
1364 [ + + + - : 355 : while (sub_->hasData() && !quit_flag_.load(std::memory_order_acquire)) {
+ + ]
1365 : 368 : sub_->take()
1366 : 184 : .and_then([this](const void* buffer) {
1367 : 184 : const auto* read_msg = static_cast<const uint8_t*>(buffer);
1368 : 184 : const auto* read_header = shm::mepoo::ChunkHeader::fromUserPayload(read_msg);
1369 : :
1370 : 184 : uint64_t channel = 0;
1371 : 184 : uint64_t seq = 0;
1372 : 184 : Bytes msg_bytes;
1373 : 184 : ShmFactory::read_data(read_msg, read_header->userPayloadSize(), channel, seq, msg_bytes);
1374 : :
1375 [ + + ]: 184 : if VUNLIKELY (is_latency_and_lost_enabled_.load(std::memory_order_acquire)) {
1376 [ + - ]: 10 : if (seq > 0) {
1377 : 10 : calc_sample_.update(seq, static_cast<uint64_t>(read_header->originId()));
1378 : : } else {
1379 : 0 : calc_sample_.update(read_header->sequenceNumber(), static_cast<uint64_t>(read_header->originId()));
1380 : : }
1381 : : }
1382 : :
1383 : 184 : bool called = false;
1384 : :
1385 [ + - + - ]: 184 : traverse_msg_callback([channel, &msg_bytes, &called](NodeImpl* impl, const auto& callback) {
1386 : 309 : const auto* conf_ptr = impl->get_target_conf<ShmConf>();
1387 : :
1388 [ + + ]: 309 : if (static_cast<uint64_t>(conf_ptr->hash_code) != channel) {
1389 : 2 : return;
1390 : : }
1391 : :
1392 : 307 : called = true;
1393 : 307 : callback(msg_bytes);
1394 : : });
1395 : :
1396 [ - + - - : 184 : if VLIKELY (!manual_unloan_.load(std::memory_order_relaxed) || !called) {
+ - ]
1397 : 184 : sub_->release(read_msg);
1398 [ + + ]: 184 : if (sem_) {
1399 [ + - ]: 1 : sem_->release();
1400 : : }
1401 : : }
1402 : 184 : })
1403 [ # # # # ]: 184 : .or_else([](auto& e) { VLOG_E("ShmFactory: Failed to take sample, error: ", e, "."); });
1404 : : }
1405 : 171 : }
1406 : :
1407 : 66 : void ShmSubscriber::subscribe() { sub_->subscribe(); }
1408 : :
1409 : 0 : void ShmSubscriber::unsubscribe() { sub_->unsubscribe(); }
1410 : :
1411 : 0 : void ShmSubscriber::set_manual_unloan(bool manual_unloan) {
1412 : 0 : manual_unloan_.store(manual_unloan, std::memory_order_relaxed);
1413 : 0 : }
1414 : :
1415 : 2 : bool ShmSubscriber::release(const Bytes& bytes) {
1416 [ + - ]: 2 : if VUNLIKELY (!bytes.is_loaned()) {
1417 : 2 : return false;
1418 : : }
1419 : :
1420 [ # # ]: 0 : if VUNLIKELY (!manual_unloan_.load(std::memory_order_relaxed)) {
1421 [ # # # # ]: 0 : VLOG_F("ShmFactory: Manual release is not supported without manual_unloan mode.");
1422 : 0 : return false;
1423 : : }
1424 : :
1425 : 0 : sub_->release(const_cast<uint8_t*>(bytes.data()) - ShmFactory::get_loaned_offset());
1426 : :
1427 [ # # ]: 0 : if (sem_) {
1428 : 0 : sem_->release();
1429 : : }
1430 : :
1431 : 0 : return true;
1432 : : }
1433 : :
1434 : 72 : void ShmSubscriber::set_latency_and_lost_enabled(bool enable) {
1435 : 72 : is_latency_and_lost_enabled_.store(enable, std::memory_order_release);
1436 : 72 : }
1437 : :
1438 : 6 : bool ShmSubscriber::is_latency_and_lost_enabled() const {
1439 : 6 : return is_latency_and_lost_enabled_.load(std::memory_order_acquire);
1440 : : }
1441 : :
1442 : 2 : const CalculateSample& ShmSubscriber::get_calculate_sample() const { return calc_sample_; }
1443 : :
1444 : 173 : void ShmSubscriber::on_msg_received(shm::popo::UntypedSubscriber*, ShmSubscriber* target) {
1445 [ + - ]: 173 : std::lock_guard callback_lock(target->callback_mtx_);
1446 : :
1447 [ - + ]: 173 : if VUNLIKELY (target->quit_flag_.load(std::memory_order_acquire)) {
1448 : 0 : return;
1449 : : }
1450 : :
1451 [ + + ]: 173 : if VUNLIKELY (target->is_suspend_.load(std::memory_order_relaxed)) {
1452 : 2 : target->sub_->releaseQueuedData();
1453 : 2 : return;
1454 : : }
1455 : :
1456 [ + - ]: 171 : auto* impl = target->get_first_impl();
1457 : :
1458 [ - + ]: 171 : if VUNLIKELY (!impl) {
1459 : 0 : return;
1460 : : }
1461 : :
1462 [ + - ]: 171 : auto* message_loop = impl->get_message_loop();
1463 : :
1464 [ + + ]: 171 : if (message_loop) {
1465 : 2 : std::weak_ptr<ShmSubscriber> weak_target = target->weak_from_this();
1466 [ + - + - ]: 2 : message_loop->post_task([weak_target]() {
1467 : 2 : auto target = weak_target.lock();
1468 : :
1469 [ - + ]: 2 : if VUNLIKELY (!target) {
1470 : 0 : return;
1471 : : }
1472 : :
1473 [ + - ]: 2 : auto* impl = target->get_first_impl();
1474 : :
1475 [ + - + - : 2 : if VUNLIKELY (!impl || !impl->get_message_loop()) {
- + - + ]
1476 : 0 : return;
1477 : : }
1478 : :
1479 : 2 : target->process_message();
1480 [ + - ]: 2 : });
1481 : 2 : } else {
1482 : 169 : target->process_message();
1483 : : }
1484 [ + + ]: 173 : }
1485 : :
1486 : : } // namespace vlink
|
