Line data Source code
1 : /*
2 : * Copyright (C) 2004-2026 Savoir-faire Linux Inc.
3 : *
4 : * This program is free software: you can redistribute it and/or modify
5 : * it under the terms of the GNU General Public License as published by
6 : * the Free Software Foundation, either version 3 of the License, or
7 : * (at your option) any later version.
8 : *
9 : * This program is distributed in the hope that it will be useful,
10 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 : * GNU General Public License for more details.
13 : *
14 : * You should have received a copy of the GNU General Public License
15 : * along with this program. If not, see <https://www.gnu.org/licenses/>.
16 : */
17 :
18 : #include "swarm_manager.h"
19 : #include <dhtnet/multiplexed_socket.h>
20 : #include <dhtnet/channel_utils.h>
21 : #include <opendht/thread_pool.h>
22 :
23 : namespace jami {
24 :
25 : using namespace swarm_protocol;
26 :
27 593 : SwarmManager::SwarmManager(const NodeId& id, bool isMobile, const std::mt19937_64& rand, ToConnectCb&& toConnectCb)
28 593 : : id_(id)
29 593 : , isMobile_(isMobile)
30 593 : , rd(rand)
31 593 : , toConnectCb_(toConnectCb)
32 : {
33 593 : routing_table.setId(id);
34 593 : }
35 :
36 593 : SwarmManager::~SwarmManager()
37 : {
38 593 : if (!isShutdown_)
39 239 : shutdown();
40 593 : }
41 :
42 : bool
43 2027 : SwarmManager::setKnownNodes(const std::vector<NodeId>& known_nodes)
44 : {
45 2027 : isShutdown_ = false;
46 2023 : std::vector<NodeId> newNodes;
47 : {
48 2023 : std::lock_guard lock(mutex);
49 4779 : for (const auto& nodeId : known_nodes) {
50 2742 : if (addKnownNode(nodeId)) {
51 692 : newNodes.emplace_back(nodeId);
52 : }
53 : }
54 2028 : }
55 :
56 2029 : if (newNodes.empty())
57 1483 : return false;
58 :
59 549 : dht::ThreadPool::io().run([w = weak(), newNodes = std::move(newNodes)] {
60 550 : auto shared = w.lock();
61 548 : if (!shared)
62 0 : return;
63 : // If we detect a new node which already got a TCP link
64 : // we can use it to speed-up the bootstrap (because opening
65 : // a new channel will be easy)
66 548 : std::set<NodeId> toConnect;
67 1241 : for (const auto& nodeId : newNodes) {
68 694 : if (shared->toConnectCb_ && shared->toConnectCb_(nodeId))
69 139 : toConnect.emplace(nodeId);
70 : }
71 549 : shared->maintainBuckets(toConnect);
72 549 : });
73 550 : return true;
74 2033 : }
75 :
76 : void
77 1270 : SwarmManager::setMobileNodes(const std::vector<NodeId>& mobile_nodes)
78 : {
79 : {
80 1270 : std::lock_guard lock(mutex);
81 1286 : for (const auto& nodeId : mobile_nodes)
82 14 : addMobileNodes(nodeId);
83 1273 : }
84 1274 : }
85 :
86 : void
87 2269 : SwarmManager::addChannel(const std::shared_ptr<dhtnet::ChannelSocketInterface>& channel)
88 : {
89 : // JAMI_WARNING("[SwarmManager {}] addChannel! with {}", fmt::ptr(this), channel->deviceId().to_view());
90 2269 : if (channel) {
91 2269 : auto emit = false;
92 : {
93 2269 : std::lock_guard lock(mutex);
94 2269 : emit = routing_table.findBucket(getId())->isEmpty();
95 2268 : auto bucket = routing_table.findBucket(channel->deviceId());
96 2268 : if (routing_table.addNode(channel, bucket)) {
97 1308 : std::error_code ec;
98 1308 : resetNodeExpiry(ec, channel, id_);
99 : }
100 2269 : }
101 2269 : receiveMessage(channel);
102 2268 : if (emit && onConnectionChanged_) {
103 : // If it's the first channel we add, we're now connected!
104 376 : JAMI_DEBUG("[SwarmManager {}] Bootstrap: Connected!", fmt::ptr(this));
105 376 : onConnectionChanged_(true);
106 : }
107 : }
108 2268 : }
109 :
110 : void
111 648 : SwarmManager::removeNode(const NodeId& nodeId)
112 : {
113 648 : std::unique_lock lk(mutex);
114 649 : if (isConnectedWith(nodeId)) {
115 519 : removeNodeInternal(nodeId);
116 518 : lk.unlock();
117 517 : maintainBuckets();
118 : }
119 650 : }
120 :
121 : void
122 210 : SwarmManager::changeMobility(const NodeId& nodeId, bool isMobile)
123 : {
124 210 : std::lock_guard lock(mutex);
125 210 : auto bucket = routing_table.findBucket(nodeId);
126 209 : bucket->changeMobility(nodeId, isMobile);
127 210 : }
128 :
129 : bool
130 1215 : SwarmManager::isConnectedWith(const NodeId& deviceId)
131 : {
132 1215 : return routing_table.hasNode(deviceId);
133 : }
134 :
135 : void
136 635 : SwarmManager::shutdown()
137 : {
138 635 : if (isShutdown_) {
139 16 : return;
140 : }
141 619 : isShutdown_ = true;
142 619 : std::lock_guard lock(mutex);
143 619 : routing_table.shutdownAllNodes();
144 619 : }
145 :
146 : void
147 16 : SwarmManager::restart()
148 : {
149 16 : isShutdown_ = false;
150 16 : }
151 :
152 : bool
153 3309 : SwarmManager::addKnownNode(const NodeId& nodeId)
154 : {
155 3309 : return routing_table.addKnownNode(nodeId);
156 : }
157 :
158 : void
159 14 : SwarmManager::addMobileNodes(const NodeId& nodeId)
160 : {
161 14 : if (id_ != nodeId) {
162 13 : routing_table.addMobileNode(nodeId);
163 : }
164 14 : }
165 :
166 : void
167 1093 : SwarmManager::maintainBuckets(const std::set<NodeId>& toConnect)
168 : {
169 1093 : std::set<NodeId> nodes = toConnect;
170 1091 : std::unique_lock lock(mutex);
171 1093 : auto& buckets = routing_table.getBuckets();
172 3977 : for (auto it = buckets.begin(); it != buckets.end(); ++it) {
173 2884 : auto& bucket = *it;
174 2878 : bool myBucket = routing_table.contains(it, id_);
175 4679 : auto connecting_nodes = myBucket ? bucket.getConnectingNodesSize()
176 1795 : : bucket.getConnectingNodesSize() + bucket.getNodesSize();
177 2885 : if (connecting_nodes < Bucket::BUCKET_MAX_SIZE) {
178 1507 : auto nodesToTry = bucket.getKnownNodesRandom(Bucket::BUCKET_MAX_SIZE - connecting_nodes, rd);
179 2325 : for (auto& node : nodesToTry)
180 818 : routing_table.addConnectingNode(node);
181 :
182 1505 : nodes.insert(nodesToTry.begin(), nodesToTry.end());
183 1508 : }
184 : }
185 1092 : lock.unlock();
186 1926 : for (const auto& node : nodes)
187 831 : tryConnect(node);
188 1094 : }
189 :
190 : void
191 1308 : SwarmManager::sendRequest(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket,
192 : const NodeId& nodeId,
193 : Query q,
194 : int numberNodes)
195 : {
196 1308 : dht::ThreadPool::io().run([socket, isMobile = isMobile_, nodeId, q, numberNodes] {
197 1308 : msgpack::sbuffer buffer;
198 1308 : msgpack::packer<msgpack::sbuffer> pk(&buffer);
199 1308 : Message msg;
200 1308 : msg.is_mobile = isMobile;
201 1308 : msg.request = Request {q, numberNodes, nodeId};
202 1307 : pk.pack(msg);
203 :
204 1305 : std::error_code ec;
205 1304 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
206 1308 : if (ec) {
207 5 : JAMI_ERROR("{}", ec.message());
208 : }
209 1308 : });
210 1308 : }
211 :
212 : void
213 1270 : SwarmManager::sendAnswer(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket, const Message& msg_)
214 : {
215 1270 : std::lock_guard lock(mutex);
216 :
217 1276 : if (msg_.request->q == Query::FIND) {
218 1271 : auto nodes = routing_table.closestNodes(msg_.request->nodeId, msg_.request->num);
219 1274 : auto bucket = routing_table.findBucket(msg_.request->nodeId);
220 1277 : const auto& m_nodes = bucket->getMobileNodes();
221 2549 : Response toResponse {Query::FOUND, nodes, {m_nodes.begin(), m_nodes.end()}};
222 :
223 1273 : Message msg;
224 1273 : msg.is_mobile = isMobile_;
225 1273 : msg.response = std::move(toResponse);
226 :
227 1274 : msgpack::sbuffer buffer((size_t) 60000);
228 1277 : msgpack::packer<msgpack::sbuffer> pk(&buffer);
229 1276 : pk.pack(msg);
230 :
231 1275 : std::error_code ec;
232 1269 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
233 1277 : if (ec) {
234 4 : JAMI_ERROR("{}", ec.message());
235 4 : return;
236 : }
237 1289 : }
238 1277 : }
239 :
240 : void
241 2269 : SwarmManager::receiveMessage(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket)
242 : {
243 4538 : socket->setOnRecv(dhtnet::buildMsgpackReader<Message>(
244 4538 : [w = weak(), wsocket = std::weak_ptr<dhtnet::ChannelSocketInterface>(socket)](Message&& msg) {
245 2541 : auto shared = w.lock();
246 2539 : auto socket = wsocket.lock();
247 2540 : if (!shared || !socket)
248 0 : return std::make_error_code(std::errc::operation_canceled);
249 :
250 2532 : if (msg.is_mobile)
251 209 : shared->changeMobility(socket->deviceId(), msg.is_mobile);
252 :
253 2533 : if (msg.request) {
254 1270 : shared->sendAnswer(socket, msg);
255 :
256 1268 : } else if (msg.response) {
257 1268 : shared->setKnownNodes(msg.response->nodes);
258 1270 : shared->setMobileNodes(msg.response->mobile_nodes);
259 : }
260 2546 : return std::error_code();
261 2545 : }));
262 :
263 2269 : socket->onShutdown([w = weak(), deviceId = socket->deviceId()](const std::error_code&) {
264 1202 : dht::ThreadPool::io().run([w, deviceId] {
265 1201 : auto shared = w.lock();
266 1197 : if (shared && !shared->isShutdown_) {
267 648 : shared->removeNode(deviceId);
268 : }
269 1202 : });
270 1203 : });
271 2268 : }
272 :
273 : void
274 1308 : SwarmManager::resetNodeExpiry(const asio::error_code& ec,
275 : const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket,
276 : NodeId node)
277 : {
278 1308 : NodeId idToFind;
279 1308 : std::list<Bucket>::iterator bucket;
280 :
281 1308 : if (ec == asio::error::operation_aborted)
282 0 : return;
283 :
284 1307 : if (!node) {
285 0 : bucket = routing_table.findBucket(socket->deviceId());
286 0 : idToFind = bucket->randomId(rd);
287 : } else {
288 1308 : bucket = routing_table.findBucket(node);
289 1308 : idToFind = node;
290 : }
291 :
292 1308 : sendRequest(socket, idToFind, Query::FIND, Bucket::BUCKET_MAX_SIZE);
293 :
294 1308 : if (!node) {
295 0 : auto& nodeTimer = bucket->getNodeTimer(socket);
296 0 : nodeTimer.expires_after(FIND_PERIOD);
297 0 : nodeTimer.async_wait(std::bind(&jami::SwarmManager::resetNodeExpiry,
298 0 : shared_from_this(),
299 : std::placeholders::_1,
300 : socket,
301 0 : NodeId {}));
302 : }
303 : }
304 :
305 : void
306 1395 : SwarmManager::tryConnect(const NodeId& nodeId, bool noNewSocket)
307 : {
308 1395 : if (needSocketCb_)
309 1392 : needSocketCb_(
310 2787 : nodeId.toString(),
311 2787 : [w = weak(), nodeId](const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket) {
312 1279 : auto shared = w.lock();
313 1279 : if (!shared || shared->isShutdown_)
314 198 : return true;
315 1081 : if (socket) {
316 993 : shared->addChannel(socket);
317 992 : return true;
318 : }
319 88 : std::unique_lock lk(shared->mutex);
320 88 : auto bucket = shared->routing_table.findBucket(nodeId);
321 88 : bucket->removeConnectingNode(nodeId);
322 88 : bucket->addKnownNode(nodeId);
323 88 : bucket = shared->routing_table.findBucket(shared->getId());
324 88 : if (bucket->getConnectingNodesSize() == 0 && bucket->isEmpty() && shared->onConnectionChanged_) {
325 51 : lk.unlock();
326 51 : JAMI_LOG("[SwarmManager {:p}] Bootstrap: all connections failed", fmt::ptr(shared.get()));
327 51 : shared->onConnectionChanged_(false);
328 : }
329 88 : return true;
330 1278 : },
331 : noNewSocket);
332 1398 : }
333 :
334 : void
335 518 : SwarmManager::removeNodeInternal(const NodeId& nodeId)
336 : {
337 518 : routing_table.removeNode(nodeId);
338 518 : }
339 :
340 : void
341 568 : SwarmManager::connectNode(const NodeId& nodeId)
342 : {
343 : {
344 568 : std::lock_guard lock(mutex);
345 568 : if (isShutdown_)
346 1 : return;
347 567 : if (isConnectedWith(nodeId))
348 2 : return;
349 565 : addKnownNode(nodeId);
350 565 : if (!routing_table.addConnectingNode(nodeId))
351 0 : return;
352 568 : }
353 565 : tryConnect(nodeId, true);
354 : }
355 :
356 : std::vector<NodeId>
357 16 : SwarmManager::getAllNodes() const
358 : {
359 16 : std::lock_guard lock(mutex);
360 32 : return routing_table.getAllNodes();
361 16 : }
362 :
363 : std::vector<NodeId>
364 1748 : SwarmManager::getConnectedNodes() const
365 : {
366 1748 : std::lock_guard lock(mutex);
367 3495 : return routing_table.getConnectedNodes();
368 1748 : }
369 :
370 : std::vector<std::map<std::string, std::string>>
371 0 : SwarmManager::getRoutingTableInfo() const
372 : {
373 0 : std::lock_guard lock(mutex);
374 0 : auto stats = routing_table.getRoutingTableStats();
375 0 : std::vector<std::map<std::string, std::string>> result;
376 0 : result.reserve(stats.size());
377 0 : for (const auto& stat : stats) {
378 0 : result.push_back({{"id", stat.id},
379 0 : {"device", stat.id},
380 0 : {"status", stat.status},
381 0 : {"remoteAddress", stat.remoteAddress},
382 0 : {"mobile", stat.isMobile ? "true" : "false"}});
383 0 : if (stat.connectionTime != std::chrono::system_clock::time_point::min()) {
384 0 : auto tt = std::chrono::system_clock::to_time_t(stat.connectionTime);
385 0 : result.back().emplace("connectionTime", std::to_string(tt));
386 : }
387 : }
388 0 : return result;
389 0 : }
390 :
391 : bool
392 2687 : SwarmManager::isConnected() const
393 : {
394 2687 : std::lock_guard lock(mutex);
395 5376 : return !routing_table.isEmpty();
396 2686 : }
397 :
398 : void
399 11 : SwarmManager::deleteNode(const std::vector<NodeId>& nodes)
400 : {
401 : {
402 11 : std::lock_guard lock(mutex);
403 22 : for (const auto& node : nodes) {
404 11 : routing_table.deleteNode(node);
405 : }
406 11 : }
407 11 : maintainBuckets();
408 11 : }
409 :
410 : } // namespace jami
|
