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>;
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 : 261 : static constexpr size_t get_loaned_offset() {
97 : : #if SHM_USE_CUSTOM_SEQ
98 : 261 : return sizeof(uint64_t) + sizeof(uint64_t);
99 : : #else
100 : : return sizeof(uint64_t);
101 : : #endif
102 : : }
103 : :
104 : 245 : static constexpr size_t get_loaned_alignment() { return sizeof(uint64_t); }
105 : :
106 : 8 : static void write_header(uint8_t* loaned_ptr, uint64_t channel, uint64_t seq) {
107 : : #if SHM_USE_CUSTOM_SEQ
108 : 8 : std::memcpy(loaned_ptr, &channel, sizeof(uint64_t));
109 : 8 : 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 : 8 : }
115 : :
116 : 234 : static void write_data(uint8_t* loaned_ptr, uint64_t channel, uint64_t seq, const Bytes& data) {
117 : : #if SHM_USE_CUSTOM_SEQ
118 : 234 : std::memcpy(loaned_ptr, &channel, sizeof(uint64_t));
119 : 234 : std::memcpy(loaned_ptr + sizeof(uint64_t), &seq, sizeof(uint64_t));
120 : 234 : 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 : 233 : }
127 : :
128 : 215 : 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 [ - + ]: 215 : if VUNLIKELY (payload_size < sizeof(uint64_t) + sizeof(uint64_t)) {
133 : 0 : return;
134 : : }
135 : :
136 : 215 : std::memcpy(&channel, loaned_ptr, sizeof(uint64_t));
137 : 215 : std::memcpy(&seq, loaned_ptr + sizeof(uint64_t), sizeof(uint64_t));
138 : 430 : data = Bytes::shallow_copy(loaned_ptr + sizeof(uint64_t) + sizeof(uint64_t),
139 : 215 : 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 = false);
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 : : shm::popo::Listener* get_listener(int32_t domain = 0);
174 : :
175 : : void try_to_destroy_listener(int32_t domain = 0, shm::popo::Listener* listener = nullptr);
176 : :
177 : : void add_detect_callback(void* node, DetectCallback&& callback);
178 : :
179 : : void remove_detect_callback(void* node);
180 : :
181 : : void start_detect_node_count();
182 : :
183 : : uint64_t get_publisher_count(const shm::capro::ServiceDescription& description);
184 : :
185 : : uint64_t get_subscriber_count(const shm::capro::ServiceDescription& description);
186 : :
187 : : int get_sub_depth() const;
188 : :
189 : : private:
190 : : std::unordered_map<int32_t, std::shared_ptr<shm::popo::Listener>> listener_map_;
191 : : MessageLoop message_loop_{MessageLoop::kNormalType};
192 : : Timer detect_timer_;
193 : : std::unordered_map<void*, DetectCallback> detect_map_;
194 : : std::shared_mutex detect_mtx_;
195 : : std::mutex listener_mtx_;
196 : : std::unordered_map<AbstractNode*, DiscoveryCallback> discovery_map_;
197 : : int sub_depth_{kDefaultSubDepth};
198 : :
199 : : std::optional<shm::popo::Subscriber<shm::roudi::PortIntrospectionFieldTopic>> port_sub_;
200 : : iox::roudi::PortIntrospectionFieldTopic topic_list_;
201 : : std::shared_mutex topic_mtx_;
202 : :
203 [ + + + - : 244 : VLINK_SINGLETON_DECLARE(ShmFactory)
+ - - - ]
204 : : };
205 : :
206 : : // ShmServer
207 : : class ShmServer final : public AbstractObject<ShmID>, public std::enable_shared_from_this<ShmServer> {
208 : : public:
209 : : explicit ShmServer(const ShmID& id);
210 : :
211 : : ~ShmServer() override;
212 : :
213 : : std::any get_native_handle() const override;
214 : :
215 : : bool suspend();
216 : :
217 : : bool resume();
218 : :
219 : : bool is_suspend() const;
220 : :
221 : : void process_message();
222 : :
223 : : void start();
224 : :
225 : : void stop();
226 : :
227 : : bool has_clients() const;
228 : :
229 : : Bytes loan(uint64_t channel, int64_t size);
230 : :
231 : : bool release(const Bytes& bytes);
232 : :
233 : : bool reply(uint64_t channel, const Bytes& resp_data);
234 : :
235 : : private:
236 : : static void on_request_received(shm::popo::UntypedServer* server, ShmServer* target);
237 : :
238 : : alignas(64) std::atomic<uint64_t> seq_{0};
239 : : std::atomic_bool is_suspend_{false};
240 : :
241 : : int32_t domain_{0};
242 : : shm::popo::Listener* listener_{nullptr};
243 : : std::optional<shm::popo::UntypedServer> server_;
244 : : const iox::popo::RequestHeader* last_req_header_{nullptr};
245 : : std::mutex mtx_;
246 : : std::atomic_bool quit_flag_{false};
247 : : std::mutex callback_mtx_;
248 : : };
249 : :
250 : : // ShmClient
251 : : class ShmClient final : public AbstractObject<ShmID>, public std::enable_shared_from_this<ShmClient> {
252 : : public:
253 : : explicit ShmClient(const ShmID& id);
254 : :
255 : : ~ShmClient() override;
256 : :
257 : : std::any get_native_handle() const override;
258 : :
259 : : void process_message();
260 : :
261 : : bool is_connected() const;
262 : :
263 : : void enable_detect_timer();
264 : :
265 : : void disable_detect_timer();
266 : :
267 : : Bytes loan(uint64_t channel, int64_t size);
268 : :
269 : : bool release(const Bytes& bytes);
270 : :
271 : : bool call(uint64_t channel, const Bytes& req_data, NodeImpl::MsgCallback&& callback = nullptr,
272 : : uint64_t* seq_out = nullptr);
273 : :
274 : : void remove_response_callback(uint64_t seq);
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 : : shm::popo::Listener* listener_{nullptr};
290 : : std::optional<shm::popo::UntypedClient> client_;
291 : : std::mutex mtx_;
292 : : std::mutex callback_mtx_;
293 : : std::unordered_map<uint64_t, Function<void(uint64_t, const Bytes&)>> callbacks_;
294 : : };
295 : :
296 : : // ShmPublisher
297 : : class ShmPublisher final : public AbstractObject<ShmID>, public std::enable_shared_from_this<ShmPublisher> {
298 : : public:
299 : : explicit ShmPublisher(const ShmID& id);
300 : :
301 : : ~ShmPublisher() override;
302 : :
303 : : std::any get_native_handle() const override;
304 : :
305 : : bool has_subscribers() const;
306 : :
307 : : Bytes loan(uint64_t channel, int64_t size);
308 : :
309 : : bool release(const Bytes& bytes);
310 : :
311 : : bool publish(uint64_t channel, const Bytes& bytes);
312 : :
313 : : void enable_detect_timer();
314 : :
315 : : void disable_detect_timer();
316 : :
317 : : private:
318 : : void detect_subscribers();
319 : :
320 : : void discovery_subscribers(bool has_subscribers);
321 : :
322 : : std::atomic_bool has_detect_timer_{false};
323 : : std::atomic_bool last_has_subscribers_{false};
324 : : std::atomic_bool quit_flag_{false};
325 : : alignas(64) std::atomic<uint64_t> seq_{0};
326 : :
327 : : int32_t domain_{0};
328 : : int32_t wait_{0};
329 : : std::optional<shm::popo::UntypedPublisher> pub_;
330 : : std::optional<SysSemaphore> sem_;
331 : : };
332 : :
333 : : // ShmSubscriber
334 : : class ShmSubscriber final : public AbstractObject<ShmID>, public std::enable_shared_from_this<ShmSubscriber> {
335 : : public:
336 : : explicit ShmSubscriber(const ShmID& id);
337 : :
338 : : ~ShmSubscriber() override;
339 : :
340 : : std::any get_native_handle() const override;
341 : :
342 : : bool suspend();
343 : :
344 : : bool resume();
345 : :
346 : : bool is_suspend() const;
347 : :
348 : : void process_message();
349 : :
350 : : void subscribe();
351 : :
352 : : void unsubscribe();
353 : :
354 : : void set_manual_unloan(bool manual_unloan);
355 : :
356 : : bool release(const Bytes& bytes);
357 : :
358 : : void set_latency_and_lost_enabled(bool enable);
359 : :
360 : : bool is_latency_and_lost_enabled() const;
361 : :
362 : : const CalculateSample& get_calculate_sample() const;
363 : :
364 : : private:
365 : : static void on_msg_received(shm::popo::UntypedSubscriber* sub, ShmSubscriber* target);
366 : :
367 : : std::atomic_bool is_suspend_{false};
368 : :
369 : : int32_t domain_{0};
370 : : int32_t wait_{0};
371 : :
372 : : shm::popo::Listener* listener_{nullptr};
373 : :
374 : : std::optional<shm::popo::UntypedSubscriber> sub_;
375 : :
376 : : std::optional<SysSemaphore> sem_;
377 : :
378 : : std::atomic_bool is_latency_and_lost_enabled_{false};
379 : :
380 : : CalculateSample calc_sample_;
381 : :
382 : : std::atomic_bool manual_unloan_{false};
383 : :
384 : : std::atomic_bool quit_flag_{false};
385 : :
386 : : std::mutex callback_mtx_;
387 : : };
388 : :
389 : : } // namespace vlink
|
