diff --git a/include/opendht/dht.h b/include/opendht/dht.h index 1c313a64..8baf1530 100644 --- a/include/opendht/dht.h +++ b/include/opendht/dht.h @@ -68,7 +68,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(const int& s, const int& s6, Config config); virtual ~Dht(); /** diff --git a/include/opendht/dhtrunner.h b/include/opendht/dhtrunner.h index fc351244..bab929b6 100644 --- a/include/opendht/dhtrunner.h +++ b/include/opendht/dhtrunner.h @@ -356,7 +356,13 @@ public: */ time_point loop() { std::lock_guard lck(dht_mtx); - return loop_(); + time_point wakeup = time_point::min(); + try { + wakeup = loop_(); + } catch (const dht::SocketException& e) { + startNetwork(bound4, bound6); + } + return wakeup; } /** @@ -420,7 +426,7 @@ private: */ void tryBootstrapContinuously(); - void doRun(const SockAddr& sin4, const SockAddr& sin6, SecureDhtConfig config); + void startNetwork(const SockAddr& sin4, const SockAddr& sin6); time_point loop_(); NodeStatus getStatus() const { @@ -487,12 +493,14 @@ private: std::queue> pending_ops {}; std::mutex storage_mtx {}; - std::atomic running {false}; + std::atomic_bool running {false}; + std::atomic_bool running_network {false}; NodeStatus status4 {NodeStatus::Disconnected}, status6 {NodeStatus::Disconnected}; StatusCallback statusCb {nullptr}; + int s4 {-1}, s6 {-1}; SockAddr bound4 {}; SockAddr bound6 {}; }; diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index ba9a06ef..2a75e9a3 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -202,8 +202,8 @@ public: using RequestCb = std::function; using RequestExpiredCb = std::function; - NetworkEngine(Logger& log, Scheduler& scheduler); - NetworkEngine(InfoHash& myid, NetId net, int s, int s6, Logger& log, Scheduler& scheduler, + NetworkEngine(Logger& log, Scheduler& scheduler, const int& s = -1, const int& s6 = -1); + NetworkEngine(InfoHash& myid, NetId net, const int& s, const int& s6, Logger& log, Scheduler& scheduler, decltype(NetworkEngine::onError) onError, decltype(NetworkEngine::onNewNode) onNewNode, decltype(NetworkEngine::onReportedAddr) onReportedAddr, @@ -505,8 +505,8 @@ private: /* DHT info */ const InfoHash& myid; const NetId network {0}; - const int dht_socket {-1}; - const int dht_socket6 {-1}; + const int& dht_socket; + const int& dht_socket6; const Logger& DHT_LOG; NodeCache cache {}; diff --git a/include/opendht/utils.h b/include/opendht/utils.h index 7904d7ee..a7621296 100644 --- a/include/opendht/utils.h +++ b/include/opendht/utils.h @@ -55,9 +55,15 @@ void erase_if(std::map& map, const Condition& condition) } class OPENDHT_PUBLIC DhtException : public std::runtime_error { - public: - DhtException(const std::string &str = "") : - std::runtime_error("DhtException occurred: " + str) {} +public: + DhtException(const std::string &str = "") : + std::runtime_error("DhtException occurred: " + str) {} +}; + +class OPENDHT_PUBLIC SocketException : public DhtException { +public: + SocketException(int err) : + DhtException(strerror(err)) {} }; // Time related definitions and utility functions diff --git a/src/dht.cpp b/src/dht.cpp index d3c150a8..f75dc479 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -1649,7 +1649,7 @@ Dht::~Dht() Dht::Dht() : store(), scheduler(DHT_LOG), network_engine(DHT_LOG, scheduler) {} -Dht::Dht(int s, int s6, Config config) +Dht::Dht(const int& s, const int& s6, Config config) : myid(config.node_id ? config.node_id : InfoHash::getRandom()), is_bootstrap(config.is_bootstrap), maintain_storage(config.maintain_storage), store(), store_quota(), diff --git a/src/dhtrunner.cpp b/src/dhtrunner.cpp index b540fd52..d5a52f2a 100644 --- a/src/dhtrunner.cpp +++ b/src/dhtrunner.cpp @@ -85,24 +85,30 @@ DhtRunner::run(const SockAddr& local4, const SockAddr& local6, DhtRunner::Config { if (running) return; - if (rcv_thread.joinable()) - rcv_thread.join(); + startNetwork(local4, local6); + + auto dht = std::unique_ptr(new Dht(s4, s6, SecureDht::getConfig(config.dht_config))); + dht_ = std::unique_ptr(new SecureDht(std::move(dht), config.dht_config)); + +#if OPENDHT_PROXY_CLIENT + if (!dht_via_proxy_) { + config_ = config; + enableProxy(use_proxy); + } +#endif + running = true; -#if OPENDHT_PROXY_CLIENT - config_ = config; -#endif //OPENDHT_PROXY_CLIENT - - doRun(local4, local6, config.dht_config); - -#if OPENDHT_PROXY_CLIENT - enableProxy(not config.proxy_server.empty()); -#endif //OPENDHT_PROXY_CLIENT if (not config.threaded) return; - dht_thread = std::thread([this]() { + dht_thread = std::thread([this, local4, local6]() { while (running) { std::unique_lock lk(dht_mtx); - auto wakeup = loop_(); + time_point wakeup; + try { + wakeup = loop_(); + } catch (const dht::SocketException& e) { + startNetwork(local4, local6); + } cv.wait_until(lk, wakeup, [this]() { if (not running) return true; @@ -141,15 +147,16 @@ DhtRunner::shutdown(ShutdownCallback cb) { void DhtRunner::join() { + running_network = false; running = false; cv.notify_all(); 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 (rcv_thread.joinable()) + rcv_thread.join(); { std::lock_guard lck(storage_mtx); @@ -161,8 +168,6 @@ DhtRunner::join() resetDht(); status4 = NodeStatus::Disconnected; status6 = NodeStatus::Disconnected; - bound4 = {}; - bound6 = {}; } } @@ -430,12 +435,13 @@ int bindSocket(const SockAddr& addr, SockAddr& bound) } void -DhtRunner::doRun(const SockAddr& sin4, const SockAddr& sin6, SecureDht::Config config) +DhtRunner::startNetwork(const SockAddr& sin4, const SockAddr& sin6) { - resetDht(); - - int s4 = -1, - s6 = -1; + running_network = false; + if (rcv_thread.joinable()) + rcv_thread.join(); + s4 = -1; + s6 = -1; bound4 = {}; if (sin4) @@ -447,18 +453,10 @@ DhtRunner::doRun(const SockAddr& sin4, const SockAddr& sin6, SecureDht::Config c s6 = bindSocket(sin6, bound6); #endif - auto dht = std::unique_ptr( - new Dht(s4, s6, SecureDht::getConfig(config)) - ); - dht_ = std::unique_ptr(new SecureDht(std::move(dht), config)); - -#if OPENDHT_PROXY_CLIENT - enableProxy(use_proxy); -#endif - - rcv_thread = std::thread([this,s4,s6]() { + running_network = true; + rcv_thread = std::thread([this]() { try { - while (true) { + while (running_network) { struct timeval tv {/*.tv_sec = */0, /*.tv_usec = */250000}; fd_set readfds; @@ -476,7 +474,7 @@ DhtRunner::doRun(const SockAddr& sin4, const SockAddr& sin6, SecureDht::Config c } } - if(!running) + if (not running_network) break; if(rc > 0) { @@ -506,6 +504,10 @@ DhtRunner::doRun(const SockAddr& sin4, const SockAddr& sin6, SecureDht::Config c close(s4); if (s6 >= 0) close(s6); + s4 = -1; + s6 = -1; + bound4 = {}; + bound6 = {}; }); } diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 4d560281..200b046d 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -126,8 +126,10 @@ 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(Logger& log, Scheduler& scheduler, const int& s, const int& s6) + : myid(zeroes), DHT_LOG(log), scheduler(scheduler), dht_socket(s), dht_socket6(s6) +{} +NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, const int& s, const int& s6, Logger& log, Scheduler& scheduler, decltype(NetworkEngine::onError) onError, decltype(NetworkEngine::onNewNode) onNewNode, decltype(NetworkEngine::onReportedAddr) onReportedAddr, @@ -589,6 +591,9 @@ NetworkEngine::send(const char *buf, size_t len, int flags, const SockAddr& addr if (sendto(s, buf, len, flags, addr.get(), addr.getLength()) == -1) { int err = errno; DHT_LOG.e("Can't send message to %s: %s", addr.toString().c_str(), strerror(err)); + if (err == EPIPE) { + throw SocketException(EPIPE); + } return err; } return 0;