Merge branch 'master' of github.com:jedmccaleb/NewCoin into notdone

This commit is contained in:
jed
2012-11-29 09:23:08 -08:00
16 changed files with 93 additions and 32 deletions

View File

@@ -71,8 +71,24 @@ static void InitDB(DatabaseCon** dbCon, const char *fileName, const char *dbInit
*dbCon = new DatabaseCon(fileName, dbInit, dbCount); *dbCon = new DatabaseCon(fileName, dbInit, dbCount);
} }
volatile bool doShutdown = false;
#ifdef SIGINT
void sigIntHandler(int)
{
doShutdown = true;
}
#endif
void Application::run() void Application::run()
{ {
#ifdef SIGINT
struct sigaction sa;
memset(&sa, 0, sizeof(sa));
sa.sa_handler = sigIntHandler;
sigaction(SIGINT, &sa, NULL);
#endif
assert(mTxnDB == NULL); assert(mTxnDB == NULL);
if (!theConfig.DEBUG_LOGFILE.empty()) if (!theConfig.DEBUG_LOGFILE.empty())
{ // Let DEBUG messages go to the file but only WARNING or higher to regular output (unless verbose) { // Let DEBUG messages go to the file but only WARNING or higher to regular output (unless verbose)

View File

@@ -35,7 +35,7 @@ public:
DatabaseCon(const std::string& name, const char *initString[], int countInit); DatabaseCon(const std::string& name, const char *initString[], int countInit);
~DatabaseCon(); ~DatabaseCon();
Database* getDB() { return mDatabase; } Database* getDB() { return mDatabase; }
ScopedLock getDBLock() { return ScopedLock(mLock); } boost::recursive_mutex& getDBLock() { return mLock; }
}; };
class Application class Application

View File

@@ -12,7 +12,7 @@ SETUP_LOG();
DECLARE_INSTANCE(HashedObject); DECLARE_INSTANCE(HashedObject);
HashedObjectStore::HashedObjectStore(int cacheSize, int cacheAge) : HashedObjectStore::HashedObjectStore(int cacheSize, int cacheAge) :
mCache("HashedObjectStore", cacheSize, cacheAge), mWritePending(false) mCache("HashedObjectStore", cacheSize, cacheAge), mWriteGeneration(0), mWritePending(false)
{ {
mWriteSet.reserve(128); mWriteSet.reserve(128);
} }
@@ -53,27 +53,29 @@ bool HashedObjectStore::store(HashedObjectType type, uint32 index,
void HashedObjectStore::waitWrite() void HashedObjectStore::waitWrite()
{ {
boost::unique_lock<boost::mutex> sl(mWriteMutex); boost::mutex::scoped_lock sl(mWriteMutex);
while (mWritePending) int gen = mWriteGeneration;
while (mWritePending && (mWriteGeneration == gen))
mWriteCondition.wait(sl); mWriteCondition.wait(sl);
} }
void HashedObjectStore::bulkWrite() void HashedObjectStore::bulkWrite()
{ {
LoadEvent::pointer event = theApp->getJobQueue().getLoadEvent(jtDISK); LoadEvent::autoptr event(theApp->getJobQueue().getLoadEventAP(jtDISK));
while (1) while (1)
{ {
std::vector< boost::shared_ptr<HashedObject> > set; std::vector< boost::shared_ptr<HashedObject> > set;
set.reserve(128); set.reserve(128);
{ {
boost::unique_lock<boost::mutex> sl(mWriteMutex); boost::mutex::scoped_lock sl(mWriteMutex);
mWriteSet.swap(set); mWriteSet.swap(set);
assert(mWriteSet.empty()); assert(mWriteSet.empty());
++mWriteGeneration;
mWriteCondition.notify_all();
if (set.empty()) if (set.empty())
{ {
mWritePending = false; mWritePending = false;
mWriteCondition.notify_all();
return; return;
} }
} }
@@ -85,7 +87,7 @@ void HashedObjectStore::bulkWrite()
Database* db = theApp->getHashNodeDB()->getDB(); Database* db = theApp->getHashNodeDB()->getDB();
{ {
ScopedLock sl = theApp->getHashNodeDB()->getDBLock(); ScopedLock sl( theApp->getHashNodeDB()->getDBLock());
db->executeSQL("BEGIN TRANSACTION;"); db->executeSQL("BEGIN TRANSACTION;");

View File

@@ -49,6 +49,7 @@ protected:
boost::mutex mWriteMutex; boost::mutex mWriteMutex;
boost::condition_variable mWriteCondition; boost::condition_variable mWriteCondition;
int mWriteGeneration;
std::vector< boost::shared_ptr<HashedObject> > mWriteSet; std::vector< boost::shared_ptr<HashedObject> > mWriteSet;
bool mWritePending; bool mWritePending;

View File

@@ -104,6 +104,8 @@ public:
LoadEvent::pointer getLoadEvent(JobType t) LoadEvent::pointer getLoadEvent(JobType t)
{ return boost::make_shared<LoadEvent>(boost::ref(mJobLoads[t]), true, 1); } { return boost::make_shared<LoadEvent>(boost::ref(mJobLoads[t]), true, 1); }
LoadEvent::autoptr getLoadEventAP(JobType t)
{ return LoadEvent::autoptr(new LoadEvent(mJobLoads[t], true, 1)); }
Json::Value getJson(int c = 0); Json::Value getJson(int c = 0);
}; };

View File

@@ -299,6 +299,29 @@ SerializedTransaction::pointer Ledger::getSTransaction(SHAMapItem::ref item, SHA
return SerializedTransaction::pointer(); return SerializedTransaction::pointer();
} }
SerializedTransaction::pointer Ledger::getSMTransaction(SHAMapItem::ref item, SHAMapTreeNode::TNType type,
TransactionMetaSet::pointer& txMeta)
{
SerializerIterator sit(item->peekSerializer());
if (type == SHAMapTreeNode::tnTRANSACTION_NM)
{
txMeta.reset();
return boost::make_shared<SerializedTransaction>(boost::ref(sit));
}
else if (type == SHAMapTreeNode::tnTRANSACTION_MD)
{
Serializer sTxn(sit.getVL());
SerializerIterator tSit(sTxn);
txMeta = boost::make_shared<TransactionMetaSet>(item->getTag(), mLedgerSeq, sit.getVL());
return boost::make_shared<SerializedTransaction>(boost::ref(tSit));
}
txMeta.reset();
return SerializedTransaction::pointer();
}
bool Ledger::getTransaction(const uint256& txID, Transaction::pointer& txn, TransactionMetaSet::pointer& meta) bool Ledger::getTransaction(const uint256& txID, Transaction::pointer& txn, TransactionMetaSet::pointer& meta)
{ {
SHAMapTreeNode::TNType type; SHAMapTreeNode::TNType type;
@@ -339,9 +362,8 @@ uint256 Ledger::getHash()
return mHash; return mHash;
} }
void Ledger::saveAcceptedLedger(bool fromConsensus) void Ledger::saveAcceptedLedger(bool fromConsensus, LoadEvent::pointer event)
{ // can be called in a different thread { // can be called in a different thread
LoadEvent::pointer event = theApp->getJobQueue().getLoadEvent(jtDISK);
cLog(lsTRACE) << "saveAcceptedLedger " << (fromConsensus ? "fromConsensus " : "fromAcquire ") << getLedgerSeq(); cLog(lsTRACE) << "saveAcceptedLedger " << (fromConsensus ? "fromConsensus " : "fromAcquire ") << getLedgerSeq();
static boost::format ledgerExists("SELECT LedgerSeq FROM Ledgers where LedgerSeq = %d;"); static boost::format ledgerExists("SELECT LedgerSeq FROM Ledgers where LedgerSeq = %d;");
static boost::format deleteLedger("DELETE FROM Ledgers WHERE LedgerSeq = %d;"); static boost::format deleteLedger("DELETE FROM Ledgers WHERE LedgerSeq = %d;");
@@ -369,7 +391,7 @@ void Ledger::saveAcceptedLedger(bool fromConsensus)
SHAMap& txSet = *peekTransactionMap(); SHAMap& txSet = *peekTransactionMap();
Database *db = theApp->getTxnDB()->getDB(); Database *db = theApp->getTxnDB()->getDB();
ScopedLock dbLock = theApp->getTxnDB()->getDBLock(); ScopedLock dbLock(theApp->getTxnDB()->getDBLock());
db->executeSQL("BEGIN TRANSACTION;"); db->executeSQL("BEGIN TRANSACTION;");
SHAMapTreeNode::TNType type; SHAMapTreeNode::TNType type;
for (SHAMapItem::pointer item = txSet.peekFirstItem(type); !!item; for (SHAMapItem::pointer item = txSet.peekFirstItem(type); !!item;
@@ -439,7 +461,7 @@ void Ledger::saveAcceptedLedger(bool fromConsensus)
} }
theApp->getLedgerMaster().setFullLedger(shared_from_this()); theApp->getLedgerMaster().setFullLedger(shared_from_this());
event = LoadEvent::pointer(); event->stop();
theApp->getOPs().pubLedger(shared_from_this()); theApp->getOPs().pubLedger(shared_from_this());
@@ -1136,7 +1158,8 @@ void Ledger::pendSave(bool fromConsensus)
if (!fromConsensus && !theApp->isNewFlag(getHash(), SF_SAVED)) if (!fromConsensus && !theApp->isNewFlag(getHash(), SF_SAVED))
return; return;
boost::thread thread(boost::bind(&Ledger::saveAcceptedLedger, shared_from_this(), fromConsensus)); boost::thread thread(boost::bind(&Ledger::saveAcceptedLedger, shared_from_this(),
fromConsensus, theApp->getJobQueue().getLoadEvent(jtDISK)));
thread.detach(); thread.detach();
boost::recursive_mutex::scoped_lock sl(sPendingSaveLock); boost::recursive_mutex::scoped_lock sl(sPendingSaveLock);

View File

@@ -19,6 +19,7 @@
#include "BitcoinUtil.h" #include "BitcoinUtil.h"
#include "SHAMap.h" #include "SHAMap.h"
#include "InstanceCounter.h" #include "InstanceCounter.h"
#include "LoadMonitor.h"
enum LedgerStateParms enum LedgerStateParms
{ {
@@ -93,7 +94,7 @@ protected:
static void incPendingSaves(); static void incPendingSaves();
static void decPendingSaves(); static void decPendingSaves();
void saveAcceptedLedger(bool fromConsensus); void saveAcceptedLedger(bool fromConsensus, LoadEvent::pointer);
public: public:
Ledger(const RippleAddress& masterID, uint64 startAmount); // used for the starting bootstrap ledger Ledger(const RippleAddress& masterID, uint64 startAmount); // used for the starting bootstrap ledger
@@ -160,7 +161,10 @@ public:
bool hasTransaction(const uint256& TransID) const { return mTransactionMap->hasItem(TransID); } bool hasTransaction(const uint256& TransID) const { return mTransactionMap->hasItem(TransID); }
Transaction::pointer getTransaction(const uint256& transID) const; Transaction::pointer getTransaction(const uint256& transID) const;
bool getTransaction(const uint256& transID, Transaction::pointer& txn, TransactionMetaSet::pointer& txMeta); bool getTransaction(const uint256& transID, Transaction::pointer& txn, TransactionMetaSet::pointer& txMeta);
static SerializedTransaction::pointer getSTransaction(SHAMapItem::ref, SHAMapTreeNode::TNType); static SerializedTransaction::pointer getSTransaction(SHAMapItem::ref, SHAMapTreeNode::TNType);
SerializedTransaction::pointer getSMTransaction(SHAMapItem::ref, SHAMapTreeNode::TNType,
TransactionMetaSet::pointer& txMeta);
// high-level functions // high-level functions
AccountState::pointer getAccountState(const RippleAddress& acctID); AccountState::pointer getAccountState(const RippleAddress& acctID);

View File

@@ -644,8 +644,16 @@ void LedgerConsensus::stateAccepted()
endConsensus(); endConsensus();
} }
extern volatile bool doShutdown;
void LedgerConsensus::timerEntry() void LedgerConsensus::timerEntry()
{ {
if (doShutdown)
{
cLog(lsFATAL) << "Shutdown requested";
theApp->stop();
}
if ((mState != lcsFINISHED) && (mState != lcsACCEPTED)) if ((mState != lcsFINISHED) && (mState != lcsACCEPTED))
checkLCL(); checkLCL();
@@ -1251,7 +1259,7 @@ void LedgerConsensus::accept(SHAMap::ref set, LoadEvent::pointer)
cLog(lsINFO) << "CNF newLCL " << newLCLHash; cLog(lsINFO) << "CNF newLCL " << newLCLHash;
Ledger::pointer newOL = boost::make_shared<Ledger>(true, boost::ref(*newLCL)); Ledger::pointer newOL = boost::make_shared<Ledger>(true, boost::ref(*newLCL));
ScopedLock sl = theApp->getLedgerMaster().getLock(); ScopedLock sl( theApp->getLedgerMaster().getLock());
// Apply disputed transactions that didn't get in // Apply disputed transactions that didn't get in
TransactionEngine engine(newOL); TransactionEngine engine(newOL);

View File

@@ -37,6 +37,7 @@ class LoadEvent
{ {
public: public:
typedef boost::shared_ptr<LoadEvent> pointer; typedef boost::shared_ptr<LoadEvent> pointer;
typedef std::auto_ptr<LoadEvent> autoptr;
protected: protected:
LoadMonitor& mMonitor; LoadMonitor& mMonitor;

View File

@@ -889,7 +889,7 @@ std::vector< std::pair<Transaction::pointer, TransactionMetaSet::pointer> >
{ {
Database* db = theApp->getTxnDB()->getDB(); Database* db = theApp->getTxnDB()->getDB();
ScopedLock dbLock = theApp->getTxnDB()->getDBLock(); ScopedLock sl(theApp->getTxnDB()->getDBLock());
SQL_FOREACH(db, sql) SQL_FOREACH(db, sql)
{ {
@@ -923,7 +923,7 @@ std::vector<RippleAddress>
RippleAddress acct; RippleAddress acct;
{ {
Database* db = theApp->getTxnDB()->getDB(); Database* db = theApp->getTxnDB()->getDB();
ScopedLock dblock = theApp->getTxnDB()->getDBLock(); ScopedLock sl(theApp->getTxnDB()->getDBLock());
SQL_FOREACH(db, sql) SQL_FOREACH(db, sql)
{ {
if (acct.setAccountID(db->getStrBinary("Account"))) if (acct.setAccountID(db->getStrBinary("Account")))
@@ -1016,7 +1016,7 @@ void NetworkOPs::pubLedger(Ledger::ref lpAccepted)
if (NetworkOPs::omDISCONNECTED == getOperatingMode()) if (NetworkOPs::omDISCONNECTED == getOperatingMode())
return; return;
LoadEvent::pointer event = theApp->getJobQueue().getLoadEvent(jtPUBLEDGER); LoadEvent::autoptr event(theApp->getJobQueue().getLoadEventAP(jtPUBLEDGER));
{ {
boost::recursive_mutex::scoped_lock sl(mMonitorLock); boost::recursive_mutex::scoped_lock sl(mMonitorLock);

View File

@@ -373,7 +373,7 @@ void Peer::processReadBuffer()
// std::cerr << "Peer::processReadBuffer: " << mIpPort.first << " " << mIpPort.second << std::endl; // std::cerr << "Peer::processReadBuffer: " << mIpPort.first << " " << mIpPort.second << std::endl;
LoadEvent::pointer event = theApp->getJobQueue().getLoadEvent(jtPEER); LoadEvent::autoptr event(theApp->getJobQueue().getLoadEventAP(jtPEER));
boost::recursive_mutex::scoped_lock sl(theApp->getMasterLock()); boost::recursive_mutex::scoped_lock sl(theApp->getMasterLock());

View File

@@ -1158,7 +1158,7 @@ Json::Value RPCHandler::doTxHistory(const Json::Value& params)
{ {
Database* db = theApp->getTxnDB()->getDB(); Database* db = theApp->getTxnDB()->getDB();
ScopedLock dbLock = theApp->getTxnDB()->getDBLock(); ScopedLock sl (theApp->getTxnDB()->getDBLock());
SQL_FOREACH(db, sql) SQL_FOREACH(db, sql)
{ {
@@ -1475,7 +1475,7 @@ Json::Value RPCHandler::doCommand(const std::string& command, Json::Value& param
cLog(lsTRACE) << "RPC:" << command; cLog(lsTRACE) << "RPC:" << command;
cLog(lsTRACE) << "RPC params:" << params; cLog(lsTRACE) << "RPC params:" << params;
LoadEvent::pointer le = theApp->getJobQueue().getLoadEvent(jtRPC); LoadEvent::autoptr le(theApp->getJobQueue().getLoadEventAP(jtRPC));
mRole = role; mRole = role;

View File

@@ -6,16 +6,18 @@
#include <boost/make_shared.hpp> #include <boost/make_shared.hpp>
#include <boost/ref.hpp> #include <boost/ref.hpp>
typedef boost::recursive_mutex::scoped_lock ScopedLock;
// A lock holder that can be returned and copied by value // A lock holder that can be returned and copied by value
// When the last reference goes away, the lock is released // When the last reference goes away, the lock is released
class ScopedLock class SharedScopedLock
{ {
protected: protected:
mutable boost::shared_ptr<boost::recursive_mutex::scoped_lock> mHolder; mutable boost::shared_ptr<boost::recursive_mutex::scoped_lock> mHolder;
public: public:
ScopedLock(boost::recursive_mutex& mutex) : SharedScopedLock(boost::recursive_mutex& mutex) :
mHolder(boost::make_shared<boost::recursive_mutex::scoped_lock>(boost::ref(mutex))) { ; } mHolder(boost::make_shared<boost::recursive_mutex::scoped_lock>(boost::ref(mutex))) { ; }
void lock() const { mHolder->lock(); } void lock() const { mHolder->lock(); }

View File

@@ -146,12 +146,13 @@ bool Transaction::save()
default: status = TXN_SQL_UNKNOWN; default: status = TXN_SQL_UNKNOWN;
} }
std::string exists = boost::str(boost::format("SELECT Status FROM Transactions WHERE TransID = '%s';") static boost::format selStat("SELECT Status FROM Transactions WHERE TransID = '%s';");
% mTransaction->getTransactionID().GetHex()); std::string exists = boost::str(selStat % mTransaction->getTransactionID().GetHex());
Database *db = theApp->getTxnDB()->getDB(); Database *db = theApp->getTxnDB()->getDB();
ScopedLock dbLock = theApp->getTxnDB()->getDBLock(); ScopedLock dbLock(theApp->getTxnDB()->getDBLock());
if (SQL_EXISTS(db, exists)) return false; if (SQL_EXISTS(db, exists))
return false;
return return
db->executeSQL(mTransaction->getSQLInsertHeader() + mTransaction->getSQL(getLedger(), status) + ";"); db->executeSQL(mTransaction->getSQLInsertHeader() + mTransaction->getSQL(getLedger(), status) + ";");
} }

View File

@@ -289,7 +289,7 @@ void ValidationCollection::condWrite()
void ValidationCollection::doWrite() void ValidationCollection::doWrite()
{ {
LoadEvent::pointer event = theApp->getJobQueue().getLoadEvent(jtDISK); LoadEvent::autoptr event(theApp->getJobQueue().getLoadEventAP(jtDISK));
static boost::format insVal("INSERT INTO LedgerValidations " static boost::format insVal("INSERT INTO LedgerValidations "
"(LedgerHash,NodePubKey,Flags,SignTime,Signature) VALUES ('%s','%s','%u','%u',%s);"); "(LedgerHash,NodePubKey,Flags,SignTime,Signature) VALUES ('%s','%s','%u','%u',%s);");

View File

@@ -154,7 +154,8 @@ inline std::string strHex(const uint64 uiHost)
inline static std::string sqlEscape(const std::string& strSrc) inline static std::string sqlEscape(const std::string& strSrc)
{ {
return str(boost::format("X'%s'") % strHex(strSrc)); static boost::format f("X'%s'");
return str(f % strHex(strSrc));
} }
template<class Iterator> template<class Iterator>