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