From 1e5963aeeb468e480ed9893add4211fb7cd4f426 Mon Sep 17 00:00:00 2001 From: JoelKatz Date: Tue, 7 Jan 2014 12:14:34 -0800 Subject: [PATCH] Move some operations from the I/O queue to the Job queue --- src/ripple_app/consensus/LedgerConsensus.cpp | 13 +++++-------- src/ripple_app/misc/NetworkOPs.cpp | 3 +-- src/ripple_app/tx/TransactionAcquire.cpp | 2 +- src/ripple_core/functional/Job.cpp | 2 +- src/ripple_core/functional/Job.h | 11 ++++++----- 5 files changed, 14 insertions(+), 17 deletions(-) diff --git a/src/ripple_app/consensus/LedgerConsensus.cpp b/src/ripple_app/consensus/LedgerConsensus.cpp index c3ae5adec..935231ef2 100644 --- a/src/ripple_app/consensus/LedgerConsensus.cpp +++ b/src/ripple_app/consensus/LedgerConsensus.cpp @@ -847,7 +847,7 @@ public: private: /** We have a new last closed ledger, process it. Final accept logic */ - void accept (SHAMap::ref set, LoadEvent::pointer) + void accept (SHAMap::pointer set) { if (set->getHash ().isNonZero ()) // put our set where others can get it later @@ -1832,14 +1832,11 @@ private: (mPeerPositions.size (), mCurrentMSeconds, mNewLedgerHash); if (synchronous) - accept (consensusSet, LoadEvent::pointer ()); + accept (consensusSet); else - { // FIXME: Post to JobQueue, not I/O service - getApp().getIOService ().post - (BIND_TYPE (&LedgerConsensusImp::accept - , shared_from_this (), consensusSet - , getApp().getJobQueue ().getLoadEvent - (jtACCEPTLEDGER, "LedgerConsensusImp::beginAccept"))); + { + getApp().getJobQueue().addJob (jtACCEPT, "acceptLedger", + BIND_TYPE (&LedgerConsensusImp::accept, shared_from_this (), consensusSet)); } } diff --git a/src/ripple_app/misc/NetworkOPs.cpp b/src/ripple_app/misc/NetworkOPs.cpp index 990d08306..35ff52be7 100644 --- a/src/ripple_app/misc/NetworkOPs.cpp +++ b/src/ripple_app/misc/NetworkOPs.cpp @@ -740,8 +740,7 @@ void NetworkOPsImp::submitTransaction (Job&, SerializedTransaction::pointer iTra } } - // FIXME: Should submit to job queue - getApp().getIOService ().post (boost::bind (&NetworkOPsImp::processTransaction, this, + getApp().getJobQueue().addJob (jtTRANSACTION, "submitTxn", boost::bind (&NetworkOPsImp::processTransaction, this, boost::make_shared (trans, false), false, false, callback)); } diff --git a/src/ripple_app/tx/TransactionAcquire.cpp b/src/ripple_app/tx/TransactionAcquire.cpp index caf9e0c35..ed1512447 100644 --- a/src/ripple_app/tx/TransactionAcquire.cpp +++ b/src/ripple_app/tx/TransactionAcquire.cpp @@ -58,7 +58,7 @@ void TransactionAcquire::done () map = mMap; } - getApp().getIOService ().post (BIND_TYPE (&TACompletionHandler, mHash, map)); + getApp().getJobQueue().addJob (jtTXN_DATA, "completeAcquire", BIND_TYPE (&TACompletionHandler, mHash, map)); } void TransactionAcquire::onTimer (bool progress, ScopedLockType& psl) diff --git a/src/ripple_core/functional/Job.cpp b/src/ripple_core/functional/Job.cpp index fb1fd4965..f505f2b32 100644 --- a/src/ripple_core/functional/Job.cpp +++ b/src/ripple_core/functional/Job.cpp @@ -119,6 +119,7 @@ const char* Job::toString (JobType t) case jtWAL: return "writeAhead"; case jtVALIDATION_t: return "trustedValidation"; case jtWRITE: return "writeObjects"; + case jtACCEPT: return "acceptLedger"; case jtPROPOSAL_t: return "trustedProposal"; case jtSWEEP: return "sweep"; case jtNETOP_CLUSTER: return "clusterReport"; @@ -129,7 +130,6 @@ const char* Job::toString (JobType t) // special types not dispatched by the job pool case jtPEER: return "peerCommand"; case jtDISK: return "diskAccess"; - case jtACCEPTLEDGER: return "acceptLedger"; case jtTXN_PROC: return "processTransaction"; case jtOB_SETUP: return "orderBookSetup"; case jtPATH_FIND: return "pathFind"; diff --git a/src/ripple_core/functional/Job.h b/src/ripple_core/functional/Job.h index b709430be..b4c7ebb8e 100644 --- a/src/ripple_core/functional/Job.h +++ b/src/ripple_core/functional/Job.h @@ -44,11 +44,12 @@ enum JobType jtWAL = 16, // Write-ahead logging jtVALIDATION_t = 17, // A validation from a trusted source jtWRITE = 18, // Write out hashed objects - jtPROPOSAL_t = 19, // A proposal from a trusted source - jtSWEEP = 20, // Sweep for stale structures - jtNETOP_CLUSTER = 21, // NetworkOPs cluster peer report - jtNETOP_TIMER = 22, // NetworkOPs net timer processing - jtADMIN = 23, // An administrative operation + jtACCEPT = 19, // Accept a consensus ledger + jtPROPOSAL_t = 20, // A proposal from a trusted source + jtSWEEP = 21, // Sweep for stale structures + jtNETOP_CLUSTER = 22, // NetworkOPs cluster peer report + jtNETOP_TIMER = 23, // NetworkOPs net timer processing + jtADMIN = 24, // An administrative operation // special types not dispatched by the job pool jtPEER = 30,