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 <utility>
27 : :
28 : : namespace vlink {
29 : :
30 : : constexpr size_t kMaxTaskSize = 10000U;
31 : :
32 : : // IntraPipeline
33 : 547 : size_t IntraPipeline::get_max_task_count() const { return kMaxTaskSize; }
34 : :
35 : : // IntraFactory
36 : 166 : IntraPipeline& IntraFactory::get_pipeline(int32_t pipeline) {
37 [ + - ]: 166 : std::lock_guard lock(pipeline_mtx_);
38 : :
39 [ + - ]: 166 : auto [iter, inserted] = pipeline_map_.try_emplace(pipeline);
40 : :
41 [ + + ]: 166 : if (inserted) {
42 [ + - + - : 15 : iter->second.set_name("INTRA-PIPELINE-" + std::to_string(pipeline));
+ - ]
43 [ + - ]: 15 : iter->second.async_run();
44 : : }
45 : :
46 : 332 : return iter->second;
47 : 166 : }
48 : :
49 [ + - ]: 14 : IntraFactory::IntraFactory() {
50 : 14 : Bytes::init_memory_pool();
51 : :
52 [ - + ]: 14 : if VUNLIKELY (IntraConf::get_thread_count() != 1) {
53 [ # # # # ]: 0 : VLOG_W("IntraFactory: Intra does not support setting thread count.");
54 : : }
55 : 14 : }
56 : :
57 : 14 : IntraFactory::~IntraFactory() {
58 [ + + ]: 29 : for (auto& [num, pipeline] : pipeline_map_) {
59 : 15 : pipeline.quit(true);
60 : 15 : pipeline.wait_for_quit();
61 : : }
62 : :
63 : 14 : pipeline_map_.clear();
64 : 14 : }
65 : :
66 : : // IntraNode
67 : 196 : IntraNode::IntraNode(const IntraID& id) {
68 : 196 : const auto& [impl_type, address, pipeline, type, channel] = id;
69 : :
70 [ + + + - : 196 : static auto& factory = IntraFactory::get();
+ - - - ]
71 : :
72 [ + + ]: 196 : if (type == IntraType::kQueue) {
73 [ + - ]: 166 : pipeline_ = &(factory.get_pipeline(pipeline));
74 : : }
75 : 196 : }
76 : :
77 : 392 : IntraNode::~IntraNode() = default;
78 : :
79 [ # # ]: 0 : std::any IntraNode::get_native_handle() const { return this; }
80 : :
81 : 521 : bool IntraNode::publish(IntraType type, uint32_t channel, const Bytes& msg_data) {
82 [ + + + - ]: 521 : if (type == IntraType::kQueue && pipeline_) {
83 [ + - ]: 500 : std::weak_ptr<IntraNode> weak_self = shared_from_this();
84 : :
85 [ + - + - ]: 1000 : return pipeline_->post_task([weak_self, channel, msg_data]() {
86 : 500 : auto self = weak_self.lock();
87 : :
88 [ + + ]: 500 : if VUNLIKELY (!self) {
89 : 7 : return;
90 : : }
91 : :
92 [ + - + - ]: 493 : self->traverse_msg_callback([channel, &msg_data](NodeImpl* impl, const auto& callback) {
93 : 547 : const auto* conf_ptr = impl->get_target_conf<IntraConf>();
94 : :
95 [ + - + + : 547 : if (conf_ptr->hash_code != channel || impl->has_suspend) {
+ + ]
96 : 1 : return;
97 : : }
98 : :
99 : 546 : callback(msg_data);
100 : : });
101 [ + + ]: 1000 : });
102 : :
103 : 500 : } else {
104 : 21 : bool ok = false;
105 : :
106 [ + - + - ]: 21 : traverse_msg_callback([channel, &msg_data, &ok](NodeImpl* impl, const auto& callback) {
107 : 23 : const auto* conf_ptr = impl->get_target_conf<IntraConf>();
108 : :
109 [ + - + + : 23 : if (conf_ptr->hash_code != channel || impl->has_suspend) {
+ + ]
110 : 2 : return;
111 : : }
112 : :
113 : 21 : callback(msg_data);
114 : :
115 : 21 : ok = true;
116 : : });
117 : :
118 : 21 : return ok;
119 : : }
120 : : }
121 : :
122 : 18 : bool IntraNode::publish(IntraType type, uint32_t channel, const IntraData& intra_data) {
123 [ + - + - ]: 18 : if (type == IntraType::kQueue && pipeline_) {
124 [ + - ]: 18 : std::weak_ptr<IntraNode> weak_self = shared_from_this();
125 : :
126 [ + - + - ]: 36 : return pipeline_->post_task([weak_self, channel, intra_data]() {
127 : 18 : auto self = weak_self.lock();
128 : :
129 [ - + ]: 18 : if VUNLIKELY (!self) {
130 : 0 : return;
131 : : }
132 : :
133 [ + - + - ]: 18 : self->traverse_intra_msg_callback([channel, &intra_data](NodeImpl* impl, const auto& callback) {
134 : 18 : const auto* conf_ptr = impl->get_target_conf<IntraConf>();
135 : :
136 [ + - - + : 18 : if (conf_ptr->hash_code != channel || impl->has_suspend) {
- + ]
137 : 0 : return;
138 : : }
139 : :
140 : 18 : callback(intra_data);
141 : : });
142 [ + - ]: 36 : });
143 : :
144 : 18 : } else {
145 : 0 : bool ok = false;
146 : :
147 [ # # # # ]: 0 : traverse_intra_msg_callback([channel, &intra_data, &ok](NodeImpl* impl, const auto& callback) {
148 : 0 : const auto* conf_ptr = impl->get_target_conf<IntraConf>();
149 : :
150 [ # # # # : 0 : if (conf_ptr->hash_code != channel || impl->has_suspend) {
# # ]
151 : 0 : return;
152 : : }
153 : :
154 : 0 : callback(intra_data);
155 : :
156 : 0 : ok = true;
157 : : });
158 : :
159 : 0 : return ok;
160 : : }
161 : : }
162 : :
163 : 46 : bool IntraNode::call(NodeImpl* requester, IntraType type, uint32_t channel, const Bytes& req_data,
164 : : NodeImpl::MsgCallback&& callback, bool inline_if_same_thread) {
165 [ + + + - : 46 : if VLIKELY (type == IntraType::kQueue && pipeline_ && !(inline_if_same_thread && pipeline_->is_in_same_thread())) {
+ + + + +
+ + + +
+ ]
166 [ + - ]: 29 : std::weak_ptr<IntraNode> weak_self = shared_from_this();
167 : :
168 [ + - + - ]: 58 : return pipeline_->post_task([weak_self, requester, channel, req_data, callback = std::move(callback)]() {
169 : 29 : auto self = weak_self.lock();
170 : :
171 [ - + ]: 29 : if VUNLIKELY (!self) {
172 : 0 : return;
173 : : }
174 : :
175 [ + - ]: 58 : self->traverse_req_resp_callback(
176 [ + - ]: 29 : [&self, requester, channel, &req_data, &callback](NodeImpl* impl, const auto& callback2) {
177 [ - + ]: 28 : if VUNLIKELY (!self->is_contains_impl(requester)) {
178 : 0 : self->ignore_called();
179 : 0 : return;
180 : : }
181 : :
182 : 28 : const auto* conf_ptr = impl->get_target_conf<IntraConf>();
183 : :
184 [ + - + + : 28 : if (conf_ptr->hash_code != channel || impl->has_suspend) {
+ + ]
185 : 1 : self->ignore_called();
186 : 1 : return;
187 : : }
188 : :
189 [ - + ]: 27 : if VUNLIKELY (self->has_called()) {
190 [ # # # # ]: 0 : VLOG_F(*conf_ptr, "Two identical service requests.");
191 : 0 : return;
192 : : }
193 : :
194 [ + + ]: 27 : if (callback) {
195 : 18 : Bytes bytes;
196 [ + - ]: 18 : callback2(0, req_data, &bytes);
197 [ + - ]: 18 : callback(bytes);
198 : 18 : } else {
199 : 9 : callback2(0, req_data, nullptr);
200 : : }
201 : : });
202 [ + - ]: 58 : });
203 : 29 : } else {
204 : 17 : bool ok = false;
205 : :
206 [ + - + - ]: 17 : traverse_req_resp_callback([this, channel, &req_data, &callback, &ok](NodeImpl* impl, const auto& callback2) {
207 : 16 : const auto* conf_ptr = impl->get_target_conf<IntraConf>();
208 : :
209 [ + - + + : 16 : if (conf_ptr->hash_code != channel || impl->has_suspend) {
+ + ]
210 : 1 : ignore_called();
211 : 1 : return;
212 : : }
213 : :
214 [ - + ]: 15 : if VUNLIKELY (has_called()) {
215 [ # # # # ]: 0 : VLOG_F(*conf_ptr, "Two identical service requests.");
216 : 0 : return;
217 : : }
218 : :
219 [ + + ]: 15 : if (callback) {
220 : 11 : Bytes bytes;
221 [ + - ]: 11 : callback2(0, req_data, &bytes);
222 [ + - ]: 11 : callback(bytes);
223 : 11 : } else {
224 : 4 : callback2(0, req_data, nullptr);
225 : : }
226 : :
227 : 15 : ok = true;
228 : : });
229 : 17 : return ok;
230 : : }
231 : : }
232 : :
233 : : } // namespace vlink
|
