ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2025-12-06 17:27:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix peer tracking and scaning model.

Browse Source
This commit is contained in:
Arthur Britto
2012-06-19 12:26:15 -07:00
parent a7f8cd1a45
commit ce16e58c62
5 changed files with 220 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

181
src/ConnectionPool.cpp
View File

@@ -48,15 +48,26 @@ void ConnectionPool::start()
}
bool ConnectionPool::getTopNAddrs(int n,std::vector<std::string>& addrs)
{
// XXX Filter out other local addresses (like ipv6)
if (!theConfig.PEER_IP.empty() && theConfig.PEER_IP != "127.0.0.1")
{
addrs.push_back(str(boost::format("%s %d") % theConfig.PEER_IP % theConfig.PEER_PORT));
}
{
Database* db = theApp->getWalletDB()->getDB();
ScopedLock sl(theApp->getWalletDB()->getDBLock());
SQL_FOREACH(db, str(boost::format("SELECT IpPort FROM PeerIps limit %d") % n) )
SQL_FOREACH(db, str(boost::format("SELECT IpPort FROM PeerIps LIMIT %d") % n) )
{
std::string str;
db->getStr(0,str);
addrs.push_back(str);
}
}
return true;
}
@@ -65,10 +76,10 @@ bool ConnectionPool::savePeer(const std::string& strIp, int iPort,char code)
{
Database* db = theApp->getWalletDB()->getDB();
std::string ipPort=db->escape(str(boost::format("%s %d") % strIp % iPort));
std::string ipPort= sqlEscape(str(boost::format("%s %d") % strIp % iPort));
ScopedLock sl(theApp->getWalletDB()->getDBLock());
std::string sql=str(boost::format("SELECT count(*) FROM PeerIps WHERE IpPort=%s;") % ipPort);
std::string sql=str(boost::format("SELECT COUNT(*) FROM PeerIps WHERE IpPort=%s;") % ipPort);
if (db->executeSQL(sql) && db->startIterRows())
{
if ( db->getInt(0)==0)
@@ -151,7 +162,7 @@ void ConnectionPool::policyLowWater()
else
{
// Try to start connection.
if (!connectTo(strIp, iPort))
if (!peerConnect(strIp, iPort))
throw std::runtime_error("Internal error: standby was already connected.");
// Check if we need more.
@@ -198,7 +209,6 @@ void ConnectionPool::policyHandler(const boost::system::error_code& ecResult)
}
}
// XXX Broken: also don't send a message to a peer if we got it from the peer.
void ConnectionPool::relayMessage(Peer* fromPeer, PackedMessage::pointer msg)
{
boost::mutex::scoped_lock sl(mPeerLock);
@@ -213,40 +223,27 @@ void ConnectionPool::relayMessage(Peer* fromPeer, PackedMessage::pointer msg)
}
}
// Inbound connection, false=reject
// Reject addresses we already have in our table.
// XXX Reject, if we have too many connections.
bool ConnectionPool::peerRegister(Peer::pointer peer, const std::string& strIp, int iPort)
// Add or modify into PeerIps as a manual entry for immediate scanning.
// Requires sane IP and port.
void ConnectionPool::connectTo(const std::string& strIp, int iPort)
{
bool bAccept;
ipPort ip = make_pair(strIp, iPort);
Database* db = theApp->getWalletDB()->getDB();
std::string ipPort = sqlEscape(str(boost::format("%s %d") % strIp % iPort));
boost::unordered_map<ipPort, Peer::pointer>::iterator it;
boost::mutex::scoped_lock sl(mPeerLock);
it = mIpMap.find(ip);
if (it == mIpMap.end())
{
// Did not find it. Not already connecting or connected.
ScopedLock sl(theApp->getWalletDB()->getDBLock());
std::cerr << "ConnectionPool::peerRegister: " << ip.first << " " << ip.second << std::endl;
// Mark as connecting.
mIpMap[ip] = peer;
bAccept = true;
}
else
{
// Found it. Already connected or connecting.
bAccept = false;
db->executeSQL(str(boost::format("REPLACE INTO PeerIps (IpPort,Score,Source,ScanNext) values (%s,%d,'%c',0);")
% ipPort
% theApp->getUNL().iSourceScore(UniqueNodeList::vsManual)
% char(UniqueNodeList::vsManual)));
}
return bAccept;
scanRefresh();
}
bool ConnectionPool::connectTo(const std::string& strIp, int iPort)
// <-- true, if already connected.
bool ConnectionPool::peerConnect(const std::string& strIp, int iPort)
{
bool bConnecting;
ipPort ip = make_pair(strIp, iPort);
@@ -260,7 +257,7 @@ bool ConnectionPool::connectTo(const std::string& strIp, int iPort)
if (it == mIpMap.end())
{
// Did not find it. Not already connecting or connected.
std::cerr << "ConnectionPool::connectTo: Connecting: "
std::cerr << "ConnectionPool::peerConnect: Connecting: "
<< strIp << " " << iPort << std::endl;
Peer::pointer peer(Peer::create(theApp->getIOService(), mCtx));
@@ -276,7 +273,7 @@ bool ConnectionPool::connectTo(const std::string& strIp, int iPort)
else
{
// Found it. Already connected.
std::cerr << "ConnectionPool::connectTo: Already connected: "
std::cerr << "ConnectionPool::peerConnect: Already connected: "
<< strIp << " " << iPort << std::endl;
bConnecting = false;
@@ -316,15 +313,15 @@ std::vector<Peer::pointer> ConnectionPool::getPeerVector()
}
// Now know peer's node public key. Determine if we want to stay connected.
bool ConnectionPool::peerConnected(Peer::pointer peer, const NewcoinAddress& na)
bool ConnectionPool::peerConnected(Peer::pointer peer, const NewcoinAddress& naPeer, const std::string& strIP, int iPort)
{
bool bSuccess;
std::cerr << "ConnectionPool::peerConnected: " << na.humanNodePublic()
<< " " << peer->getIP() << " " << peer->getPort()
<< std::endl;
std::cerr << "ConnectionPool::peerConnected: "
<< naPeer.humanNodePublic() << " " << strIP << " " << iPort << std::endl;
assert(!!peer);
if (na == theApp->getWallet().getNodePublic())
if (naPeer == theApp->getWallet().getNodePublic())
{
std::cerr << "ConnectionPool::peerConnected: To self." << std::endl;
bSuccess = false;
@@ -332,25 +329,42 @@ bool ConnectionPool::peerConnected(Peer::pointer peer, const NewcoinAddress& na)
else
{
boost::mutex::scoped_lock sl(mPeerLock);
boost::unordered_map<NewcoinAddress, Peer::pointer>::iterator itCm = mConnectedMap.find(naPeer);
mConnectedMap[na] = peer;
if (itCm == mConnectedMap.end())
{
// New connection.
mConnectedMap[naPeer] = peer;
bSuccess = true;
}
else
{
// Found in map, already connected.
if (!strIP.empty())
{
// Was an outbound connection, we know IP and port.
// Note in previous connection how to reconnect.
itCm->second->peerIpPort(strIP, iPort);
}
bSuccess = false; // Don't need a redundant connection.
}
}
return bSuccess;
}
void ConnectionPool::peerDisconnected(Peer::pointer peer, const ipPort& ipPeer, const NewcoinAddress& naPeer)
// We maintain a map of public key to peer for connectted and verified peers. Maintain it.
void ConnectionPool::peerDisconnected(Peer::pointer peer, const NewcoinAddress& naPeer)
{
std::cerr << "ConnectionPool::peerDisconnected: " << ipPeer.first << " " << ipPeer.second << std::endl;
boost::mutex::scoped_lock sl(mPeerLock);
std::cerr << "ConnectionPool::peerDisconnected: " << peer->getIP() << " " << peer->getPort() << std::endl;
if (naPeer.isValid())
{
boost::unordered_map<NewcoinAddress, Peer::pointer>::iterator itCm;
boost::mutex::scoped_lock sl(mPeerLock);
itCm = mConnectedMap.find(naPeer);
if (itCm == mConnectedMap.end())
@@ -365,7 +379,66 @@ void ConnectionPool::peerDisconnected(Peer::pointer peer, const ipPort& ipPeer,
mConnectedMap.erase(itCm);
}
}
}
void ConnectionPool::peerScanSet(const std::string& strIp, int iPort)
{
std::cerr << "ConnectionPool::peerScanSet: " << strIp << " " << iPort << std::endl;
std::string strIpPort = str(boost::format("%s %d") % strIp % iPort);
ScopedLock sl(theApp->getWalletDB()->getDBLock());
Database* db = theApp->getWalletDB()->getDB();
if (db->executeSQL(str(boost::format("SELECT ScanNext FROM PeerIps WHERE IpPort=%s;")
% sqlEscape(strIpPort)))
&& db->startIterRows())
{
if (db->getNull("ScanNext"))
{
// Non-scanning connection terminated. Schedule for scanning.
int iInterval = theConfig.PEER_SCAN_INTERVAL_MIN;
boost::posix_time::ptime tpNow = boost::posix_time::second_clock::universal_time();
boost::posix_time::ptime tpNext = tpNow + boost::posix_time::seconds(iInterval);
std::cerr << str(boost::format("peerScanSet: scan schedule: %s %s (next %s, delay=%s)")
% mScanIp % mScanPort % tpNext % iInterval) << std::endl;
db->executeSQL(str(boost::format("UPDATE PeerIps SET ScanNext=%d,ScanInterval=%d WHERE IpPort=%s;")
% iToSeconds(tpNext)
% iInterval
% db->escape(strIpPort)));
}
else
{
// Scanning connection terminate, already scheduled for retry.
nothing();
}
}
else
{
std::cerr << "peerScanSet: peer wasn't in PeerIps: " << strIp << " " << iPort << std::endl;
}
}
void ConnectionPool::peerFailed(const std::string& strIp, int iPort)
{
std::cerr << "ConnectionPool::peerFailed: " << strIp << " " << iPort << std::endl;
ipPort ipPeer = make_pair(strIp, iPort);
// If the fail was our scan, we are no longer scanning.
if (bScanning && !mScanIp.compare(strIp) && mScanPort == iPort)
{
bScanning = false;
// Look for more to scan.
scanRefresh();
}
bool bScanSet = false;
{
boost::mutex::scoped_lock sl(mPeerLock);
boost::unordered_map<ipPort, Peer::pointer>::iterator itIp;
itIp = mIpMap.find(ipPeer);
@@ -381,18 +454,15 @@ void ConnectionPool::peerDisconnected(Peer::pointer peer, const ipPort& ipPeer,
{
// Found it. Delete it.
mIpMap.erase(itIp);
bScanSet = true;
}
}
void ConnectionPool::peerFailed(const std::string& strIp, int iPort)
if (bScanSet)
{
// If the fail was our scan, we are no longer scanning.
if (bScanning && !mScanIp.compare(strIp) && mScanPort == iPort)
{
bScanning = false;
// Look for more to scan.
scanRefresh();
// Schedule for scanning.
peerScanSet(ipPeer.first, ipPeer.second);
}
}
@@ -401,6 +471,9 @@ void ConnectionPool::peerVerified(const std::string& strIp, int iPort)
if (bScanning && !mScanIp.compare(strIp), mScanPort == iPort)
{
std::string strIpPort = str(boost::format("%s %d") % strIp % iPort);
std::cerr << str(boost::format("peerVerified: %s %s (scan off)") % mScanIp % mScanPort) << std::endl;
// Scan completed successfully.
{
ScopedLock sl(theApp->getWalletDB()->getDBLock());
@@ -486,7 +559,6 @@ void ConnectionPool::scanRefresh()
(void) mScanTimer.cancel();
std::cerr << "scanRefresh: scanning: " << mScanIp << " " << mScanPort << std::endl;
bScanning = true;
iInterval *= 2;
@@ -494,6 +566,9 @@ void ConnectionPool::scanRefresh()
tpNext = tpNow + boost::posix_time::seconds(iInterval);
std::cerr << str(boost::format("scanRefresh: now scanning: %s %s (next %s, delay=%s)")
% mScanIp % mScanPort % tpNext % iInterval) << std::endl;
{
ScopedLock sl(theApp->getWalletDB()->getDBLock());
Database *db=theApp->getWalletDB()->getDB();
@@ -505,7 +580,7 @@ void ConnectionPool::scanRefresh()
// XXX Check error.
}
if (!connectTo(mScanIp, mScanPort))
if (!peerConnect(mScanIp, mScanPort))
{
// Already connected. Try again.
scanRefresh();

20
src/ConnectionPool.h
View File

@@ -20,9 +20,14 @@ private:
typedef std::pair<ipPort, Peer::pointer> pipPeer;
// Peers we are connecting with and non-thin peers we are connected to.
// Only peers we know the connection ip for are listed.
// We know the ip and port for:
// - All outbound connections
// - Some inbound connections (which we figured out).
boost::unordered_map<ipPort, Peer::pointer> mIpMap;
// Non-thin peers which we are connected to.
// Peers we have the public key for.
boost::unordered_map<NewcoinAddress, Peer::pointer> mConnectedMap;
boost::asio::ssl::context mCtx;
@@ -42,6 +47,9 @@ private:
// int miConnectStarting;
bool peerAvailable(std::string& strIp, int& iPort);
void peerScanSet(const std::string& strIp, int iPort);
bool peerConnect(const std::string& strIp, int iPort);
public:
ConnectionPool(boost::asio::io_service& io_service);
@@ -54,7 +62,7 @@ public:
// Manual connection request.
// Queue for immediate scanning.
bool connectTo(const std::string& strIp, int iPort);
void connectTo(const std::string& strIp, int iPort);
//
// Peer connectivity notification.
@@ -62,14 +70,12 @@ public:
bool getTopNAddrs(int n,std::vector<std::string>& addrs);
bool savePeer(const std::string& strIp, int iPort, char code);
// Inbound connection, false=reject
bool peerRegister(Peer::pointer peer, const std::string& strIp, int iPort);
// We know peers node public key. false=reject
bool peerConnected(Peer::pointer peer, const NewcoinAddress& na);
// We know peers node public key.
// <-- bool: false=reject
bool peerConnected(Peer::pointer peer, const NewcoinAddress& naPeer, const std::string& strIP, int iPort);
// No longer connected.
void peerDisconnected(Peer::pointer peer, const ipPort& ipPeer, const NewcoinAddress& naPeer);
void peerDisconnected(Peer::pointer peer, const NewcoinAddress& naPeer);
// As client accepted.
void peerVerified(const std::string& strIp, int iPort);

52
src/Peer.cpp
View File

@@ -60,22 +60,32 @@ void Peer::handle_write(const boost::system::error_code& error, size_t bytes_tra
void Peer::detach(const char *rsn)
{
#ifdef DEBUG
std::cerr << "DETACHING PEER: " << rsn << std::endl;
std::cerr << "DETACHING PEER: " << rsn
<< ": "
<< (mNodePublic.isValid() ? mNodePublic.humanNodePublic() : "-")
<< " " << getIP() << " " << getPort() << std::endl;
#endif
boost::system::error_code ecCancel;
(void) mVerifyTimer.cancel();
mSendQ.clear();
if (mNodePublic.isValid())
{
theApp->getConnectionPool().peerDisconnected(shared_from_this(), mNodePublic);
mNodePublic.clear(); // Be idompotent.
}
if (!mIpPort.first.empty())
{
if (mClientConnect)
// Connection might be part of scanning. Inform connect failed.
// Might need to scan. Inform connection disconnected.
theApp->getConnectionPool().peerFailed(mIpPort.first, mIpPort.second);
theApp->getConnectionPool().peerDisconnected(shared_from_this(), mIpPort, mNodePublic);
mIpPort.first.clear();
mIpPort.first.empty(); // Be idompotent.
}
}
@@ -191,7 +201,8 @@ void Peer::handleConnect(const boost::system::error_code& error, boost::asio::ip
}
}
// Connect ssl as server.
// Connect ssl as server to an inbound connection.
// - We don't bother remembering the inbound IP or port. Only useful for debugging.
void Peer::connected(const boost::system::error_code& error)
{
boost::asio::ip::tcp::endpoint ep = mSocketSsl.lowest_layer().remote_endpoint();
@@ -208,21 +219,12 @@ void Peer::connected(const boost::system::error_code& error)
std::cerr << "Remote peer: accept error: " << strIp << " " << iPort << " : " << error << std::endl;
detach("ctd");
}
else if (!theApp->getConnectionPool().peerRegister(shared_from_this(), strIp, iPort))
{
std::cerr << "Remote peer: rejecting: " << strIp << " " << iPort << std::endl;
// XXX Reject with a rejection message: already connected
detach("ctd2");
}
else
{
// Not redundant ip and port, add to connection list.
std::cerr << "Remote peer: accepted: " << strIp << " " << iPort << std::endl;
mIpPort = make_pair(strIp, iPort);
assert(!mIpPort.first.empty());
mSocketSsl.set_verify_mode(boost::asio::ssl::verify_none);
mSocketSsl.async_handshake(boost::asio::ssl::stream<boost::asio::ip::tcp::socket>::server,
@@ -307,7 +309,6 @@ void Peer::handle_read_body(const boost::system::error_code& error)
}
}
void Peer::processReadBuffer()
{
int type = PackedMessage::getType(mReadbuf);
@@ -529,18 +530,20 @@ void Peer::recvHello(newcoin::TMHello& packet)
{ // Unable to verify they have private key for claimed public key.
std::cerr << "Recv(Hello): Disconnect: Failed to verify session." << std::endl;
}
else if (!theApp->getConnectionPool().peerConnected(shared_from_this(), mNodePublic))
else if (!theApp->getConnectionPool().peerConnected(shared_from_this(), mNodePublic, getIP(), getPort()))
{ // Already connected, self, or some other reason.
std::cerr << "Recv(Hello): Disconnect: Extraneous connection." << std::endl;
}
else
{ // Successful connection.
std::cerr << "Recv(Hello): Connect: " << mNodePublic.humanNodePublic() << std::endl;
// Cancel verification timeout.
(void) mVerifyTimer.cancel();
if (mClientConnect)
{
// If we connected due to scan, no longer need to scan.
theApp->getConnectionPool().peerVerified(mIpPort.first, mIpPort.second);
// No longer connecting as client.
@@ -568,9 +571,6 @@ void Peer::recvHello(newcoin::TMHello& packet)
mClosedLedgerTime = boost::posix_time::second_clock::universal_time();
}
theApp->getConnectionPool().savePeer(getIP(),packet.ipv4port(),'I');
bDetach = false;
}
@@ -712,7 +712,9 @@ void Peer::recvGetContacts(newcoin::TMGetContacts& packet)
void Peer::recvGetPeers(newcoin::TMGetPeers& packet)
{
std::vector<std::string> addrs;
theApp->getConnectionPool().getTopNAddrs(30,addrs);
if (addrs.size())
{
newcoin::TMPeers peers;
@@ -721,7 +723,9 @@ void Peer::recvGetPeers(newcoin::TMGetPeers& packet)
{
std::string strIP;
int port;
splitIpPort(addrs[n], strIP, port);
newcoin::TMIPv4EndPoint* addr=peers.add_nodes();
addr->set_ipv4(inet_addr(strIP.c_str()));
addr->set_ipv4port(port);
@@ -734,6 +738,7 @@ void Peer::recvGetPeers(newcoin::TMGetPeers& packet)
sendPacket(message);
}
}
// TODO: filter out all the LAN peers
void Peer::recvPeers(newcoin::TMPeers& packet)
{
@@ -744,11 +749,18 @@ void Peer::recvPeers(newcoin::TMPeers& packet)
std::string strIP(inet_ntoa(addr));
int port=packet.nodes(i).ipv4port();
if (strIP == "0.0.0.0")
{
strIP = mSocketSsl.lowest_layer().remote_endpoint().address().to_string();
}
// if (strIP != "127.0.0.1")
{
std::cout << "Learning about: " << strIP << std::endl;
theApp->getConnectionPool().savePeer(strIP, port, 'T');
}
}
}
void Peer::recvIndexedObject(newcoin::TMIndexedObject& packet)

6
src/Peer.h
View File

@@ -96,8 +96,10 @@ public:
//bool operator == (const Peer& other);
std::string& getIP(){ return(mIpPort.first); }
int getPort(){ return(mIpPort.second); }
std::string& getIP() { return mIpPort.first; }
int getPort() { return mIpPort.second; }
void peerIpPort(const std::string& strIP, int iPort) { mIpPort = make_pair(strIP, iPort); }
static pointer create(boost::asio::io_service& io_service, boost::asio::ssl::context& ctx)
{

4
src/RPCServer.cpp
View File

@@ -728,8 +728,8 @@ Json::Value RPCServer::doConnect(Json::Value& params)
iPort = boost::lexical_cast<int>(strPort);
}
if (!theApp->getConnectionPool().connectTo(strIp, iPort))
return "connected";
// XXX Validate legal IP and port
theApp->getConnectionPool().connectTo(strIp, iPort);
return "connecting";
}