62class OPENDHT_PUBLIC
Dht final :
public DhtInterface {
68 Dht(std::unique_ptr<net::DatagramSocket>&& sock,
const Config& config,
const Sp<Logger>& l = {}, std::unique_ptr<std::mt19937_64>&& rd = {});
75 inline const InfoHash&
getNodeId()
const override {
return myid; }
76 void setOnPublicAddressChanged(PublicAddressChangedCb cb)
override {
77 publicAddressChangedCb_ = std::move(cb);
86 return dht(af).status;
90 return std::max(getStatus(AF_INET), getStatus(AF_INET6));
93 net::DatagramSocket* getSocket()
const override {
return network_engine.getSocket(); };
98 void shutdown(ShutdownCallback cb,
bool stop =
false)
override;
108 virtual void registerType(
const ValueType& type)
override {
109 types.registerType(type);
111 const ValueType& getType(ValueType::Id type_id)
const override {
112 return types.getType(type_id);
115 void addBootstrap(
const std::string& host,
const std::string& service)
override {
116 bootstrap_nodes.emplace_back(host, service);
120 void clearBootstrap()
override {
121 bootstrap_nodes.clear();
130 void insertNode(
const NodeExport& n)
override {
134 void pingNode(
SockAddr, DoneCallbackSimple&& cb={})
override;
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);
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));
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));
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));
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));
184 Sp<Value>
getLocalById(
const InfoHash& key, Value::Id vid)
const override;
192 void put(
const InfoHash& key,
194 DoneCallback cb=
nullptr,
195 time_point created=time_point::max(),
196 bool permanent =
false)
override;
197 void put(
const InfoHash& key,
199 DoneCallbackSimple cb,
200 time_point created=time_point::max(),
201 bool permanent =
false)
override
203 put(key, v, bindDoneCb(cb), created, permanent);
206 void put(
const InfoHash& key,
208 DoneCallback cb=
nullptr,
209 time_point created=time_point::max(),
210 bool permanent =
false)
override
212 put(key, std::make_shared<Value>(std::move(v)), cb, created, permanent);
214 void put(
const InfoHash& key,
216 DoneCallbackSimple cb,
217 time_point created=time_point::max(),
218 bool permanent =
false)
override
220 put(key, std::forward<Value>(v), bindDoneCb(cb), created, permanent);
226 std::vector<Sp<Value>>
getPut(
const InfoHash&)
const override;
231 Sp<Value>
getPut(
const InfoHash&,
const Value::Id&)
const override;
237 bool cancelPut(
const InfoHash&,
const Value::Id&)
override;
249 return listen(key, [cb](
const std::vector<Sp<Value>>& vals,
bool expired){
253 }, std::forward<Value::Filter>(f), std::forward<Where>(w));
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));
259 bool cancelListen(
const InfoHash&,
size_t token)
override;
278 std::vector<ValuesExport> exportValues()
const override;
279 void importValues(
const std::vector<ValuesExport>&)
override;
281 void saveState(
const std::string& path)
const;
282 void loadState(
const std::string& path);
284 NodeStats getNodesStats(sa_family_t af)
const override;
286 std::string getStorageLog()
const override;
287 std::string getStorageLog(
const InfoHash&)
const override;
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;
293 void dumpTables()
const override;
294 std::vector<unsigned> getNodeMessageStats(
bool in =
false)
override {
295 return network_engine.getNodeMessageStats(in);
302 max_store_size = limit;
304 size_t getStorageLimit()
const override {
305 return max_store_size;
313 return {total_store_size, total_values};
316 std::vector<SockAddr> getPublicAddress(sa_family_t family = 0)
override;
319 void resubscribe(
unsigned) {}
326 static constexpr unsigned SEARCH_NODES {14};
336 static constexpr unsigned SEARCH_MAX_BAD_NODES {25};
339 static constexpr unsigned MAX_REQUESTED_SEARCH_NODES {4};
342 static constexpr unsigned LISTEN_NODES {4};
345 static constexpr unsigned MAX_HASHES {1024 * 1024 * 1024};
348 static constexpr unsigned MAX_SEARCHES {1024 * 1024};
350 static constexpr std::chrono::minutes MAX_STORAGE_MAINTENANCE_EXPIRE_TIME {10};
353 static constexpr std::chrono::minutes SEARCH_EXPIRE_TIME {62};
356 static constexpr duration LISTEN_EXPIRE_TIME {std::chrono::seconds(30)};
357 static constexpr duration LISTEN_EXPIRE_TIME_PUBLIC {std::chrono::minutes(5)};
359 static constexpr duration REANNOUNCE_MARGIN {std::chrono::seconds(10)};
361 static constexpr std::chrono::seconds BOOTSTRAP_PERIOD {10};
363 static constexpr size_t TOKEN_SIZE {32};
372 Dht(
const Dht&) =
delete;
373 Dht& operator=(
const Dht&) =
delete;
380 uint64_t oldsecret {};
385 using SearchMap = std::map<InfoHash, Sp<Search>>;
386 using ReportedAddr = std::pair<unsigned, SockAddr>;
389 RoutingTable buckets {};
390 SearchMap searches {};
391 unsigned pending_pings {0};
393 std::vector<ReportedAddr> reported_addr;
396 NodeStats getNodesStats(time_point now,
const InfoHash& myid)
const;
401 PublicAddressChangedCb publicAddressChangedCb_ {};
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 {};
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};
414 size_t max_searches {MAX_SEARCHES};
415 size_t search_id {0};
419 std::map<size_t, std::tuple<size_t, size_t, size_t>> listeners {};
420 size_t listener_token {1};
425 Sp<Scheduler::Job> nextNodesConfirmation {};
426 Sp<Scheduler::Job> nextStorageMaintenance {};
428 net::NetworkEngine network_engine;
430 std::string persistPath;
436 const bool is_bootstrap {
false};
437 const bool maintain_storage {
false};
438 const bool public_stable {
false};
440 inline const duration& getListenExpiration()
const {
441 return public_stable ? LISTEN_EXPIRE_TIME_PUBLIC : LISTEN_EXPIRE_TIME;
444 void rotateSecrets();
446 Blob makeToken(
const SockAddr&,
bool old)
const;
447 bool tokenMatch(
const Blob& token,
const SockAddr&)
const;
449 void reportedAddr(
const SockAddr&);
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);
456 void expireStorage(InfoHash h);
457 void expireStore(
decltype(store)::iterator);
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;
468 void dataPersistence(InfoHash
id);
469 size_t maintainStorage(
decltype(store)::value_type&,
bool force=
false,
const DoneCallback& donecb={});
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);
481 const Bucket* findBucket(
const InfoHash&
id, sa_family_t af)
const {
482 return const_cast<Dht*
>(
this)->findBucket(
id, af);
485 void expireBuckets(RoutingTable&);
486 void sendCachedPing(Bucket& b);
487 bool bucketMaintenance(RoutingTable&);
488 void dumpBucket(
const Bucket& b, std::ostream& out)
const;
490 void startBootstrap();
491 void stopBootstrap();
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);
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; }
506 Sp<Search> search(
const InfoHash&
id, sa_family_t af, GetCallback = {}, QueryCallback = {}, DoneCallback = {}, Value::Filter = {},
const Sp<Query>& q = {});
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 = {});
518 unsigned refill(Search& sr);
519 void expireSearches();
525 void onDisconnected();
535 void searchNodeGetDone(
const net::Request& status,
536 net::RequestAnswer&& answer,
537 std::weak_ptr<Search> ws,
549 void searchNodeGetExpired(
const net::Request& status,
bool over, std::weak_ptr<Search> ws, Sp<Query> query);
558 void paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n);
563 SearchNode* searchSendGetValues(Sp<Search> sr, SearchNode *n =
nullptr,
bool update =
true);
571 void searchSendAnnounceValue(
const Sp<Search>& sr,
unsigned syncLevel = TARGET_NODES);
580 void searchStep(std::weak_ptr<Search> ws);
582 void searchSynchedNodeListen(
const Sp<Search>&, SearchNode&);
584 void dumpSearch(
const Search& sr, std::ostream& out)
const;
586 bool neighbourhoodMaintenance(RoutingTable&);
588 void onError(Sp<net::Request> node, net::DhtProtocolException e);
590 void onReportedAddr(
const InfoHash&
id,
const SockAddr&);
592 net::RequestAnswer onPing(Sp<Node> node);
594 net::RequestAnswer onFindNode(Sp<Node> node,
const InfoHash& hash, want_t want);
595 void onFindNodeDone(
const Sp<Node>& status,
596 net::RequestAnswer& a,
599 net::RequestAnswer onGetValues(Sp<Node> node,
600 const InfoHash& hash,
603 void onGetValuesDone(
const Sp<Node>& status,
604 net::RequestAnswer& a,
606 const Sp<Query>& orig_query);
608 net::RequestAnswer onListen(Sp<Node> node,
609 const InfoHash& hash,
614 void onListenDone(
const Sp<Node>& status,
615 net::RequestAnswer& a,
618 net::RequestAnswer onAnnounce(Sp<Node> node,
619 const InfoHash& hash,
621 const std::vector<Sp<Value>>& v,
622 const time_point& created);
623 net::RequestAnswer onRefresh(Sp<Node> node,
624 const InfoHash& hash,
626 const Value::Id& vid);
627 void onAnnounceDone(
const Sp<Node>& status,
628 net::RequestAnswer& a,