My Project 3.2.0
C++ Distributed Hash Table
Loading...
Searching...
No Matches
dht.h
1/*
2 * Copyright (C) 2014-2023 Savoir-faire Linux Inc.
3 * Authors: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
4 * Simon Désaulniers <simon.desaulniers@savoirfairelinux.com>
5 * Sébastien Blin <sebastien.blin@savoirfairelinux.com>
6 *
7 * This program is free software; you can redistribute it and/or modify
8 * it under the terms of the GNU General Public License as published by
9 * the Free Software Foundation; either version 3 of the License, or
10 * (at your option) any later version.
11 *
12 * This program is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU General Public License for more details.
16 *
17 * You should have received a copy of the GNU General Public License
18 * along with this program. If not, see <https://www.gnu.org/licenses/>.
19 */
20
21#pragma once
22
23#include "infohash.h"
24#include "value.h"
25#include "utils.h"
26#include "network_engine.h"
27#include "scheduler.h"
28#include "routing_table.h"
29#include "callbacks.h"
30#include "dht_interface.h"
31
32#include <string>
33#include <array>
34#include <vector>
35#include <map>
36#include <functional>
37#include <memory>
38
39#ifdef _WIN32
40#include <iso646.h>
41#endif
42
43namespace dht {
44
45namespace net {
46struct Request;
47} /* namespace net */
48
49struct Storage;
50struct ValueStorage;
51class StorageBucket;
52struct Listener;
53struct LocalListener;
54
62class OPENDHT_PUBLIC Dht final : public DhtInterface {
63public:
68 Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, const Sp<Logger>& l = {}, std::unique_ptr<std::mt19937_64>&& rd = {});
69
70 virtual ~Dht();
71
75 inline const InfoHash& getNodeId() const override { return myid; }
76 void setOnPublicAddressChanged(PublicAddressChangedCb cb) override {
77 publicAddressChangedCb_ = std::move(cb);
78 }
79
80 NodeStatus updateStatus(sa_family_t af) override;
81
85 NodeStatus getStatus(sa_family_t af) const override {
86 return dht(af).status;
87 }
88
89 NodeStatus getStatus() const override {
90 return std::max(getStatus(AF_INET), getStatus(AF_INET6));
91 }
92
93 net::DatagramSocket* getSocket() const override { return network_engine.getSocket(); };
94
98 void shutdown(ShutdownCallback cb, bool stop = false) override;
99
106 bool isRunning(sa_family_t af = 0) const override;
107
108 virtual void registerType(const ValueType& type) override {
109 types.registerType(type);
110 }
111 const ValueType& getType(ValueType::Id type_id) const override {
112 return types.getType(type_id);
113 }
114
115 void addBootstrap(const std::string& host, const std::string& service) override {
116 bootstrap_nodes.emplace_back(host, service);
117 startBootstrap();
118 }
119
120 void clearBootstrap() override {
121 bootstrap_nodes.clear();
122 }
123
129 void insertNode(const InfoHash& id, const SockAddr&) override;
130 void insertNode(const NodeExport& n) override {
131 insertNode(n.id, n.addr);
132 }
133
134 void pingNode(SockAddr, DoneCallbackSimple&& cb={}) override;
135
136 time_point periodic(const uint8_t *buf, size_t buflen, SockAddr, const time_point& now) override;
137 time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen, const time_point& now) override {
138 return periodic(buf, buflen, SockAddr(from, fromlen), now);
139 }
140
151 virtual void get(const InfoHash& key, GetCallback cb, DoneCallback donecb={}, Value::Filter&& f={}, Where&& w = {}) override;
152 virtual void get(const InfoHash& key, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter&& f={}, Where&& w = {}) override {
153 get(key, cb, bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
154 }
155 virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter&& f={}, Where&& w = {}) override {
156 get(key, bindGetCb(cb), donecb, std::forward<Value::Filter>(f), std::forward<Where>(w));
157 }
158 virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallbackSimple donecb, Value::Filter&& f={}, Where&& w = {}) override {
159 get(key, bindGetCb(cb), bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
160 }
171 virtual void query(const InfoHash& key, QueryCallback cb, DoneCallback done_cb = {}, Query&& q = {}) override;
172 virtual void query(const InfoHash& key, QueryCallback cb, DoneCallbackSimple done_cb = {}, Query&& q = {}) override {
173 query(key, cb, bindDoneCb(done_cb), std::forward<Query>(q));
174 }
175
179 std::vector<Sp<Value>> getLocal(const InfoHash& key, const Value::Filter& f = {}) const override;
180
184 Sp<Value> getLocalById(const InfoHash& key, Value::Id vid) const override;
185
192 void put(const InfoHash& key,
193 Sp<Value>,
194 DoneCallback cb=nullptr,
195 time_point created=time_point::max(),
196 bool permanent = false) override;
197 void put(const InfoHash& key,
198 const Sp<Value>& v,
199 DoneCallbackSimple cb,
200 time_point created=time_point::max(),
201 bool permanent = false) override
202 {
203 put(key, v, bindDoneCb(cb), created, permanent);
204 }
205
206 void put(const InfoHash& key,
207 Value&& v,
208 DoneCallback cb=nullptr,
209 time_point created=time_point::max(),
210 bool permanent = false) override
211 {
212 put(key, std::make_shared<Value>(std::move(v)), cb, created, permanent);
213 }
214 void put(const InfoHash& key,
215 Value&& v,
216 DoneCallbackSimple cb,
217 time_point created=time_point::max(),
218 bool permanent = false) override
219 {
220 put(key, std::forward<Value>(v), bindDoneCb(cb), created, permanent);
221 }
222
226 std::vector<Sp<Value>> getPut(const InfoHash&) const override;
227
231 Sp<Value> getPut(const InfoHash&, const Value::Id&) const override;
232
237 bool cancelPut(const InfoHash&, const Value::Id&) override;
238
246 size_t listen(const InfoHash&, ValueCallback, Value::Filter={}, Where={}) override;
247
248 size_t listen(const InfoHash& key, GetCallback cb, Value::Filter f={}, Where w={}) override {
249 return listen(key, [cb](const std::vector<Sp<Value>>& vals, bool expired){
250 if (not expired)
251 return cb(vals);
252 return true;
253 }, std::forward<Value::Filter>(f), std::forward<Where>(w));
254 }
255 size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w={}) override {
256 return listen(key, bindGetCb(cb), std::forward<Value::Filter>(f), std::forward<Where>(w));
257 }
258
259 bool cancelListen(const InfoHash&, size_t token) override;
260
266 void connectivityChanged(sa_family_t) override;
267 void connectivityChanged() override {
268 connectivityChanged(AF_INET);
269 connectivityChanged(AF_INET6);
270 }
271
276 std::vector<NodeExport> exportNodes() const override;
277
278 std::vector<ValuesExport> exportValues() const override;
279 void importValues(const std::vector<ValuesExport>&) override;
280
281 void saveState(const std::string& path) const;
282 void loadState(const std::string& path);
283
284 NodeStats getNodesStats(sa_family_t af) const override;
285
286 std::string getStorageLog() const override;
287 std::string getStorageLog(const InfoHash&) const override;
288
289 std::string getRoutingTablesLog(sa_family_t) const override;
290 std::string getSearchesLog(sa_family_t) const override;
291 std::string getSearchLog(const InfoHash&, sa_family_t af = AF_UNSPEC) const override;
292
293 void dumpTables() const override;
294 std::vector<unsigned> getNodeMessageStats(bool in = false) override {
295 return network_engine.getNodeMessageStats(in);
296 }
297
301 void setStorageLimit(size_t limit = DEFAULT_STORAGE_LIMIT) override {
302 max_store_size = limit;
303 }
304 size_t getStorageLimit() const override {
305 return max_store_size;
306 }
307
312 std::pair<size_t, size_t> getStoreSize() const override {
313 return {total_store_size, total_values};
314 }
315
316 std::vector<SockAddr> getPublicAddress(sa_family_t family = 0) override;
317
318 void pushNotificationReceived(const std::map<std::string, std::string>&) override {}
319 void resubscribe(unsigned) {}
320
321private:
322
323 /* When performing a search, we search for up to SEARCH_NODES closest nodes
324 to the destination, and use the additional ones to backtrack if any of
325 the target 8 turn out to be dead. */
326 static constexpr unsigned SEARCH_NODES {14};
327
328 /* The number of bad nodes is limited in order to help determine
329 * presence of connectivity changes. See
330 * https://github.com/savoirfairelinux/opendht/issues/137 for details.
331 *
332 * According to the tables, 25 is a good average value for big networks. If
333 * the network is small, normal search expiration process will handle the
334 * situation.
335 * */
336 static constexpr unsigned SEARCH_MAX_BAD_NODES {25};
337
338 /* Concurrent search nodes requested count */
339 static constexpr unsigned MAX_REQUESTED_SEARCH_NODES {4};
340
341 /* Number of listening nodes */
342 static constexpr unsigned LISTEN_NODES {4};
343
344 /* The maximum number of hashes we're willing to track. */
345 static constexpr unsigned MAX_HASHES {1024 * 1024 * 1024};
346
347 /* The maximum number of searches we keep data about. */
348 static constexpr unsigned MAX_SEARCHES {1024 * 1024};
349
350 static constexpr std::chrono::minutes MAX_STORAGE_MAINTENANCE_EXPIRE_TIME {10};
351
352 /* The time after which we consider a search to be expirable. */
353 static constexpr std::chrono::minutes SEARCH_EXPIRE_TIME {62};
354
355 /* Timeout for listen */
356 static constexpr duration LISTEN_EXPIRE_TIME {std::chrono::seconds(30)};
357 static constexpr duration LISTEN_EXPIRE_TIME_PUBLIC {std::chrono::minutes(5)};
358
359 static constexpr duration REANNOUNCE_MARGIN {std::chrono::seconds(10)};
360
361 static constexpr std::chrono::seconds BOOTSTRAP_PERIOD {10};
362
363 static constexpr size_t TOKEN_SIZE {32};
364
365 // internal structures
366 struct SearchNode;
367 struct Get;
368 struct Announce;
369 struct Search;
370
371 // prevent copy
372 Dht(const Dht&) = delete;
373 Dht& operator=(const Dht&) = delete;
374
375 std::mt19937_64 rd;
376
377 InfoHash myid {};
378
379 uint64_t secret {};
380 uint64_t oldsecret {};
381
382 // registred types
383 TypeStore types;
384
385 using SearchMap = std::map<InfoHash, Sp<Search>>;
386 using ReportedAddr = std::pair<unsigned, SockAddr>;
387
388 struct Kad {
389 RoutingTable buckets {};
390 SearchMap searches {};
391 unsigned pending_pings {0};
392 NodeStatus status;
393 std::vector<ReportedAddr> reported_addr;
394
395 NodeStatus getStatus(time_point now) const;
396 NodeStats getNodesStats(time_point now, const InfoHash& myid) const;
397 };
398
399 Kad dht4 {};
400 Kad dht6 {};
401 PublicAddressChangedCb publicAddressChangedCb_ {};
402
403 std::vector<std::pair<std::string,std::string>> bootstrap_nodes {};
404 std::chrono::steady_clock::duration bootstrap_period {BOOTSTRAP_PERIOD};
405 Sp<Scheduler::Job> bootstrapJob {};
406
407 std::map<InfoHash, Storage> store;
408 std::map<SockAddr, StorageBucket, SockAddr::ipCmp> store_quota;
409 size_t total_values {0};
410 size_t total_store_size {0};
411 size_t max_store_keys {MAX_HASHES};
412 size_t max_store_size {DEFAULT_STORAGE_LIMIT};
413
414 size_t max_searches {MAX_SEARCHES};
415 size_t search_id {0};
416
417 // map a global listen token to IPv4, IPv6 specific listen tokens.
418 // 0 is the invalid token.
419 std::map<size_t, std::tuple<size_t, size_t, size_t>> listeners {};
420 size_t listener_token {1};
421
422
423 // timing
424 Scheduler scheduler;
425 Sp<Scheduler::Job> nextNodesConfirmation {};
426 Sp<Scheduler::Job> nextStorageMaintenance {};
427
428 net::NetworkEngine network_engine;
429
430 std::string persistPath;
431
432 // are we a bootstrap node ?
433 // note: Any running node can be used as a bootstrap node.
434 // Only nodes running only as bootstrap nodes should
435 // be put in bootstrap mode.
436 const bool is_bootstrap {false};
437 const bool maintain_storage {false};
438 const bool public_stable {false};
439
440 inline const duration& getListenExpiration() const {
441 return public_stable ? LISTEN_EXPIRE_TIME_PUBLIC : LISTEN_EXPIRE_TIME;
442 }
443
444 void rotateSecrets();
445
446 Blob makeToken(const SockAddr&, bool old) const;
447 bool tokenMatch(const Blob& token, const SockAddr&) const;
448
449 void reportedAddr(const SockAddr&);
450
451 // Storage
452 void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {}, int version = 0);
453 bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {}, bool permanent = false);
454 bool storageRefresh(const InfoHash& id, Value::Id vid);
455 void expireStore();
456 void expireStorage(InfoHash h);
457 void expireStore(decltype(store)::iterator);
458
459 void storageRemoved(const InfoHash& id, Storage& st, const std::vector<Sp<Value>>& values, size_t totalSize);
460 void storageChanged(const InfoHash& id, Storage& st, const Sp<Value>&, bool newValue);
461 std::string printStorageLog(const decltype(store)::value_type&) const;
462
468 void dataPersistence(InfoHash id);
469 size_t maintainStorage(decltype(store)::value_type&, bool force=false, const DoneCallback& donecb={});
470
471 // Buckets
472 Kad& dht(sa_family_t af) { return af == AF_INET ? dht4 : dht6; }
473 const Kad& dht(sa_family_t af) const { return af == AF_INET ? dht4 : dht6; }
474 RoutingTable& buckets(sa_family_t af) { return dht(af).buckets; }
475 const RoutingTable& buckets(sa_family_t af) const { return dht(af).buckets; }
476 Bucket* findBucket(const InfoHash& id, sa_family_t af) {
477 auto& b = buckets(af);
478 auto it = b.findBucket(id);
479 return it == b.end() ? nullptr : &(*it);
480 }
481 const Bucket* findBucket(const InfoHash& id, sa_family_t af) const {
482 return const_cast<Dht*>(this)->findBucket(id, af);
483 }
484
485 void expireBuckets(RoutingTable&);
486 void sendCachedPing(Bucket& b);
487 bool bucketMaintenance(RoutingTable&);
488 void dumpBucket(const Bucket& b, std::ostream& out) const;
489 void bootstrap();
490 void startBootstrap();
491 void stopBootstrap();
492
493 // Nodes
494 void onNewNode(const Sp<Node>& node, int confirm);
495 const Sp<Node> findNode(const InfoHash& id, sa_family_t af) const;
496 bool trySearchInsert(const Sp<Node>& node);
497
498 // Searches
499 inline SearchMap& searches(sa_family_t af) { return dht(af).searches; }
500 inline const SearchMap& searches(sa_family_t af) const { return dht(af).searches; }
501
506 Sp<Search> search(const InfoHash& id, sa_family_t af, GetCallback = {}, QueryCallback = {}, DoneCallback = {}, Value::Filter = {}, const Sp<Query>& q = {});
507
508 void announce(const InfoHash& id, sa_family_t af, Sp<Value> value, DoneCallback callback, time_point created=time_point::max(), bool permanent = false);
509 size_t listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f = {}, const Sp<Query>& q = {});
510
518 unsigned refill(Search& sr);
519 void expireSearches();
520
521 void confirmNodes();
522 void expire();
523
524 void onConnected();
525 void onDisconnected();
526
535 void searchNodeGetDone(const net::Request& status,
536 net::RequestAnswer&& answer,
537 std::weak_ptr<Search> ws,
538 Sp<Query> query);
539
549 void searchNodeGetExpired(const net::Request& status, bool over, std::weak_ptr<Search> ws, Sp<Query> query);
550
558 void paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n);
559
563 SearchNode* searchSendGetValues(Sp<Search> sr, SearchNode *n = nullptr, bool update = true);
564
571 void searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel = TARGET_NODES);
572
580 void searchStep(std::weak_ptr<Search> ws);
581
582 void searchSynchedNodeListen(const Sp<Search>&, SearchNode&);
583
584 void dumpSearch(const Search& sr, std::ostream& out) const;
585
586 bool neighbourhoodMaintenance(RoutingTable&);
587
588 void onError(Sp<net::Request> node, net::DhtProtocolException e);
589 /* when our address is reported by a distant peer. */
590 void onReportedAddr(const InfoHash& id, const SockAddr&);
591 /* when we receive a ping request */
592 net::RequestAnswer onPing(Sp<Node> node);
593 /* when we receive a "find node" request */
594 net::RequestAnswer onFindNode(Sp<Node> node, const InfoHash& hash, want_t want);
595 void onFindNodeDone(const Sp<Node>& status,
596 net::RequestAnswer& a,
597 Sp<Search> sr);
598 /* when we receive a "get values" request */
599 net::RequestAnswer onGetValues(Sp<Node> node,
600 const InfoHash& hash,
601 want_t want,
602 const Query& q);
603 void onGetValuesDone(const Sp<Node>& status,
604 net::RequestAnswer& a,
605 Sp<Search>& sr,
606 const Sp<Query>& orig_query);
607 /* when we receive a listen request */
608 net::RequestAnswer onListen(Sp<Node> node,
609 const InfoHash& hash,
610 const Blob& token,
611 size_t socket_id,
612 const Query& query,
613 int version = 0);
614 void onListenDone(const Sp<Node>& status,
615 net::RequestAnswer& a,
616 Sp<Search>& sr);
617 /* when we receive an announce request */
618 net::RequestAnswer onAnnounce(Sp<Node> node,
619 const InfoHash& hash,
620 const Blob& token,
621 const std::vector<Sp<Value>>& v,
622 const time_point& created);
623 net::RequestAnswer onRefresh(Sp<Node> node,
624 const InfoHash& hash,
625 const Blob& token,
626 const Value::Id& vid);
627 void onAnnounceDone(const Sp<Node>& status,
628 net::RequestAnswer& a,
629 Sp<Search>& sr);
630};
631
632}
Definition dht.h:62
Dht(std::unique_ptr< net::DatagramSocket > &&sock, const Config &config, const Sp< Logger > &l={}, std::unique_ptr< std::mt19937_64 > &&rd={})
void insertNode(const InfoHash &id, const SockAddr &) override
void setStorageLimit(size_t limit=DEFAULT_STORAGE_LIMIT) override
Definition dht.h:301
NodeStatus updateStatus(sa_family_t af) override
NodeStatus getStatus(sa_family_t af) const override
Definition dht.h:85
std::pair< size_t, size_t > getStoreSize() const override
Definition dht.h:312
Sp< Value > getLocalById(const InfoHash &key, Value::Id vid) const override
std::vector< Sp< Value > > getPut(const InfoHash &) const override
std::vector< NodeExport > exportNodes() const override
size_t listen(const InfoHash &, ValueCallback, Value::Filter={}, Where={}) override
bool cancelPut(const InfoHash &, const Value::Id &) override
void put(const InfoHash &key, Sp< Value >, DoneCallback cb=nullptr, time_point created=time_point::max(), bool permanent=false) override
void connectivityChanged(sa_family_t) override
void pushNotificationReceived(const std::map< std::string, std::string > &) override
Definition dht.h:318
size_t listen(const InfoHash &key, GetCallback cb, Value::Filter f={}, Where w={}) override
Definition dht.h:248
bool isRunning(sa_family_t af=0) const override
Sp< Value > getPut(const InfoHash &, const Value::Id &) const override
void shutdown(ShutdownCallback cb, bool stop=false) override
virtual void get(const InfoHash &key, GetCallback cb, DoneCallback donecb={}, Value::Filter &&f={}, Where &&w={}) override
std::vector< Sp< Value > > getLocal(const InfoHash &key, const Value::Filter &f={}) const override
virtual void query(const InfoHash &key, QueryCallback cb, DoneCallback done_cb={}, Query &&q={}) override
const InfoHash & getNodeId() const override
Definition dht.h:75
std::vector< uint8_t > Blob
Definition utils.h:151
NodeStatus
Definition callbacks.h:42
Describes a query destined to another peer.
Definition value.h:924
Serializable dht::Value filter.
Definition value.h:797