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 "./intra_factory.h"
25 : :
26 : : #include <algorithm>
27 : : #include <type_traits>
28 : : #include <utility>
29 : :
30 : : #include "./base/memory_resource.h"
31 : : #include "./base/utils.h"
32 : :
33 : : namespace vlink {
34 : :
35 : : constexpr size_t kMaxTaskSize = 10000U;
36 : :
37 : : // IntraPipeline
38 : 547 : size_t IntraPipeline::get_max_task_count() const { return kMaxTaskSize; }
39 : :
40 : : // IntraFactory
41 : 174 : IntraPipeline& IntraFactory::get_pipeline(int32_t pipeline) {
42 [ + - ]: 174 : std::lock_guard lock(pipeline_mtx_);
43 : :
44 [ + - ]: 174 : auto [iter, inserted] = pipeline_map_.try_emplace(pipeline);
45 : :
46 [ + + ]: 174 : if (inserted) {
47 [ + - + - : 16 : iter->second.set_name("INTRA-PIPELINE-" + std::to_string(pipeline));
+ - ]
48 [ + - ]: 16 : iter->second.async_run();
49 : : }
50 : :
51 : 348 : return iter->second;
52 : 174 : }
53 : :
54 [ + - ]: 14 : IntraFactory::IntraFactory() {
55 : 14 : Bytes::init_memory_pool();
56 : :
57 [ - + ]: 14 : if VUNLIKELY (IntraConf::get_thread_count() != 1) {
58 [ # # # # : 0 : VLOG_W("IntraFactory: Intra does not support setting thread count.");
# # ]
59 : : }
60 : 14 : }
61 : :
62 : 14 : IntraFactory::~IntraFactory() {
63 : : #ifdef _WIN32
64 : : if (Utils::is_terminating()) {
65 : : return;
66 : : }
67 : : #endif
68 : :
69 [ + + ]: 30 : for (auto& [num, pipeline] : pipeline_map_) {
70 : 16 : pipeline.quit(true);
71 : 16 : pipeline.wait_for_quit();
72 : : }
73 : :
74 : 14 : pipeline_map_.clear();
75 : 14 : }
76 : :
77 : : // IntraNode
78 : 226 : IntraNode::IntraNode(const IntraID& id) {
79 : 226 : const auto& [impl_type, address, pipeline, type, channel] = id;
80 : :
81 [ + + + - : 226 : static auto& factory = IntraFactory::get();
+ - - - ]
82 : :
83 [ + + ]: 226 : if (type == IntraType::kQueue) {
84 [ + - ]: 174 : pipeline_ = &(factory.get_pipeline(pipeline));
85 : : }
86 : 226 : }
87 : :
88 : 452 : IntraNode::~IntraNode() = default;
89 : :
90 [ # # ]: 0 : std::any IntraNode::get_native_handle() const { return this; }
91 : :
92 : : template <typename DataT>
93 : 518 : void IntraNode::deliver_data(MessageLoop* target_loop, uint32_t channel, const DataT& data) {
94 : 3383 : auto deliver = [this, target_loop, channel, &data](NodeImpl* impl, const auto& callback) {
95 : 576 : const auto* conf = impl->get_target_conf<IntraConf>();
96 : 576 : auto* loop = impl->get_message_loop();
97 : :
98 [ + + + + ]: 576 : if (!loop) {
99 : 567 : loop = pipeline_;
100 : : }
101 : :
102 [ + - - + : 576 : if VUNLIKELY (conf->hash_code != channel || impl->has_suspend || loop != target_loop) {
+ - + + +
+ + - + +
+ + + + +
+ ]
103 : 5 : return;
104 : : }
105 : :
106 : 571 : callback(data);
107 : : };
108 : :
109 : : if constexpr (std::is_same_v<DataT, Bytes>) {
110 [ + - + - ]: 498 : traverse_msg_callback(deliver);
111 : : } else {
112 [ + - + - ]: 20 : traverse_intra_msg_callback(deliver);
113 : : }
114 : 518 : }
115 : :
116 : : template <typename DataT>
117 : 570 : bool IntraNode::publish_data(IntraType type, uint32_t channel, const DataT& data) {
118 : 570 : MessageLoop* first_loop = nullptr;
119 : 570 : std::vector<MessageLoop*> other_loops;
120 : 571 : bool delivered = false;
121 : :
122 : 4040 : auto route = [this, channel, &data, &first_loop, &other_loops, &delivered](NodeImpl* impl, const auto& callback) {
123 [ + - + - ]: 695 : const auto* conf = impl->get_target_conf<IntraConf>();
124 : :
125 [ + - - + : 695 : if VUNLIKELY (conf->hash_code != channel || impl->has_suspend) {
- + + - +
+ + + ]
126 : 3 : return;
127 : : }
128 : :
129 [ + - + - ]: 692 : auto* loop = impl->get_message_loop();
130 : :
131 [ + + + + ]: 692 : if (!loop) {
132 : 687 : loop = pipeline_;
133 : : }
134 : :
135 [ - + + + ]: 692 : if (!loop) {
136 [ - - + - ]: 51 : callback(data);
137 : 51 : delivered = true;
138 [ + + + + ]: 641 : } else if (!first_loop) {
139 : 514 : first_loop = loop;
140 [ + - + - : 127 : } else if (loop != first_loop && std::find(other_loops.begin(), other_loops.end(), loop) == other_loops.end()) {
+ - + - +
+ + - + -
+ + ]
141 [ + - + - ]: 2 : other_loops.push_back(loop);
142 : : }
143 : : };
144 : :
145 : : if constexpr (std::is_same_v<DataT, Bytes>) {
146 [ + - + - ]: 552 : traverse_msg_callback(route);
147 : : } else {
148 [ + - + - ]: 19 : traverse_intra_msg_callback(route);
149 : : }
150 : :
151 [ + + ]: 571 : if (!first_loop) {
152 [ + + + + ]: 57 : if (delivered || type != IntraType::kQueue) {
153 : 49 : return delivered;
154 : : }
155 : :
156 : 8 : first_loop = pipeline_;
157 : : }
158 : :
159 : 522 : auto weak_self = weak_from_this();
160 : :
161 [ + + ]: 522 : if (other_loops.empty()) {
162 [ + - + - ]: 2074 : return first_loop->post_task([weak_self, first_loop, channel, data]() {
163 : 520 : auto self = weak_self.lock();
164 : :
165 [ - + + + ]: 520 : if VUNLIKELY (!self) {
166 : 6 : return;
167 : : }
168 : :
169 [ + - + - ]: 514 : self->deliver_data(first_loop, channel, data);
170 [ + - + + ]: 1040 : });
171 : : }
172 : :
173 : : if constexpr (std::is_same_v<DataT, Bytes>) {
174 [ + - ]: 1 : auto shared_data = MemoryResource::make_shared<Bytes>(data);
175 : 5 : auto post = [&weak_self, channel, &shared_data](MessageLoop* loop) {
176 [ + - + - ]: 4 : return loop->post_task([weak_self, loop, channel, shared_data]() {
177 : 2 : auto self = weak_self.lock();
178 : :
179 [ - + ]: 2 : if VUNLIKELY (!self) {
180 : 0 : return;
181 : : }
182 : :
183 [ + - ]: 2 : self->deliver_data(loop, channel, *shared_data);
184 [ + - ]: 6 : });
185 : : };
186 : :
187 [ + - ]: 1 : bool result = post(first_loop);
188 : :
189 [ + + ]: 2 : for (auto* loop : other_loops) {
190 [ + - + - : 1 : result = post(loop) && result;
+ - ]
191 : : }
192 : :
193 : 1 : return result;
194 : 1 : } else {
195 : 5 : auto post = [&weak_self, channel, &data](MessageLoop* loop) {
196 [ + - + - ]: 4 : return loop->post_task([weak_self, loop, channel, data]() {
197 : 2 : auto self = weak_self.lock();
198 : :
199 [ - + ]: 2 : if VUNLIKELY (!self) {
200 : 0 : return;
201 : : }
202 : :
203 [ + - ]: 2 : self->deliver_data(loop, channel, data);
204 [ + - ]: 6 : });
205 : : };
206 : :
207 [ + - ]: 1 : bool result = post(first_loop);
208 : :
209 [ + + ]: 2 : for (auto* loop : other_loops) {
210 [ + - + - : 1 : result = post(loop) && result;
+ - ]
211 : : }
212 : :
213 : 1 : return result;
214 : : }
215 : 571 : }
216 : :
217 : 551 : bool IntraNode::publish(IntraType type, uint32_t channel, const Bytes& msg_data) {
218 : 551 : return publish_data(type, channel, msg_data);
219 : : }
220 : :
221 : 19 : bool IntraNode::publish(IntraType type, uint32_t channel, const IntraData& intra_data) {
222 : 19 : return publish_data(type, channel, intra_data);
223 : : }
224 : :
225 : 91 : void IntraNode::deliver_request(NodeImpl* server, NodeImpl* requester, MessageLoop* target_loop, const Bytes& request,
226 : : const NodeImpl::MsgCallback& callback) {
227 : 91 : Bytes response;
228 : 91 : bool replied = false;
229 : :
230 [ + - + - ]: 91 : invoke_req_resp_callback(server, [&, this](NodeImpl* impl, const auto& handler) {
231 : 91 : auto* loop = impl->get_message_loop();
232 : :
233 [ + + ]: 91 : if (!loop) {
234 : 85 : loop = pipeline_;
235 : : }
236 : :
237 [ + - - + : 91 : if VUNLIKELY (!is_contains_impl(requester) || impl->has_suspend || loop != target_loop) {
+ - - + -
+ ]
238 : 0 : return;
239 : : }
240 : :
241 [ + + ]: 91 : if (callback) {
242 : 78 : handler(0, request, &response);
243 : 78 : replied = true;
244 : : } else {
245 : 13 : handler(0, request, nullptr);
246 : : }
247 : : });
248 : :
249 [ + + ]: 91 : if (replied) {
250 [ + - ]: 78 : callback(response);
251 : : }
252 : 91 : }
253 : :
254 : 97 : bool IntraNode::call(NodeImpl* requester, IntraType type, uint32_t channel, const Bytes& req_data,
255 : : NodeImpl::MsgCallback&& callback, bool inline_if_same_thread) {
256 : 97 : auto weak_self = weak_from_this();
257 : :
258 [ + + ]: 97 : if (callback) {
259 : 80 : auto response_callback = std::move(callback);
260 : :
261 : 78 : callback = [weak_self, requester, inline_if_same_thread,
262 : 167 : response_callback = std::move(response_callback)](const Bytes& response) mutable {
263 : 78 : auto self = weak_self.lock();
264 : :
265 [ - + ]: 78 : if VUNLIKELY (!self) {
266 : 0 : return;
267 : : }
268 : :
269 : 78 : MessageLoop* loop = nullptr;
270 : :
271 [ + - + - ]: 78 : self->invoke_callback(requester, [&]() {
272 : 78 : loop = requester->get_message_loop();
273 : :
274 [ + + + + ]: 78 : if (!loop || inline_if_same_thread) {
275 : 75 : response_callback(response);
276 : : }
277 : 78 : });
278 : :
279 [ + + + + ]: 78 : if (loop && !inline_if_same_thread) {
280 [ + - + - ]: 3 : loop->post_task([weak_self, requester, loop, response, response_callback = std::move(response_callback)]() {
281 : 3 : auto self = weak_self.lock();
282 : :
283 [ - + ]: 3 : if VUNLIKELY (!self) {
284 : 0 : return;
285 : : }
286 : :
287 : 3 : bool attached = false;
288 [ + - + - ]: 6 : self->invoke_callback(requester, [&]() { attached = requester->get_message_loop() == loop; });
289 : :
290 [ + - ]: 3 : if (attached) {
291 [ + - ]: 3 : response_callback(response);
292 : : }
293 [ + - ]: 3 : });
294 : : }
295 [ + - + - ]: 158 : };
296 : 80 : }
297 : :
298 : 97 : bool found = false;
299 : 97 : bool result = false;
300 : 97 : NodeImpl* server = nullptr;
301 : 97 : MessageLoop* target_loop = nullptr;
302 : :
303 [ + - + - ]: 97 : traverse_req_resp_callback([&](NodeImpl* impl, const auto&) {
304 : 94 : const auto* conf = impl->get_target_conf<IntraConf>();
305 : :
306 [ + - + + : 94 : if VUNLIKELY (conf->hash_code != channel || impl->has_suspend) {
+ + ]
307 : 3 : return;
308 : : }
309 : :
310 [ - + ]: 91 : if VUNLIKELY (found) {
311 [ # # # # : 0 : VLOG_F(*conf, "Two identical service requests.");
# # ]
312 : 0 : return;
313 : : }
314 : :
315 : 91 : found = true;
316 : 91 : server = impl;
317 : 91 : target_loop = impl->get_message_loop();
318 : :
319 [ + + ]: 91 : if (!target_loop) {
320 : 85 : target_loop = pipeline_;
321 : : }
322 : : });
323 : :
324 [ + + ]: 97 : if (server) {
325 [ + + + + : 91 : if (!target_loop || (inline_if_same_thread && target_loop->is_in_same_thread())) {
+ - + + +
+ ]
326 [ + - ]: 60 : deliver_request(server, requester, target_loop, req_data, callback);
327 : 60 : result = true;
328 : : } else {
329 [ + - + - ]: 62 : result = target_loop->post_task(
330 : 31 : [weak_self, server, requester, target_loop, req_data, callback = std::move(callback)]() {
331 : 31 : auto self = weak_self.lock();
332 : :
333 [ - + ]: 31 : if VUNLIKELY (!self) {
334 : 0 : return;
335 : : }
336 : :
337 [ + - ]: 31 : self->deliver_request(server, requester, target_loop, req_data, callback);
338 [ + - ]: 31 : });
339 : : }
340 : : }
341 : :
342 [ + + + + ]: 97 : if (!found && type == IntraType::kQueue) {
343 [ + - + - ]: 4 : return pipeline_->post_task([weak_self, requester, channel, req_data, callback = std::move(callback)]() mutable {
344 [ + - + - : 2 : if (auto self = weak_self.lock(); self && self->is_contains_impl(requester)) {
+ - + - ]
345 [ + - ]: 2 : self->call(requester, IntraType::kDirect, channel, req_data, std::move(callback), true);
346 : 2 : }
347 : 4 : });
348 : : }
349 : :
350 : 95 : return result;
351 : 97 : }
352 : :
353 : : } // namespace vlink
|
