ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2026-08-21 14:20:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fetch pack cleanups:

Browse Source 
* Limit backed up fetch pack requests
* Cleanups
* Dispatch fetch packs sooner
This commit is contained in:
JoelKatz
2014-03-14 12:38:47 -07:00
committed by Vinnie Falco
parent 0075f36bbc
commit db3a387224
4 changed files with 58 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
src/ripple_app/misc/NetworkOPs.cpp
View File

@@ -243,7 +243,7 @@ public:
void mapComplete (uint256 const& hash, SHAMap::ref map);
bool stillNeedTXSet (uint256 const& hash);
void makeFetchPack (Job&, boost::weak_ptr<Peer> peer, boost::shared_ptr<protocol::TMGetObjectByHash> request,
Ledger::pointer wantLedger, Ledger::pointer haveLedger, std::uint32_t uUptime);
uint256 haveLedger, std::uint32_t uUptime);
bool shouldFetchPack (std::uint32_t seq);
void gotFetchPack (bool progress, std::uint32_t seq);
void addFetchPack (uint256 const& hash, boost::shared_ptr< Blob >& data);
@@ -3191,8 +3191,7 @@ static void fpAppender (protocol::TMGetObjectByHash* reply, std::uint32_t ledger
void NetworkOPsImp::makeFetchPack (Job&, boost::weak_ptr<Peer> wPeer,
boost::shared_ptr<protocol::TMGetObjectByHash> request,
Ledger::pointer wantLedger, Ledger::pointer haveLedger,
std::uint32_t uUptime)
uint256 haveLedgerHash, std::uint32_t uUptime)
{
if (UptimeTimer::getInstance ().getElapsedSeconds () > (uUptime + 1))
{
@@ -3200,18 +3199,54 @@ void NetworkOPsImp::makeFetchPack (Job&, boost::weak_ptr<Peer> wPeer,
return;
}
if (getApp().getFeeTrack ().isLoadedLocal ())
if (getApp().getFeeTrack ().isLoadedLocal () ||
(m_ledgerMaster.getValidatedLedgerAge() > 40))
{
m_journal.info << "Too busy to make fetch pack";
return;
}
Peer::ptr peer = wPeer.lock ();
if (!peer)
return;
Ledger::pointer haveLedger = getLedgerByHash (haveLedgerHash);
if (!haveLedger)
{
m_journal.info << "Peer requests fetch pack for ledger we don't have: " << haveLedger;
peer->charge (Resource::feeRequestNoReply);
return;
}
if (!haveLedger->isClosed ())
{
m_journal.warning << "Peer requests fetch pack from open ledger: " << haveLedger;
peer->charge (Resource::feeInvalidRequest);
return;
}
if (haveLedger->getLedgerSeq() < m_ledgerMaster.getEarliestFetch())
{
m_journal.debug << "Peer requests fetch pack that is too early";
peer->charge (Resource::feeInvalidRequest);
return;
}
Ledger::pointer wantLedger = getLedgerByHash (haveLedger->getParentHash ());
if (!wantLedger)
{
m_journal.info
<< "Peer requests fetch pack for ledger whose predecessor we don't have: "
<< haveLedger;
peer->charge (Resource::feeRequestNoReply);
return;
}
try
{
Peer::ptr peer = wPeer.lock ();
if (!peer)
return;
protocol::TMGetObjectByHash reply;
reply.set_query (false);
@@ -3234,7 +3269,8 @@ void NetworkOPsImp::makeFetchPack (Job&, boost::weak_ptr<Peer> wPeer,
newObj.set_data (s.getDataPtr (), s.getLength ());
newObj.set_ledgerseq (lSeq);
wantLedger->peekAccountStateMap ()->getFetchPack (haveLedger->peekAccountStateMap ().get (), true, 1024,
wantLedger->peekAccountStateMap ()->getFetchPack
(haveLedger->peekAccountStateMap ().get (), true, 1024,
BIND_TYPE (fpAppender, &reply, lSeq, P_1, P_2));
if (wantLedger->getTransHash ().isNonZero ())
@@ -3244,7 +3280,7 @@ void NetworkOPsImp::makeFetchPack (Job&, boost::weak_ptr<Peer> wPeer,
if (reply.objects ().size () >= 256)
break;
// VFALCO NOTE Why use move?
// move may save a ref/unref
haveLedger = std::move (wantLedger);
wantLedger = getLedgerByHash (haveLedger->getParentHash ());
}

2
src/ripple_app/misc/NetworkOPs.h
View File

@@ -228,7 +228,7 @@ public:
// Fetch packs
virtual void makeFetchPack (Job&, boost::weak_ptr<Peer> peer,
boost::shared_ptr<protocol::TMGetObjectByHash> request,
Ledger::pointer wantLedger, Ledger::pointer haveLedger, std::uint32_t uUptime) = 0;
uint256 wantLedger, std::uint32_t uUptime) = 0;
virtual bool shouldFetchPack (std::uint32_t seq) = 0;
virtual void gotFetchPack (bool progress, std::uint32_t seq) = 0;

9
src/ripple_app/shamap/SHAMapSync.cpp
View File

@@ -635,16 +635,13 @@ std::list<SHAMap::fetchPackEntry_t> SHAMap::getFetchPack (SHAMap* have, bool inc
void SHAMap::getFetchPack (SHAMap* have, bool includeLeaves, int max,
std::function<void (const uint256&, const Blob&)> func)
{
ScopedReadLockType ul1 (mLock);
std::unique_ptr <ScopedReadLockType> ul2;
ScopedReadLockType ul1 (mLock), ul2;
if (have)
{
// VFALCO NOTE This looks like a mess. A dynamically allocated scoped lock?
ul2.reset (new ScopedReadLockType (have->mLock, boost::try_to_lock));
ul2 = std::move (ScopedReadLockType (have->mLock, boost::try_to_lock));
if (! ul2->owns_lock ())
if (! ul2.owns_lock ())
{
WriteLog (lsINFO, SHAMap) << "Unable to create pack due to lock";
return;

43
src/ripple_overlay/impl/PeerImp.h
View File

@@ -2556,7 +2556,11 @@ private:
void doFetchPack (const boost::shared_ptr<protocol::TMGetObjectByHash>& packet)
{
// VFALCO TODO Invert this dependency using an observer and shared state object.
if (getApp().getFeeTrack ().isLoadedLocal ())
// Don't queue fetch pack jobs if we're under load or we already have
// some queued.
if (getApp().getFeeTrack ().isLoadedLocal () ||
(getApp().getLedgerMaster().getValidatedLedgerAge() > 40) ||
(getApp().getJobQueue().getJobCount(jtPACK) > 10))
{
m_journal.info << "Too busy to make fetch pack";
return;
@@ -2572,41 +2576,10 @@ private:
uint256 hash;
memcpy (hash.begin (), packet->ledgerhash ().data (), 32);
Ledger::pointer haveLedger = getApp().getOPs ().getLedgerByHash (hash);
if (!haveLedger)
{
m_journal.info << "Peer requests fetch pack for ledger we don't have: " << hash;
charge (Resource::feeRequestNoReply);
return;
}
if (!haveLedger->isClosed ())
{
m_journal.warning << "Peer requests fetch pack from open ledger: " << hash;
charge (Resource::feeInvalidRequest);
return;
}
if (haveLedger->getLedgerSeq() < getApp().getLedgerMaster().getEarliestFetch())
{
m_journal.debug << "Peer requests fetch pack that is too early";
charge (Resource::feeInvalidRequest);
return;
}
Ledger::pointer wantLedger = getApp().getOPs ().getLedgerByHash (haveLedger->getParentHash ());
if (!wantLedger)
{
m_journal.info << "Peer requests fetch pack for ledger whose predecessor we don't have: " << hash;
charge (Resource::feeRequestNoReply);
return;
}
getApp().getJobQueue ().addJob (jtPACK, "MakeFetchPack",
BIND_TYPE (&NetworkOPs::makeFetchPack, &getApp().getOPs (), P_1,
boost::weak_ptr<Peer> (shared_from_this ()), packet, wantLedger, haveLedger, UptimeTimer::getInstance ().getElapsedSeconds ()));
BIND_TYPE (&NetworkOPs::makeFetchPack, &getApp().getOPs (), P_1,
boost::weak_ptr<Peer> (shared_from_this ()), packet,
hash, UptimeTimer::getInstance ().getElapsedSeconds ()));
}
void doProofOfWork (Job&, boost::weak_ptr <Peer> peer, ProofOfWork::pointer pow)