[rhmessaging-commits] rhmessaging commits: r1402 - in store/trunk/cpp: lib/jrnl and 1 other directories.

Author: kpvdr Date: 2007-11-29 17:16:50 -0500 (Thu, 29 Nov 2007) New Revision: 1402 Modified: store/trunk/cpp/lib/BdbMessageStore.cpp store/trunk/cpp/lib/JournalImpl.cpp store/trunk/cpp/lib/StoreException.h store/trunk/cpp/lib/TxnCtxt.h store/trunk/cpp/lib/jrnl/data_tok.hpp store/trunk/cpp/lib/jrnl/jcntl.cpp store/trunk/cpp/lib/jrnl/rmgr.cpp store/trunk/cpp/lib/jrnl/wmgr.cpp store/trunk/cpp/tests/jrnl/unit_test_jdir.cpp store/trunk/cpp/tests/jrnl/unit_test_jinf.cpp Log: BZ403201: Solves problem of messages being prematurely deleted when an exception is thrown. This was achieved by extending the boost::intrusive_ptr<PersistableMessage> into data_token. Thus as long as the data_token exists, the message will not be deleted. Also solved the St9exception message mystery, and tidied up exception handling - part of the cause of this mystery. Modified: store/trunk/cpp/lib/BdbMessageStore.cpp =================================================================== --- store/trunk/cpp/lib/BdbMessageStore.cpp 2007-11-29 21:46:43 UTC (rev 1401) +++ store/trunk/cpp/lib/BdbMessageStore.cpp 2007-11-29 22:16:50 UTC (rev 1402) @@ -93,12 +93,12 @@ open(dequeueXidDb, txn.get(), "dequeue_xid.db", true); open(prepareXidDb, txn.get(), "prepare_xid.db", false); txn.commit(); - } catch (DbException& e) { + } catch (const DbException& e) { txn.abort(); THROW_STORE_EXCEPTION_2("Error opening databases", e); - } catch (std::exception& e) { + } catch (const std::exception& e) { txn.abort(); - throw e; + throw; } ret = mode(async, force); if (!ret) return false; @@ -169,10 +169,10 @@ for (std::list<Db*>::iterator i = dbs.begin(); i != dbs.end(); i++) { (*i)->close(0); } - } catch (DbException& e) { + } catch (const DbException& e) { THROW_STORE_EXCEPTION_2("Error closing databases", e); - } catch (std::exception& e) { - throw e; + } catch (const std::exception& e) { + throw; } } @@ -190,7 +190,7 @@ try{ journal::jdir::delete_dir(getJrnlBaseDir(),true); } - catch ( journal::jexception& e) { + catch (const journal::jexception& e) { THROW_STORE_EXCEPTION(std::string("truncate() failed: ") + e.what() ); } } @@ -207,7 +207,7 @@ try { // init will create the deque's for the init... jQueue->initialize(); - } catch (journal::jexception& e) { + } catch (const journal::jexception& e) { THROW_STORE_EXCEPTION(std::string("Queue ") + queue.getName() + ": create() failed: " + e.what()); } } @@ -216,7 +216,7 @@ if (!create(queueDb, queueIdSequence, queue)) { THROW_STORE_EXCEPTION("Queue already exists: " + queue.getName()); } - } catch (DbException& e) { + } catch (const DbException& e) { THROW_STORE_EXCEPTION_2("Error creating queue named " + queue.getName(), e); } } @@ -246,7 +246,7 @@ if (!create(exchangeDb, exchangeIdSequence, exchange)) { THROW_STORE_EXCEPTION("Exchange already exists: " + exchange.getName()); } - } catch (DbException& e) { + } catch (const DbException& e) { THROW_STORE_EXCEPTION_2("Error creating exchange named " + exchange.getName(), e); } } @@ -334,7 +334,7 @@ txn.commit(); - } catch (DbException& e) { + } catch (const DbException& e) { txn.abort(); THROW_STORE_EXCEPTION_2("Error on recovery", e); } @@ -546,7 +546,7 @@ assert( "Store Error: Unexpected msg state"); } // switch } // while - } catch (rhm::journal::jexception& e) { + } catch (const journal::jexception& e) { THROW_STORE_EXCEPTION(std::string("Queue ") + queue->getName() + ": recoverMessages() failed: " + e.what()); } @@ -737,9 +737,9 @@ store(NULL, &txn, key, msg, true); msg->setPersistenceId(messageId); txn.commit(); - } catch (std::exception& e) { + } catch (const std::exception& e) { txn.abort(); - throw e; + throw; } } } @@ -754,7 +754,7 @@ try { deleteIfUnused(txn.get(), key); txn.commit(); - } catch (DbException& e) { + } catch (const DbException& e) { txn.abort(); THROW_STORE_EXCEPTION_2("Error destroying message", e); } @@ -771,7 +771,7 @@ if (status != DB_BUFFER_SMALL) { THROW_STORE_EXCEPTION("Unexpected status code when determining record length: " + string(DbEnv::strerror(status))); } - } catch (DbMemoryException& expected) { + } catch (const DbMemoryException& expected) { //api doc indicates may throw exception instead of status = DB_BUFFER_SMALL; } return peek.get_size(); @@ -798,7 +798,7 @@ value.set_doff(offset); value.set_dlen(size); messageDb.put(0, &key, &value, DB_AUTO_COMMIT); - } catch (DbException& e) { + } catch (const DbException& e) { THROW_STORE_EXCEPTION_2("Error appending content", e); } } else { @@ -836,9 +836,9 @@ data.assign(buffer, value.get_size()); delete [] buffer; } - } catch (DbException& e) { + } catch (const DbException& e) { THROW_STORE_EXCEPTION_2("Error loading content", e); - } catch (journal::jexception& e) { + } catch (const journal::jexception& e) { THROW_STORE_EXCEPTION(std::string("Queue ") + queue.getName() + ": loadContent() failed: " + e.what()); } @@ -856,7 +856,7 @@ if (jc){ jc->flush(); } - }catch ( journal::jexception& e) { + }catch (const journal::jexception& e) { THROW_STORE_EXCEPTION(std::string("Queue ") + queue.getName() + ": flush() failed: " + e.what() ); } } @@ -908,9 +908,9 @@ } if (!ctxt) txn->commit(); - } catch (std::exception& e) { + } catch (const std::exception& e) { if (!ctxt) txn->abort(); - throw e; + throw; } } @@ -936,7 +936,7 @@ //std::cout << "E" << std::flush; boost::intrusive_ptr<DataTokenImpl> dtokp(new DataTokenImpl); dtokp->ref(); - dtokp->setSourceMessage (message.get()); + dtokp->setSourceMessage(message); dtokp->set_rid(message->getPersistenceId()); // set the messageID into the Journal header (record-id) bool written = false; @@ -998,10 +998,10 @@ messageDb.put(txn->get(), &messageId, &data, DB_NOOVERWRITE); } } - }catch ( journal::jexception& e) { + } catch (const journal::jexception& e) { THROW_STORE_EXCEPTION(std::string("Queue ") + queue->getName() + ": store() failed: " + e.what()); - }catch (DbException& e) { + } catch (const DbException& e) { THROW_STORE_EXCEPTION_2("Error storing message", e); } } @@ -1060,12 +1060,12 @@ } if (!ctxt) txn->commit(); - } catch (DbException& e) { + } catch (const DbException& e) { if (!ctxt) txn->abort(); THROW_STORE_EXCEPTION_2("Error dequeing message", e); - } catch (std::exception& e) { + } catch (const std::exception& e) { if (!ctxt) txn->abort(); - throw e; + throw; } } @@ -1076,7 +1076,7 @@ bool written = false; boost::intrusive_ptr<DataTokenImpl> ddtokp(new DataTokenImpl); ddtokp->ref(); - ddtokp->setSourceMessage (msg.get()); + ddtokp->setSourceMessage(msg); ddtokp->set_rid(messageIdSequence.next()); ddtokp->set_dequeue_rid(msg->getPersistenceId()); ddtokp->set_wstate(DataTokenImpl::ENQ); @@ -1098,7 +1098,7 @@ } else { dres = jc->dequeue_txn_data_record(ddtokp.get(), tid); } - } catch (rhm::journal::jexception& e) { + } catch (const journal::jexception& e) { THROW_STORE_EXCEPTION(std::string("Queue ") + queue.getName() + ": async_dequeue() failed: " + e.what()); } switch (dres) @@ -1234,10 +1234,10 @@ prepareXidDb.del(txn.get(), &key, 0); txn.complete(commit); - } catch (std::exception& e) { + } catch (const std::exception& e) { std::cout << "Error completing xid " << txn.getXid() << ": " << e.what() << std::endl; txn.abort(); - throw e; + throw; } } @@ -1279,9 +1279,9 @@ prepareXidDb.put(txn->get(), &key, &value, 0); txn->commit(); - } catch (std::exception& e) { + } catch (const std::exception& e) { txn->abort(); - throw e; + throw; } } @@ -1333,7 +1333,7 @@ } else if (status) { THROW_STORE_EXCEPTION(DbEnv::strerror(status)); } - } catch (DbException& e) { + } catch (const DbException& e) { THROW_STORE_EXCEPTION(e.what()); } } Modified: store/trunk/cpp/lib/JournalImpl.cpp =================================================================== --- store/trunk/cpp/lib/JournalImpl.cpp 2007-11-29 21:46:43 UTC (rev 1401) +++ store/trunk/cpp/lib/JournalImpl.cpp 2007-11-29 22:16:50 UTC (rev 1402) @@ -99,7 +99,7 @@ } catch (const jexception& e) { if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) - throw e; + throw; } } } @@ -257,7 +257,7 @@ { // Another thread has already called get_wr_events() and it is still busy, ignore if (e.err_code() != jerrno::JERR__PTHREAD) { - throw e; + throw; } } } Modified: store/trunk/cpp/lib/StoreException.h =================================================================== --- store/trunk/cpp/lib/StoreException.h 2007-11-29 21:46:43 UTC (rev 1401) +++ store/trunk/cpp/lib/StoreException.h 2007-11-29 22:16:50 UTC (rev 1402) @@ -32,7 +32,7 @@ std::string text; public: StoreException(const std::string& _text) : text(_text) {} - StoreException(const std::string& _text, DbException& cause) : text(_text + ": " + cause.what()) {} + StoreException(const std::string& _text, const DbException& cause) : text(_text + ": " + cause.what()) {} virtual ~StoreException() throw() {} virtual const char* what() const throw() { return text.c_str(); } }; @@ -41,7 +41,7 @@ { public: StoreFullException(const std::string& _text) : StoreException(_text) {} - StoreFullException(const std::string& _text, DbException& cause) : StoreException(_text, cause) {} + StoreFullException(const std::string& _text, const DbException& cause) : StoreException(_text, cause) {} virtual ~StoreFullException() throw() {} }; Modified: store/trunk/cpp/lib/TxnCtxt.h =================================================================== --- store/trunk/cpp/lib/TxnCtxt.h 2007-11-29 21:46:43 UTC (rev 1401) +++ store/trunk/cpp/lib/TxnCtxt.h 2007-11-29 22:16:50 UTC (rev 1402) @@ -75,7 +75,7 @@ { jc->txn_abort(dtokp.get(), getXid()); } - } catch (rhm::journal::jexception& e) { + } catch (const journal::jexception& e) { //std::cout << "Error commit" << e << std::endl; THROW_STORE_EXCEPTION(std::string("Error commit") + e.what()); } @@ -115,7 +115,7 @@ allWritten = false; jc->get_wr_events(); } - }catch (rhm::journal::jexception& e) { + } catch (const journal::jexception& e) { //std::cout << " TxnCtxt: Error sync 3" << e << std::flush; THROW_STORE_EXCEPTION(std::string("Error sync") + e.what()); } Modified: store/trunk/cpp/lib/jrnl/data_tok.hpp =================================================================== --- store/trunk/cpp/lib/jrnl/data_tok.hpp 2007-11-29 21:46:43 UTC (rev 1401) +++ store/trunk/cpp/lib/jrnl/data_tok.hpp 2007-11-29 22:16:50 UTC (rev 1402) @@ -49,7 +49,9 @@ } } +#include <boost/intrusive_ptr.hpp> #include <pthread.h> +#include <qpid/broker/PersistableMessage.h> #include <sys/types.h> #include <jrnl/jexception.hpp> @@ -113,7 +115,7 @@ u_int64_t _rid; ///< RID of data set by enqueue operation std::string _xid; ///< XID set by enqueue operation u_int64_t _dequeue_rid; ///< RID of data set by dequeue operation - qpid::broker::PersistableMessage* _sourceMsg; ///< Pointer back to source Message in Broker + boost::intrusive_ptr<qpid::broker::PersistableMessage> _sourceMsg; ///< Pointer back to source Message in Broker public: data_tok(); @@ -122,8 +124,10 @@ inline size_t refcnt(void) { return _ref_cnt;} inline void ref(void) { _ref_cnt++; } inline void unref(void) { _ref_cnt--; } - inline qpid::broker::PersistableMessage* getSourceMessage(){return _sourceMsg;} - inline void setSourceMessage(qpid::broker::PersistableMessage* msg) {_sourceMsg = msg;} + inline boost::intrusive_ptr<qpid::broker::PersistableMessage>& getSourceMessage() + { return _sourceMsg; } + inline void setSourceMessage(boost::intrusive_ptr<qpid::broker::PersistableMessage>& msg) + { _sourceMsg = msg; } inline const u_int64_t id() const { return _icnt; } inline const write_state wstate() const { return _wstate; } Modified: store/trunk/cpp/lib/jrnl/jcntl.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/jcntl.cpp 2007-11-29 21:46:43 UTC (rev 1401) +++ store/trunk/cpp/lib/jrnl/jcntl.cpp 2007-11-29 22:16:50 UTC (rev 1402) @@ -209,7 +209,7 @@ res = _wmgr.enqueue(data_buff, tot_data_len, this_data_len, dtokp, NULL, 0, transient, false); } - catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw e; } + catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw; } pthread_mutex_unlock(&_mutex); return res; } @@ -225,7 +225,7 @@ { res = _wmgr.enqueue(NULL, tot_data_len, 0, dtokp, NULL, 0, transient, true); } - catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw e; } + catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw; } pthread_mutex_unlock(&_mutex); return res; } @@ -243,7 +243,7 @@ res = _wmgr.enqueue(data_buff, tot_data_len, this_data_len, dtokp, xid.data(), xid.size(), transient, false); } - catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw e; } + catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw; } pthread_mutex_unlock(&_mutex); return res; } @@ -259,7 +259,7 @@ { res = _wmgr.enqueue(NULL, tot_data_len, 0, dtokp, xid.data(), xid.size(), transient, true); } - catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw e; } + catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw; } pthread_mutex_unlock(&_mutex); return res; } @@ -294,7 +294,7 @@ check_wstatus("dequeue_data"); pthread_mutex_lock(&_mutex); try { res = _wmgr.dequeue(dtokp, NULL, 0); } - catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw e; } + catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw; } pthread_mutex_unlock(&_mutex); return res; } @@ -306,7 +306,7 @@ check_wstatus("dequeue_data"); pthread_mutex_lock(&_mutex); try { res = _wmgr.dequeue(dtokp, xid.data(), xid.size()); } - catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw e; } + catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw; } pthread_mutex_unlock(&_mutex); return res; } @@ -318,7 +318,7 @@ check_wstatus("txn_abort"); pthread_mutex_lock(&_mutex); try { res = _wmgr.abort(dtokp, xid.data(), xid.size()); } - catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw e; } + catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw; } pthread_mutex_unlock(&_mutex); return res; } @@ -330,7 +330,7 @@ check_wstatus("txn_commit"); pthread_mutex_lock(&_mutex); try { res = _wmgr.commit(dtokp, xid.data(), xid.size()); } - catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw e; } + catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw; } pthread_mutex_unlock(&_mutex); return res; } @@ -357,7 +357,7 @@ return 0; // already locked, return immediately } try { res = _wmgr.get_events(pmgr::UNUSED); } - catch (const std::exception& e) { pthread_mutex_unlock(&_mutex); throw e; } + catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw; } pthread_mutex_unlock(&_mutex); return res; } @@ -393,7 +393,7 @@ throw jexception(jerrno::JERR_JCNTL_READONLY, "jcntl", "flush"); pthread_mutex_lock(&_mutex); try { _wmgr.flush(); } - catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw e; } + catch(const std::exception& e) { pthread_mutex_unlock(&_mutex); throw; } pthread_mutex_unlock(&_mutex); } @@ -461,8 +461,7 @@ } catch (const jexception& e) { - if (e.err_code() != jerrno::JERR_JINF_JDATEMPTY) - throw e; + if (e.err_code() != jerrno::JERR_JINF_JDATEMPTY) throw; } // Restore all read and write pointers and transactions @@ -550,8 +549,7 @@ try { _emap.lock(dr.deq_rid()); } catch(const jexception& e) { - if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) - throw e; + if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw; } dr.get_xid(&xidp); assert(xidp != NULL); @@ -567,10 +565,9 @@ //std::cout << enq_fid; rd._enq_cnt_list[enq_fid]--; } - catch (const jexception& e) + catch(const jexception& e) { - if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) - throw e; + if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw; } } } @@ -601,8 +598,7 @@ } catch(const jexception& e) { - if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) - throw e; + if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw; } if (itr->_enq_flag) rd._enq_cnt_list[itr->_fid]--; Modified: store/trunk/cpp/lib/jrnl/rmgr.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/rmgr.cpp 2007-11-29 21:46:43 UTC (rev 1401) +++ store/trunk/cpp/lib/jrnl/rmgr.cpp 2007-11-29 22:16:50 UTC (rev 1402) @@ -290,11 +290,11 @@ //std::cout << ":ok" << std::flush; is_enq = true; } - catch (jexception& e) + catch (const jexception& e) { // Block read for transactionally locked record (only when not recovering) if (e.err_code() == jerrno::JERR_MAP_LOCKED && !_jc->is_read_only()) - throw e; + throw; //if (e.err_code() == jerrno::JERR_MAP_LOCKED) std::cout << ":locked" << std::flush; // (Recover mode only) Ok, not in emap - now search tmap, if present then read @@ -333,10 +333,9 @@ if (_emap.is_locked(_hdr._rid) && !_jc->is_read_only()) return RHM_IORES_TXPENDING; } - catch (jexception e) + catch (const jexception& e) { - if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) - throw e; + if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw; } #endif if (dtokp->rid()) Modified: store/trunk/cpp/lib/jrnl/wmgr.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/wmgr.cpp 2007-11-29 21:46:43 UTC (rev 1401) +++ store/trunk/cpp/lib/jrnl/wmgr.cpp 2007-11-29 22:16:50 UTC (rev 1402) @@ -311,7 +311,10 @@ { // If the enqueue is part of a pending txn, it will not yet be in emap try { _emap.lock(dequeue_rid); } - catch(jexception e) { if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw e; } + catch(const jexception& e) + { + if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw; + } std::string xid((char*)xid_ptr, xid_len); _tmap.insert_txn_data(xid, txn_data(rid, dequeue_rid, dtokp->fid(), false)); } @@ -452,7 +455,10 @@ if (!itr->_enq_flag) _emap.unlock(itr->_drid); } - catch(jexception e) { if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw e; } + catch(const jexception& e) + { + if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw; + } if (itr->_enq_flag) _wrfc.decr_enqcnt(itr->_fid); } @@ -866,10 +872,9 @@ bool is_synced = true; // Check for outstanding enqueues/dequeues try { is_synced = _tmap.is_txn_synced(xid); } - catch (jexception e) + catch (const jexception& e) { - if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) - throw e; + if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw; } if (!is_synced) return false; Modified: store/trunk/cpp/tests/jrnl/unit_test_jdir.cpp =================================================================== --- store/trunk/cpp/tests/jrnl/unit_test_jdir.cpp 2007-11-29 21:46:43 UTC (rev 1401) +++ store/trunk/cpp/tests/jrnl/unit_test_jdir.cpp 2007-11-29 22:16:50 UTC (rev 1402) @@ -126,7 +126,7 @@ jdir::is_dir("/tmp/D"); BOOST_ERROR("jdir::is_dir() failed to throw jexeption for non-existent directory."); } - catch(const jexception e) + catch(const jexception& e) { BOOST_CHECK_EQUAL(e.err_code(), jerrno::JERR_JDIR_STAT); } @@ -243,7 +243,7 @@ test_dir_3.clear_dir(false); BOOST_ERROR("jdir::clear_dir(flase) failed to throw jexeption for non-existent directory."); } - catch(const jexception e) + catch(const jexception& e) { BOOST_CHECK_EQUAL(e.err_code(), jerrno::JERR_JDIR_OPENDIR); } Modified: store/trunk/cpp/tests/jrnl/unit_test_jinf.cpp =================================================================== --- store/trunk/cpp/tests/jrnl/unit_test_jinf.cpp 2007-11-29 21:46:43 UTC (rev 1401) +++ store/trunk/cpp/tests/jrnl/unit_test_jinf.cpp 2007-11-29 22:16:50 UTC (rev 1402) @@ -126,7 +126,7 @@ fn << jdir << "/" << base_filename << "." << JRNL_INFO_EXTENSION; jinf ji(fn.str(), false); try { ji.analyze(); } - catch (jexception e) + catch (const jexception& e) { if (e.err_code() != jerrno::JERR_JINF_JDATEMPTY) BOOST_ERROR("Failed to throw expected exception jerrno::JERR_JINF_JDATEMPTY");