diff options
Diffstat (limited to 'cpp/src')
| -rw-r--r-- | cpp/src/Freeze/DBI.cpp | 669 | ||||
| -rw-r--r-- | cpp/src/Freeze/DBI.h | 54 |
2 files changed, 411 insertions, 312 deletions
diff --git a/cpp/src/Freeze/DBI.cpp b/cpp/src/Freeze/DBI.cpp index c6364c0ba97..32f341628e6 100644 --- a/cpp/src/Freeze/DBI.cpp +++ b/cpp/src/Freeze/DBI.cpp @@ -25,16 +25,308 @@ using namespace Freeze; # define FREEZE_DB_MODE (S_IRUSR | S_IWUSR) #endif -Freeze::DBI::DBI(const CommunicatorPtr& communicator, const string& name, const DBEnvIPtr& dbenvObj, ::DB_ENV* dbenv, - ::DB* db) : +Freeze::DBEnvI::DBEnvI(const CommunicatorPtr& communicator, const string& name) : _communicator(communicator), - _name(name), + _logger(communicator->getLogger()), + _trace(0), + _dbenv(0), + _name(name) +{ + _errorPrefix = "Freeze::DBEnv(\"" + _name + "\"): "; + + PropertiesPtr properties = _communicator->getProperties(); + string value; + + value = properties->getProperty("Freeze.Trace.DB"); + if (!value.empty()) + { + _trace = atoi(value.c_str()); + } + + int ret; + + ret = db_env_create(&_dbenv, 0); + + if (ret != 0) + { + ostringstream s; + s << _errorPrefix << "db_env_create: " << db_strerror(ret); + DBException ex; + ex.message = s.str(); + throw ex; + } + + if (_trace >= 1) + { + ostringstream s; + s << "opening database environment \"" << _name << "\""; + _logger->trace("DB", s.str()); + } + + ret = _dbenv->open(_dbenv, _name.c_str(), + DB_CREATE | + DB_INIT_LOCK | + DB_INIT_LOG | + DB_INIT_MPOOL | + //DB_INIT_TXN | + DB_RECOVER | + DB_THREAD, + FREEZE_DB_MODE); + if (ret != 0) + { + ostringstream s; + s << _errorPrefix << "DB_ENV->open: " << db_strerror(ret); + DBException ex; + ex.message = s.str(); + throw ex; + } +} + +Freeze::DBEnvI::~DBEnvI() +{ + if (_dbenv) + { + ostringstream s; + s << _errorPrefix << "\"" << _name << "\" has not been closed"; + _communicator->getLogger()->warning(s.str()); + } +} + +string +Freeze::DBEnvI::getName() +{ + // No mutex lock necessary, _name is immutable + return _name; +} + +DBPtr +Freeze::DBEnvI::openDB(const string& name) +{ + JTCSyncT<JTCRecursiveMutex> sync(*this); + + if (!_dbenv) + { + ostringstream s; + s << _errorPrefix << "\"" << _name << "\" has been closed"; + DBException ex; + ex.message = s.str(); + throw ex; + } + + map<string, DBPtr>::iterator p = _dbmap.find(name); + if (p != _dbmap.end()) + { + return p->second; + } + + ::DB* db; + int ret = db_create(&db, _dbenv, 0); + + if (ret != 0) + { + ostringstream s; + s << _errorPrefix << "db_create: " << db_strerror(ret); + DBException ex; + ex.message = s.str(); + throw ex; + } + + return new DBI(_communicator, this, db, name); +} + +TXNPtr +Freeze::DBEnvI::startTXN() +{ + JTCSyncT<JTCRecursiveMutex> sync(*this); + + return new TXNI(_communicator, _dbenv, _name); +} + +void +Freeze::DBEnvI::close() +{ + JTCSyncT<JTCRecursiveMutex> sync(*this); + + if (!_dbenv) + { + return; + } + + while(!_dbmap.empty()) + { + _dbmap.begin()->second->close(); + } + + if (_trace >= 1) + { + ostringstream s; + s << "closing database environment \"" << _name << "\""; + _logger->trace("DB", s.str()); + } + + int ret = _dbenv->close(_dbenv, 0); + + if (ret != 0) + { + ostringstream s; + s << _errorPrefix << "DB_ENV->close: " << db_strerror(ret); + DBException ex; + ex.message = s.str(); + throw ex; + } + + _dbenv = 0; +} + +void +Freeze::DBEnvI::add(const string& name, const DBPtr& db) +{ + JTCSyncT<JTCRecursiveMutex> sync(*this); + + _dbmap[name] = db; +} + +void +Freeze::DBEnvI::remove(const string& name) +{ + JTCSyncT<JTCRecursiveMutex> sync(*this); + + _dbmap.erase(name); +} + +DBEnvPtr +Freeze::initialize(const CommunicatorPtr& communicator, const string& name) +{ + return new DBEnvI(communicator, name); +} + +Freeze::TXNI::TXNI(const CommunicatorPtr& communicator, ::DB_ENV* dbenv, const string& name) : + _communicator(communicator), + _logger(communicator->getLogger()), + _trace(0), + _tid(0), + _name(name) +{ + _errorPrefix = "Freeze::TXN(\"" + _name + "\"): "; + + PropertiesPtr properties = _communicator->getProperties(); + string value; + + value = properties->getProperty("Freeze.Trace.TXN"); + if (!value.empty()) + { + _trace = atoi(value.c_str()); + } + + if (_trace >= 1) + { + ostringstream s; + s << "starting transaction for environment \"" << _name << "\""; + _logger->trace("DB", s.str()); + } + + int ret = txn_begin(dbenv, 0, &_tid, 0); + + if (ret != 0) + { + ostringstream s; + s << _errorPrefix << "txn_begin: " << db_strerror(ret); + DBException ex; + ex.message = s.str(); + throw ex; + } +} + +Freeze::TXNI::~TXNI() +{ + if (_tid) + { + ostringstream s; + s << _errorPrefix << "transaction has not been committed or aborted"; + _communicator->getLogger()->warning(s.str()); + } +} + +void +Freeze::TXNI::commit() +{ + JTCSyncT<JTCMutex> sync(*this); + + if (!_tid) + { + ostringstream s; + s << _errorPrefix << "transaction has already been committed or aborted"; + DBException ex; + ex.message = s.str(); + throw ex; + } + + if (_trace >= 1) + { + ostringstream s; + s << "committing transaction for environment \"" << _name << "\""; + _logger->trace("DB", s.str()); + } + + int ret = txn_commit(_tid, 0); + + if (ret != 0) + { + ostringstream s; + s << _errorPrefix << "txn_commit: " << db_strerror(ret); + DBException ex; + ex.message = s.str(); + throw ex; + } + + _tid = 0; +} + +void +Freeze::TXNI::abort() +{ + JTCSyncT<JTCMutex> sync(*this); + + if (!_tid) + { + ostringstream s; + s << _errorPrefix << "transaction has already been committed or aborted"; + DBException ex; + ex.message = s.str(); + throw ex; + } + + if (_trace >= 1) + { + ostringstream s; + s << "aborting transaction for environment \"" << _name << "\" due to deadlock"; + _logger->trace("DB", s.str()); + } + + int ret = txn_abort(_tid); + + if (ret != 0) + { + ostringstream s; + s << _errorPrefix << "txn_abort: " << db_strerror(ret); + DBException ex; + ex.message = s.str(); + throw ex; + } + + _tid = 0; +} + +Freeze::DBI::DBI(const CommunicatorPtr& communicator, const DBEnvIPtr& dbenvObj, ::DB* db, const string& name) : + _communicator(communicator), + _logger(communicator->getLogger()), + _trace(0), _dbenvObj(dbenvObj), - _dbenv(dbenv), _db(db), - _logger(communicator->getLogger()), - _trace(0) + _name(name) { + _errorPrefix = "Freeze::DB(\"" + _name + "\"): "; + PropertiesPtr properties = _communicator->getProperties(); string value; @@ -43,6 +335,24 @@ Freeze::DBI::DBI(const CommunicatorPtr& communicator, const string& name, const { _trace = atoi(value.c_str()); } + + if (_trace >= 1) + { + ostringstream s; + s << "opening database \"" << _name << "\" in environment \"" << _dbenvObj->getName() << "\""; + _logger->trace("DB", s.str()); + } + + int ret = _db->open(_db, name.c_str(), 0, DB_BTREE, DB_CREATE, FREEZE_DB_MODE); + + if (ret != 0) + { + ostringstream s; + s << _errorPrefix << "DB->open: " << db_strerror(ret); + DBException ex; + ex.message = s.str(); + throw ex; + } } Freeze::DBI::~DBI() @@ -50,27 +360,27 @@ Freeze::DBI::~DBI() if (_db) { ostringstream s; - s << "Freeze::DB(\"" << _name << "\"): "; - s << "\"" << _name << "\" has not been closed"; + s << _errorPrefix << "\"" << _name << "\" has not been closed"; _communicator->getLogger()->warning(s.str()); } } +string +Freeze::DBI::getName() +{ + // No mutex lock necessary, _name is immutable + return _name; +} + void -Freeze::DBI::put(const string& key, const ObjectPtr& servant) +Freeze::DBI::put(const string& key, const ObjectPtr& servant, bool txn) { - // - // TODO: Is synchronization necessary here? I really don't - // understand what the Berekely DB documentation says with "free - // threaded". - // JTCSyncT<JTCMutex> sync(*this); if (!_db) { ostringstream s; - s << "Freeze::DB(\"" << _name << "\"): "; - s << "\"" << _name << "\" has been closed"; + s << _errorPrefix << "\"" << _name << "\" has been closed"; DBException ex; ex.message = s.str(); throw ex; @@ -86,8 +396,6 @@ Freeze::DBI::put(const string& key, const ObjectPtr& servant) stream.write(servant); DBT dbKey, dbData; - DB_TXN *tid; - int ret; memset(&dbKey, 0, sizeof(dbKey)); memset(&dbData, 0, sizeof(dbData)); @@ -98,23 +406,10 @@ Freeze::DBI::put(const string& key, const ObjectPtr& servant) while (true) { - if (_trace >= 2) - { - ostringstream s; - s << "starting transaction for database \"" << _name << "\""; - _logger->trace("DB", s.str()); - } - - ret = txn_begin(_dbenv, 0, &tid, 0); - - if (ret != 0) + TXNPtr txnObj; + if (txn) { - ostringstream s; - s << "Freeze::DB(\"" << _name << "\"): "; - s << "txn_begin: " << db_strerror(ret); - DBException ex; - ex.message = s.str(); - throw ex; + txnObj = _dbenvObj->startTXN(); } if (_trace >= 1) @@ -124,70 +419,55 @@ Freeze::DBI::put(const string& key, const ObjectPtr& servant) _logger->trace("DB", s.str()); } - ret = _db->put(_db, tid, &dbKey, &dbData, 0); + int ret = _db->put(_db, 0, &dbKey, &dbData, 0); switch (ret) { case 0: { - // - // Everything ok, commit the transaction - // - if (_trace >= 2) + if (txnObj) { - ostringstream s; - s << "committing transaction for database \"" << _name << "\""; - _logger->trace("DB", s.str()); + // + // Everything ok, commit the transaction + // + txnObj->commit(); } - ret = txn_commit(tid, 0); - - if (ret != 0) - { - ostringstream s; - s << "Freeze::DB(\"" << _name << "\"): "; - s << "txn_commit: " << db_strerror(ret); - DBException ex; - ex.message = s.str(); - throw ex; - } return; // We're done } case DB_LOCK_DEADLOCK: { - // - // Deadlock, abort the transaction and retry - // - if (_trace >= 2) + if (txnObj) { - ostringstream s; - s << "aborting transaction for database \"" << _name << "\" due to deadlock"; - _logger->trace("DB", s.str()); + // + // Deadlock, abort the transaction and retry + // + txnObj->abort(); + break; // Repeat } - - ret = txn_abort(tid); - - if (ret != 0) + else { ostringstream s; - s << "Freeze::DB(\"" << _name << "\"): "; - s << "txn_abort: " << db_strerror(ret); + s << _errorPrefix << "DB->put: " << db_strerror(ret); DBException ex; ex.message = s.str(); throw ex; } - break; // Repeat } default: { - // - // Error, run recovery - // + if (txnObj) + { + // + // Error, run recovery + // + txnObj->abort(); + } + ostringstream s; - s << "Freeze::DB(\"" << _name << "\"): "; - s << "DB->put: " << db_strerror(ret); + s << _errorPrefix << "DB->put: " << db_strerror(ret); DBException ex; ex.message = s.str(); throw ex; @@ -199,34 +479,22 @@ Freeze::DBI::put(const string& key, const ObjectPtr& servant) ObjectPtr Freeze::DBI::get(const string& key) { - // - // TODO: Is synchronization necessary here? I really don't - // understand what the Berekely DB documentation says with "free - // threaded". - // JTCSyncT<JTCMutex> sync(*this); if (!_db) { ostringstream s; - s << "Freeze::DB(\"" << _name << "\"): "; - s << "\"" << _name << "\" has been closed"; + s << _errorPrefix << "\"" << _name << "\" has been closed"; DBException ex; ex.message = s.str(); throw ex; } - // - // TODO: Do I need transactions for get()? - // DBT dbKey, dbData; - int ret; - memset(&dbKey, 0, sizeof(dbKey)); memset(&dbData, 0, sizeof(dbData)); dbKey.data = const_cast<void*>(static_cast<const void*>(key.c_str())); dbKey.size = key.size(); - dbData.flags = DB_DBT_MALLOC; if (_trace >= 1) { @@ -235,7 +503,7 @@ Freeze::DBI::get(const string& key) _logger->trace("DB", s.str()); } - ret = _db->get(_db, 0, &dbKey, &dbData, 0); + int ret = _db->get(_db, 0, &dbKey, &dbData, 0); switch (ret) { @@ -244,22 +512,14 @@ Freeze::DBI::get(const string& key) // // Everything ok // + IceInternal::InstancePtr instance = IceInternal::getInstance(_communicator); + IceInternal::Stream stream(instance); + stream.b.resize(dbData.size); + stream.i = stream.b.begin(); + memcpy(stream.b.begin(), dbData.data, dbData.size); + ObjectPtr servant; - try - { - IceInternal::InstancePtr instance = IceInternal::getInstance(_communicator); - IceInternal::Stream stream(instance); - stream.b.resize(dbData.size); - stream.i = stream.b.begin(); - memcpy(stream.b.begin(), dbData.data, dbData.size); - stream.read(servant, "::Ice::Object"); - } - catch(...) - { - free(dbData.data); - throw; - } - free(dbData.data); + stream.read(servant, "::Ice::Object"); if (!servant) { @@ -280,8 +540,7 @@ Freeze::DBI::get(const string& key) default: { ostringstream s; - s << "Freeze::DB(\"" << _name << "\"): "; - s << "DB->get: " << db_strerror(ret); + s << _errorPrefix << "DB->get: " << db_strerror(ret); DBException ex; ex.message = s.str(); throw ex; @@ -292,29 +551,18 @@ Freeze::DBI::get(const string& key) void Freeze::DBI::del(const string& key) { - // - // TODO: Is synchronization necessary here? I really don't - // understand what the Berekely DB documentation says with "free - // threaded". - // JTCSyncT<JTCMutex> sync(*this); if (!_db) { ostringstream s; - s << "Freeze::DB(\"" << _name << "\"): "; - s << "\"" << _name << "\" has been closed"; + s << _errorPrefix << "\"" << _name << "\" has been closed"; DBException ex; ex.message = s.str(); throw ex; } - // - // TODO: Do I need transactions for del()? - // DBT dbKey; - int ret; - memset(&dbKey, 0, sizeof(dbKey)); dbKey.data = const_cast<void*>(static_cast<const void*>(key.c_str())); dbKey.size = key.size(); @@ -326,7 +574,7 @@ Freeze::DBI::del(const string& key) _logger->trace("DB", s.str()); } - ret = _db->del(_db, 0, &dbKey, 0); + int ret = _db->del(_db, 0, &dbKey, 0); switch (ret) { @@ -341,8 +589,7 @@ Freeze::DBI::del(const string& key) default: { ostringstream s; - s << "Freeze::DB(\"" << _name << "\"): "; - s << "DB->del: " << db_strerror(ret); + s << _errorPrefix << "DB->del: " << db_strerror(ret); DBException ex; ex.message = s.str(); throw ex; @@ -372,8 +619,7 @@ Freeze::DBI::close() if (ret != 0) { ostringstream s; - s << "Freeze::DB(\"" << _name << "\"): "; - s << "DB->close: " << db_strerror(ret); + s << _errorPrefix << "DB->close: " << db_strerror(ret); DBException ex; ex.message = s.str(); throw ex; @@ -381,7 +627,6 @@ Freeze::DBI::close() _dbenvObj->remove(_name); _dbenvObj = 0; - _dbenv = 0; _db = 0; } @@ -393,8 +638,7 @@ Freeze::DBI::createEvictor() if (!_db) { ostringstream s; - s << "Freeze::DB(\"" << _name << "\"): "; - s << "\"" << _name << "\" has been closed"; + s << _errorPrefix << "\"" << _name << "\" has been closed"; DBException ex; ex.message = s.str(); throw ex; @@ -402,182 +646,3 @@ Freeze::DBI::createEvictor() return new EvictorI(this, _communicator); } - -Freeze::DBEnvI::DBEnvI(const CommunicatorPtr& communicator, const string& directory) : - _communicator(communicator), - _directory(directory), - _dbenv(0), - _logger(communicator->getLogger()), - _trace(0) -{ - PropertiesPtr properties = _communicator->getProperties(); - string value; - - value = properties->getProperty("Freeze.Trace.DB"); - if (!value.empty()) - { - _trace = atoi(value.c_str()); - } - - int ret; - - ret = db_env_create(&_dbenv, 0); - - if (ret != 0) - { - ostringstream s; - s << "Freeze::DBEnv(\"" << _directory << "\"): "; - s << "db_env_create: " << db_strerror(ret); - DBException ex; - ex.message = s.str(); - throw ex; - } - - if (_trace >= 1) - { - ostringstream s; - s << "opening database environment \"" << _directory << "\""; - _logger->trace("DB", s.str()); - } - - ret = _dbenv->open(_dbenv, _directory.c_str(), - DB_CREATE | - DB_INIT_LOCK | - DB_INIT_LOG | - DB_INIT_MPOOL | - DB_INIT_TXN | - DB_RECOVER | - DB_THREAD, - FREEZE_DB_MODE); - if (ret != 0) - { - ostringstream s; - s << "Freeze::DBEnv(\"" << _directory << "\"): "; - s << "DB_ENV->open: " << db_strerror(ret); - DBException ex; - ex.message = s.str(); - throw ex; - } -} - -Freeze::DBEnvI::~DBEnvI() -{ - if (_dbenv) - { - ostringstream s; - s << "Freeze::DBEnv(\"" << _directory << "\"): "; - s << "\"" << _directory << "\" has not been closed"; - _communicator->getLogger()->warning(s.str()); - } -} - -DBPtr -Freeze::DBEnvI::open(const string& name) -{ - JTCSyncT<JTCRecursiveMutex> sync(*this); - - if (!_dbenv) - { - ostringstream s; - s << "Freeze::DBEnv(\"" << _directory << "\"): "; - s << "\"" << _directory << "\" has been closed"; - DBException ex; - ex.message = s.str(); - throw ex; - } - - map<string, DBPtr>::iterator p = _dbmap.find(name); - if (p != _dbmap.end()) - { - return p->second; - } - - int ret; - - ::DB* db; - ret = db_create(&db, _dbenv, 0); - - if (ret != 0) - { - ostringstream s; - s << "Freeze::DBEnv(\"" << _directory << "\"): "; - s << "db_create: " << db_strerror(ret); - DBException ex; - ex.message = s.str(); - throw ex; - } - - if (_trace >= 1) - { - ostringstream s; - s << "opening database \"" << name << "\" in environment \"" << _directory << "\""; - _logger->trace("DB", s.str()); - } - - ret = db->open(db, name.c_str(), 0, DB_BTREE, DB_CREATE | DB_THREAD, FREEZE_DB_MODE); - - if (ret != 0) - { - ostringstream s; - s << "Freeze::DBEnv(\"" << _directory << "\"): "; - s << "DB->open: " << db_strerror(ret); - DBException ex; - ex.message = s.str(); - throw ex; - } - - DBPtr dbp = new DBI(_communicator, name, this, _dbenv, db); - _dbmap[name] = dbp; - return dbp; -} - -void -Freeze::DBEnvI::close() -{ - JTCSyncT<JTCRecursiveMutex> sync(*this); - - if (!_dbenv) - { - return; - } - - while(!_dbmap.empty()) - { - _dbmap.begin()->second->close(); - } - - if (_trace >= 1) - { - ostringstream s; - s << "closing database environment \"" << _directory << "\""; - _logger->trace("DB", s.str()); - } - - int ret = _dbenv->close(_dbenv, 0); - - if (ret != 0) - { - ostringstream s; - s << "Freeze::DBEnv(\"" << _directory << "\"): "; - s << "DB_ENV->close: " << db_strerror(ret); - DBException ex; - ex.message = s.str(); - throw ex; - } - - _dbenv = 0; -} - -void -Freeze::DBEnvI::remove(const string& name) -{ - JTCSyncT<JTCRecursiveMutex> sync(*this); - - _dbmap.erase(name); -} - -DBEnvPtr -Freeze::initialize(const CommunicatorPtr& communicator, const string& directory) -{ - return new DBEnvI(communicator, directory); -} diff --git a/cpp/src/Freeze/DBI.h b/cpp/src/Freeze/DBI.h index b98aed480e8..5e03136033f 100644 --- a/cpp/src/Freeze/DBI.h +++ b/cpp/src/Freeze/DBI.h @@ -26,10 +26,11 @@ class DBI : public DB, public JTCMutex { public: - DBI(const ::Ice::CommunicatorPtr&, const std::string&, const DBEnvIPtr&, ::DB_ENV*, ::DB*); + DBI(const ::Ice::CommunicatorPtr&, const DBEnvIPtr&, ::DB*, const std::string&); virtual ~DBI(); - virtual void put(const std::string&, const ::Ice::ObjectPtr&); + virtual std::string getName(); + virtual void put(const std::string&, const ::Ice::ObjectPtr&, bool); virtual ::Ice::ObjectPtr get(const std::string&); virtual void del(const std::string&); virtual void close(); @@ -38,12 +39,14 @@ public: private: ::Ice::CommunicatorPtr _communicator; - std::string _name; - DBEnvIPtr _dbenvObj; - ::DB_ENV* _dbenv; - ::DB* _db; ::Ice::LoggerPtr _logger; int _trace; + + DBEnvIPtr _dbenvObj; + ::DB* _db; + + std::string _name; + std::string _errorPrefix; }; class DBEnvI : public DBEnv, public JTCRecursiveMutex @@ -53,19 +56,50 @@ public: DBEnvI(const ::Ice::CommunicatorPtr&, const std::string&); virtual ~DBEnvI(); - virtual DBPtr open(const std::string&); + virtual std::string getName(); + virtual DBPtr openDB(const std::string&); + virtual TXNPtr startTXN(); virtual void close(); - void remove(const std::string&); - private: + // DBI needs access to add and remove + friend class DBI; + void add(const std::string&, const DBPtr&); + void remove(const std::string&); + ::Ice::CommunicatorPtr _communicator; - std::string _directory; + ::Ice::LoggerPtr _logger; + int _trace; + ::DB_ENV* _dbenv; + + std::string _name; + std::string _errorPrefix; + std::map<std::string, DBPtr> _dbmap; +}; + +class TXNI : public TXN, public JTCMutex +{ +public: + + TXNI(const ::Ice::CommunicatorPtr&, ::DB_ENV*, const std::string&); + virtual ~TXNI(); + + virtual void commit(); + virtual void abort(); + +private: + + ::Ice::CommunicatorPtr _communicator; ::Ice::LoggerPtr _logger; int _trace; + + ::DB_TXN* _tid; + + std::string _name; + std::string _errorPrefix; }; } |