From de099b2d8b6c59ad297ffa372ca684ca77a5adec Mon Sep 17 00:00:00 2001 From: subtly Date: Wed, 19 Nov 2014 06:14:56 +0100 Subject: [PATCH] remove network pause. fix client repeat-disconnect loop. removed resolver from getifaddr dance (not reqd) and updated getifaddr code (needs windowz/linux testing). don't accept new connections if listenport bind fails (possible due to OS firewall settings). only use upnp w/ipv4 addresses. re: disconnect loop. Looks like ethhost session isn't deallocating once stopped. Thinking UI might have shared_ptr. For now, letting dealloc/quit continue as long as isOpen returns false -- maybe okay since network will be halted. Also, it maybe possible the issue is due to network thread not running after disconnect so packets aren't tx/rx to kill peer -- either way, it's possible remote end doesn't ack disconnect and timeout/force-close-dealloc is required, so will need more attention later. --- libdevcore/Worker.cpp | 3 +- libdevcore/Worker.h | 18 +- libethereum/EthereumHost.h | 1 + libp2p/Host.cpp | 376 +++++++++++++++++++++---------------- libp2p/Host.h | 42 +++-- 5 files changed, 256 insertions(+), 184 deletions(-) diff --git a/libdevcore/Worker.cpp b/libdevcore/Worker.cpp index b2660305a..d9246f9fd 100644 --- a/libdevcore/Worker.cpp +++ b/libdevcore/Worker.cpp @@ -41,7 +41,8 @@ void Worker::startWorking() startedWorking(); while (!m_stop) { - this_thread::sleep_for(chrono::milliseconds(30)); + if (m_idlewaitms) + this_thread::sleep_for(chrono::milliseconds(m_idlewaitms)); doWork(); } cdebug << "Finishing up worker thread"; diff --git a/libdevcore/Worker.h b/libdevcore/Worker.h index a4a998dd7..f8d694681 100644 --- a/libdevcore/Worker.h +++ b/libdevcore/Worker.h @@ -31,7 +31,7 @@ namespace dev class Worker { protected: - Worker(std::string const& _name = "anon"): m_name(_name) {} + Worker(std::string const& _name = "anon", unsigned _idlewaitms = 30): m_name(_name), m_idlewaitms(_idlewaitms) {} /// Move-constructor. Worker(Worker&& _m) { std::swap(m_name, _m.m_name); } @@ -41,20 +41,34 @@ protected: virtual ~Worker() { stopWorking(); } + /// Allows changing worker name if work is stopped. void setName(std::string _n) { if (!isWorking()) m_name = _n; } + /// Starts worker thread; causes startedWorking() to be called. void startWorking(); + + /// Stop worker thread; causes call to stopWorking(). void stopWorking(); + + /// Returns if worker thread is present. bool isWorking() const { Guard l(x_work); return !!m_work; } + + /// Called after thread is started from startWorking(). virtual void startedWorking() {} + + /// Called continuously following sleep for m_idlewaitms. virtual void doWork() = 0; + + /// Called when is to be stopped, just prior to thread being joined. virtual void doneWorking() {} private: + std::string m_name; + unsigned m_idlewaitms; + mutable Mutex x_work; ///< Lock for the network existance. std::unique_ptr m_work; ///< The network thread. bool m_stop = false; - std::string m_name; }; } diff --git a/libethereum/EthereumHost.h b/libethereum/EthereumHost.h index 18ba765aa..c732f2dc1 100644 --- a/libethereum/EthereumHost.h +++ b/libethereum/EthereumHost.h @@ -51,6 +51,7 @@ class BlockQueue; /** * @brief The EthereumHost class * @warning None of this is thread-safe. You have been warned. + * @doWork Syncs to peers and sends new blocks and transactions. */ class EthereumHost: public p2p::HostCapability, Worker { diff --git a/libp2p/Host.cpp b/libp2p/Host.cpp index 996e219db..4c6973ae3 100644 --- a/libp2p/Host.cpp +++ b/libp2p/Host.cpp @@ -46,19 +46,17 @@ using namespace std; using namespace dev; using namespace dev::p2p; -// Addresses we will skip during network interface discovery -// Use a vector as the list is small -// Why this and not names? -// Under MacOSX loopback (127.0.0.1) can be named lo0 and br0 are bridges (0.0.0.0) +// Addresses skipped during network interface discovery +// @todo: filter out ivp6 link-local network mess on macos, ex: fe80::1%lo0 static const set c_rejectAddresses = { {bi::address_v4::from_string("127.0.0.1")}, - {bi::address_v6::from_string("::1")}, {bi::address_v4::from_string("0.0.0.0")}, + {bi::address_v6::from_string("::1")}, {bi::address_v6::from_string("::")} }; Host::Host(std::string const& _clientVersion, NetworkPreferences const& _n, bool _start): - Worker("p2p"), + Worker("p2p", 0), m_clientVersion(_clientVersion), m_netPrefs(_n), m_ioService(new ba::io_service), @@ -79,70 +77,17 @@ Host::~Host() void Host::start() { - // if there's no ioService, it means we've had quit() called - bomb out - we're not allowed in here. - if (!m_ioService) - return; - - if (isWorking()) - stop(); - - for (unsigned i = 0; i < 2; ++i) - { - bi::tcp::endpoint endpoint(bi::tcp::v4(), i ? 0 : m_netPrefs.listenPort); - try - { - m_acceptor->open(endpoint.protocol()); - m_acceptor->set_option(ba::socket_base::reuse_address(true)); - m_acceptor->bind(endpoint); - m_acceptor->listen(); - m_listenPort = i ? m_acceptor->local_endpoint().port() : m_netPrefs.listenPort; - break; - } - catch (...) - { - if (i) - { - cwarn << "Couldn't start accepting connections on host. Something very wrong with network?\n" << boost::current_exception_diagnostic_information(); - return; - } - m_acceptor->close(); - continue; - } - } - - for (auto const& h: m_capabilities) - h.second->onStarting(); - startWorking(); } void Host::stop() { - // if there's no ioService, it means we've had quit() called - bomb out - we're not allowed in here. - if (!m_ioService) - return; - - for (auto const& h: m_capabilities) - h.second->onStopping(); - + // flag transition to shutdown network + // once m_run is false the scheduler will shutdown network and stopWorking() + m_run = false; + while (m_timer) + this_thread::sleep_for(chrono::milliseconds(100)); stopWorking(); - - if (m_acceptor->is_open()) - { - if (m_accepting) - m_acceptor->cancel(); - m_acceptor->close(); - m_accepting = false; - } - if (m_socket->is_open()) - m_socket->close(); - disconnectPeers(); - - if (!!m_ioService) - { - m_ioService->stop(); - m_ioService->reset(); - } } void Host::quit() @@ -150,7 +95,8 @@ void Host::quit() // called to force io_service to kill any remaining tasks it might have - // such tasks may involve socket reads from Capabilities that maintain references // to resources we're about to free. - stop(); + if (isWorking()) + stop(); m_acceptor.reset(); m_socket.reset(); m_ioService.reset(); @@ -183,33 +129,6 @@ void Host::registerPeer(std::shared_ptr _s, CapDescs const& _caps) } } -void Host::disconnectPeers() -{ - // if there's no ioService, it means we've had quit() called - bomb out - we're not allowed in here. - if (!m_ioService) - return; - - for (unsigned n = 0;; n = 0) - { - { - RecursiveGuard l(x_peers); - for (auto i: m_peers) - if (auto p = i.second.lock()) - { - p->disconnect(ClientQuit); - n++; - } - } - if (!n) - break; - m_ioService->poll(); - this_thread::sleep_for(chrono::milliseconds(100)); - } - - delete m_upnp; - m_upnp = nullptr; -} - void Host::seal(bytes& _b) { _b[0] = 0x22; @@ -225,10 +144,6 @@ void Host::seal(bytes& _b) void Host::determinePublic(string const& _publicAddress, bool _upnp) { - // if there's no ioService, it means we've had quit() called - bomb out - we're not allowed in here. - if (!m_ioService) - return; - if (_upnp) try { @@ -236,11 +151,12 @@ void Host::determinePublic(string const& _publicAddress, bool _upnp) } catch (NoUPnPDevice) {} // let m_upnp continue as null - we handle it properly. - bi::tcp::resolver r(*m_ioService); if (m_upnp && m_upnp->isValid() && m_peerAddresses.size()) { clog(NetNote) << "External addr:" << m_upnp->externalIP(); int p; + + // iterate m_peerAddresses (populated by populateAddresses()) for (auto const& addr : m_peerAddresses) if ((p = m_upnp->addRedirect(addr.to_string().c_str(), m_listenPort))) break; @@ -258,7 +174,7 @@ void Host::determinePublic(string const& _publicAddress, bool _upnp) m_public = bi::tcp::endpoint(bi::address(), (unsigned short)p); else { - bi::address adr = adr = bi::address::from_string(eip); + bi::address adr = bi::address::from_string(eip); try { adr = bi::address::from_string(_publicAddress); @@ -271,7 +187,21 @@ void Host::determinePublic(string const& _publicAddress, bool _upnp) else { // No UPnP - fallback on given public address or, if empty, the assumed peer address. - bi::address adr = m_peerAddresses.size() ? m_peerAddresses[0] : bi::address(); + bi::address adr; // = m_peerAddresses.size() ? m_peerAddresses[0] : bi::address(); + if (m_peerAddresses.size()) + { + // prefer local ipv4 over local ipv6 + for (auto const& ip: m_peerAddresses) + if (ip.is_v4()) + { + adr = ip; + break; + } + + if (adr.is_unspecified()) + adr = m_peerAddresses[0]; + } + try { adr = bi::address::from_string(_publicAddress); @@ -328,57 +258,43 @@ void Host::populateAddresses() if (getifaddrs(&ifaddr) == -1) BOOST_THROW_EXCEPTION(NoNetworking()); - bi::tcp::resolver r(*m_ioService); - - for (ifaddrs* ifa = ifaddr; ifa; ifa = ifa->ifa_next) - { - if (!ifa->ifa_addr) + for (auto ifa = ifaddr; ifa != NULL; ifa = ifa->ifa_next) { + if (!ifa->ifa_addr || (strlen(ifa->ifa_name) > 2 && !strncmp(ifa->ifa_name, "lo0", 3))) { continue; + } + if (ifa->ifa_addr->sa_family == AF_INET) { - char host[NI_MAXHOST]; - if (getnameinfo(ifa->ifa_addr, sizeof(struct sockaddr_in), host, NI_MAXHOST, NULL, 0, NI_NUMERICHOST)) - continue; - try - { - auto it = r.resolve({host, "30303"}); - bi::tcp::endpoint ep = it->endpoint(); - bi::address ad = ep.address(); - m_addresses.push_back(ad.to_v4()); - bool isLocal = std::find(c_rejectAddresses.begin(), c_rejectAddresses.end(), ad) != c_rejectAddresses.end(); - if (!isLocal) - m_peerAddresses.push_back(ad.to_v4()); - clog(NetNote) << "Address: " << host << " = " << m_addresses.back() << (isLocal ? " [LOCAL]" : " [PEER]"); - } - catch (...) - { - clog(NetNote) << "Couldn't resolve: " << host; - } + in_addr addr = ((struct sockaddr_in *)ifa->ifa_addr)->sin_addr; + boost::asio::ip::address_v4 address(boost::asio::detail::socket_ops::network_to_host_long(addr.s_addr)); + if (std::find(c_rejectAddresses.begin(), c_rejectAddresses.end(), address) == c_rejectAddresses.end()) + m_peerAddresses.push_back(address); + + // Log IPv4 Address: + auto addr4 = &((struct sockaddr_in *)ifa->ifa_addr)->sin_addr; + char addressBuffer[INET_ADDRSTRLEN]; + inet_ntop(AF_INET, addr4, addressBuffer, INET_ADDRSTRLEN); + printf("%s IP Address %s\n", ifa->ifa_name, addressBuffer); } else if (ifa->ifa_addr->sa_family == AF_INET6) { - char host[NI_MAXHOST]; - if (getnameinfo(ifa->ifa_addr, sizeof(struct sockaddr_in6), host, NI_MAXHOST, NULL, 0, NI_NUMERICHOST)) - continue; - try - { - auto it = r.resolve({host, "30303"}); - bi::tcp::endpoint ep = it->endpoint(); - bi::address ad = ep.address(); - m_addresses.push_back(ad.to_v6()); - bool isLocal = std::find(c_rejectAddresses.begin(), c_rejectAddresses.end(), ad) != c_rejectAddresses.end(); - if (!isLocal) - m_peerAddresses.push_back(ad); - clog(NetNote) << "Address: " << host << " = " << m_addresses.back() << (isLocal ? " [LOCAL]" : " [PEER]"); - } - catch (...) - { - clog(NetNote) << "Couldn't resolve: " << host; - } + sockaddr_in6* sockaddr = ((struct sockaddr_in6 *)ifa->ifa_addr); + in6_addr addr = sockaddr->sin6_addr; + boost::asio::ip::address_v6::bytes_type bytes; + memcpy(&bytes[0], addr.s6_addr, 16); + boost::asio::ip::address_v6 address(bytes, sockaddr->sin6_scope_id); + if (std::find(c_rejectAddresses.begin(), c_rejectAddresses.end(), address) == c_rejectAddresses.end()) + m_peerAddresses.push_back(address); + + // Log IPv6 Address: + auto addr6 = &((struct sockaddr_in6 *)ifa->ifa_addr)->sin6_addr; + char addressBuffer[INET6_ADDRSTRLEN]; + inet_ntop(AF_INET6, addr6, addressBuffer, INET6_ADDRSTRLEN); + printf("%s IP Address %s\n", ifa->ifa_name, addressBuffer); } } + if (ifaddr!=NULL) freeifaddrs(ifaddr); - freeifaddrs(ifaddr); #endif } @@ -461,8 +377,8 @@ Nodes Host::potentialPeers(RangeMask const& _known) void Host::ensureAccepting() { - // if there's no ioService, it means we've had quit() called - bomb out - we're not allowed in here. - if (!m_ioService) + // return if there's no io-server (quit called) or we're not listening + if (!m_ioService || m_listenPort < 1) return; if (!m_accepting) @@ -654,12 +570,9 @@ void Host::growPeers() return; } else - { - ensureAccepting(); for (auto const& i: m_peers) if (auto p = i.second.lock()) p->ensureNodesRequested(); - } } } @@ -717,36 +630,150 @@ PeerInfos Host::peers(bool _updatePing) const ret.push_back(j->m_info); return ret; } - -void Host::startedWorking() -{ - determinePublic(m_netPrefs.publicIP, m_netPrefs.upnp); - ensureAccepting(); - - if (!m_public.address().is_unspecified() && (m_nodes.empty() || m_nodes[m_nodesList[0]]->id != id())) - noteNode(id(), m_public, Origin::Perfect, false); - - clog(NetNote) << "Id:" << id().abridged(); -} - -void Host::doWork() + +void Host::run(boost::system::error_code const& error) { - // if there's no ioService, it means we've had quit() called - bomb out - we're not allowed in here. - if (asserts(!!m_ioService)) + static unsigned s_lasttick = 0; + s_lasttick += c_timerInterval; + + if (error) + // tood: error handling. + { + m_timer.reset(); return; + } + + // no timer means this is first run and network must be started + if (!m_timer) + // run once when host worker thread calls startedWorking() + { + // reset io service and create deadline timer + m_ioService->reset(); + m_timer.reset(new boost::asio::deadline_timer(*m_ioService)); + m_run = true; + + // try to open acceptor (ipv4; todo: update for ipv6) + for (unsigned i = 0; i < 2; ++i) + { + // try to connect w/listenPort, else attempt net-allocated port + bi::tcp::endpoint endpoint(bi::tcp::v4(), i ? 0 : m_netPrefs.listenPort); + try + { + m_acceptor->open(endpoint.protocol()); + m_acceptor->set_option(ba::socket_base::reuse_address(true)); + m_acceptor->bind(endpoint); + m_acceptor->listen(); + m_listenPort = i ? m_acceptor->local_endpoint().port() : m_netPrefs.listenPort; + break; + } + catch (...) + { + if (i) + { + // both attempts failed + cwarn << "Couldn't start accepting connections on host. Something very wrong with network?\n" << boost::current_exception_diagnostic_information(); + m_listenPort = -1; + } + + // first attempt failed + m_acceptor->close(); + continue; + } + } + + // start capability threads + for (auto const& h: m_capabilities) + h.second->onStarting(); + + // determine public IP, but only if we're able to listen for connections + // todo: visualize when listen is unavailable in UI + // tood: only punch hole for ipv4 + if (m_listenPort) + { + determinePublic(m_netPrefs.publicIP, m_netPrefs.upnp); + ensureAccepting(); + } + + // if m_public address is valid then add us to node list + // todo: abstract empty() and emplace logic + if (!m_public.address().is_unspecified() && (m_nodes.empty() || m_nodes[m_nodesList[0]]->id != id())) + noteNode(id(), m_public, Origin::Perfect, false); + + clog(NetNote) << "Id:" << id().abridged(); + } + + // io service went away, so stop here + if (!m_ioService) + { + m_timer.reset(); + return; + } - growPeers(); - prunePeers(); - + // network stopped; disconnect peers + if (!m_run) + { + // close acceptor + if (m_acceptor->is_open()) + { + if (m_accepting) + m_acceptor->cancel(); + m_acceptor->close(); + m_accepting = false; + } + + // stop capabilities (eth: stops syncing or block/tx broadcast) + for (auto const& h: m_capabilities) + h.second->onStopping(); + + // disconnect peers + for (unsigned n = 0;; n = 0) + { + { + RecursiveGuard l(x_peers); + for (auto i: m_peers) + if (auto p = i.second.lock()) + if (p->isOpen()) + { + p->disconnect(ClientQuit); + n++; + } + } + if (!n) + break; + this_thread::sleep_for(chrono::milliseconds(100)); + } + + if (m_socket->is_open()) + m_socket->close(); + + if (m_upnp != nullptr) + delete m_upnp; + + // m_run is false, so we're stopping; kill timer + s_lasttick = 0; + m_timer.reset(); + if (!!m_ioService) + m_ioService->stop(); + return; + } + + + if (s_lasttick == c_timerInterval * 100) + { + growPeers(); + prunePeers(); + s_lasttick = 0; + } + if (m_hadNewNodes) { for (auto p: m_peers) if (auto pp = p.second.lock()) pp->serviceNodesRequest(); - + m_hadNewNodes = false; } - + if (chrono::steady_clock::now() - m_lastPing > chrono::seconds(30)) // ping every 30s. { for (auto p: m_peers) @@ -756,7 +783,22 @@ void Host::doWork() pingAll(); } - m_ioService->poll(); + auto runcb = [this](boost::system::error_code const& error)->void{ run(error); }; + m_timer->expires_from_now(boost::posix_time::milliseconds(c_timerInterval)); + m_timer->async_wait(runcb); +} + +void Host::startedWorking() +{ + run(boost::system::error_code()); +} + +void Host::doWork() +{ + // no ioService means we've had quit() called - bomb out - we're not allowed in here. + if (asserts(!!m_ioService)) + return; + m_ioService->run(); } void Host::pingAll() diff --git a/libp2p/Host.h b/libp2p/Host.h index 7722905ab..7c4fbe1ce 100644 --- a/libp2p/Host.h +++ b/libp2p/Host.h @@ -110,7 +110,7 @@ struct NetworkPreferences bool upnp = true; bool localNetworking = false; }; - + /** * @brief The Host class * Capabilities should be registered prior to startNetwork, since m_capabilities is not thread-safe. @@ -147,7 +147,7 @@ public: void connect(bi::tcp::endpoint const& _ep); void connect(std::shared_ptr const& _n); - /// @returns true iff we have the a peer of the given id. + /// @returns true if we have the a peer of the given id. bool havePeer(NodeId _id) const; /// Set ideal number of peers. @@ -175,10 +175,16 @@ public: void setNetworkPreferences(NetworkPreferences const& _p) { auto had = isStarted(); if (had) stop(); m_netPrefs = _p; if (had) start(); } + /// Start network. void start(); + + /// Stop network. void stop(); + + /// @returns if network is running bool isStarted() const { return isWorking(); } + /// Reset acceptor, socket, and IO service. Called by deallocator. Maybe called by implementation when ordered deallocation is required. void quit(); NodeId id() const { return m_key.pub(); } @@ -190,17 +196,23 @@ public: private: void seal(bytes& _b); void populateAddresses(); + + /// Try UPNP or listen to assumed address. Requires valid m_listenPort. void determinePublic(std::string const& _publicAddress, bool _upnp); + + void ensureAccepting(); void growPeers(); void prunePeers(); + /// Called by Worker. Not thread-safe; to be called only by worker. virtual void startedWorking(); + /// Called by startedWorking. Not thread-safe; to be called only be worker callback. + void run(boost::system::error_code const& error); ///< Run network. Called serially via ASIO deadline timer. Manages connection state transitions. + bool m_run = false; - /// Conduct I/O, polling, syncing, whatever. - /// Ideally all time-consuming I/O is done in a background thread or otherwise asynchronously, but you get this call every 100ms or so anyway. - /// This won't touch alter the blockchain. + /// Run network virtual void doWork(); std::shared_ptr noteNode(NodeId _id, bi::tcp::endpoint _a, Origin _o, bool _ready, NodeId _oldId = NodeId()); @@ -208,18 +220,20 @@ private: std::string m_clientVersion; ///< Our version string. - NetworkPreferences m_netPrefs; ///< Network settings. + NetworkPreferences m_netPrefs; ///< Network settings. - static const int NetworkStopped = -1; ///< The value meaning we're not actually listening. - int m_listenPort = NetworkStopped; ///< What port are we listening on? + int m_listenPort = -1; ///< What port are we listening on. -1 means binding failed or acceptor hasn't been initialized. std::unique_ptr m_ioService; ///< IOService for network stuff. - std::unique_ptr m_acceptor; ///< Listening acceptor. - std::unique_ptr m_socket; ///< Listening socket. - - UPnP* m_upnp = nullptr; ///< UPnP helper. - bi::tcp::endpoint m_public; ///< Our public listening endpoint. - KeyPair m_key; ///< Our unique ID. + std::unique_ptr m_acceptor; ///< Listening acceptor. + std::unique_ptr m_socket; ///< Listening socket. + + std::unique_ptr m_timer; ///< Timer which, when network is running, calls scheduler() every c_timerInterval ms. + static const unsigned c_timerInterval = 100; ///< Interval which m_timer is run when network is connected. + + UPnP* m_upnp = nullptr; ///< UPnP helper. + bi::tcp::endpoint m_public; ///< Our public listening endpoint. + KeyPair m_key; ///< Our unique ID. bool m_hadNewNodes = false;