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 586 : SwarmManager::SwarmManager(const NodeId& id, const std::mt19937_64& rand, ToConnectCb&& toConnectCb)
28 586 : : id_(id)
29 586 : , rd(rand)
30 586 : , toConnectCb_(toConnectCb)
31 : {
32 586 : routing_table.setId(id);
33 586 : }
34 :
35 586 : SwarmManager::~SwarmManager()
36 : {
37 586 : if (!isShutdown_)
38 242 : shutdown();
39 586 : }
40 :
41 : bool
42 2030 : SwarmManager::setKnownNodes(const std::vector<NodeId>& known_nodes)
43 : {
44 2030 : isShutdown_ = false;
45 2029 : std::vector<NodeId> newNodes;
46 : {
47 2029 : std::lock_guard lock(mutex);
48 4847 : for (const auto& nodeId : known_nodes) {
49 2813 : if (addKnownNode(nodeId)) {
50 707 : newNodes.emplace_back(nodeId);
51 : }
52 : }
53 2026 : }
54 :
55 2036 : if (newNodes.empty())
56 1472 : return false;
57 :
58 561 : dht::ThreadPool::io().run([w = weak(), newNodes = std::move(newNodes)] {
59 563 : auto shared = w.lock();
60 563 : if (!shared)
61 0 : return;
62 : // If we detect a new node which already got a TCP link
63 : // we can use it to speed-up the bootstrap (because opening
64 : // a new channel will be easy)
65 563 : std::set<NodeId> toConnect;
66 1273 : for (const auto& nodeId : newNodes) {
67 712 : if (shared->toConnectCb_ && shared->toConnectCb_(nodeId))
68 130 : toConnect.emplace(nodeId);
69 : }
70 559 : shared->maintainBuckets(toConnect);
71 563 : });
72 563 : return true;
73 2035 : }
74 :
75 : void
76 1303 : SwarmManager::setMobileNodes(const std::vector<NodeId>& mobile_nodes)
77 : {
78 : {
79 1303 : std::lock_guard lock(mutex);
80 1320 : for (const auto& nodeId : mobile_nodes)
81 16 : addMobileNodes(nodeId);
82 1303 : }
83 1305 : }
84 :
85 : void
86 2310 : SwarmManager::addChannel(const std::shared_ptr<dhtnet::ChannelSocketInterface>& channel)
87 : {
88 : // JAMI_WARNING("[SwarmManager {}] addChannel! with {}", fmt::ptr(this), channel->deviceId().to_view());
89 2310 : if (channel) {
90 2310 : auto emit = false;
91 : {
92 2310 : std::lock_guard lock(mutex);
93 2310 : emit = routing_table.findBucket(getId())->isEmpty();
94 2310 : auto bucket = routing_table.findBucket(channel->deviceId());
95 2310 : if (routing_table.addNode(channel, bucket)) {
96 1336 : std::error_code ec;
97 1335 : resetNodeExpiry(ec, channel, id_);
98 : }
99 2310 : }
100 2310 : receiveMessage(channel);
101 2308 : if (emit && onConnectionChanged_) {
102 : // If it's the first channel we add, we're now connected!
103 1493 : JAMI_DEBUG("[SwarmManager {}] Bootstrap: Connected!", fmt::ptr(this));
104 374 : onConnectionChanged_(true);
105 : }
106 : }
107 2309 : }
108 :
109 : void
110 601 : SwarmManager::removeNode(const NodeId& nodeId)
111 : {
112 601 : std::unique_lock lk(mutex);
113 601 : if (isConnectedWith(nodeId)) {
114 496 : removeNodeInternal(nodeId);
115 495 : lk.unlock();
116 495 : maintainBuckets();
117 : }
118 601 : }
119 :
120 : void
121 228 : SwarmManager::changeMobility(const NodeId& nodeId, bool isMobile)
122 : {
123 228 : std::lock_guard lock(mutex);
124 227 : auto bucket = routing_table.findBucket(nodeId);
125 226 : bucket->changeMobility(nodeId, isMobile);
126 227 : }
127 :
128 : bool
129 1207 : SwarmManager::isConnectedWith(const NodeId& deviceId)
130 : {
131 1207 : return routing_table.hasNode(deviceId);
132 : }
133 :
134 : void
135 628 : SwarmManager::shutdown()
136 : {
137 628 : if (isShutdown_) {
138 13 : return;
139 : }
140 615 : isShutdown_ = true;
141 615 : std::lock_guard lock(mutex);
142 615 : routing_table.shutdownAllNodes();
143 615 : }
144 :
145 : void
146 22 : SwarmManager::restart()
147 : {
148 22 : isShutdown_ = false;
149 22 : }
150 :
151 : bool
152 3417 : SwarmManager::addKnownNode(const NodeId& nodeId)
153 : {
154 3417 : return routing_table.addKnownNode(nodeId);
155 : }
156 :
157 : void
158 16 : SwarmManager::addMobileNodes(const NodeId& nodeId)
159 : {
160 16 : if (id_ != nodeId) {
161 15 : routing_table.addMobileNode(nodeId);
162 : }
163 16 : }
164 :
165 : void
166 1089 : SwarmManager::maintainBuckets(const std::set<NodeId>& toConnect)
167 : {
168 1089 : std::set<NodeId> nodes = toConnect;
169 1087 : std::unique_lock lock(mutex);
170 1091 : auto& buckets = routing_table.getBuckets();
171 3976 : for (auto it = buckets.begin(); it != buckets.end(); ++it) {
172 2885 : auto& bucket = *it;
173 2889 : bool myBucket = routing_table.contains(it, id_);
174 4690 : auto connecting_nodes = myBucket ? bucket.getConnectingNodesSize()
175 1802 : : bucket.getConnectingNodesSize() + bucket.getNodesSize();
176 2887 : if (connecting_nodes < Bucket::BUCKET_MAX_SIZE) {
177 1435 : auto nodesToTry = bucket.getKnownNodesRandom(Bucket::BUCKET_MAX_SIZE - connecting_nodes, rd);
178 2215 : for (auto& node : nodesToTry)
179 787 : routing_table.addConnectingNode(node);
180 :
181 1429 : nodes.insert(nodesToTry.begin(), nodesToTry.end());
182 1434 : }
183 : }
184 1089 : lock.unlock();
185 1896 : for (const auto& node : nodes)
186 805 : tryConnect(node);
187 1087 : }
188 :
189 : void
190 1336 : SwarmManager::sendRequest(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket,
191 : const NodeId& nodeId,
192 : Query q,
193 : int numberNodes)
194 : {
195 1336 : dht::ThreadPool::io().run([socket, isMobile = isMobile_, nodeId, q, numberNodes] {
196 1336 : msgpack::sbuffer buffer;
197 1336 : msgpack::packer<msgpack::sbuffer> pk(&buffer);
198 1336 : Message msg;
199 1336 : msg.is_mobile = isMobile;
200 1336 : msg.request = Request {q, numberNodes, nodeId};
201 1336 : pk.pack(msg);
202 :
203 1334 : std::error_code ec;
204 1334 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
205 1336 : if (ec) {
206 12 : JAMI_ERROR("{}", ec.message());
207 : }
208 1336 : });
209 1336 : }
210 :
211 : void
212 1301 : SwarmManager::sendAnswer(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket, const Message& msg_)
213 : {
214 1301 : std::lock_guard lock(mutex);
215 :
216 1302 : if (msg_.request->q == Query::FIND) {
217 1298 : auto nodes = routing_table.closestNodes(msg_.request->nodeId, msg_.request->num);
218 1307 : auto bucket = routing_table.findBucket(msg_.request->nodeId);
219 1308 : const auto& m_nodes = bucket->getMobileNodes();
220 2613 : Response toResponse {Query::FOUND, nodes, {m_nodes.begin(), m_nodes.end()}};
221 :
222 1303 : Message msg;
223 1303 : msg.is_mobile = isMobile_;
224 1303 : msg.response = std::move(toResponse);
225 :
226 1304 : msgpack::sbuffer buffer((size_t) 60000);
227 1308 : msgpack::packer<msgpack::sbuffer> pk(&buffer);
228 1307 : pk.pack(msg);
229 :
230 1305 : std::error_code ec;
231 1304 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
232 1308 : if (ec) {
233 12 : JAMI_ERROR("{}", ec.message());
234 3 : return;
235 : }
236 1317 : }
237 1308 : }
238 :
239 : void
240 2310 : SwarmManager::receiveMessage(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket)
241 : {
242 4620 : socket->setOnRecv(dhtnet::buildMsgpackReader<Message>(
243 4620 : [w = weak(), wsocket = std::weak_ptr<dhtnet::ChannelSocketInterface>(socket)](Message&& msg) {
244 2597 : auto shared = w.lock();
245 2594 : auto socket = wsocket.lock();
246 2589 : if (!shared || !socket)
247 0 : return std::make_error_code(std::errc::operation_canceled);
248 :
249 2584 : if (msg.is_mobile)
250 228 : shared->changeMobility(socket->deviceId(), msg.is_mobile);
251 :
252 2584 : if (msg.request) {
253 1301 : shared->sendAnswer(socket, msg);
254 :
255 1297 : } else if (msg.response) {
256 1304 : shared->setKnownNodes(msg.response->nodes);
257 1300 : shared->setMobileNodes(msg.response->mobile_nodes);
258 : }
259 2608 : return std::error_code();
260 2606 : }));
261 :
262 2309 : socket->onShutdown([w = weak(), deviceId = socket->deviceId()](const std::error_code&) {
263 1191 : dht::ThreadPool::io().run([w, deviceId] {
264 1189 : auto shared = w.lock();
265 1190 : if (shared && !shared->isShutdown_) {
266 601 : shared->removeNode(deviceId);
267 : }
268 1191 : });
269 1194 : });
270 2308 : }
271 :
272 : void
273 1336 : SwarmManager::resetNodeExpiry(const asio::error_code& ec,
274 : const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket,
275 : NodeId node)
276 : {
277 1336 : NodeId idToFind;
278 1336 : std::list<Bucket>::iterator bucket;
279 :
280 1335 : if (ec == asio::error::operation_aborted)
281 0 : return;
282 :
283 1336 : if (!node) {
284 0 : bucket = routing_table.findBucket(socket->deviceId());
285 0 : idToFind = bucket->randomId(rd);
286 : } else {
287 1336 : bucket = routing_table.findBucket(node);
288 1336 : idToFind = node;
289 : }
290 :
291 1336 : sendRequest(socket, idToFind, Query::FIND, Bucket::BUCKET_MAX_SIZE);
292 :
293 1336 : if (!node) {
294 0 : auto& nodeTimer = bucket->getNodeTimer(socket);
295 0 : nodeTimer.expires_after(FIND_PERIOD);
296 0 : nodeTimer.async_wait(std::bind(&jami::SwarmManager::resetNodeExpiry,
297 0 : shared_from_this(),
298 : std::placeholders::_1,
299 : socket,
300 0 : NodeId {}));
301 : }
302 : }
303 :
304 : void
305 1405 : SwarmManager::tryConnect(const NodeId& nodeId, bool noNewSocket)
306 : {
307 1405 : if (needSocketCb_)
308 1404 : needSocketCb_(
309 2807 : nodeId.toString(),
310 2803 : [w = weak(), nodeId](const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket) {
311 1303 : auto shared = w.lock();
312 1303 : if (!shared || shared->isShutdown_)
313 189 : return true;
314 1114 : if (socket) {
315 1035 : shared->addChannel(socket);
316 1035 : return true;
317 : }
318 79 : std::unique_lock lk(shared->mutex);
319 79 : auto bucket = shared->routing_table.findBucket(nodeId);
320 79 : bucket->removeConnectingNode(nodeId);
321 79 : bucket->addKnownNode(nodeId);
322 79 : bucket = shared->routing_table.findBucket(shared->getId());
323 79 : if (bucket->getConnectingNodesSize() == 0 && bucket->isEmpty() && shared->onConnectionChanged_) {
324 41 : lk.unlock();
325 164 : JAMI_LOG("[SwarmManager {:p}] Bootstrap: all connections failed", fmt::ptr(shared.get()));
326 41 : shared->onConnectionChanged_(false);
327 : }
328 79 : return true;
329 1303 : },
330 : noNewSocket);
331 1407 : }
332 :
333 : void
334 496 : SwarmManager::removeNodeInternal(const NodeId& nodeId)
335 : {
336 496 : routing_table.removeNode(nodeId);
337 494 : }
338 :
339 : void
340 607 : SwarmManager::connectNode(const NodeId& nodeId)
341 : {
342 : {
343 607 : std::lock_guard lock(mutex);
344 607 : if (isShutdown_)
345 1 : return;
346 606 : if (isConnectedWith(nodeId))
347 5 : return;
348 601 : addKnownNode(nodeId);
349 601 : if (!routing_table.addConnectingNode(nodeId))
350 0 : return;
351 607 : }
352 601 : tryConnect(nodeId, true);
353 : }
354 :
355 : std::vector<NodeId>
356 16 : SwarmManager::getAllNodes() const
357 : {
358 16 : std::lock_guard lock(mutex);
359 32 : return routing_table.getAllNodes();
360 16 : }
361 :
362 : std::vector<NodeId>
363 1769 : SwarmManager::getConnectedNodes() const
364 : {
365 1769 : std::lock_guard lock(mutex);
366 3538 : return routing_table.getConnectedNodes();
367 1768 : }
368 :
369 : std::vector<std::map<std::string, std::string>>
370 0 : SwarmManager::getRoutingTableInfo() const
371 : {
372 0 : std::lock_guard lock(mutex);
373 0 : auto stats = routing_table.getRoutingTableStats();
374 0 : std::vector<std::map<std::string, std::string>> result;
375 0 : result.reserve(stats.size());
376 0 : for (const auto& stat : stats) {
377 0 : result.push_back({{"id", stat.id},
378 0 : {"device", stat.id},
379 0 : {"status", stat.status},
380 0 : {"remoteAddress", stat.remoteAddress},
381 0 : {"mobile", stat.isMobile ? "true" : "false"}});
382 0 : if (stat.connectionTime != std::chrono::system_clock::time_point::min()) {
383 0 : auto tt = std::chrono::system_clock::to_time_t(stat.connectionTime);
384 0 : result.back().emplace("connectionTime", std::to_string(tt));
385 : }
386 : }
387 0 : return result;
388 0 : }
389 :
390 : bool
391 2690 : SwarmManager::isConnected() const
392 : {
393 2690 : std::lock_guard lock(mutex);
394 5380 : return !routing_table.isEmpty();
395 2690 : }
396 :
397 : void
398 10 : SwarmManager::deleteNode(const std::vector<NodeId>& nodes)
399 : {
400 : {
401 10 : std::lock_guard lock(mutex);
402 20 : for (const auto& node : nodes) {
403 10 : routing_table.deleteNode(node);
404 : }
405 10 : }
406 10 : maintainBuckets();
407 10 : }
408 :
409 : } // namespace jami
|
