mirrors/opendht
1
0 Fork 0
mirror of https://github.com/savoirfairelinux/opendht.git synced 2026-11-08 04:21:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

use libuv for networking and event scheduling

Browse Source
This commit is contained in:
Adrien Beraud committed 2017-09-18 20:33:38 +02:00
1 parent 22f54e7335
commit d644a9fe01
18 files changed
+776 -452
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

CMakeLists.txt
+4 -4
View File
@@ -19,6 +19,7 @@ option (OPENDHT_LTO "Build with LTO" OFF)
# Dependencies
list (APPEND CMAKE_MODULE_PATH "${PROJECT_SOURCE_DIR}/cmake")
find_package (PkgConfig)
pkg_search_module(libuv libuv)
find_package (GnuTLS 3.3 REQUIRED)
pkg_search_module (Nettle nettle)
find_package (Msgpack 1.2 REQUIRED)
@@ -26,7 +27,6 @@ if (OPENDHT_TOOLS)
find_package (Readline 6 REQUIRED)
endif ()
if (NOT OPENDHT_ARGON2)
find_package(PkgConfig)
pkg_search_module(argon2 libargon2)
if (argon2_FOUND)
message("Argon2 found " ${argon2_req})
@@ -38,7 +38,7 @@ if (NOT OPENDHT_ARGON2)
endif ()
# Build flags
set (CMAKE_CXX_STANDARD 11)
set (CMAKE_CXX_STANDARD 14)
set (CMAKE_CXX_STANDARD_REQUIRED on)
set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pthread -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -pedantic-errors -fvisibility=hidden")
set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
@@ -167,7 +167,7 @@ if (OPENDHT_STATIC)
target_link_libraries(opendht-static ${argon2_LIBRARIES})
target_include_directories(opendht-static SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
endif ()
target_link_libraries(opendht-static ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES})
target_link_libraries(opendht-static ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${libuv_LIBRARIES})
install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
endif ()
@@ -188,7 +188,7 @@ if (OPENDHT_SHARED)
target_link_libraries(opendht ${argon2_LIBRARIES})
target_include_directories(opendht SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
endif ()
target_link_libraries(opendht ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES})
target_link_libraries(opendht ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${libuv_LIBRARIES})
install (TARGETS opendht DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
endif ()
include/opendht/callbacks.h
+9
View File
@@ -49,6 +49,12 @@ struct OPENDHT_PUBLIC NodeStats {
std::string toString() const;
};
struct OPENDHT_PUBLIC NetworkConfig {
std::string bind_addr {"::0"};
in_port_t bind_port {0};
};
/**
* Dht configuration.
*/
@@ -68,6 +74,8 @@ struct OPENDHT_PUBLIC Config {
/** Makes the DHT responsible to maintain its stored values. Consumes more ressources. */
bool maintain_storage;
NetworkConfig network_config;
};
/**
@@ -87,6 +95,7 @@ using QueryCallback = std::function<bool(const std::vector<std::shared_ptr<Field
using GetCallback = std::function<bool(const std::vector<std::shared_ptr<Value>>& values)>;
using GetCallbackSimple = std::function<bool(std::shared_ptr<Value> value)>;
using ShutdownCallback = std::function<void()>;
using StatusCallback = std::function<void(NodeStatus, NodeStatus)>;
using CertificateStoreQuery = std::function<std::vector<std::shared_ptr<crypto::Certificate>>(const InfoHash& pk_id)>;
include/opendht/dht.h
+21 -14
View File
@@ -54,9 +54,6 @@ struct LocalListener;
/**
* Main Dht class.
* Provides a Distributed Hash Table node.
*
* Must be given open UDP sockets and ::periodic must be
* called regularly.
*/
class OPENDHT_PUBLIC Dht {
public:
@@ -73,7 +70,7 @@ public:
* Initialise the Dht with two open sockets (for IPv4 and IP6)
* and an ID for the node.
*/
Dht(int s, int s6, Config config);
Dht(uv_loop_t* loop, Config config);
virtual ~Dht();
/**
@@ -89,6 +86,9 @@ public:
NodeStatus getStatus() const {
return std::max(getStatus(AF_INET), getStatus(AF_INET6));
}
void setOnStatusChanged(StatusCallback&& cb) {
statusCb = std::move(cb);
}
/**
* Performs final operations before quitting.
@@ -136,13 +136,13 @@ public:
insertNode(n.id, SockAddr(n.ss, n.sslen));
}
void pingNode(const sockaddr*, socklen_t, DoneCallbackSimple&& cb={});
time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr&);
time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) {
return periodic(buf, buflen, SockAddr(from, fromlen));
void bootstrap(const std::string& host, const std::string& service);
void clearBootstrap() {
bootstrap_nodes.clear();
}
void pingNode(const SockAddr&, DoneCallbackSimple&& cb={});
/**
* Get a value by searching on all available protocols (IPv4, IPv6),
* and call the provided get callback when values are found at key.
@@ -307,6 +307,10 @@ public:
std::vector<SockAddr> getPublicAddress(sa_family_t family = 0);
SockAddr getBoundAddr() {
return network_engine.getBoundAddr();
}
protected:
Logger DHT_LOG;
bool logFilerEnable_ {};
@@ -378,6 +382,8 @@ private:
const bool is_bootstrap {false};
const bool maintain_storage {false};
std::vector<std::pair<std::string,std::string>> bootstrap_nodes {};
// the stuff
RoutingTable buckets4 {};
RoutingTable buckets6 {};
@@ -400,8 +406,7 @@ private:
// timing
Scheduler scheduler;
Sp<Scheduler::Job> nextNodesConfirmation {};
Sp<Scheduler::Job> nextStorageMaintenance {};
Sp<Job> nextNodesConfirmation {};
time_point mybucket_grow_time {time_point::min()}, mybucket6_grow_time {time_point::min()};
net::NetworkEngine network_engine;
@@ -410,9 +415,13 @@ private:
using ReportedAddr = std::pair<unsigned, SockAddr>;
std::vector<ReportedAddr> reported_addr;
std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};
NodeStatus status4 {NodeStatus::Disconnected},
status6 {NodeStatus::Disconnected};
StatusCallback statusCb {nullptr};
void updateStatus();
void rotateSecrets();
Blob makeToken(const SockAddr&, bool old) const;
@@ -554,8 +563,6 @@ private:
bool neighbourhoodMaintenance(RoutingTable&);
void processMessage(const uint8_t *buf, size_t buflen, const SockAddr&);
void onError(Sp<net::Request> node, net::DhtProtocolException e);
/* when our address is reported by a distant peer. */
void onReportedAddr(const InfoHash& id, const SockAddr&);
include/opendht/dhtrunner.h
+39 -33
View File
@@ -26,6 +26,7 @@
#include "sockaddr.h"
#include "log_enable.h"
#include "def.h"
#include "uv_utils.h"
#include <thread>
#include <mutex>
@@ -51,8 +52,6 @@ struct SecureDhtConfig;
class OPENDHT_PUBLIC DhtRunner {
public:
typedef std::function<void(NodeStatus, NodeStatus)> StatusCallback;
DhtRunner();
virtual ~DhtRunner();
@@ -241,16 +240,14 @@ public:
* Returns the currently bound address.
* @param f: address family of the bound address to retreive.
*/
const SockAddr& getBound(sa_family_t f = AF_INET) const {
return (f == AF_INET) ? bound4 : bound6;
}
SockAddr getBound() const;
/**
* Returns the currently bound port, in host byte order.
* @param f: address family of the bound port to retreive.
*/
in_port_t getBoundPort(sa_family_t f = AF_INET) const {
return ntohs(((sockaddr_in*)&getBound(f).first)->sin_port);
in_port_t getBoundPort() const {
return getBound().getPort();
}
std::pair<size_t, size_t> getStoreSize() const;
@@ -306,20 +303,24 @@ public:
* @param cb: Optional callback to receive general state information.
*/
void run(in_port_t port, const crypto::Identity identity, bool threaded = false, NetId network = 0) {
run(port, {
run({
/*.dht_config = */{
/*.node_config = */{
/*.node_id = */{},
/*.network = */network,
/*.is_bootstrap = */false,
/*.maintain_storage*/false
/*.maintain_storage = */false,
/* .network_config = */{
/*.addr = */ "::0",
/*.port = */ port
}
},
/*.id = */identity
},
/*.threaded = */threaded
});
}
void run(in_port_t port, Config config);
void run(Config config);
/**
* @param local4: Local IPv4 address and port to bind. Can be null.
@@ -329,25 +330,22 @@ public:
* @param threaded: If false, loop() must be called periodically. Otherwise a thread is launched.
* @param cb: Optional callback to receive general state information.
*/
void run(const sockaddr_in* local4, const sockaddr_in6* local6, Config config);
//void run(const sockaddr_in* local4, const sockaddr_in6* local6, Config config);
/**
* Same as @run(sockaddr_in, sockaddr_in6, Identity, bool, StatusCallback), but with string IP addresses and service (port).
*/
void run(const char* ip4, const char* ip6, const char* service, Config config);
//void run(const char* ip4, const char* ip6, const char* service, Config config);
void setOnStatusChanged(StatusCallback&& cb) {
statusCb = std::move(cb);
}
void setOnStatusChanged(StatusCallback&& cb);
/**
* In non-threaded mode, the user should call this method
* regularly and everytime a new packet is received.
* @return the next op
*/
time_point loop() {
std::lock_guard<std::mutex> lck(dht_mtx);
return loop_();
void loop() {
uv_loop_.runOnce();
}
/**
@@ -373,33 +371,41 @@ private:
*/
void tryBootstrapContinuously();
void doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDhtConfig config);
time_point loop_();
void doRun(SecureDhtConfig config);
void loop_();
static std::vector<std::pair<sockaddr_storage, socklen_t>> getAddrInfo(const std::string& host, const std::string& service);
NodeStatus getStatus() const {
return std::max(status4, status6);
void loopCallback() {
std::lock_guard<std::mutex> lck(dht_mtx);
if (running)
loop_();
else {
uv_close((uv_handle_t*)&uv_async_, nullptr);
uv_loop_.stop();
}
}
static void loop_callback(uv_async_t* handle) {
if (handle->data)
static_cast<DhtRunner*>(handle->data)->loopCallback();
}
std::unique_ptr<SecureDht> dht_;
EventLoop uv_loop_;
uv_async_t uv_async_;
mutable std::mutex dht_mtx {};
std::thread dht_thread {};
std::condition_variable cv {};
std::thread rcv_thread {};
std::mutex sock_mtx {};
std::vector<std::pair<Blob, SockAddr>> rcv {};
/** true if currently actively boostraping */
std::atomic_bool bootstraping {false};
/* bootstrap nodes given as (host, service) pairs */
std::vector<std::pair<std::string,std::string>> bootstrap_nodes_all {};
/*std::vector<std::pair<std::string,std::string>> bootstrap_nodes_all {};
std::vector<std::pair<std::string,std::string>> bootstrap_nodes {};
std::thread bootstrap_thread {};
std::thread bootstrap_thread {};*/
/** protects bootstrap_nodes, bootstrap_thread */
std::mutex bootstrap_mtx {};
std::condition_variable bootstrap_cv {};
/*std::mutex bootstrap_mtx {};
std::condition_variable bootstrap_cv {};*/
std::queue<std::function<void(SecureDht&)>> pending_ops_prio {};
std::queue<std::function<void(SecureDht&)>> pending_ops {};
@@ -407,12 +413,12 @@ private:
std::atomic<bool> running {false};
NodeStatus status4 {NodeStatus::Disconnected},
/*NodeStatus status4 {NodeStatus::Disconnected},
status6 {NodeStatus::Disconnected};
StatusCallback statusCb {nullptr};
SockAddr bound4 {};
SockAddr bound6 {};
SockAddr bound6 {};*/
};
}
include/opendht/network_engine.h
+17 -8
View File
@@ -28,6 +28,8 @@
#include "rng.h"
#include "rate_limiter.h"
#include "net.h"
#include "callbacks.h"
#include "uv_utils.h"
#include <vector>
#include <string>
@@ -210,8 +212,8 @@ public:
using RequestCb = std::function<void(const Request&, RequestAnswer&&)>;
using RequestExpiredCb = std::function<void(const Request&, bool)>;
NetworkEngine(Logger& log, Scheduler& scheduler);
NetworkEngine(InfoHash& myid, NetId net, int s, int s6, Logger& log, Scheduler& scheduler,
NetworkEngine(uv_loop_t*, Logger& log, Scheduler& scheduler);
NetworkEngine(uv_loop_t*, const NetworkConfig& config, InfoHash& myid, NetId net, Logger& log, Scheduler& scheduler,
decltype(NetworkEngine::onError) onError,
decltype(NetworkEngine::onNewNode) onNewNode,
decltype(NetworkEngine::onReportedAddr) onReportedAddr,
@@ -225,6 +227,9 @@ public:
virtual ~NetworkEngine();
void clear();
void close(OnClose cb) {
sock->close(cb);
}
/**
* Sends values (with closest nodes) to a listenner.
@@ -244,7 +249,7 @@ public:
std::vector<Sp<Value>>&& values, const Query& q);
bool isRunning(sa_family_t af) const;
inline want_t want () const { return dht_socket >= 0 && dht_socket6 >= 0 ? (WANT4 | WANT6) : -1; }
inline want_t want () const { /*return dht_socket >= 0 && dht_socket6 >= 0 ? (WANT4 | WANT6) : -1;*/return WANT4 | WANT6; }
/**
* Cancel a request. Setting req->cancelled = true is not enough in the case
@@ -280,8 +285,8 @@ public:
* @return the request with information concerning its success.
*/
Sp<Request>
sendPing(const sockaddr* sa, socklen_t salen, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
return sendPing(std::make_shared<Node>(zeroes, sa, salen),
sendPing(const SockAddr& sa, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
return sendPing(std::make_shared<Node>(zeroes, sa),
std::forward<RequestCb>(on_done),
std::forward<RequestExpiredCb>(on_expired));
}
@@ -440,6 +445,10 @@ public:
return cache.getCachedNodes(id, sa_f, count);
}
SockAddr getBoundAddr() {
return sock->getBoundAddr();
}
private:
struct PartialMessage;
@@ -505,7 +514,8 @@ private:
// basic wrapper for socket sendto function
int send(const char *buf, size_t len, int flags, const SockAddr& addr);
int send(msgpack::sbuffer& msg, const SockAddr& addr);
int send(const Blob& msg, int flags, const SockAddr& addr);
void sendValueParts(TransId tid, const std::vector<Blob>& svals, const SockAddr& addr);
std::vector<Blob> packValueHeader(msgpack::sbuffer&, const std::vector<Sp<Value>>&);
@@ -547,10 +557,9 @@ private:
/* DHT info */
const InfoHash& myid;
const NetId network {0};
const int dht_socket {-1};
const int dht_socket6 {-1};
const Logger& DHT_LOG;
std::shared_ptr<UdpSocket> sock;
NodeCache cache {};
/**
include/opendht/scheduler.h
+86 -41
View File
@@ -22,6 +22,9 @@
#include "utils.h"
#include "log_enable.h"
#include "uv_utils.h"
#include <uv.h>
#include <functional>
#include <map>
@@ -36,12 +39,7 @@ namespace dht {
*/
class Scheduler {
public:
Scheduler(const Logger& l) : DHT_LOG(l) {}
struct Job {
Job(std::function<void()>&& f) : do_(std::move(f)) {}
std::function<void()> do_;
};
Scheduler(uv_loop_t* loop, const Logger& l) : loop_(loop), DHT_LOG(l) {}
/**
* Adds another job to the queue.
@@ -51,10 +49,32 @@ public:
*
* @return pointer to the newly scheduled job.
*/
Sp<Scheduler::Job> add(time_point t, std::function<void()>&& job_func) {
auto job = std::make_shared<Job>(std::move(job_func));
if (t != time_point::max())
timers.emplace(std::move(t), job);
Sp<Job> add(const time_point& t, std::function<void()>&& job_func) {
if (not loop_)
return {};
auto job = Job::make(loop_, std::move(job_func));
if (t != time_point::max()) {
garbageCollection();
pending_jobs.emplace(job);
job->run(t);
} else
job->cancel();
return job;
}
Sp<Job> add(const duration& t, std::function<void()>&& job_func) {
if (not loop_)
return {};
auto job = Job::make(loop_, std::move(job_func));
garbageCollection();
pending_jobs.emplace(job);
job->run(t);
return job;
}
Sp<Job> run(std::function<void()>&& job_func) {
auto job = Job::make(loop_, std::move(job_func));
garbageCollection();
pending_jobs.emplace(job);
job->run();
return job;
}
@@ -66,7 +86,9 @@ public:
*
* @return pointer to the newly scheduled job.
*/
void edit(Sp<Scheduler::Job>& job, time_point t) {
void edit(Sp<Job>& job, const duration& t = {}) {
if (not loop_)
return;
if (not job) {
DHT_LOG.ERR("editing an empty job");
return;
@@ -75,50 +97,73 @@ public:
// Force clearing old value.
auto task = std::move(job->do_);
job->do_ = {};
job->cancel();
job = add(t, std::move(task));
}
void edit(Sp<Job>& job, const time_point& t) {
if (not loop_)
return;
if (not job) {
DHT_LOG.ERR("editing an empty job");
return;
}
auto task = std::move(job->do_);
job->do_ = {};
job->cancel();
job = add(t, std::move(task));
}
/**
* Runs the jobs to do up to now.
*
* @return The time for the next job to run.
*/
time_point run() {
syncTime();
while (not timers.empty()) {
auto timer = timers.begin();
/*
* Running jobs scheduled before "now" prevents run+rescheduling
* loops before this method ends. It is garanteed by the fact that a
* job will at least be scheduled for "now" and not before.
*/
if (timer->first > now)
break;
auto job = std::move(timer->second);
timers.erase(timer);
if (job->do_)
job->do_();
}
return getNextJobTime();
void stop() {
loop_ = nullptr;
size_t del {0};
for (auto& w : pending_jobs)
if (auto job = w.lock()) {
job->cancel();
del++;
}
pending_jobs.clear();
DHT_LOG.DEBUG("stopping scheduler, %lu events canceled", del);
}
inline time_point getNextJobTime() const {
return timers.empty() ? time_point::max() : timers.begin()->first;
~Scheduler(){
stop();
}
/**
* Accessors for the common time reference used for synchronizing
* operations.
*/
inline const time_point& time() const { return now; }
inline time_point syncTime() { return (now = clock::now()); }
inline time_point time() const {
return time_point(std::chrono::duration_cast<duration>(std::chrono::milliseconds(uv_now(loop_))));
}
inline time_point syncTime() {
return time();
}
uv_loop_t* getLoop() const {
return loop_;
}
private:
time_point now {clock::now()};
std::multimap<time_point, Sp<Job>> timers {}; /* the jobs ordered by time */
std::set<std::weak_ptr<Job>, std::owner_less<std::weak_ptr<Job>>> pending_jobs {};
size_t lastSize_ {0};
uv_loop_t* loop_;
const Logger& DHT_LOG;
void garbageCollection() {
if (pending_jobs.size() > lastSize_ * 2) {
unsigned del {0};
for (auto it = pending_jobs.begin(); it != pending_jobs.end();) {
if (it->expired()) {
it = pending_jobs.erase(it);
del++;
} else
++it;
}
//DHT_LOG.DEBUG("pending_jobs cleanup: %u removed, %zu remaining", del, pending_jobs.size());
lastSize_ = pending_jobs.size();
}
}
};
}
include/opendht/securedht.h
+1 -1
View File
@@ -44,7 +44,7 @@ public:
* id: the identity to use for the crypto layer and to compute
* our own hash on the Dht.
*/
SecureDht(int s, int s6, Config config);
SecureDht(uv_loop_t* loop, Config config);
virtual ~SecureDht();
include/opendht/sockaddr.h
+18
View File
@@ -50,6 +50,15 @@ public:
throw std::runtime_error("Socket address length is too large");
std::copy_n((uint8_t*)sa, len, (uint8_t*)&first);
}
SockAddr(const sockaddr* sa) : pair<sockaddr_storage, socklen_t>::pair({},0) {
if (sa) {
if (sa->sa_family == AF_INET)
second = sizeof(sockaddr_in);
else if(sa->sa_family == AF_INET6)
second = sizeof(sockaddr_in6);
}
std::copy_n((uint8_t*)sa, second, (uint8_t*)&first);
}
SockAddr(const sockaddr_storage& ss, socklen_t len) : SockAddr((const sockaddr*)&ss, len) {}
bool operator<(const SockAddr& o) const {
@@ -117,6 +126,12 @@ public:
sockaddr_in6& getIPv6() {
return *reinterpret_cast<sockaddr_in6*>(&first);
}
sockaddr* get() {
return reinterpret_cast<sockaddr*>(&first);
}
const sockaddr* get() const {
return reinterpret_cast<const sockaddr*>(&first);
}
/**
* Return true if address is a loopback IP address.
@@ -130,6 +145,9 @@ public:
bool isUnspecified() const;
bool isMappedIPv4() const;
SockAddr getMappedIPv4() const;
/**
* A comparator to classify IP addresses, only considering the
* first 64 bits in IPv6.
include/opendht/uv_utils.h
+319
View File
@@ -0,0 +1,319 @@
#pragma once
#include "utils.h"
#include <uv.h>
namespace dht {
struct EventLoop {
using OnSignal = std::function<void(int)>;
EventLoop() {
uv_loop_init(&loop);
uv_signal_init(&loop, &sig_handler);
sig_handler.data = this;
uv_signal_start(&sig_handler, &EventLoop::signal_handler, SIGINT);
}
~EventLoop() {
uv_loop_close(&loop);
}
auto run() {
return uv_run(&loop, UV_RUN_DEFAULT);
}
auto runOnce() {
return uv_run(&loop, UV_RUN_NOWAIT);
}
void stop() {
uv_signal_stop(&sig_handler);
uv_walk(&loop, &EventLoop::close_walk_cb, this);
}
uv_loop_t* get() { return &loop; }
void setOnSignal(OnSignal cb) {
onSignal_ = cb;
}
private:
uv_loop_t loop;
uv_signal_t sig_handler;
OnSignal onSignal_;
void onSignal(int sig) {
if (onSignal_)
onSignal_(sig);
}
static void close_walk_cb(uv_handle_t* handle, void* /*arg*/) {
if (not uv_is_closing(handle))
uv_close(handle, [](uv_handle_t* handle) {
std::cout << "handle " << handle << " of type " << handle->type << " closed" << std::endl;
});
}
static void signal_handler(uv_signal_t* handle, int signum) {
if (handle->data) {
static_cast<EventLoop*>(handle->data)->onSignal(signum);
}
}
};
using GetAddrInfoCb = std::function<void(std::vector<SockAddr>&&)>;
void getAddrInfo(uv_loop_t* loop, const char* node, const char* service, GetAddrInfoCb&& cb);
struct Job : std::enable_shared_from_this<Job> {
Job(uv_loop_t* loop, std::function<void()>&& f = {}) : do_(std::move(f)) {
uv_timer_init(loop, &timer_);
timer_.data = nullptr;
}
void run(uint64_t t = 0) {
uv_timer_start(&timer_, &Job::timer_cb, t, 0);
}
inline void run(const duration& t) {
run(getTimeout(t));
}
inline void run(const time_point& t) {
run(getTimeout(t));
}
void cancel() {
if (timer_.data and not uv_is_closing((uv_handle_t*)&timer_)) {
uv_timer_stop(&timer_);
uv_close((uv_handle_t*)&timer_, &Job::close_cb);
}
}
static Sp<Job> make(uv_loop_t* loop, std::function<void()>&& f = {}) {
auto job = std::make_shared<Job>(loop, std::move(f));
job->timer_.data = new Sp<Job>(job);
return job;
}
static uint64_t getTimeout(const duration& d) {
return std::chrono::duration_cast<std::chrono::milliseconds>(d).count();
}
uint64_t getTimeout(const time_point& tp) const {
uint64_t t = getTimeout(tp.time_since_epoch());
uint64_t now = uv_now(timer_.loop);
return (t < now) ? 0 : t - now;
}
std::function<void()> do_;
private:
uv_timer_t timer_;
void onTimer() {
if (do_)
do_();
cancel();
}
void onClosed() {
if (timer_.data) {
delete static_cast<Sp<Job>*>(timer_.data);
timer_.data = nullptr;
}
}
static void timer_cb(uv_timer_t* handle) {
if (handle->data) {
auto timer = *static_cast<Sp<Job>*>(handle->data);
timer->onTimer();
}
}
static void close_cb(uv_handle_t* handle) {
if (handle->data) {
auto timer = *static_cast<Sp<Job>*>(handle->data);
timer->onClosed();
} else {
std::cerr << "Job::onClosed(): no data! " << handle << std::endl;
}
}
};
static inline void alloc_buffer(uv_handle_t* /*handle*/, size_t suggested_size, uv_buf_t *buf) {
buf->base = (char*)malloc(suggested_size);
buf->len = suggested_size;
}
using OnClose = std::function<void()>;
struct UdpSocket : public std::enable_shared_from_this<UdpSocket>
{
using OnReceive = std::function<void(const uint8_t* data, size_t size, const SockAddr& addr)>;
UdpSocket(uv_loop_t* loop) {
uv_udp_init(loop, &sock);
sock.data = nullptr;
}
void open(const char* bind_addr, in_port_t port, OnReceive cb) {
if (not sock.data)
sock.data = new std::shared_ptr<UdpSocket>(shared_from_this());
receive_cb = cb;
struct sockaddr_in6 addr;
uv_ip6_addr(bind_addr, port, &addr);
uv_udp_bind(&sock, (const struct sockaddr*)&addr, 0);
uv_udp_recv_start(&sock, alloc_buffer, &UdpSocket::on_receive);
}
int send(uint8_t* data, size_t size, const SockAddr& addr) {
auto req = new SendReq;
req->msg = data;
req->req.data = req;
const uv_buf_t buf {
/*.base =*/ (char*)data,
/*.len =*/ size
};
return uv_udp_send(&req->req, &sock, &buf, 1, addr.get(), &UdpSocket::on_sent);
}
void close(OnClose cb) {
close_cb = cb;
uv_close((uv_handle_t*) &sock, &UdpSocket::on_close);
}
bool isRunning() const {
return uv_is_active((uv_handle_t*) &sock);
}
SockAddr getBoundAddr() {
SockAddr addr;
int size = sizeof(addr.first);
uv_udp_getsockname(&sock, addr.get(), &size);
addr.second = size;
return addr;
}
private:
struct SendReq {
uint8_t* msg;
uv_udp_send_t req;
};
uv_udp_t sock;
OnReceive receive_cb;
OnClose close_cb;
void onReceive(ssize_t nread, const uv_buf_t* buf, const struct sockaddr* addr, unsigned /*flags*/) {
if (nread > 0 and addr and receive_cb) {
receive_cb((const uint8_t*)buf->base, nread, SockAddr(addr));
}
}
void onClosed() {
if (close_cb)
close_cb();
if (sock.data) {
delete static_cast<std::shared_ptr<UdpSocket>*>(sock.data);
sock.data = nullptr;
}
}
void onSent(SendReq& req, int /*status*/) {
// if (status) fprintf(stderr, "onSent %s\n", uv_strerror(status));
free(req.msg);
}
static std::shared_ptr<UdpSocket> get(uv_handle_t* handle) {
return handle->data ? *static_cast<std::shared_ptr<UdpSocket>*>(handle->data) : std::shared_ptr<UdpSocket>();
}
static void on_close(uv_handle_t* handle) {
if (auto node = get(handle))
node->onClosed();
}
static void on_sent(uv_udp_send_t* req, int status) {
if (auto data = static_cast<SendReq*>(req->data)) {
if (auto node = get((uv_handle_t*)req->handle))
node->onSent(*data, status);
delete data;
}
}
static void on_receive(uv_udp_t* handle, ssize_t nread, const uv_buf_t* buf, const struct sockaddr* addr, unsigned flags) {
if (auto node = get((uv_handle_t*)handle))
node->onReceive(nread, buf, addr, flags);
if (buf->base)
free(buf->base);
}
};
struct TcpSocket : public std::enable_shared_from_this<TcpSocket>
{
using OnConnect = std::function<void(const std::shared_ptr<TcpSocket>& new_socket)>;
TcpSocket(uv_loop_t* loop) {
uv_tcp_init(loop, &sock);
sock.data = nullptr;
}
void listen(in_port_t port, OnConnect cb) {
if (not sock.data)
sock.data = new std::shared_ptr<TcpSocket>(shared_from_this());
connect_cb = cb;
struct sockaddr_in6 addr;
uv_ip6_addr("::0", port, &addr);
uv_tcp_bind(&sock, (const struct sockaddr*)&addr, 0);
if (int r = uv_listen((uv_stream_t*)&sock, 128, &TcpSocket::on_new_connection)) {
fprintf(stderr, "Listen error %s\n", uv_strerror(r));
}
}
void accept(TcpSocket& server) {
if (not sock.data)
sock.data = new std::shared_ptr<TcpSocket>(shared_from_this());
if (uv_accept((uv_stream_t*)&server.sock, (uv_stream_t*) &sock) == 0) {
fprintf(stderr, "uv_read_start\n");
uv_read_start((uv_stream_t*)&sock, alloc_buffer, &TcpSocket::echo_read);
} else {
close();
}
}
void close(OnClose onclose = {}) {
fprintf(stderr, "uv_close\n");
close_cb = onclose;
uv_close((uv_handle_t*) &sock, &TcpSocket::on_close);
}
private:
uv_tcp_t sock;
OnConnect connect_cb;
OnClose close_cb;
void onNewConnection(int status) {
if (status < 0) {
fprintf(stderr, "New connection error %s\n", uv_strerror(status));
return;
}
auto s = std::make_shared<TcpSocket>(sock.loop);
s->accept(*this);
connect_cb(s);
fprintf(stderr, "on_new_connection end\n");
}
void onRead(ssize_t nread, const uv_buf_t *buf) {
if (nread < 0) {
if (nread != UV_EOF) {
fprintf(stderr, "Read error %s\n", uv_err_name(nread));
} else {
fprintf(stderr, "EOF\n");
}
close();
} else if (nread > 0) {
uv_write_t *req = (uv_write_t *) malloc(sizeof(uv_write_t));
uv_buf_t wrbuf = uv_buf_init(buf->base, nread);
fprintf(stderr, "uv_write\n");
uv_write(req, (uv_stream_t*) &sock, &wrbuf, 1, &TcpSocket::echo_write);
}
}
void onClosed() {
if (close_cb)
close_cb();
if (sock.data) {
delete static_cast<std::shared_ptr<TcpSocket>*>(sock.data);
sock.data = nullptr;
}
}
static std::shared_ptr<TcpSocket> get(uv_handle_t* handle) {
return handle->data ? *static_cast<std::shared_ptr<TcpSocket>*>(handle->data) : std::shared_ptr<TcpSocket>();
}
static void on_new_connection(uv_stream_t* handle, int status) {
if (auto node = get((uv_handle_t*)handle))
node->onNewConnection(status);
}
static void on_close(uv_handle_t* handle) {
if (auto node = get(handle))
node->onClosed();
}
static void echo_read(uv_stream_t* client, ssize_t nread, const uv_buf_t *buf) {
if (auto node = get((uv_handle_t*)client))
node->onRead(nread, buf);
if (buf->base)
free(buf->base);
}
static void echo_write(uv_write_t *req, int status) {
if (status) {
fprintf(stderr, "Write error %s\n", uv_strerror(status));
}
free(req);
}
};
}
python/opendht.pyx
+10 -6
View File
@@ -300,6 +300,8 @@ cdef class NodeSet(object):
return s
def __iter__(self):
return NodeSetIter(self)
def __len__(self):
return self.size()
cdef class PrivateKey(_WithID):
cdef shared_ptr[cpp.PrivateKey] _key
@@ -442,6 +444,10 @@ cdef class DhtConfig(object):
self._config.dht_config.node_config.node_id = id._infohash
def setNetwork(self, unsigned netid):
self._config.dht_config.node_config.network = netid
def setPort(self, cpp.in_port_t port):
self._config.dht_config.node_config.network_config.bind_port = port
def setBindAddr(self, string addr):
self._config.dht_config.node_config.network_config.bind_addr = addr
def setMaintainStorage(self, bool maintain_storage):
self._config.dht_config.node_config.maintain_storage = maintain_storage
@@ -484,12 +490,10 @@ cdef class DhtRunner(_WithID):
def run(self, Identity id=None, is_bootstrap=False, cpp.in_port_t port=0, str ipv4="", str ipv6="", DhtConfig config=DhtConfig()):
if id:
config.setIdentity(id)
if ipv4 or ipv6:
bind4 = ipv4.encode() if ipv4 else b''
bind6 = ipv6.encode() if ipv6 else b''
self.thisptr.get().run(bind4, bind6, str(port).encode(), config._config)
else:
self.thisptr.get().run(port, config._config)
config.setPort(port)
if ipv6:
config.setBindAddr(ipv6)
self.thisptr.get().run(config._config)
def join(self):
self.thisptr.get().join()
def shutdown(self, shutdown_cb=None):
python/opendht_cpp.pxd
+5 -2
View File
@@ -198,11 +198,15 @@ cdef extern from "opendht/callbacks.h" namespace "dht":
cdef DoneCallback bindDoneCb(DoneCallbackRaw cb, void *user_data)
cdef DoneCallbackSimple bindDoneCbSimple(DoneCallbackSimpleRaw cb, void *user_data)
cppclass NetworkConfig:
string bind_addr
in_port_t bind_port
cppclass Config:
InfoHash node_id
uint32_t network
bool is_bootstrap
bool maintain_storage
NetworkConfig network_config
cppclass SecureDhtConfig:
Config node_config
Identity id
@@ -219,8 +223,7 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
InfoHash getNodeId() const
void bootstrap(const_char*, const_char*)
void bootstrap(const SockAddr&, DoneCallbackSimple done_cb)
void run(in_port_t, Config config)
void run(const_char*, const_char*, const_char*, Config config)
void run(Config config)
void join()
void shutdown(ShutdownCallback)
bool isRunning()
python/setup.py.in
+2 -2
View File
@@ -50,8 +50,8 @@ setup(name="opendht",
["@CURRENT_SOURCE_DIR@/opendht.pyx"],
include_dirs = ['@PROJECT_SOURCE_DIR@/include'],
language="c++",
extra_compile_args=["-std=c++11"],
extra_link_args=["-std=c++11"],
extra_compile_args=["-std=c++14"],
extra_link_args=["-std=c++14"],
libraries=["opendht"],
library_dirs = ['@CURRENT_BINARY_DIR@', '@PROJECT_BINARY_DIR@']
))
src/dht.cpp
+77 -68
View File
@@ -62,8 +62,16 @@ Dht::getStatus(sa_family_t af) const
void
Dht::shutdown(ShutdownCallback cb)
{
auto final_shutdown = [this,cb](){
scheduler.stop();
DHT_LOG.w("closing socket...");
network_engine.close([=](){
if (cb)
cb();
});
};
if (not maintain_storage) {
if (cb) cb();
final_shutdown();
return;
}
@@ -73,7 +81,7 @@ Dht::shutdown(ShutdownCallback cb)
auto str_donecb = [=](bool, const std::vector<Sp<Node>>&) {
--*remaining;
DHT_LOG.w("shuting down node: %u ops remaining", *remaining);
if (!*remaining && cb) { cb(); }
if (!*remaining) { final_shutdown(); }
};
for (auto& str : store)
@@ -81,8 +89,7 @@ Dht::shutdown(ShutdownCallback cb)
if (!*remaining) {
DHT_LOG.w("shuting down node: %u ops remaining", *remaining);
if (cb)
cb();
final_shutdown();
}
}
@@ -241,8 +248,11 @@ Dht::onNewNode(const Sp<Node>& node, int confirm)
if (confirm or not b->cached)
b->cached = node;
} else {
auto empty = b->nodes.empty();
/* Create a new node. */
b->nodes.emplace_front(node);
if (empty)
updateStatus();
}
}
@@ -767,7 +777,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
if (sr->nextSearchStep)
scheduler.edit(sr->nextSearchStep, sr->getNextStepTime(scheduler.time()));
else
sr->nextSearchStep = scheduler.add(scheduler.time(), std::bind(&Dht::searchStep, this, sr));
sr->nextSearchStep = scheduler.run(std::bind(&Dht::searchStep, this, sr));
return sr;
}
@@ -1048,6 +1058,20 @@ bool callbackWrapper(Cb get_cb,
return !op->status.ok;
}
void
Dht::bootstrap(const std::string& host, const std::string& service)
{
for (const auto& bn : bootstrap_nodes)
if (bn.first == host and bn.second == service)
return;
bootstrap_nodes.emplace_back(host, service);
getAddrInfo(scheduler.getLoop(), host.c_str(), service.c_str(), [this](std::vector<SockAddr>&& res) {
for (const auto& addr : res)
pingNode(addr);
});
}
void
Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filter&& filter, Where&& where)
{
@@ -1391,6 +1415,7 @@ Dht::expireStore()
void
Dht::connectivityChanged(sa_family_t af)
{
updateStatus();
const auto& now = scheduler.time();
scheduler.edit(nextNodesConfirmation, now);
auto& bucket_grow_time = (af == AF_INET) ? mybucket_grow_time : mybucket6_grow_time;
@@ -1418,14 +1443,13 @@ Dht::rotateSecrets()
secret = std::uniform_int_distribution<uint64_t>{}(rdev);
}
uniform_duration_distribution<> time_dist(std::chrono::minutes(15), std::chrono::minutes(45));
auto rotate_secrets_time = scheduler.time() + time_dist(rd);
scheduler.add(rotate_secrets_time, std::bind(&Dht::rotateSecrets, this));
scheduler.add(time_dist(rd), std::bind(&Dht::rotateSecrets, this));
}
Blob
Dht::makeToken(const SockAddr& addr, bool old) const
{
const void *ip;
const void* ip;
size_t iplen;
in_port_t port;
@@ -1434,12 +1458,12 @@ Dht::makeToken(const SockAddr& addr, bool old) const
const auto& sin = addr.getIPv4();
ip = &sin.sin_addr;
iplen = 4;
port = sin.sin_port;
port = ntohs(sin.sin_port);
} else if (family == AF_INET6) {
const auto& sin6 = addr.getIPv6();
ip = &sin6.sin6_addr;
iplen = 16;
port = sin6.sin6_port;
port = ntohs(sin6.sin6_port);
} else {
return {};
}
@@ -1745,14 +1769,14 @@ Dht::~Dht()
s.second->clear();
}
Dht::Dht() : store(), scheduler(DHT_LOG), network_engine(DHT_LOG, scheduler) {}
Dht::Dht() : store(), scheduler(nullptr, DHT_LOG), network_engine(nullptr, DHT_LOG, scheduler) {}
Dht::Dht(int s, int s6, Config config)
Dht::Dht(uv_loop_t* loop, Config config)
: myid(config.node_id != zeroes ? config.node_id : InfoHash::getRandom()),
is_bootstrap(config.is_bootstrap),
maintain_storage(config.maintain_storage), store(), store_quota(),
scheduler(DHT_LOG),
network_engine(myid, config.network, s, s6, DHT_LOG, scheduler,
is_bootstrap(config.is_bootstrap), maintain_storage(config.maintain_storage),
buckets4{Bucket {AF_INET}},buckets6{Bucket {AF_INET6}}, store(), store_quota(),
scheduler(loop, DHT_LOG),
network_engine(loop, config.network_config, myid, config.network, DHT_LOG, scheduler,
std::bind(&Dht::onError, this, _1, _2),
std::bind(&Dht::onNewNode, this, _1, _2),
std::bind(&Dht::onReportedAddr, this, _1, _2),
@@ -1764,19 +1788,10 @@ Dht::Dht(int s, int s6, Config config)
std::bind(&Dht::onRefresh, this, _1, _2, _3, _4))
{
scheduler.syncTime();
if (s < 0 && s6 < 0)
return;
if (s >= 0)
buckets4 = {Bucket {AF_INET}};
if (s6 >= 0)
buckets6 = {Bucket {AF_INET6}};
search_id = std::uniform_int_distribution<decltype(search_id)>{}(rd);
uniform_duration_distribution<> time_dis {std::chrono::seconds(3), std::chrono::seconds(5)};
auto confirm_nodes_time = scheduler.time() + time_dis(rd);
DHT_LOG.d(myid, "Scheduling %s", myid.toString().c_str());
nextNodesConfirmation = scheduler.add(confirm_nodes_time, std::bind(&Dht::confirmNodes, this));
nextNodesConfirmation = scheduler.add(time_dis(rd), std::bind(&Dht::confirmNodes, this));
// Fill old secret
{
@@ -1823,7 +1838,13 @@ Dht::neighbourhoodMaintenance(RoutingTable& list)
n->toString().c_str(), id.toString().c_str());
/* Since our node-id is the same in both DHTs, it's probably
profitable to query both families. */
network_engine.sendFindNode(n, id, network_engine.want(), nullptr, nullptr);
network_engine.sendFindNode(n, id, network_engine.want(), nullptr, [this,n](const net::Request&, bool over) {
if (over) {
DHT_LOG.d(n->id, "[node %s] neighborhood maintenance op expired", n->toString().c_str());
scheduler.edit(nextNodesConfirmation, Node::MAX_RESPONSE_TIME);
updateStatus();
}
});
}
return true;
@@ -1876,10 +1897,9 @@ Dht::bucketMaintenance(RoutingTable& list)
auto start = scheduler.time();
network_engine.sendFindNode(n, id, want, nullptr, [this,start,n](const net::Request&, bool over) {
if (over) {
const auto& end = scheduler.time();
using namespace std::chrono;
DHT_LOG.d(n->id, "[node %s] bucket maintenance op expired after %llu ms", n->toString().c_str(), duration_cast<milliseconds>(end-start).count());
scheduler.edit(nextNodesConfirmation, end + Node::MAX_RESPONSE_TIME);
DHT_LOG.d(n->id, "[node %s] bucket maintenance op expired", n->toString().c_str());
scheduler.edit(nextNodesConfirmation, Node::MAX_RESPONSE_TIME);
updateStatus();
}
});
sent = true;
@@ -1952,45 +1972,34 @@ Dht::maintainStorage(decltype(store)::value_type& storage, bool force, DoneCallb
}
void
Dht::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& from)
Dht::updateStatus()
{
if (buflen == 0)
return;
try {
network_engine.processMessage(buf, buflen, from);
} catch (const std::exception& e) {
DHT_LOG.e("Can't parse message from %s: %s", from.toString().c_str(), e.what());
//auto code = e.getCode();
//if (code == net::DhtProtocolException::INVALID_TID_SIZE or code == net::DhtProtocolException::WRONG_NODE_INFO_BUF_LEN) {
/* This is really annoying, as it means that we will
time-out all our searches that go through this node.
Kill it. */
//const auto& id = e.getNodeId();
//blacklistNode(&id, from, fromlen);
///}
NodeStatus nstatus4 = getStatus(AF_INET);
NodeStatus nstatus6 = getStatus(AF_INET6);
if (nstatus4 != status4 || nstatus6 != status6) {
status4 = nstatus4;
status6 = nstatus6;
if (status4 == NodeStatus::Disconnected and status6 == NodeStatus::Disconnected) {
// We have lost connection with the DHT. Try to recover using bootstrap nodes.
DHT_LOG.e(myid, "DHT disconnected", myid.toString().c_str());
} else {
//bootstrap_nodes.clear();
}
if (statusCb)
statusCb(status4, status6);
}
}
time_point
Dht::periodic(const uint8_t *buf, size_t buflen, const SockAddr& from)
{
scheduler.syncTime();
processMessage(buf, buflen, from);
return scheduler.run();
}
void
Dht::expire()
{
uniform_duration_distribution<> time_dis(std::chrono::minutes(2), std::chrono::minutes(6));
auto expire_stuff_time = scheduler.time() + duration(time_dis(rd));
expireBuckets(buckets4);
expireBuckets(buckets6);
expireStore();
expireSearches();
scheduler.add(expire_stuff_time, std::bind(&Dht::expire, this));
scheduler.add(time_dis(rd), std::bind(&Dht::expire, this));
}
void
@@ -2000,11 +2009,13 @@ Dht::confirmNodes()
bool soon = false;
const auto& now = scheduler.time();
if (searches4.empty() and getStatus(AF_INET) == NodeStatus::Connected) {
updateStatus();
if (searches4.empty() and status4 == NodeStatus::Connected) {
DHT_LOG.d(myid, "[confirm nodes] initial IPv4 'get' for my id (%s)", myid.toString().c_str());
search(myid, AF_INET);
}
if (searches6.empty() and getStatus(AF_INET6) == NodeStatus::Connected) {
if (searches6.empty() and status6 == NodeStatus::Connected) {
DHT_LOG.d(myid, "[confirm nodes] initial IPv6 'get' for my id (%s)", myid.toString().c_str());
search(myid, AF_INET6);
}
@@ -2026,9 +2037,7 @@ Dht::confirmNodes()
auto time_dis = soon
? uniform_duration_distribution<> {seconds(5) , seconds(25)}
: uniform_duration_distribution<> {seconds(60), seconds(180)};
auto confirm_nodes_time = now + time_dis(rd);
scheduler.edit(nextNodesConfirmation, confirm_nodes_time);
scheduler.edit(nextNodesConfirmation, time_dis(rd));
}
std::vector<ValuesExport>
@@ -2135,13 +2144,13 @@ Dht::insertNode(const InfoHash& id, const SockAddr& addr)
}
void
Dht::pingNode(const sockaddr* sa, socklen_t salen, DoneCallbackSimple&& cb)
Dht::pingNode(const SockAddr& sa, DoneCallbackSimple&& cb)
{
scheduler.syncTime();
DHT_LOG.d("Sending ping to %s", print_addr(sa, salen).c_str());
auto& count = sa->sa_family == AF_INET ? pending_pings4 : pending_pings6;
DHT_LOG.d("Sending ping to %s", sa.toString().c_str());
auto& count = sa.getFamily() == AF_INET ? pending_pings4 : pending_pings6;
count++;
network_engine.sendPing(sa, salen, [&count,cb](const net::Request&, net::NetworkEngine::RequestAnswer&&) {
network_engine.sendPing(sa, [&count,cb](const net::Request&, net::NetworkEngine::RequestAnswer&&) {
count--;
if (cb)
cb(true);
@@ -2311,7 +2320,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
net::DhtProtocolException::LISTEN_NO_INFOHASH
};
}
if (!tokenMatch(token, node->addr)) {
if (not tokenMatch(token, node->addr)) {
DHT_LOG.w(hash, node->id, "[node %s] incorrect token %s for 'listen'", node->toString().c_str(), hash.toString().c_str());
throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::LISTEN_WRONG_TOKEN};
}
@@ -2349,7 +2358,7 @@ Dht::onAnnounce(Sp<Node> node,
net::DhtProtocolException::PUT_NO_INFOHASH
};
}
if (!tokenMatch(token, node->addr)) {
if (not tokenMatch(token, node->addr)) {
DHT_LOG.w(hash, node->id, "[node %s] incorrect token %s for 'put'", node->toString().c_str(), hash.toString().c_str());
throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::PUT_WRONG_TOKEN};
}
src/dhtrunner.cpp
+72 -225
View File
@@ -20,98 +20,32 @@
#include "dhtrunner.h"
#include "securedht.h"
#ifndef _WIN32
#include <unistd.h>
#else
#include <io.h>
#endif
#ifndef _WIN32
#include <sys/socket.h>
#else
#include <winsock2.h>
#include <ws2tcpip.h>
#define close(x) closesocket(x)
#endif
namespace dht {
constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;
DhtRunner::DhtRunner() : dht_()
{
#ifdef _WIN32
WSADATA wsd;
if (WSAStartup(MAKEWORD(2,2), &wsd) != 0)
throw DhtException("Can't initialize Winsock2");
#endif
uv_async_.data = this;
uv_async_init(uv_loop_.get(), &uv_async_, &DhtRunner::loop_callback);
}
DhtRunner::~DhtRunner()
{
join();
#ifdef _WIN32
WSACleanup();
#endif
}
void
DhtRunner::run(in_port_t port, DhtRunner::Config config)
{
sockaddr_in sin4;
std::fill_n((uint8_t*)&sin4, sizeof(sin4), 0);
sin4.sin_family = AF_INET;
sin4.sin_port = htons(port);
sockaddr_in6 sin6;
std::fill_n((uint8_t*)&sin6, sizeof(sin6), 0);
sin6.sin6_family = AF_INET6;
sin6.sin6_port = htons(port);
run(&sin4, &sin6, config);
}
void
DhtRunner::run(const char* ip4, const char* ip6, const char* service, DhtRunner::Config config)
{
auto res4 = getAddrInfo(ip4, service);
auto res6 = getAddrInfo(ip6, service);
run(res4.empty() ? nullptr : (sockaddr_in*) &res4.front().first,
res6.empty() ? nullptr : (sockaddr_in6*)&res6.front().first, config);
}
void
DhtRunner::run(const sockaddr_in* local4, const sockaddr_in6* local6, DhtRunner::Config config)
DhtRunner::run(DhtRunner::Config config)
{
if (running)
return;
if (rcv_thread.joinable())
rcv_thread.join();
running = true;
doRun(local4, local6, config.dht_config);
doRun(config.dht_config);
if (not config.threaded)
return;
dht_thread = std::thread([this]() {
while (running) {
std::unique_lock<std::mutex> lk(dht_mtx);
auto wakeup = loop_();
cv.wait_until(lk, wakeup, [this]() {
if (not running)
return true;
{
std::lock_guard<std::mutex> lck(sock_mtx);
if (not rcv.empty())
return true;
}
{
std::lock_guard<std::mutex> lck(storage_mtx);
if (not pending_ops_prio.empty())
return true;
auto s = getStatus();
if (not pending_ops.empty() and (s == NodeStatus::Connected or (s == NodeStatus::Disconnected and not bootstraping)))
return true;
}
return false;
});
}
uv_loop_.run();
});
}
@@ -121,21 +55,19 @@ DhtRunner::shutdown(ShutdownCallback cb) {
pending_ops_prio.emplace([=](SecureDht& dht) mutable {
dht.shutdown(cb);
});
cv.notify_all();
uv_async_send(&uv_async_);
}
void
DhtRunner::join()
{
running = false;
cv.notify_all();
bootstrap_cv.notify_all();
uv_async_send(&uv_async_);
//bootstrap_cv.notify_all();
if (dht_thread.joinable())
dht_thread.join();
if (rcv_thread.joinable())
rcv_thread.join();
if (bootstrap_thread.joinable())
bootstrap_thread.join();
/*if (bootstrap_thread.joinable())
bootstrap_thread.join();*/
{
std::lock_guard<std::mutex> lck(storage_mtx);
@@ -145,10 +77,10 @@ DhtRunner::join()
{
std::lock_guard<std::mutex> lck(dht_mtx);
dht_.reset();
status4 = NodeStatus::Disconnected;
/*status4 = NodeStatus::Disconnected;
status6 = NodeStatus::Disconnected;
bound4 = {};
bound6 = {};
bound6 = {};*/
}
}
@@ -175,6 +107,14 @@ DhtRunner::getNodeId() const
return dht_->getNodeId();
}
void
DhtRunner::setOnStatusChanged(StatusCallback&& cb)
{
std::lock_guard<std::mutex> lck(dht_mtx);
if (not dht_)
throw std::runtime_error("dht is not running");
dht_->setOnStatusChanged(std::move(cb));
}
std::pair<size_t, size_t>
DhtRunner::getStoreSize() const {
@@ -307,6 +247,11 @@ DhtRunner::getPublicAddressStr(sa_family_t af)
return ret;
}
SockAddr
DhtRunner::getBound() const {
return dht_->getBoundAddr();
}
void
DhtRunner::registerCertificate(std::shared_ptr<crypto::Certificate> cert) {
std::lock_guard<std::mutex> lck(dht_mtx);
@@ -318,42 +263,12 @@ DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) {
dht_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
}
time_point
void
DhtRunner::loop_()
{
if (!dht_)
return {};
decltype(pending_ops) ops {};
{
std::lock_guard<std::mutex> lck(storage_mtx);
auto s = getStatus();
ops = (pending_ops_prio.empty() && (s == NodeStatus::Connected or (s == NodeStatus::Disconnected and not bootstraping))) ?
std::move(pending_ops) : std::move(pending_ops_prio);
}
while (not ops.empty()) {
ops.front()(*dht_);
ops.pop();
}
time_point wakeup {};
decltype(rcv) received {};
{
std::lock_guard<std::mutex> lck(sock_mtx);
// move to stack
received = std::move(rcv);
}
if (not received.empty()) {
for (const auto& pck : received) {
auto& buf = pck.first;
auto& from = pck.second;
wakeup = dht_->periodic(buf.data(), buf.size()-1, from);
}
received.clear();
} else {
wakeup = dht_->periodic(nullptr, 0, nullptr, 0);
}
return;
/*
NodeStatus nstatus4 = dht_->getStatus(AF_INET);
NodeStatus nstatus6 = dht_->getStatus(AF_INET6);
if (nstatus4 != status4 || nstatus6 != status6) {
@@ -371,102 +286,24 @@ DhtRunner::loop_()
if (statusCb)
statusCb(status4, status6);
}
return wakeup;
*/
decltype(pending_ops) ops {};
{
std::lock_guard<std::mutex> lck(storage_mtx);
auto s = dht_->getStatus();
ops = (pending_ops_prio.empty() && (s == NodeStatus::Connected or (s == NodeStatus::Disconnected and not bootstraping))) ?
std::move(pending_ops) : std::move(pending_ops_prio);
}
while (not ops.empty()) {
ops.front()(*dht_);
ops.pop();
}
}
void
DhtRunner::doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDht::Config config)
DhtRunner::doRun(SecureDht::Config config)
{
dht_.reset();
int s4 = -1,
s6 = -1;
bound4 = {};
if (sin4) {
s4 = socket(PF_INET, SOCK_DGRAM, 0);
if(s4 >= 0) {
int rc = bind(s4, (sockaddr*)sin4, sizeof(sockaddr_in));
if(rc < 0)
throw DhtException("Can't bind IPv4 socket on " + dht::print_addr((sockaddr*)sin4, sizeof(sockaddr_in)));
bound4.second = sizeof(bound4.first);
getsockname(s4, (sockaddr*)&bound4.first, &bound4.second);
}
}
#if 1
bound6 = {};
if (sin6) {
s6 = socket(PF_INET6, SOCK_DGRAM, 0);
if(s6 >= 0) {
int val = 1;
int rc = setsockopt(s6, IPPROTO_IPV6, IPV6_V6ONLY, (char *)&val, sizeof(val));
if(rc < 0)
throw DhtException("Can't set IPV6_V6ONLY");
rc = bind(s6, (sockaddr*)sin6, sizeof(sockaddr_in6));
if(rc < 0)
throw DhtException("Can't bind IPv6 socket on " + dht::print_addr((sockaddr*)sin6, sizeof(sockaddr_in6)));
bound6.second = sizeof(bound6.first);
getsockname(s6, (sockaddr*)&bound6.first, &bound6.second);
}
}
#endif
dht_ = std::unique_ptr<SecureDht>(new SecureDht {s4, s6, config});
rcv_thread = std::thread([this,s4,s6]() {
try {
while (true) {
struct timeval tv {/*.tv_sec = */0, /*.tv_usec = */250000};
fd_set readfds;
FD_ZERO(&readfds);
if(s4 >= 0)
FD_SET(s4, &readfds);
if(s6 >= 0)
FD_SET(s6, &readfds);
int rc = select(s4 > s6 ? s4 + 1 : s6 + 1, &readfds, nullptr, nullptr, &tv);
if(rc < 0) {
if(errno != EINTR) {
perror("select");
std::this_thread::sleep_for( std::chrono::seconds(1) );
}
}
if(!running)
break;
if(rc > 0) {
std::array<uint8_t, 1024 * 64> buf;
SockAddr from;
from.second = sizeof(from.first);
if(s4 >= 0 && FD_ISSET(s4, &readfds))
rc = recvfrom(s4, (char*)buf.data(), buf.size(), 0, (struct sockaddr*)&from.first, &from.second);
else if(s6 >= 0 && FD_ISSET(s6, &readfds))
rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (struct sockaddr*)&from.first, &from.second);
else
break;
if (rc > 0) {
{
std::lock_guard<std::mutex> lck(sock_mtx);
rcv.emplace_back(Blob {buf.begin(), buf.begin()+rc+1}, from);
}
cv.notify_all();
}
}
}
} catch (const std::exception& e) {
std::cerr << "Error in DHT networking thread: " << e.what() << std::endl;
}
if (s4 >= 0)
close(s4);
if (s6 >= 0)
close(s6);
});
dht_ = std::unique_ptr<SecureDht>(new SecureDht {uv_loop_.get(), config});
}
void
@@ -475,10 +312,11 @@ DhtRunner::get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f
{
std::lock_guard<std::mutex> lck(storage_mtx);
pending_ops.emplace([=](SecureDht& dht) mutable {
std::cout << "DhtRunner dht.get " << hash << std::endl;
dht.get(hash, vcb, dcb, std::move(f), std::move(w));
});
}
cv.notify_all();
uv_async_send(&uv_async_);
}
void
@@ -493,7 +331,7 @@ void DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_
dht.query(hash, cb, done_cb, std::move(q));
});
}
cv.notify_all();
uv_async_send(&uv_async_);
}
std::future<size_t>
@@ -506,7 +344,7 @@ DhtRunner::listen(InfoHash hash, GetCallback vcb, Value::Filter f, Where w)
ret_token->set_value(dht.listen(hash, vcb, std::move(f), std::move(w)));
});
}
cv.notify_all();
uv_async_send(&uv_async_);
return ret_token->get_future();
}
@@ -525,7 +363,7 @@ DhtRunner::cancelListen(InfoHash h, size_t token)
dht.cancelListen(h, token);
});
}
cv.notify_all();
uv_async_send(&uv_async_);
}
void
@@ -538,7 +376,7 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> token)
dht.cancelListen(h, tk);
});
}
cv.notify_all();
uv_async_send(&uv_async_);
}
void
@@ -551,7 +389,7 @@ DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created
dht.put(hash, sv, cb, created, permanent);
});
}
cv.notify_all();
uv_async_send(&uv_async_);
}
void
@@ -563,7 +401,7 @@ DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, tim
dht.put(hash, value, cb, created, permanent);
});
}
cv.notify_all();
uv_async_send(&uv_async_);
}
void
@@ -581,7 +419,7 @@ DhtRunner::cancelPut(const InfoHash& h , const Value::Id& id)
dht.cancelPut(h, id);
});
}
cv.notify_all();
uv_async_send(&uv_async_);
}
void
@@ -593,7 +431,7 @@ DhtRunner::putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallback c
dht.putSigned(hash, value, cb);
});
}
cv.notify_all();
uv_async_send(&uv_async_);
}
void
@@ -617,7 +455,7 @@ DhtRunner::putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value
dht.putEncrypted(hash, to, value, cb);
});
}
cv.notify_all();
uv_async_send(&uv_async_);
}
void
@@ -632,6 +470,7 @@ DhtRunner::putEncrypted(const std::string& key, InfoHash to, Value&& value, Done
putEncrypted(InfoHash::get(key), to, std::forward<Value>(value), cb);
}
/*
void
DhtRunner::tryBootstrapContinuously()
{
@@ -689,7 +528,7 @@ DhtRunner::tryBootstrapContinuously()
}
} while (bootstraping);
});
}
}*/
std::vector<std::pair<sockaddr_storage, socklen_t>>
DhtRunner::getAddrInfo(const std::string& host, const std::string& service)
@@ -719,17 +558,25 @@ DhtRunner::getAddrInfo(const std::string& host, const std::string& service)
void
DhtRunner::bootstrap(const std::string& host, const std::string& service)
{
std::lock_guard<std::mutex> lck(bootstrap_mtx);
/*std::lock_guard<std::mutex> lck(bootstrap_mtx);
bootstrap_nodes_all.emplace_back(host, service);
bootstrap_nodes.emplace_back(host, service);
tryBootstrapContinuously();
tryBootstrapContinuously();*/
pending_ops_prio.emplace([=](SecureDht& dht) mutable {
dht.bootstrap(host, service);
});
uv_async_send(&uv_async_);
}
void
DhtRunner::clearBootstrap()
{
std::lock_guard<std::mutex> lck(bootstrap_mtx);
bootstrap_nodes_all.clear();
//std::lock_guard<std::mutex> lck(bootstrap_mtx);
//bootstrap_nodes_all.clear();
pending_ops_prio.emplace([=](SecureDht& dht) mutable {
dht.clearBootstrap();
});
uv_async_send(&uv_async_);
}
void
@@ -739,7 +586,7 @@ DhtRunner::bootstrap(const std::vector<std::pair<sockaddr_storage, socklen_t>>&
pending_ops_prio.emplace([=](SecureDht& dht) mutable {
auto rem = cb ? std::make_shared<std::pair<size_t, bool>>(nodes.size(), false) : nullptr;
for (auto& node : nodes)
dht.pingNode((sockaddr*)&node.first, node.second, cb ? [rem,cb](bool ok) {
dht.pingNode(SockAddr((sockaddr*)&node.first, node.second), cb ? [rem,cb](bool ok) {
auto& r = *rem;
r.first--;
r.second |= ok;
@@ -747,7 +594,7 @@ DhtRunner::bootstrap(const std::vector<std::pair<sockaddr_storage, socklen_t>>&
cb(r.second);
} : DoneCallbackSimple{});
});
cv.notify_all();
uv_async_send(&uv_async_);
}
void
@@ -770,7 +617,7 @@ DhtRunner::bootstrap(const std::vector<NodeExport>& nodes)
dht.insertNode(node);
});
}
cv.notify_all();
uv_async_send(&uv_async_);
}
void
@@ -782,7 +629,7 @@ DhtRunner::connectivityChanged()
dht.connectivityChanged();
});
}
cv.notify_all();
uv_async_send(&uv_async_);
}
void
@@ -793,7 +640,7 @@ DhtRunner::findCertificate(InfoHash hash, std::function<void(const std::shared_p
dht.findCertificate(hash, cb);
});
}
cv.notify_all();
uv_async_send(&uv_async_);
}
}
src/network_engine.cpp
+35 -44
View File
@@ -173,8 +173,8 @@ NetworkEngine::RequestAnswer::RequestAnswer(ParsedMessage&& msg)
: ntoken(std::move(msg.token)), values(std::move(msg.values)), fields(std::move(msg.fields)),
nodes4(std::move(msg.nodes4)), nodes6(std::move(msg.nodes6)) {}
NetworkEngine::NetworkEngine(Logger& log, Scheduler& scheduler) : myid(zeroes), DHT_LOG(log), scheduler(scheduler) {}
NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, int s, int s6, Logger& log, Scheduler& scheduler,
NetworkEngine::NetworkEngine(uv_loop_t*, Logger& log, Scheduler& scheduler) : myid(zeroes), DHT_LOG(log), scheduler(scheduler) {}
NetworkEngine::NetworkEngine(uv_loop_t* loop, const NetworkConfig& config, InfoHash& myid, NetId net, Logger& log, Scheduler& scheduler,
decltype(NetworkEngine::onError) onError,
decltype(NetworkEngine::onNewNode) onNewNode,
decltype(NetworkEngine::onReportedAddr) onReportedAddr,
@@ -186,7 +186,7 @@ NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, int s, int s6, Logger& l
decltype(NetworkEngine::onRefresh) onRefresh) :
onError(onError), onNewNode(onNewNode), onReportedAddr(onReportedAddr), onPing(onPing), onFindNode(onFindNode),
onGetValues(onGetValues), onListen(onListen), onAnnounce(onAnnounce), onRefresh(onRefresh), myid(myid),
network(net), dht_socket(s), dht_socket6(s6), DHT_LOG(log), scheduler(scheduler)
network(net), DHT_LOG(log), sock(std::make_shared<UdpSocket>(loop)), scheduler(scheduler)
{
if (dht_socket >= 0) {
if (!set_nonblocking(dht_socket, 1))
@@ -197,10 +197,11 @@ NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, int s, int s6, Logger& l
throw DhtException("Can't set socket to non-blocking mode");
}
transaction_id = std::uniform_int_distribution<decltype(transaction_id)>{1}(rd_device);
using namespace std::placeholders;
sock->open(config.bind_addr.c_str(), config.bind_port, std::bind(&NetworkEngine::processMessage, this, _1, _2, _3));
}
NetworkEngine::~NetworkEngine() {
clear();
}
void
@@ -218,18 +219,9 @@ NetworkEngine::tellListener(Sp<Node> node, uint32_t socket_id, const InfoHash& h
}
bool
NetworkEngine::isRunning(sa_family_t af) const
NetworkEngine::isRunning(sa_family_t /*af*/) const
{
switch (af) {
case 0:
return dht_socket >= 0 || dht_socket6 >= 0;
case AF_INET:
return dht_socket >= 0;
case AF_INET6:
return dht_socket6 >= 0;
default:
return false;
}
return sock->isRunning();
}
Sp<Socket>
@@ -294,7 +286,7 @@ NetworkEngine::requestStep(Sp<Request> sreq)
req.on_expired(req, false);
}
auto err = send((char*)req.msg.data(), req.msg.size(),
auto err = send(req.msg, req.msg.size(),
(node.reply_time >= now - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM,
node.addr);
if (err == ENETUNREACH ||
@@ -410,8 +402,9 @@ NetworkEngine::isNodeBlacklisted(const SockAddr& addr) const
}
void
NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& from)
NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& from_raw)
{
SockAddr from = from_raw.isMappedIPv4() ? from_raw.getMappedIPv4() : from_raw;
if (isMartian(from)) {
DHT_LOG.w("Received packet from martian node %s", from.toString().c_str());
return;
@@ -461,7 +454,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
process(std::move(pmsg_it->second.msg), from);
partial_messages.erase(pmsg_it);
} else
scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, msg->tid));
scheduler.add(RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, msg->tid));
}
return;
}
@@ -490,8 +483,8 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
pmsg.last_part = now;
auto wmsg = partial_messages.emplace(pmsg.msg->tid, std::move(pmsg));
if (wmsg.second) {
scheduler.add(now + RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, wmsg.first->first));
scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, wmsg.first->first));
scheduler.add(RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, wmsg.first->first));
scheduler.add(RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, wmsg.first->first));
} else
DHT_LOG.e("Partial message with given TID already exists");
}
@@ -508,8 +501,10 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
auto rsocket_it = opened_sockets.end();
if (req_it == requests.end())
rsocket_it = opened_sockets.find(msg->tid);
if (req_it == requests.end() and rsocket_it == opened_sockets.end())
throw DhtProtocolException {DhtProtocolException::UNKNOWN_TID, "Can't find transaction", msg->id};
if (req_it == requests.end() and rsocket_it == opened_sockets.end()) {
DHT_LOG.e("Can't find transaction.");
return;
}
auto req = req_it != requests.end() ? req_it->second : nullptr;
auto rsocket = rsocket_it != opened_sockets.end() ? rsocket_it->second : nullptr;
@@ -661,27 +656,23 @@ insertAddr(msgpack::packer<msgpack::sbuffer>& pk, const SockAddr& addr)
}
int
NetworkEngine::send(const char *buf, size_t len, int flags, const SockAddr& addr)
NetworkEngine::send(msgpack::sbuffer& msg, const SockAddr& addr)
{
if (addr.second == 0)
return EFAULT;
int s;
if (addr.getFamily() == AF_INET)
s = dht_socket;
else if (addr.getFamily() == AF_INET6)
s = dht_socket6;
else
s = -1;
// move data
size_t size = msg.size();
uint8_t* data = (uint8_t*)msg.release();
return sock->send(data, size, addr);
}
if (s < 0)
return EAFNOSUPPORT;
if (sendto(s, buf, len, flags, (const sockaddr*)&addr.first, addr.second) == -1) {
int err = errno;
DHT_LOG.e("Can't send message to %s: %s", addr.toString().c_str(), strerror(err));
return err;
}
return 0;
int
NetworkEngine::send(const Blob& msg, int /*flags*/, const SockAddr& addr)
{
auto data = (uint8_t*)malloc(msg.size());
memcpy(data, msg.data(), msg.size());
return sock->send(data, msg.size(), addr);
}
Sp<Request>
@@ -740,7 +731,7 @@ NetworkEngine::sendPong(const SockAddr& addr, TransId tid) {
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer.data(), buffer.size(), 0, addr);
send(buffer, addr);
}
Sp<Request>
@@ -945,7 +936,7 @@ NetworkEngine::sendValueParts(TransId tid, const std::vector<Blob>& svals, const
pk.pack(std::string("o")); pk.pack(start);
pk.pack(std::string("d")); pk.pack_bin(end-start);
pk.pack_bin_body((const char*)v.data()+start, end-start);
send(buffer.data(), buffer.size(), 0, addr);
send(buffer, addr);
start = end;
} while (start != v.size());
i++;
@@ -1003,7 +994,7 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, TransId tid, const Blob& no
}
// send response
send(buffer.data(), buffer.size(), 0, addr);
send(buffer, addr);
// send parts
if (not svals.empty())
@@ -1149,7 +1140,7 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, TransId tid) {
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer.data(), buffer.size(), 0, addr);
send(buffer, addr);
}
Sp<Request>
@@ -1282,7 +1273,7 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, TransId tid, Value::Id v
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer.data(), buffer.size(), 0, addr);
send(buffer, addr);
}
void
@@ -1313,7 +1304,7 @@ NetworkEngine::sendError(const SockAddr& addr,
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer.data(), buffer.size(), 0, addr);
send(buffer, addr);
}
void
src/securedht.cpp
+3 -3
View File
@@ -40,10 +40,10 @@ Config& getConfig(SecureDht::Config& conf)
return c;
}
SecureDht::SecureDht(int s, int s6, SecureDht::Config conf)
: Dht(s, s6, getConfig(conf)), key_(conf.id.first), certificate_(conf.id.second)
SecureDht::SecureDht(uv_loop_t* loop, SecureDht::Config conf)
: Dht(loop, getConfig(conf)), key_(conf.id.first), certificate_(conf.id.second)
{
if (s < 0 && s6 < 0)
if (not loop)
return;
for (const auto& type : DEFAULT_TYPES)
src/utils.cpp
+54
View File
@@ -19,12 +19,15 @@
#include "utils.h"
#include "sockaddr.h"
#include "default_types.h"
#include "uv_utils.h"
/* An IPv4 equivalent to IN6_IS_ADDR_UNSPECIFIED */
#ifndef IN_IS_ADDR_UNSPECIFIED
#define IN_IS_ADDR_UNSPECIFIED(a) (((long int) (a)->s_addr) == 0x00000000)
#endif /* IN_IS_ADDR_UNSPECIFIED */
static constexpr std::array<uint8_t, 12> MAPPED_IPV4_PREFIX {{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff}};
namespace dht {
std::string
@@ -116,6 +119,57 @@ SockAddr::isPrivate() const
}
}
bool
SockAddr::isMappedIPv4() const
{
if (getFamily() != AF_INET6)
return false;
const uint8_t* addr6 = reinterpret_cast<const uint8_t*>(&getIPv6().sin6_addr);
return std::equal(MAPPED_IPV4_PREFIX.begin(), MAPPED_IPV4_PREFIX.end(), addr6);
}
SockAddr
SockAddr::getMappedIPv4() const
{
if (getFamily() != AF_INET6)
return *this;
SockAddr ret;
ret.setFamily(AF_INET);
ret.setPort(getPort());
auto addr6 = reinterpret_cast<const uint8_t*>(&getIPv6().sin6_addr);
auto addr4 = reinterpret_cast<uint8_t*>(&ret.getIPv4().sin_addr);
addr6 += MAPPED_IPV4_PREFIX.size();
std::copy_n(addr6, sizeof(in_addr), addr4);
return ret;
}
void get_addr_info_cb(uv_getaddrinfo_t* req, int status, struct addrinfo* res) {
if (auto cb = static_cast<GetAddrInfoCb*>(req->data)) {
size_t count {0};
for (auto addr = res; addr; addr = addr->ai_next)
count++;
std::vector<SockAddr> ret;
ret.reserve(count);
for (auto addr = res; addr; addr = addr->ai_next)
ret.emplace_back(addr->ai_addr, addr->ai_addrlen);
(*cb)(std::move(ret));
delete cb;
}
delete req;
uv_freeaddrinfo(res);
}
void getAddrInfo(uv_loop_t* loop, const char* node, const char* service, GetAddrInfoCb&& cb)
{
auto req = new uv_getaddrinfo_t;
req->data = new GetAddrInfoCb(std::move(cb));
addrinfo hints {};
hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG;
hints.ai_family = AF_UNSPEC;
hints.ai_socktype = SOCK_DGRAM;
uv_getaddrinfo(loop, req, get_addr_info_cb, node, service, &hints);
}
bool operator==(const SockAddr& a, const SockAddr& b) {
return a.equals(b);
}
tools/dhtnode.cpp
+4 -1
View File
@@ -39,7 +39,7 @@ void print_id_req() {
}
void print_node_info(const std::shared_ptr<DhtRunner>& dht, const dht_params& params) {
std::cout << "OpenDht node " << dht->getNodeId() << " running on port " << dht->getBoundPort() << std::endl;
std::cout << "OpenDht node " << dht->getNodeId() << " running on port " << dht->getBound().toString() << std::endl;
if (params.generate_identity)
std::cout << "Public key ID " << dht->getId() << std::endl;
}
@@ -370,6 +370,9 @@ main(int argc, char **argv)
}
dht->run(params.port, crt, true, params.network);
dht->setOnStatusChanged([](dht::NodeStatus status4, dht::NodeStatus status6){
std::cout << "Node status changed : IPv4:" << (unsigned)status4 << " IPv6:" << (unsigned)status6 << std::endl;
});
if (params.log) {
if (params.syslog or (params.daemonize and params.logfile.empty()))