Browse Source

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.
cl-refactor
subtly 10 years ago
parent
commit
de099b2d8b
  1. 3
      libdevcore/Worker.cpp
  2. 18
      libdevcore/Worker.h
  3. 1
      libethereum/EthereumHost.h
  4. 376
      libp2p/Host.cpp
  5. 42
      libp2p/Host.h

3
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";

18
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<std::thread> m_work; ///< The network thread.
bool m_stop = false;
std::string m_name;
};
}

1
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<EthereumPeer>, Worker
{

376
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<bi::address> 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<Session> _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<unsigned> 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()

42
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<Node> 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<Node> 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<ba::io_service> m_ioService; ///< IOService for network stuff.
std::unique_ptr<bi::tcp::acceptor> m_acceptor; ///< Listening acceptor.
std::unique_ptr<bi::tcp::socket> 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<bi::tcp::acceptor> m_acceptor; ///< Listening acceptor.
std::unique_ptr<bi::tcp::socket> m_socket; ///< Listening socket.
std::unique_ptr<boost::asio::deadline_timer> 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;

Loading…
Cancel
Save