mirror of
https://github.com/XRPLF/rippled.git
synced 2025-12-06 17:27:55 +00:00
Make InboundLedgers, LedgerConsensus abstract
This commit is contained in:
NATTSiM
File diff suppressed because it is too large
Load Diff
@@ -26,148 +26,56 @@
|
||||
is destroyed when the process is complete.
|
||||
*/
|
||||
class LedgerConsensus
|
||||
: public boost::enable_shared_from_this <LedgerConsensus>
|
||||
, public CountedObject <LedgerConsensus>
|
||||
{
|
||||
public:
|
||||
static char const* getCountedObjectName () { return "LedgerConsensus"; }
|
||||
public:
|
||||
static shared_ptr <LedgerConsensus> New(
|
||||
LedgerHash const & prevLCLHash, Ledger::ref previousLedger,
|
||||
uint32 closeTime);
|
||||
|
||||
LedgerConsensus (LedgerHash const & prevLCLHash, Ledger::ref previousLedger, uint32 closeTime);
|
||||
virtual int startup () = 0;
|
||||
|
||||
int startup ();
|
||||
virtual Json::Value getJson (bool full) = 0;
|
||||
|
||||
Json::Value getJson (bool full);
|
||||
virtual Ledger::ref peekPreviousLedger () = 0;
|
||||
|
||||
Ledger::ref peekPreviousLedger ()
|
||||
{
|
||||
return mPreviousLedger;
|
||||
}
|
||||
virtual uint256 getLCL () = 0;
|
||||
|
||||
uint256 getLCL ()
|
||||
{
|
||||
return mPrevLedgerHash;
|
||||
}
|
||||
virtual SHAMap::pointer getTransactionTree (uint256 const & hash,
|
||||
bool doAcquire) = 0;
|
||||
|
||||
SHAMap::pointer getTransactionTree (uint256 const & hash, bool doAcquire);
|
||||
virtual void mapComplete (uint256 const & hash, SHAMap::ref map,
|
||||
bool acquired) = 0;
|
||||
|
||||
TransactionAcquire::pointer getAcquiring (uint256 const & hash);
|
||||
virtual bool stillNeedTXSet (uint256 const & hash) = 0;
|
||||
|
||||
void mapComplete (uint256 const & hash, SHAMap::ref map, bool acquired);
|
||||
virtual void checkLCL () = 0;
|
||||
|
||||
bool stillNeedTXSet (uint256 const & hash);
|
||||
virtual void handleLCL (uint256 const & lclHash) = 0;
|
||||
|
||||
void checkLCL ();
|
||||
|
||||
void handleLCL (uint256 const & lclHash);
|
||||
|
||||
void timerEntry ();
|
||||
virtual void timerEntry () = 0;
|
||||
|
||||
// state handlers
|
||||
void statePreClose ();
|
||||
void stateEstablish ();
|
||||
void stateCutoff ();
|
||||
void stateFinished ();
|
||||
void stateAccepted ();
|
||||
virtual void statePreClose () = 0;
|
||||
virtual void stateEstablish () = 0;
|
||||
virtual void stateFinished () = 0;
|
||||
virtual void stateAccepted () = 0;
|
||||
|
||||
bool haveConsensus (bool forReal);
|
||||
virtual bool haveConsensus (bool forReal) = 0;
|
||||
|
||||
bool peerPosition (LedgerProposal::ref);
|
||||
virtual bool peerPosition (LedgerProposal::ref) = 0;
|
||||
|
||||
bool peerHasSet (Peer::ref peer, uint256 const & set, protocol::TxSetStatus status);
|
||||
virtual bool peerHasSet (Peer::ref peer, uint256 const & set,
|
||||
protocol::TxSetStatus status) = 0;
|
||||
|
||||
SHAMapAddNode peerGaveNodes (Peer::ref peer, uint256 const & setHash,
|
||||
const std::list<SHAMapNode>& nodeIDs, const std::list< Blob >& nodeData);
|
||||
virtual SHAMapAddNode peerGaveNodes (Peer::ref peer,
|
||||
uint256 const & setHash,
|
||||
const std::list<SHAMapNode>& nodeIDs,
|
||||
const std::list< Blob >& nodeData) = 0;
|
||||
|
||||
bool isOurPubKey (const RippleAddress & k)
|
||||
{
|
||||
return k == mValPublic;
|
||||
}
|
||||
virtual bool isOurPubKey (const RippleAddress & k) = 0;
|
||||
|
||||
// test/debug
|
||||
void simulate ();
|
||||
|
||||
private:
|
||||
// final accept logic
|
||||
void accept (SHAMap::ref txSet, LoadEvent::pointer);
|
||||
|
||||
void weHave (uint256 const & id, Peer::ref avoidPeer);
|
||||
void startAcquiring (TransactionAcquire::pointer);
|
||||
SHAMap::pointer find (uint256 const & hash);
|
||||
|
||||
void createDisputes (SHAMap::ref, SHAMap::ref);
|
||||
void addDisputedTransaction (uint256 const& , Blob const & transaction);
|
||||
void adjustCount (SHAMap::ref map, const std::vector<uint160>& peers);
|
||||
void propose ();
|
||||
|
||||
void addPosition (LedgerProposal&, bool ours);
|
||||
void removePosition (LedgerProposal&, bool ours);
|
||||
void sendHaveTxSet (uint256 const & set, bool direct);
|
||||
void applyTransactions (SHAMap::ref transactionSet, Ledger::ref targetLedger,
|
||||
Ledger::ref checkLedger, CanonicalTXSet & failedTransactions, bool openLgr);
|
||||
int applyTransaction (TransactionEngine & engine, SerializedTransaction::ref txn, Ledger::ref targetLedger,
|
||||
bool openLgr, bool retryAssured);
|
||||
|
||||
uint32 roundCloseTime (uint32 closeTime);
|
||||
|
||||
// manipulating our own position
|
||||
void statusChange (protocol::NodeEvent, Ledger & ledger);
|
||||
void takeInitialPosition (Ledger & initialLedger);
|
||||
void updateOurPositions ();
|
||||
void playbackProposals ();
|
||||
int getThreshold ();
|
||||
void closeLedger ();
|
||||
void checkOurValidation ();
|
||||
|
||||
void beginAccept (bool synchronous);
|
||||
void endConsensus ();
|
||||
|
||||
void addLoad (SerializedValidation::ref val);
|
||||
|
||||
private:
|
||||
// VFALCO TODO Rename these to look pretty
|
||||
enum LCState
|
||||
{
|
||||
lcsPRE_CLOSE, // We haven't closed our ledger yet, but others might have
|
||||
lcsESTABLISH, // Establishing consensus
|
||||
lcsFINISHED, // We have closed on a transaction set
|
||||
lcsACCEPTED, // We have accepted/validated a new last closed ledger
|
||||
};
|
||||
|
||||
LCState mState;
|
||||
uint32 mCloseTime; // The wall time this ledger closed
|
||||
uint256 mPrevLedgerHash, mNewLedgerHash;
|
||||
Ledger::pointer mPreviousLedger;
|
||||
InboundLedger::pointer mAcquiringLedger;
|
||||
LedgerProposal::pointer mOurPosition;
|
||||
RippleAddress mValPublic, mValPrivate;
|
||||
bool mProposing, mValidating, mHaveCorrectLCL, mConsensusFail;
|
||||
|
||||
int mCurrentMSeconds, mClosePercent, mCloseResolution;
|
||||
bool mHaveCloseTimeConsensus;
|
||||
|
||||
boost::posix_time::ptime mConsensusStartTime;
|
||||
int mPreviousProposers;
|
||||
int mPreviousMSeconds;
|
||||
|
||||
// Convergence tracking, trusted peers indexed by hash of public key
|
||||
boost::unordered_map<uint160, LedgerProposal::pointer> mPeerPositions;
|
||||
|
||||
// Transaction Sets, indexed by hash of transaction tree
|
||||
boost::unordered_map<uint256, SHAMap::pointer> mAcquired;
|
||||
boost::unordered_map<uint256, TransactionAcquire::pointer> mAcquiring;
|
||||
|
||||
// Peer sets
|
||||
boost::unordered_map<uint256, std::vector< boost::weak_ptr<Peer> > > mPeerData;
|
||||
|
||||
// Disputed transactions
|
||||
boost::unordered_map<uint256, DisputedTx::pointer> mDisputes;
|
||||
boost::unordered_set<uint256> mCompares;
|
||||
|
||||
// Close time estimates
|
||||
std::map<uint32, int> mCloseTimes;
|
||||
|
||||
// nodes that have bowed out of this consensus process
|
||||
boost::unordered_set<uint160> mDeadNodes;
|
||||
virtual void simulate () = 0;
|
||||
};
|
||||
|
||||
|
||||
|
||||
@@ -17,347 +17,418 @@
|
||||
*/
|
||||
//==============================================================================
|
||||
|
||||
typedef std::pair<uint256, InboundLedger::pointer> u256_acq_pair;
|
||||
|
||||
InboundLedgers::InboundLedgers (Stoppable& parent)
|
||||
: Stoppable ("InboundLedgers", parent)
|
||||
, mLock (this, "InboundLedger", __FILE__, __LINE__)
|
||||
, mRecentFailures ("LedgerAcquireRecentFailures", 0, kReacquireIntervalSeconds)
|
||||
class InboundLedgersImp
|
||||
: public InboundLedgers
|
||||
, public Stoppable
|
||||
, public LeakChecked <InboundLedgers>
|
||||
{
|
||||
}
|
||||
|
||||
InboundLedger::pointer InboundLedgers::findCreate (uint256 const& hash, uint32 seq, bool couldBeNew)
|
||||
{
|
||||
assert (hash.isNonZero ());
|
||||
InboundLedger::pointer ret;
|
||||
public:
|
||||
typedef std::pair<uint256, InboundLedger::pointer> u256_acq_pair;
|
||||
// How long before we try again to acquire the same ledger
|
||||
static const int kReacquireIntervalSeconds = 300;
|
||||
|
||||
explicit InboundLedgersImp (Stoppable& parent)
|
||||
: Stoppable ("InboundLedgers", parent)
|
||||
, mLock (this, "InboundLedger", __FILE__, __LINE__)
|
||||
, mRecentFailures ("LedgerAcquireRecentFailures", 0,
|
||||
kReacquireIntervalSeconds)
|
||||
{
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
}
|
||||
|
||||
// VFALCO TODO Should this be called findOrAdd ?
|
||||
//
|
||||
InboundLedger::pointer findCreate (uint256 const& hash, uint32 seq,
|
||||
bool couldBeNew)
|
||||
{
|
||||
assert (hash.isNonZero ());
|
||||
InboundLedger::pointer ret;
|
||||
|
||||
if (! isStopping ())
|
||||
{
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
|
||||
if (! isStopping ())
|
||||
{
|
||||
boost::unordered_map<uint256, InboundLedger::pointer>::iterator it = mLedgers.find (hash);
|
||||
if (it != mLedgers.end ())
|
||||
{
|
||||
ret = it->second;
|
||||
// FIXME: Should set the sequence if it's not set
|
||||
}
|
||||
else
|
||||
{
|
||||
ret = boost::make_shared<InboundLedger> (hash, seq);
|
||||
assert (ret);
|
||||
mLedgers.insert (std::make_pair (hash, ret));
|
||||
ret->init (sl, couldBeNew);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
InboundLedger::pointer find (uint256 const& hash)
|
||||
{
|
||||
assert (hash.isNonZero ());
|
||||
|
||||
InboundLedger::pointer ret;
|
||||
|
||||
{
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
|
||||
boost::unordered_map<uint256, InboundLedger::pointer>::iterator it = mLedgers.find (hash);
|
||||
if (it != mLedgers.end ())
|
||||
{
|
||||
ret = it->second;
|
||||
// FIXME: Should set the sequence if it's not set
|
||||
}
|
||||
else
|
||||
{
|
||||
ret = boost::make_shared<InboundLedger> (hash, seq);
|
||||
assert (ret);
|
||||
mLedgers.insert (std::make_pair (hash, ret));
|
||||
ret->init (sl, couldBeNew);
|
||||
}
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
InboundLedger::pointer InboundLedgers::find (uint256 const& hash)
|
||||
{
|
||||
assert (hash.isNonZero ());
|
||||
|
||||
InboundLedger::pointer ret;
|
||||
|
||||
bool hasLedger (LedgerHash const& hash)
|
||||
{
|
||||
assert (hash.isNonZero ());
|
||||
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
|
||||
boost::unordered_map<uint256, InboundLedger::pointer>::iterator it = mLedgers.find (hash);
|
||||
if (it != mLedgers.end ())
|
||||
{
|
||||
ret = it->second;
|
||||
}
|
||||
return mLedgers.find (hash) != mLedgers.end ();
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
bool InboundLedgers::hasLedger (LedgerHash const& hash)
|
||||
{
|
||||
assert (hash.isNonZero ());
|
||||
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
return mLedgers.find (hash) != mLedgers.end ();
|
||||
}
|
||||
|
||||
void InboundLedgers::dropLedger (LedgerHash const& hash)
|
||||
{
|
||||
assert (hash.isNonZero ());
|
||||
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
mLedgers.erase (hash);
|
||||
|
||||
}
|
||||
|
||||
bool InboundLedgers::awaitLedgerData (LedgerHash const& ledgerHash)
|
||||
{
|
||||
InboundLedger::pointer ledger = find (ledgerHash);
|
||||
|
||||
if (!ledger)
|
||||
return false;
|
||||
|
||||
ledger->awaitData ();
|
||||
return true;
|
||||
}
|
||||
|
||||
/*
|
||||
This gets called when
|
||||
"We got some data from an inbound ledger"
|
||||
|
||||
inboundLedgerTrigger:
|
||||
"What do we do with this partial data?"
|
||||
Figures out what to do with the responses to our requests for information.
|
||||
|
||||
*/
|
||||
// means "We got some data from an inbound ledger"
|
||||
void InboundLedgers::gotLedgerData (Job&, LedgerHash hash,
|
||||
boost::shared_ptr<protocol::TMLedgerData> packet_ptr, boost::weak_ptr<Peer> wPeer)
|
||||
{
|
||||
protocol::TMLedgerData& packet = *packet_ptr;
|
||||
Peer::pointer peer = wPeer.lock ();
|
||||
|
||||
WriteLog (lsTRACE, InboundLedger) << "Got data (" << packet.nodes ().size () << ") for acquiring ledger: " << hash;
|
||||
|
||||
InboundLedger::pointer ledger = find (hash);
|
||||
|
||||
if (!ledger)
|
||||
void dropLedger (LedgerHash const& hash)
|
||||
{
|
||||
WriteLog (lsTRACE, InboundLedger) << "Got data for ledger we're not acquiring";
|
||||
assert (hash.isNonZero ());
|
||||
|
||||
if (peer)
|
||||
{
|
||||
peer->charge (Resource::feeInvalidRequest);
|
||||
}
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
mLedgers.erase (hash);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
ledger->noAwaitData ();
|
||||
|
||||
if (!peer)
|
||||
return;
|
||||
|
||||
if (packet.type () == protocol::liBASE)
|
||||
bool awaitLedgerData (LedgerHash const& ledgerHash)
|
||||
{
|
||||
if (packet.nodes_size () < 1)
|
||||
{
|
||||
WriteLog (lsWARNING, InboundLedger) << "Got empty base data";
|
||||
peer->charge (Resource::feeInvalidRequest);
|
||||
return;
|
||||
}
|
||||
InboundLedger::pointer ledger = find (ledgerHash);
|
||||
|
||||
if (!ledger->takeBase (packet.nodes (0).nodedata ()))
|
||||
{
|
||||
WriteLog (lsWARNING, InboundLedger) << "Got invalid base data";
|
||||
peer->charge (Resource::feeInvalidRequest);
|
||||
return;
|
||||
}
|
||||
if (!ledger)
|
||||
return false;
|
||||
|
||||
SHAMapAddNode san = SHAMapAddNode::useful ();
|
||||
|
||||
if ((packet.nodes ().size () > 1) && !ledger->takeAsRootNode (strCopy (packet.nodes (1).nodedata ()), san))
|
||||
{
|
||||
WriteLog (lsWARNING, InboundLedger) << "Included ASbase invalid";
|
||||
}
|
||||
|
||||
if ((packet.nodes ().size () > 2) && !ledger->takeTxRootNode (strCopy (packet.nodes (2).nodedata ()), san))
|
||||
{
|
||||
WriteLog (lsWARNING, InboundLedger) << "Included TXbase invalid";
|
||||
}
|
||||
|
||||
if (!san.isInvalid ())
|
||||
{
|
||||
ledger->progress ();
|
||||
ledger->trigger (peer);
|
||||
}
|
||||
else
|
||||
WriteLog (lsDEBUG, InboundLedger) << "Peer sends invalid base data";
|
||||
|
||||
return;
|
||||
ledger->awaitData ();
|
||||
return true;
|
||||
}
|
||||
/*
|
||||
This gets called when
|
||||
"We got some data from an inbound ledger"
|
||||
|
||||
if ((packet.type () == protocol::liTX_NODE) || (packet.type () == protocol::liAS_NODE))
|
||||
inboundLedgerTrigger:
|
||||
"What do we do with this partial data?"
|
||||
Figures out what to do with the responses to our requests for information.
|
||||
|
||||
*/
|
||||
// means "We got some data from an inbound ledger"
|
||||
|
||||
// VFALCO TODO Why is hash passed by value?
|
||||
// VFALCO TODO Remove the dependency on the Peer object.
|
||||
//
|
||||
void gotLedgerData (Job& job,
|
||||
LedgerHash hash,
|
||||
boost::shared_ptr<protocol::TMLedgerData> packet_ptr,
|
||||
boost::weak_ptr<Peer> wPeer)
|
||||
{
|
||||
std::list<SHAMapNode> nodeIDs;
|
||||
std::list< Blob > nodeData;
|
||||
protocol::TMLedgerData& packet = *packet_ptr;
|
||||
Peer::pointer peer = wPeer.lock ();
|
||||
|
||||
if (packet.nodes ().size () <= 0)
|
||||
WriteLog (lsTRACE, InboundLedger) << "Got data (" << packet.nodes ().size () << ") for acquiring ledger: " << hash;
|
||||
|
||||
InboundLedger::pointer ledger = find (hash);
|
||||
|
||||
if (!ledger)
|
||||
{
|
||||
WriteLog (lsINFO, InboundLedger) << "Got response with no nodes";
|
||||
peer->charge (Resource::feeInvalidRequest);
|
||||
return;
|
||||
}
|
||||
WriteLog (lsTRACE, InboundLedger) << "Got data for ledger we're not acquiring";
|
||||
|
||||
for (int i = 0; i < packet.nodes ().size (); ++i)
|
||||
{
|
||||
const protocol::TMLedgerNode& node = packet.nodes (i);
|
||||
|
||||
if (!node.has_nodeid () || !node.has_nodedata ())
|
||||
if (peer)
|
||||
{
|
||||
WriteLog (lsWARNING, InboundLedger) << "Got bad node";
|
||||
peer->charge (Resource::feeInvalidRequest);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
ledger->noAwaitData ();
|
||||
|
||||
if (!peer)
|
||||
return;
|
||||
|
||||
if (packet.type () == protocol::liBASE)
|
||||
{
|
||||
if (packet.nodes_size () < 1)
|
||||
{
|
||||
WriteLog (lsWARNING, InboundLedger) << "Got empty base data";
|
||||
peer->charge (Resource::feeInvalidRequest);
|
||||
return;
|
||||
}
|
||||
|
||||
nodeIDs.push_back (SHAMapNode (node.nodeid ().data (), node.nodeid ().size ()));
|
||||
nodeData.push_back (Blob (node.nodedata ().begin (), node.nodedata ().end ()));
|
||||
}
|
||||
|
||||
SHAMapAddNode ret;
|
||||
|
||||
if (packet.type () == protocol::liTX_NODE)
|
||||
ledger->takeTxNode (nodeIDs, nodeData, ret);
|
||||
else
|
||||
ledger->takeAsNode (nodeIDs, nodeData, ret);
|
||||
|
||||
if (!ret.isInvalid ())
|
||||
{
|
||||
ledger->progress ();
|
||||
ledger->trigger (peer);
|
||||
}
|
||||
else
|
||||
WriteLog (lsDEBUG, InboundLedger) << "Peer sends invalid node data";
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
WriteLog (lsWARNING, InboundLedger) << "Not sure what ledger data we got";
|
||||
peer->charge (Resource::feeInvalidRequest);
|
||||
}
|
||||
|
||||
void InboundLedgers::sweep ()
|
||||
{
|
||||
mRecentFailures.sweep ();
|
||||
|
||||
int const now = UptimeTimer::getInstance ().getElapsedSeconds ();
|
||||
|
||||
// Make a list of things to sweep, while holding the lock
|
||||
std::vector <MapType::mapped_type> stuffToSweep;
|
||||
std::size_t total;
|
||||
{
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
MapType::iterator it (mLedgers.begin ());
|
||||
total = mLedgers.size ();
|
||||
stuffToSweep.reserve (total);
|
||||
|
||||
while (it != mLedgers.end ())
|
||||
{
|
||||
if (it->second->getLastAction () > now)
|
||||
if (!ledger->takeBase (packet.nodes (0).nodedata ()))
|
||||
{
|
||||
it->second->touch ();
|
||||
++it;
|
||||
WriteLog (lsWARNING, InboundLedger) << "Got invalid base data";
|
||||
peer->charge (Resource::feeInvalidRequest);
|
||||
return;
|
||||
}
|
||||
else if ((it->second->getLastAction () + 60) < now)
|
||||
|
||||
SHAMapAddNode san = SHAMapAddNode::useful ();
|
||||
|
||||
if ((packet.nodes ().size () > 1) && !ledger->takeAsRootNode (strCopy (packet.nodes (1).nodedata ()), san))
|
||||
{
|
||||
stuffToSweep.push_back (it->second);
|
||||
// shouldn't cause the actual final delete
|
||||
// since we are holding a reference in the vector.
|
||||
it = mLedgers.erase (it);
|
||||
WriteLog (lsWARNING, InboundLedger) << "Included ASbase invalid";
|
||||
}
|
||||
|
||||
if ((packet.nodes ().size () > 2) && !ledger->takeTxRootNode (strCopy (packet.nodes (2).nodedata ()), san))
|
||||
{
|
||||
WriteLog (lsWARNING, InboundLedger) << "Included TXbase invalid";
|
||||
}
|
||||
|
||||
if (!san.isInvalid ())
|
||||
{
|
||||
ledger->progress ();
|
||||
ledger->trigger (peer);
|
||||
}
|
||||
else
|
||||
WriteLog (lsDEBUG, InboundLedger) << "Peer sends invalid base data";
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
if ((packet.type () == protocol::liTX_NODE) || (packet.type () == protocol::liAS_NODE))
|
||||
{
|
||||
std::list<SHAMapNode> nodeIDs;
|
||||
std::list< Blob > nodeData;
|
||||
|
||||
if (packet.nodes ().size () <= 0)
|
||||
{
|
||||
++it;
|
||||
WriteLog (lsINFO, InboundLedger) << "Got response with no nodes";
|
||||
peer->charge (Resource::feeInvalidRequest);
|
||||
return;
|
||||
}
|
||||
|
||||
for (int i = 0; i < packet.nodes ().size (); ++i)
|
||||
{
|
||||
const protocol::TMLedgerNode& node = packet.nodes (i);
|
||||
|
||||
if (!node.has_nodeid () || !node.has_nodedata ())
|
||||
{
|
||||
WriteLog (lsWARNING, InboundLedger) << "Got bad node";
|
||||
peer->charge (Resource::feeInvalidRequest);
|
||||
return;
|
||||
}
|
||||
|
||||
nodeIDs.push_back (SHAMapNode (node.nodeid ().data (), node.nodeid ().size ()));
|
||||
nodeData.push_back (Blob (node.nodedata ().begin (), node.nodedata ().end ()));
|
||||
}
|
||||
|
||||
SHAMapAddNode ret;
|
||||
|
||||
if (packet.type () == protocol::liTX_NODE)
|
||||
ledger->takeTxNode (nodeIDs, nodeData, ret);
|
||||
else
|
||||
ledger->takeAsNode (nodeIDs, nodeData, ret);
|
||||
|
||||
if (!ret.isInvalid ())
|
||||
{
|
||||
ledger->progress ();
|
||||
ledger->trigger (peer);
|
||||
}
|
||||
else
|
||||
WriteLog (lsDEBUG, InboundLedger) << "Peer sends invalid node data";
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
WriteLog (lsWARNING, InboundLedger) << "Not sure what ledger data we got";
|
||||
peer->charge (Resource::feeInvalidRequest);
|
||||
}
|
||||
|
||||
int getFetchCount (int& timeoutCount)
|
||||
{
|
||||
timeoutCount = 0;
|
||||
int ret = 0;
|
||||
|
||||
std::vector<u256_acq_pair> inboundLedgers;
|
||||
|
||||
{
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
|
||||
inboundLedgers.reserve(mLedgers.size());
|
||||
BOOST_FOREACH (const u256_acq_pair & it, mLedgers)
|
||||
{
|
||||
inboundLedgers.push_back(it);
|
||||
}
|
||||
}
|
||||
|
||||
BOOST_FOREACH (const u256_acq_pair & it, inboundLedgers)
|
||||
{
|
||||
if (it.second->isActive ())
|
||||
{
|
||||
++ret;
|
||||
timeoutCount += it.second->getTimeouts ();
|
||||
}
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
WriteLog (lsDEBUG, InboundLedger) <<
|
||||
"Sweeped " << stuffToSweep.size () <<
|
||||
" out of " << total << " inbound ledgers.";
|
||||
}
|
||||
void logFailure (uint256 const& h)
|
||||
{
|
||||
mRecentFailures.add (h);
|
||||
}
|
||||
|
||||
int InboundLedgers::getFetchCount (int& timeoutCount)
|
||||
{
|
||||
timeoutCount = 0;
|
||||
int ret = 0;
|
||||
|
||||
std::vector<u256_acq_pair> inboundLedgers;
|
||||
bool isFailure (uint256 const& h)
|
||||
{
|
||||
return mRecentFailures.isPresent (h, false);
|
||||
}
|
||||
|
||||
void clearFailures ()
|
||||
{
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
|
||||
inboundLedgers.reserve(mLedgers.size());
|
||||
BOOST_FOREACH (const u256_acq_pair & it, mLedgers)
|
||||
mRecentFailures.clear();
|
||||
mLedgers.clear();
|
||||
}
|
||||
|
||||
Json::Value getInfo()
|
||||
{
|
||||
Json::Value ret(Json::objectValue);
|
||||
|
||||
std::vector<u256_acq_pair> acquires;
|
||||
{
|
||||
inboundLedgers.push_back(it);
|
||||
}
|
||||
}
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
|
||||
BOOST_FOREACH (const u256_acq_pair & it, inboundLedgers)
|
||||
{
|
||||
if (it.second->isActive ())
|
||||
acquires.reserve (mLedgers.size ());
|
||||
BOOST_FOREACH (const u256_acq_pair & it, mLedgers)
|
||||
{
|
||||
assert (it.second);
|
||||
acquires.push_back (it);
|
||||
}
|
||||
}
|
||||
|
||||
BOOST_FOREACH (const u256_acq_pair& it, acquires)
|
||||
{
|
||||
++ret;
|
||||
timeoutCount += it.second->getTimeouts ();
|
||||
uint32 seq = it.second->getSeq();
|
||||
if (seq > 1)
|
||||
ret[lexicalCastThrow <std::string>(seq)] = it.second->getJson(0);
|
||||
else
|
||||
ret[it.first.GetHex()] = it.second->getJson(0);
|
||||
}
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
void InboundLedgers::gotFetchPack (Job&)
|
||||
{
|
||||
std::vector<InboundLedger::pointer> acquires;
|
||||
{
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
|
||||
acquires.reserve (mLedgers.size ());
|
||||
BOOST_FOREACH (const u256_acq_pair & it, mLedgers)
|
||||
{
|
||||
assert (it.second);
|
||||
acquires.push_back (it.second);
|
||||
}
|
||||
}
|
||||
|
||||
BOOST_FOREACH (const InboundLedger::pointer & acquire, acquires)
|
||||
{
|
||||
acquire->checkLocal ();
|
||||
}
|
||||
}
|
||||
|
||||
void InboundLedgers::clearFailures ()
|
||||
{
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
|
||||
mRecentFailures.clear();
|
||||
mLedgers.clear();
|
||||
}
|
||||
|
||||
Json::Value InboundLedgers::getInfo()
|
||||
{
|
||||
Json::Value ret(Json::objectValue);
|
||||
|
||||
std::vector<u256_acq_pair> acquires;
|
||||
{
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
|
||||
acquires.reserve (mLedgers.size ());
|
||||
BOOST_FOREACH (const u256_acq_pair & it, mLedgers)
|
||||
{
|
||||
assert (it.second);
|
||||
acquires.push_back (it);
|
||||
}
|
||||
}
|
||||
|
||||
BOOST_FOREACH (const u256_acq_pair& it, acquires)
|
||||
{
|
||||
uint32 seq = it.second->getSeq();
|
||||
if (seq > 1)
|
||||
ret[lexicalCastThrow <std::string>(seq)] = it.second->getJson(0);
|
||||
else
|
||||
ret[it.first.GetHex()] = it.second->getJson(0);
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
}
|
||||
|
||||
void InboundLedgers::onStop ()
|
||||
void gotFetchPack (Job&)
|
||||
{
|
||||
std::vector<InboundLedger::pointer> acquires;
|
||||
{
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
|
||||
acquires.reserve (mLedgers.size ());
|
||||
BOOST_FOREACH (const u256_acq_pair & it, mLedgers)
|
||||
{
|
||||
assert (it.second);
|
||||
acquires.push_back (it.second);
|
||||
}
|
||||
}
|
||||
|
||||
BOOST_FOREACH (const InboundLedger::pointer & acquire, acquires)
|
||||
{
|
||||
acquire->checkLocal ();
|
||||
}
|
||||
}
|
||||
|
||||
void sweep ()
|
||||
{
|
||||
mRecentFailures.sweep ();
|
||||
|
||||
int const now = UptimeTimer::getInstance ().getElapsedSeconds ();
|
||||
|
||||
// Make a list of things to sweep, while holding the lock
|
||||
std::vector <MapType::mapped_type> stuffToSweep;
|
||||
std::size_t total;
|
||||
{
|
||||
ScopedLockType sl (mLock, __FILE__, __LINE__);
|
||||
MapType::iterator it (mLedgers.begin ());
|
||||
total = mLedgers.size ();
|
||||
stuffToSweep.reserve (total);
|
||||
|
||||
while (it != mLedgers.end ())
|
||||
{
|
||||
if (it->second->getLastAction () > now)
|
||||
{
|
||||
it->second->touch ();
|
||||
++it;
|
||||
}
|
||||
else if ((it->second->getLastAction () + 60) < now)
|
||||
{
|
||||
stuffToSweep.push_back (it->second);
|
||||
// shouldn't cause the actual final delete
|
||||
// since we are holding a reference in the vector.
|
||||
it = mLedgers.erase (it);
|
||||
}
|
||||
else
|
||||
{
|
||||
++it;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
WriteLog (lsDEBUG, InboundLedger) <<
|
||||
"Sweeped " << stuffToSweep.size () <<
|
||||
" out of " << total << " inbound ledgers.";
|
||||
}
|
||||
|
||||
void onStop ()
|
||||
{
|
||||
ScopedLockType lock (mLock, __FILE__, __LINE__);
|
||||
|
||||
mLedgers.clear();
|
||||
mRecentFailures.clear();
|
||||
|
||||
stopped();
|
||||
}
|
||||
|
||||
private:
|
||||
typedef boost::unordered_map <uint256, InboundLedger::pointer> MapType;
|
||||
|
||||
typedef RippleRecursiveMutex LockType;
|
||||
typedef LockType::ScopedLockType ScopedLockType;
|
||||
LockType mLock;
|
||||
|
||||
MapType mLedgers;
|
||||
KeyCache <uint256, UptimeTimerAdapter> mRecentFailures;
|
||||
};
|
||||
|
||||
//------------------------------------------------------------------------------
|
||||
|
||||
InboundLedgers::~InboundLedgers()
|
||||
{
|
||||
ScopedLockType lock (mLock, __FILE__, __LINE__);
|
||||
|
||||
mLedgers.clear();
|
||||
mRecentFailures.clear();
|
||||
|
||||
stopped();
|
||||
}
|
||||
|
||||
InboundLedgers* InboundLedgers::New (Stoppable& parent)
|
||||
{
|
||||
return new InboundLedgersImp (parent);
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -24,68 +24,48 @@
|
||||
|
||||
@see InboundLedger
|
||||
*/
|
||||
// VFALCO TODO Rename to InboundLedgers
|
||||
// VFALCO TODO Create abstract interface
|
||||
class InboundLedgers
|
||||
: public Stoppable
|
||||
, public LeakChecked <InboundLedger>
|
||||
{
|
||||
public:
|
||||
// How long before we try again to acquire the same ledger
|
||||
static const int kReacquireIntervalSeconds = 300;
|
||||
virtual ~InboundLedgers() = 0;
|
||||
|
||||
explicit InboundLedgers (Stoppable& parent);
|
||||
static InboundLedgers* New (Stoppable& parent);
|
||||
|
||||
// VFALCO TODO Should this be called findOrAdd ?
|
||||
//
|
||||
InboundLedger::pointer findCreate (uint256 const& hash, uint32 seq, bool bCouldBeNew);
|
||||
virtual InboundLedger::pointer findCreate (uint256 const& hash,
|
||||
uint32 seq,
|
||||
bool bCouldBeNew) = 0;
|
||||
|
||||
InboundLedger::pointer find (uint256 const& hash);
|
||||
virtual InboundLedger::pointer find (uint256 const& hash) = 0;
|
||||
|
||||
bool hasLedger (LedgerHash const& ledgerHash);
|
||||
virtual bool hasLedger (LedgerHash const& ledgerHash) = 0;
|
||||
|
||||
void dropLedger (LedgerHash const& ledgerHash);
|
||||
virtual void dropLedger (LedgerHash const& ledgerHash) = 0;
|
||||
|
||||
bool awaitLedgerData (LedgerHash const& ledgerHash);
|
||||
virtual bool awaitLedgerData (LedgerHash const& ledgerHash) = 0;
|
||||
|
||||
// VFALCO TODO Why is hash passed by value?
|
||||
// VFALCO TODO Remove the dependency on the Peer object.
|
||||
//
|
||||
void gotLedgerData (Job&,
|
||||
virtual void gotLedgerData (Job& job,
|
||||
LedgerHash hash,
|
||||
boost::shared_ptr <protocol::TMLedgerData> packet,
|
||||
boost::weak_ptr<Peer> peer);
|
||||
boost::weak_ptr<Peer> peer) = 0;
|
||||
|
||||
int getFetchCount (int& timeoutCount);
|
||||
virtual int getFetchCount (int& timeoutCount) = 0;
|
||||
|
||||
void logFailure (uint256 const& h)
|
||||
{
|
||||
mRecentFailures.add (h);
|
||||
}
|
||||
virtual void logFailure (uint256 const& h) = 0;
|
||||
|
||||
bool isFailure (uint256 const& h)
|
||||
{
|
||||
return mRecentFailures.isPresent (h, false);
|
||||
}
|
||||
virtual bool isFailure (uint256 const& h) = 0;
|
||||
|
||||
void clearFailures();
|
||||
virtual void clearFailures() = 0;
|
||||
|
||||
Json::Value getInfo();
|
||||
virtual Json::Value getInfo() = 0;
|
||||
|
||||
void gotFetchPack (Job&);
|
||||
void sweep ();
|
||||
virtual void gotFetchPack (Job&) = 0;
|
||||
virtual void sweep () = 0;
|
||||
|
||||
void onStop ();
|
||||
|
||||
private:
|
||||
typedef boost::unordered_map <uint256, InboundLedger::pointer> MapType;
|
||||
|
||||
typedef RippleRecursiveMutex LockType;
|
||||
typedef LockType::ScopedLockType ScopedLockType;
|
||||
LockType mLock;
|
||||
|
||||
MapType mLedgers;
|
||||
KeyCache <uint256, UptimeTimerAdapter> mRecentFailures;
|
||||
};
|
||||
|
||||
#endif
|
||||
|
||||
@@ -131,7 +131,7 @@ public:
|
||||
|
||||
, m_sntpClient (SNTPClient::New (*this))
|
||||
|
||||
, m_inboundLedgers (*m_jobQueue)
|
||||
, m_inboundLedgers (InboundLedgers::New(*m_jobQueue))
|
||||
|
||||
, m_txQueue (TxQueue::New ())
|
||||
|
||||
@@ -215,7 +215,7 @@ public:
|
||||
|
||||
InboundLedgers& getInboundLedgers ()
|
||||
{
|
||||
return m_inboundLedgers;
|
||||
return *m_inboundLedgers;
|
||||
}
|
||||
|
||||
TransactionMaster& getMasterTransaction ()
|
||||
@@ -914,7 +914,7 @@ private:
|
||||
NodeStoreScheduler m_nodeStoreScheduler;
|
||||
ScopedPointer <NodeStore::Database> m_nodeStore;
|
||||
ScopedPointer <SNTPClient> m_sntpClient;
|
||||
InboundLedgers m_inboundLedgers;
|
||||
beast::unique_ptr <InboundLedgers> m_inboundLedgers;
|
||||
ScopedPointer <TxQueue> m_txQueue;
|
||||
ScopedPointer <Validators::Manager> m_validators;
|
||||
ScopedPointer <IFeatures> mFeatures;
|
||||
|
||||
@@ -1403,8 +1403,10 @@ int NetworkOPsImp::beginConsensus (uint256 const& networkClosed, Ledger::pointer
|
||||
// Create a consensus object to get consensus on this ledger
|
||||
assert (!mConsensus);
|
||||
prevLedger->setImmutable ();
|
||||
mConsensus = boost::make_shared<LedgerConsensus> (
|
||||
networkClosed, prevLedger, m_ledgerMaster.getCurrentLedger ()->getCloseTimeNC ());
|
||||
|
||||
mConsensus = LedgerConsensus::New(
|
||||
networkClosed, prevLedger,
|
||||
m_ledgerMaster.getCurrentLedger ()->getCloseTimeNC ());
|
||||
|
||||
m_journal.debug << "Initiating consensus engine";
|
||||
return mConsensus->startup ();
|
||||
|
||||
Reference in New Issue
Block a user