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 : : #pragma once
25 : :
26 : : #if __has_include(<unistd.h>)
27 : : #include <unistd.h>
28 : : #endif
29 : :
30 : : #ifdef _WIN32
31 : : #define _WINSOCKAPI_ // NOLINT(bugprone-reserved-identifier, readability-identifier-naming)
32 : : #define WIN32_LEAN_AND_MEAN
33 : : #include <Windows.h>
34 : : #include <Winsock2.h>
35 : : #include <ws2tcpip.h>
36 : : #endif
37 : :
38 : : #include <atomic>
39 : : #include <memory>
40 : : #include <mutex>
41 : : #include <optional>
42 : : #include <shared_mutex>
43 : : #include <string>
44 : : #include <thread>
45 : : #include <tuple>
46 : : #include <unordered_map>
47 : : #include <vector>
48 : :
49 : : // iox
50 : : #include <iceoryx_hoofs/cxx/optional.hpp>
51 : : #include <iceoryx_hoofs/cxx/scoped_static.hpp>
52 : : #include <iceoryx_posh/mepoo/mepoo_config.hpp>
53 : : #include <iceoryx_posh/popo/listener.hpp>
54 : : #include <iceoryx_posh/popo/untyped_client.hpp>
55 : : #include <iceoryx_posh/popo/untyped_publisher.hpp>
56 : : #include <iceoryx_posh/popo/untyped_server.hpp>
57 : : #include <iceoryx_posh/popo/untyped_subscriber.hpp>
58 : : #include <iceoryx_posh/roudi/iceoryx_roudi_components.hpp>
59 : : #include <iceoryx_posh/roudi/introspection_types.hpp>
60 : : #include <iceoryx_posh/roudi/roudi_config_toml_file_provider.hpp>
61 : : #include <iceoryx_posh/runtime/service_discovery.hpp>
62 : : #include <iceoryx_versions.hpp>
63 : :
64 : : #include "./base/message_loop.h"
65 : : #include "./base/sys_semaphore.h"
66 : : #include "./base/utils.h"
67 : : #include "./impl/abstract_factory.h"
68 : : #include "./impl/calculate_sample.h"
69 : : #include "./modules/shm_conf.h"
70 : :
71 : : #define SHM_USE_CUSTOM_SEQ 1
72 : :
73 : : namespace vlink {
74 : :
75 : : namespace shm = iox;
76 : :
77 : : using ShmID = std::tuple<uint8_t, std::string, int32_t, int32_t, int32_t, int32_t, std::string, const void*>;
78 : :
79 : : [[maybe_unused]] static constexpr int kDefaultReqDepth = 50;
80 : : [[maybe_unused]] static constexpr int kDefaultRespDepth = 10;
81 : : [[maybe_unused]] static constexpr int kDefaultSubDepth = 5;
82 : :
83 : : // ShmFactory
84 : : class ShmFactory final : public AbstractFactory<ShmID> {
85 : : public:
86 : : using DetectCallback = Function<void()>;
87 : : using DiscoveryCallback = Function<void(shm::runtime::ServiceDiscovery*)>;
88 : : using ShmId = std::tuple<shm::capro::IdString_t, shm::capro::IdString_t, shm::capro::IdString_t>;
89 : :
90 : : private:
91 : : ShmFactory();
92 : :
93 : : ~ShmFactory() override;
94 : :
95 : : public:
96 : 4868 : static constexpr size_t get_loaned_offset() {
97 : : #if SHM_USE_CUSTOM_SEQ
98 : 4868 : return sizeof(uint64_t) + sizeof(uint64_t);
99 : : #else
100 : : return sizeof(uint64_t);
101 : : #endif
102 : : }
103 : :
104 : 4523 : static constexpr size_t get_loaned_alignment() { return sizeof(uint64_t); }
105 : :
106 : 169 : static void write_header(uint8_t* loaned_ptr, uint64_t channel, uint64_t seq) {
107 : : #if SHM_USE_CUSTOM_SEQ
108 : 169 : std::memcpy(loaned_ptr, &channel, sizeof(uint64_t));
109 : 169 : std::memcpy(loaned_ptr + sizeof(uint64_t), &seq, sizeof(uint64_t));
110 : : #else
111 : : (void)seq;
112 : : std::memcpy(loaned_ptr, &channel, sizeof(uint64_t));
113 : : #endif
114 : 169 : }
115 : :
116 : 4354 : static void write_data(uint8_t* loaned_ptr, uint64_t channel, uint64_t seq, const Bytes& data) {
117 : : #if SHM_USE_CUSTOM_SEQ
118 : 4354 : std::memcpy(loaned_ptr, &channel, sizeof(uint64_t));
119 : 4354 : std::memcpy(loaned_ptr + sizeof(uint64_t), &seq, sizeof(uint64_t));
120 : 4354 : std::memcpy(loaned_ptr + sizeof(uint64_t) + sizeof(uint64_t), data.data(), data.size());
121 : : #else
122 : : (void)seq;
123 : : std::memcpy(loaned_ptr, &channel, sizeof(uint64_t));
124 : : std::memcpy(loaned_ptr + sizeof(uint64_t), data.data(), data.size());
125 : : #endif
126 : 4353 : }
127 : :
128 : 4237 : static void read_data(const uint8_t* loaned_ptr, uint64_t payload_size, uint64_t& channel, uint64_t& seq,
129 : : Bytes& data) {
130 : : #if SHM_USE_CUSTOM_SEQ
131 : :
132 [ - + ]: 4237 : if VUNLIKELY (payload_size < sizeof(uint64_t) + sizeof(uint64_t)) {
133 : 0 : return;
134 : : }
135 : :
136 : 4237 : std::memcpy(&channel, loaned_ptr, sizeof(uint64_t));
137 : 4237 : std::memcpy(&seq, loaned_ptr + sizeof(uint64_t), sizeof(uint64_t));
138 : 8474 : data = Bytes::shallow_copy(loaned_ptr + sizeof(uint64_t) + sizeof(uint64_t),
139 : 4237 : payload_size - sizeof(uint64_t) - sizeof(uint64_t));
140 : : #else
141 : : (void)seq;
142 : :
143 : : if VUNLIKELY (payload_size < sizeof(uint64_t)) {
144 : : return;
145 : : }
146 : :
147 : : std::memcpy(&channel, loaned_ptr, sizeof(uint64_t));
148 : : data = Bytes::shallow_copy(loaned_ptr + sizeof(uint64_t), payload_size - sizeof(uint64_t));
149 : : #endif
150 : : }
151 : :
152 : : static bool has_roudi_inited();
153 : :
154 : : static bool has_runtime_inited();
155 : :
156 : : static bool has_roudi_running();
157 : :
158 : : static bool auto_init_roudi(bool same_process_from_roudi, int memory_strategy);
159 : :
160 : : static shm::capro::ServiceDescription get_description(const std::string& service, const std::string& instance,
161 : : const std::string& event);
162 : :
163 : : static void init_log_level(bool wait_roudi);
164 : :
165 : : static void init_roudi(const std::string& config_path = "", int memory_strategy = 0, bool monitoring_enable = true);
166 : :
167 : : static void init_runtime(std::string name = "", bool same_process_from_roudi = false);
168 : :
169 : : static void deinit_runtime();
170 : :
171 : : static void deinit_roudi();
172 : :
173 : : [[nodiscard]] std::shared_ptr<shm::popo::Listener> get_listener(int32_t domain = 0);
174 : :
175 : : void add_detect_callback(void* node, DetectCallback&& callback);
176 : :
177 : : void remove_detect_callback(void* node);
178 : :
179 : : void start_detect_node_count();
180 : :
181 : : uint64_t get_publisher_count(const shm::capro::ServiceDescription& description);
182 : :
183 : : uint64_t get_subscriber_count(const shm::capro::ServiceDescription& description);
184 : :
185 : : int get_sub_depth() const;
186 : :
187 : : private:
188 : : std::unordered_map<int32_t, std::weak_ptr<shm::popo::Listener>> listener_map_;
189 : : MessageLoop message_loop_{MessageLoop::kNormalType};
190 : : Timer detect_timer_;
191 : : std::unordered_map<void*, DetectCallback> detect_map_;
192 : : std::shared_mutex detect_mtx_;
193 : : std::mutex listener_mtx_;
194 : : std::unordered_map<AbstractNode*, DiscoveryCallback> discovery_map_;
195 : : int sub_depth_{kDefaultSubDepth};
196 : :
197 : : std::optional<shm::popo::Subscriber<shm::roudi::PortIntrospectionFieldTopic>> port_sub_;
198 : : iox::roudi::PortIntrospectionFieldTopic topic_list_;
199 : : std::shared_mutex topic_mtx_;
200 : :
201 [ + + + - : 223 : VLINK_SINGLETON_DECLARE(ShmFactory)
+ - - - ]
202 : : };
203 : :
204 : : // ShmServer
205 : : class ShmServer final : public AbstractObject<ShmID>, public std::enable_shared_from_this<ShmServer> {
206 : : public:
207 : : explicit ShmServer(const ShmID& id);
208 : :
209 : : ~ShmServer() override;
210 : :
211 : : std::any get_native_handle() const override;
212 : :
213 : : bool suspend();
214 : :
215 : : bool resume();
216 : :
217 : : bool is_suspend() const;
218 : :
219 : : void process_message();
220 : :
221 : : void start();
222 : :
223 : : void stop();
224 : :
225 : : bool has_clients() const;
226 : :
227 : : Bytes loan(uint64_t channel, int64_t size);
228 : :
229 : : bool release(const Bytes& bytes);
230 : :
231 : : bool reply(uint64_t channel, const Bytes& resp_data);
232 : :
233 : : private:
234 : : static void on_request_received(shm::popo::UntypedServer* server, ShmServer* target);
235 : :
236 : : alignas(64) std::atomic<uint64_t> seq_{0};
237 : : std::atomic_bool is_suspend_{false};
238 : :
239 : : int32_t domain_{0};
240 : : std::shared_ptr<shm::popo::Listener> listener_;
241 : : std::optional<shm::popo::UntypedServer> server_;
242 : : const iox::popo::RequestHeader* last_req_header_{nullptr};
243 : : std::mutex mtx_;
244 : : std::atomic_bool quit_flag_{false};
245 : : std::mutex callback_mtx_;
246 : : };
247 : :
248 : : // ShmClient
249 : : class ShmClient final : public AbstractObject<ShmID>, public std::enable_shared_from_this<ShmClient> {
250 : : public:
251 : : explicit ShmClient(const ShmID& id);
252 : :
253 : : ~ShmClient() override;
254 : :
255 : : std::any get_native_handle() const override;
256 : :
257 : : void process_message();
258 : :
259 : : bool is_connected() const;
260 : :
261 : : void enable_detect_timer();
262 : :
263 : : void disable_detect_timer();
264 : :
265 : : Bytes loan(uint64_t channel, int64_t size);
266 : :
267 : : bool release(const Bytes& bytes);
268 : :
269 : : bool call(NodeImpl* owner, uint64_t channel, const Bytes& req_data, NodeImpl::MsgCallback&& callback = nullptr,
270 : : uint64_t* seq_out = nullptr, bool dispatch = true);
271 : :
272 : : void remove_response_callback(uint64_t seq);
273 : :
274 : : void cancel_calls(NodeImpl* owner);
275 : :
276 : : private:
277 : : void detect_server();
278 : :
279 : : void discovery_server(bool connect);
280 : :
281 : : static void on_response_received(shm::popo::UntypedClient*, ShmClient* target);
282 : :
283 : : std::atomic_bool has_detect_timer_{false};
284 : : std::atomic_bool last_connected_{false};
285 : : std::atomic_bool quit_flag_{false};
286 : : alignas(64) std::atomic<uint64_t> seq_{0};
287 : :
288 : : int32_t domain_{0};
289 : : std::shared_ptr<shm::popo::Listener> listener_;
290 : : std::optional<shm::popo::UntypedClient> client_;
291 : : std::mutex mtx_;
292 : : std::mutex callback_mtx_;
293 : : struct ResponseCallback final {
294 : : NodeImpl* owner{nullptr};
295 : : bool dispatch{true};
296 : : NodeImpl::MsgCallback callback;
297 : : };
298 : : std::unordered_map<uint64_t, ResponseCallback> callbacks_;
299 : : };
300 : :
301 : : // ShmPublisher
302 : : class ShmPublisher final : public AbstractObject<ShmID>, public std::enable_shared_from_this<ShmPublisher> {
303 : : public:
304 : : explicit ShmPublisher(const ShmID& id);
305 : :
306 : : ~ShmPublisher() override;
307 : :
308 : : std::any get_native_handle() const override;
309 : :
310 : : bool has_subscribers() const;
311 : :
312 : : Bytes loan(uint64_t channel, int64_t size);
313 : :
314 : : bool release(const Bytes& bytes);
315 : :
316 : : bool publish(uint64_t channel, const Bytes& bytes);
317 : :
318 : : void enable_detect_timer();
319 : :
320 : : void disable_detect_timer();
321 : :
322 : : private:
323 : : void detect_subscribers();
324 : :
325 : : void discovery_subscribers(bool has_subscribers);
326 : :
327 : : std::atomic_bool has_detect_timer_{false};
328 : : std::atomic_bool last_has_subscribers_{false};
329 : : std::atomic_bool quit_flag_{false};
330 : : alignas(64) std::atomic<uint64_t> seq_{0};
331 : :
332 : : int32_t domain_{0};
333 : : int32_t wait_{0};
334 : : std::optional<shm::popo::UntypedPublisher> pub_;
335 : : std::optional<SysSemaphore> sem_;
336 : : std::mutex mtx_;
337 : : };
338 : :
339 : : // ShmSubscriber
340 : : class ShmSubscriber final : public AbstractObject<ShmID>, public std::enable_shared_from_this<ShmSubscriber> {
341 : : public:
342 : : explicit ShmSubscriber(const ShmID& id);
343 : :
344 : : ~ShmSubscriber() override;
345 : :
346 : : std::any get_native_handle() const override;
347 : :
348 : : bool suspend();
349 : :
350 : : bool resume();
351 : :
352 : : bool is_suspend() const;
353 : :
354 : : void process_message(MessageLoop* dispatched_loop = nullptr);
355 : :
356 : : void subscribe();
357 : :
358 : : void unsubscribe();
359 : :
360 : : void set_latency_and_lost_enabled(bool enable);
361 : :
362 : : bool is_latency_and_lost_enabled() const;
363 : :
364 : : const CalculateSample& get_calculate_sample() const;
365 : :
366 : : private:
367 : : static void on_msg_received(shm::popo::UntypedSubscriber* sub, ShmSubscriber* target);
368 : :
369 : : std::atomic_bool is_suspend_{false};
370 : :
371 : : int32_t domain_{0};
372 : : int32_t wait_{0};
373 : :
374 : : std::shared_ptr<shm::popo::Listener> listener_;
375 : :
376 : : std::optional<shm::popo::UntypedSubscriber> sub_;
377 : :
378 : : std::optional<SysSemaphore> sem_;
379 : :
380 : : std::atomic_bool is_latency_and_lost_enabled_{false};
381 : :
382 : : CalculateSample calc_sample_;
383 : :
384 : : std::atomic_bool quit_flag_{false};
385 : :
386 : : std::mutex callback_mtx_;
387 : : std::mutex receive_mtx_;
388 : : };
389 : :
390 : : } // namespace vlink
|
