From 40527cca2b53656e14f3f10a5ac4033a43d1bcce Mon Sep 17 00:00:00 2001 From: JoelKatz Date: Fri, 9 Nov 2012 14:14:47 -0800 Subject: [PATCH] Start adding support for concurrent I/O. --- src/cpp/ripple/Application.h | 3 +++ src/cpp/ripple/LedgerConsensus.cpp | 1 + src/cpp/ripple/NetworkOPs.cpp | 5 ++++- src/cpp/ripple/Peer.cpp | 4 ++++ src/cpp/ripple/RPCHandler.cpp | 4 +++- 5 files changed, 15 insertions(+), 2 deletions(-) diff --git a/src/cpp/ripple/Application.h b/src/cpp/ripple/Application.h index a10019096..3e68a8b14 100644 --- a/src/cpp/ripple/Application.h +++ b/src/cpp/ripple/Application.h @@ -43,6 +43,8 @@ class Application boost::asio::io_service mIOService, mAuxService; boost::asio::io_service::work mIOWork, mAuxWork; + boost::recursive_mutex mMasterLock; + Wallet mWallet; UniqueNodeList mUNL; LedgerMaster mMasterLedger; @@ -99,6 +101,7 @@ public: JobQueue& getJobQueue() { return mJobQueue; } SuppressionTable& getSuppression() { return mSuppressions; } RPCHandler& getRPCHandler() { return mRPCHandler; } + boost::recursive_mutex& getMasterLock() { return mMasterLock; } bool isNew(const uint256& s) { return mSuppressions.addSuppression(s); } diff --git a/src/cpp/ripple/LedgerConsensus.cpp b/src/cpp/ripple/LedgerConsensus.cpp index 0fc07a4aa..9f9e09945 100644 --- a/src/cpp/ripple/LedgerConsensus.cpp +++ b/src/cpp/ripple/LedgerConsensus.cpp @@ -1141,6 +1141,7 @@ uint32 LedgerConsensus::roundCloseTime(uint32 closeTime) void LedgerConsensus::accept(SHAMap::ref set) { + boost::recursive_mutex::scoped_lock masterLock(theApp->getMasterLock()); assert(set->getHash() == mOurPosition->getCurrentHash()); uint32 closeTime = roundCloseTime(mOurPosition->getCloseTime()); diff --git a/src/cpp/ripple/NetworkOPs.cpp b/src/cpp/ripple/NetworkOPs.cpp index 44259523f..882551cc8 100644 --- a/src/cpp/ripple/NetworkOPs.cpp +++ b/src/cpp/ripple/NetworkOPs.cpp @@ -161,7 +161,6 @@ Transaction::pointer NetworkOPs::submitTransactionSync(const Transaction::pointe Transaction::pointer NetworkOPs::processTransaction(Transaction::pointer trans, stCallback callback) { - Transaction::pointer dbtx = theApp->getMasterTransaction().fetch(trans->getID(), true); int newFlags = theApp->getSuppression().getFlags(trans->getID()); if ((newFlags & SF_BAD) != 0) @@ -182,6 +181,8 @@ Transaction::pointer NetworkOPs::processTransaction(Transaction::pointer trans, theApp->isNewFlag(trans->getID(), SF_SIGGOOD); } + boost::recursive_mutex::scoped_lock sl(theApp->getMasterLock()); + Transaction::pointer dbtx = theApp->getMasterTransaction().fetch(trans->getID(), true); TER r = mLedgerMaster->doTransaction(*trans->getSTransaction(), tapOPEN_LEDGER | tapNO_CHECK_SIGN); trans->setResult(r); @@ -762,6 +763,8 @@ uint256 NetworkOPs::getConsensusLCL() void NetworkOPs::processTrustedProposal(LedgerProposal::pointer proposal, boost::shared_ptr set, RippleAddress nodePublic, uint256 checkLedger, bool sigGood) { + boost::recursive_mutex::scoped_lock sl(theApp->getMasterLock()); + bool relay = true; if (!haveConsensusObject()) diff --git a/src/cpp/ripple/Peer.cpp b/src/cpp/ripple/Peer.cpp index a5f34665f..f364df838 100644 --- a/src/cpp/ripple/Peer.cpp +++ b/src/cpp/ripple/Peer.cpp @@ -41,6 +41,8 @@ void Peer::handle_write(const boost::system::error_code& error, size_t bytes_tra // std::cerr << "Peer::handle_write bytes: "<< bytes_transferred << std::endl; #endif + boost::recursive_mutex::scoped_lock sl(theApp->getMasterLock()); + mSendingPacket.reset(); if (mDetaching) @@ -357,6 +359,7 @@ void Peer::handle_read_body(const boost::system::error_code& error) else { cLog(lsINFO) << "Peer: Body: Error: " << ADDRESS(this) << ": " << error.category().name() << ": " << error.message() << ": " << error; + boost::recursive_mutex::scoped_lock sl(theApp->getMasterLock()); detach("hrb"); } } @@ -371,6 +374,7 @@ void Peer::processReadBuffer() // std::cerr << "Peer::processReadBuffer: " << mIpPort.first << " " << mIpPort.second << std::endl; // If connected and get a mtHELLO or if not connected and get a non-mtHELLO, wrong message was sent. + boost::recursive_mutex::scoped_lock sl(theApp->getMasterLock()); if (mHelloed == (type == ripple::mtHELLO)) { cLog(lsWARNING) << "Wrong message type: " << type; diff --git a/src/cpp/ripple/RPCHandler.cpp b/src/cpp/ripple/RPCHandler.cpp index bd70776c4..fbd24d555 100644 --- a/src/cpp/ripple/RPCHandler.cpp +++ b/src/cpp/ripple/RPCHandler.cpp @@ -1384,7 +1384,9 @@ Json::Value RPCHandler::doCommand(const std::string& command, Json::Value& param return rpcError(rpcNO_NETWORK); } // XXX Should verify we have a current ledger. - else if ((commandsA[i].iOptions & optCurrent) && false) + + boost::recursive_mutex::scoped_lock sl(theApp->getMasterLock()); + if ((commandsA[i].iOptions & optCurrent) && false) { return rpcError(rpcNO_CURRENT); }