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

Author: kpvdr Date: 2007-11-05 17:51:24 -0500 (Mon, 05 Nov 2007) New Revision: 1239 Modified: store/trunk/cpp/lib/BdbMessageStore.cpp store/trunk/cpp/lib/BdbMessageStore.h store/trunk/cpp/lib/JournalImpl.cpp store/trunk/cpp/lib/JournalImpl.h store/trunk/cpp/lib/TxnCtxt.h store/trunk/cpp/lib/jrnl/jcntl.cpp store/trunk/cpp/lib/jrnl/jcntl.hpp store/trunk/cpp/lib/jrnl/jexception.cpp store/trunk/cpp/lib/jrnl/jexception.hpp store/trunk/cpp/tests/jrnl/JournalSystemTests.cpp Log: Added inactivity timer to JournalImpl which forces a flush in the event of write activity that leaves unwritten data in the journal page cache. Note that the test that specifically tests this ability is not yet available. Modified: store/trunk/cpp/lib/BdbMessageStore.cpp =================================================================== --- store/trunk/cpp/lib/BdbMessageStore.cpp 2007-11-05 22:23:04 UTC (rev 1238) +++ store/trunk/cpp/lib/BdbMessageStore.cpp 2007-11-05 22:51:24 UTC (rev 1239) @@ -44,8 +44,8 @@ static const u_int8_t MESSAGE_MESSAGE = 1; static const u_int8_t BASIC_MESSAGE = 2; bool BdbMessageStore::useAsync; -qpid::sys::Duration BdbMessageStore::defJournalTimeout(1000); // 1ms (timeout value is in nanoseconds) - +qpid::sys::Duration BdbMessageStore::defJournalGetEventsTimeout(1000000); // 1ms +qpid::sys::Duration BdbMessageStore::defJournalFlushTimeout(1000000000); // 1 sec unsigned int TxnCtxt::count = 0; @@ -190,8 +190,7 @@ journal::jdir::delete_dir(getJrnlBaseDir(),true); } catch ( journal::jexception& e) { - std::string str; - THROW_STORE_EXCEPTION("Truncate clean up failed: " +e.to_string(str) ); + THROW_STORE_EXCEPTION("Truncate clean up failed: " +e.to_string() ); } } @@ -202,14 +201,13 @@ THROW_STORE_EXCEPTION("Queue already created: " + queue.getName()); } if (usingJrnl()) { - JournalImpl* jQueue = new JournalImpl(queue.getName(), getJrnlDir(queue), string("JournalData"), defJournalTimeout); + JournalImpl* jQueue = new JournalImpl(queue.getName(), getJrnlDir(queue), string("JournalData"), defJournalGetEventsTimeout, defJournalFlushTimeout); queue.setExternalQueueStore(dynamic_cast<ExternalQueueStore*>(jQueue)); try { // init will create the deque's for the init... jQueue->initialize(); } catch (journal::jexception& e) { - std::string s; - THROW_STORE_EXCEPTION(e.to_string(s) + queue.getName()); + THROW_STORE_EXCEPTION(e.to_string() + queue.getName()); } } @@ -381,7 +379,7 @@ if (usingJrnl()) { const char* queueName = queue->getName().c_str(); - JournalImpl* jQueue = new JournalImpl(queueName, getJrnlDir(queueName), string("JournalData"), defJournalTimeout); + JournalImpl* jQueue = new JournalImpl(queueName, getJrnlDir(queueName), string("JournalData"), defJournalGetEventsTimeout, defJournalFlushTimeout); queue->setExternalQueueStore(dynamic_cast<ExternalQueueStore*>(jQueue)); try @@ -390,8 +388,7 @@ recoverMessages(txn, registry, queue, prepared, messages); jQueue->recover_complete(); // start journal. } catch (const journal::jexception& e) { - std::string s; - THROW_STORE_EXCEPTION(e.to_string(s) + queueName); + THROW_STORE_EXCEPTION(e.to_string() + queueName); } //read all messages: done on a per queue basis if using Journal } @@ -819,8 +816,7 @@ // jc->get_wr_events(); } }catch ( journal::jexception& e) { - std::string str; - THROW_STORE_EXCEPTION("Flush failed: " + e.to_string(str) ); + THROW_STORE_EXCEPTION("Flush failed: " + e.to_string() ); } } @@ -931,9 +927,8 @@ } } }catch ( journal::jexception& e) { - std::string str; // std::cout << "-------------" << e << std::endl; - THROW_STORE_EXCEPTION("Enqueue failed: " +e.to_string(str) ); + THROW_STORE_EXCEPTION("Enqueue failed: " +e.to_string() ); }catch (DbException& e) { THROW_STORE_EXCEPTION_2("Error storing message", e); } @@ -1021,8 +1016,7 @@ dres = jc->dequeue_txn_data_record(ddtokp.get(), tid); } } catch (rhm::journal::jexception& e) { - std::string str; - THROW_STORE_EXCEPTION("Error dequeuing message" + e.to_string(str)); + THROW_STORE_EXCEPTION("Error dequeuing message" + e.to_string()); } switch (dres) { Modified: store/trunk/cpp/lib/BdbMessageStore.h =================================================================== --- store/trunk/cpp/lib/BdbMessageStore.h 2007-11-05 22:23:04 UTC (rev 1238) +++ store/trunk/cpp/lib/BdbMessageStore.h 2007-11-05 22:51:24 UTC (rev 1239) @@ -77,7 +77,8 @@ std::string storeDir; bool isInit; const char* envPath; - static qpid::sys::Duration defJournalTimeout; + static qpid::sys::Duration defJournalGetEventsTimeout; + static qpid::sys::Duration defJournalFlushTimeout; bool mode(const bool mode, const bool force); void recoverQueues(TxnCtxt& txn, qpid::broker::RecoveryManager& recovery, queue_index& index, Modified: store/trunk/cpp/lib/JournalImpl.cpp =================================================================== --- store/trunk/cpp/lib/JournalImpl.cpp 2007-11-05 22:23:04 UTC (rev 1238) +++ store/trunk/cpp/lib/JournalImpl.cpp 2007-11-05 22:51:24 UTC (rev 1239) @@ -28,30 +28,38 @@ using namespace rhm::bdbstore; using namespace rhm::journal; -qpid::broker::TimerA JournalImpl::journalGetEventsTimer; +qpid::broker::TimerA JournalImpl::journalTimer; +void InactivityFireEvent::fire() { if (parent) parent->flushFire(); } +void GetEventsFireEvent::fire() { if (parent) parent->getEventsFire(); unref(); } JournalImpl::JournalImpl(const std::string& journalId, const std::string& journalDirectory, const std::string& journalBaseFilename, - const qpid::sys::Duration timeout): + const qpid::sys::Duration getEventsTimeout, + const qpid::sys::Duration flushTimeout): jcntl(journalId, journalDirectory, journalBaseFilename), - timerSetFlag(false) + getEventsTimerSetFlag(false), + writeActivityFlag(false), + flushTriggeredFlag(true) { - fireEventPtr = new FireEvent(this, timeout); - journalGetEventsTimer.start(); + getEventsFireEventsPtr = new GetEventsFireEvent(this, getEventsTimeout); + inactivityFireEventPtr = new InactivityFireEvent(this, flushTimeout); + journalTimer.start(); + journalTimer.add(inactivityFireEventPtr); } JournalImpl::~JournalImpl() { - //fireEventPtr->cancel(); + (dynamic_cast<GetEventsFireEvent*>(getEventsFireEventsPtr.get()))->cancel(); + (dynamic_cast<InactivityFireEvent*>(inactivityFireEventPtr.get()))->cancel(); } void -JournalImpl::recover(std::deque<journal::data_tok*>* rd_dtokl, const journal::aio_cb rd_cb, - std::deque<journal::data_tok*>* wr_dtokl, const journal::aio_cb wr_cb, +JournalImpl::recover(std::deque<data_tok*>* rd_dtokl, const aio_cb rd_cb, + std::deque<data_tok*>* wr_dtokl, const aio_cb wr_cb, boost::ptr_list<bdbstore::PreparedTransaction>& prep_tx_list, u_int64_t queue_id) - throw (journal::jexception) + throw (jexception) { // Create list of prepared xids std::vector<std::string> prep_xid_list; @@ -60,15 +68,15 @@ prep_xid_list.push_back(i->xid); } - journal::jcntl::recover(rd_dtokl, rd_cb, wr_dtokl, wr_cb, prep_xid_list); + jcntl::recover(rd_dtokl, rd_cb, wr_dtokl, wr_cb, prep_xid_list); // Populate PreparedTransaction lists from _tmap for (bdbstore::PreparedTransaction::list::iterator i = prep_tx_list.begin(); i != prep_tx_list.end(); i++) { try { - journal::txn_data_list tdl = _tmap.get_tdata_list(i->xid); + txn_data_list tdl = _tmap.get_tdata_list(i->xid); assert(tdl.size()); // should never be empty - for (journal::tdl_itr tdl_itr = tdl.begin(); tdl_itr < tdl.end(); tdl_itr++) { + for (tdl_itr tdl_itr = tdl.begin(); tdl_itr < tdl.end(); tdl_itr++) { if (tdl_itr->_enq_flag) { // enqueue op i->enqueues->add(queue_id, tdl_itr->_rid); } else { // dequeue op @@ -76,42 +84,121 @@ } } } - catch (const journal::jexception& e) { - if (e.err_code() != journal::jerrno::JERR_MAP_NOTFOUND) + catch (const jexception& e) { + if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw e; } } } +const iores +JournalImpl::enqueue_data_record(const void* const data_buff, const size_t tot_data_len, + const size_t this_data_len, data_tok* dtokp, const bool transient) + throw (jexception) +{ +//std::cout << " " << _jid << ":E" << std::flush; + return handleInactivityTimer(jcntl::enqueue_data_record(data_buff, tot_data_len, + this_data_len, dtokp, transient)); +} + +const iores +JournalImpl::enqueue_txn_data_record(const void* const data_buff, const size_t tot_data_len, + const size_t this_data_len, data_tok* dtokp, const std::string& xid, + const bool transient) throw (jexception) +{ +//std::cout << " " << _jid << ":Etx" << std::flush; + return handleInactivityTimer(jcntl::enqueue_txn_data_record(data_buff, tot_data_len, + this_data_len, dtokp, xid, transient)); +} + +const iores +JournalImpl::dequeue_data_record(data_tok* const dtokp) throw (jexception) +{ +//std::cout << " " << _jid << ":D" << std::flush; + return handleInactivityTimer(jcntl::dequeue_data_record(dtokp)); +} + +const iores +JournalImpl::dequeue_txn_data_record(data_tok* const dtokp, const std::string& xid) + throw (jexception) +{ +//std::cout << " " << _jid << ":Dtx" << std::flush; + return handleInactivityTimer(jcntl::dequeue_txn_data_record(dtokp, xid)); +} + +const iores +JournalImpl::txn_abort(data_tok* const dtokp, const std::string& xid) throw (jexception) +{ +//std::cout << " " << _jid << ":Atx" << std::flush; + return handleInactivityTimer(jcntl::txn_abort(dtokp, xid)); +} + +const iores +JournalImpl::txn_commit(data_tok* const dtokp, const std::string& xid) throw (jexception) +{ +//std::cout << " " << _jid << ":Ctx" << std::flush; + return handleInactivityTimer(jcntl::txn_commit(dtokp, xid)); +} + void -JournalImpl::flush() throw (journal::jexception) +JournalImpl::stop(bool block_till_aio_cmpl) throw (jexception) { + (dynamic_cast<InactivityFireEvent*>(inactivityFireEventPtr.get()))->cancel(); + jcntl::stop(block_till_aio_cmpl); +} + +void +JournalImpl::flush() throw (jexception) +{ jcntl::flush(); - handleEventTimer((const iores)0); + if (_wmgr.get_aio_evt_rem() && !getEventsTimerSetFlag) { + intrusive_ptr_add_ref(getEventsFireEventsPtr.get()); + journalTimer.add(getEventsFireEventsPtr); + getEventsTimerSetFlag = true; + } } void -JournalImpl::fire() +JournalImpl::getEventsFire() { if (_wmgr.get_aio_evt_rem()) { - jcntl::get_wr_events(); + try { jcntl::get_wr_events(); } + catch (const jexception& e) + { + // Another thread has already called get_wr_events() and it is still busy, ignore + if (e.err_code() != jerrno::JERR__PTHREAD) { + throw e; + } + } } - timerSetFlag = false; + getEventsTimerSetFlag = false; if (_wmgr.get_aio_evt_rem()) { - intrusive_ptr_add_ref(fireEventPtr.get()); - journalGetEventsTimer.add(fireEventPtr); - timerSetFlag = true; + intrusive_ptr_add_ref(getEventsFireEventsPtr.get()); + journalTimer.add(getEventsFireEventsPtr); + getEventsTimerSetFlag = true; } } -const iores -JournalImpl::handleEventTimer(const iores res) +void +JournalImpl::flushFire() { - if (_wmgr.get_aio_evt_rem() && !timerSetFlag) { - intrusive_ptr_add_ref(fireEventPtr.get()); - journalGetEventsTimer.add(fireEventPtr); - timerSetFlag = true; + if (writeActivityFlag) { + writeActivityFlag = false; + flushTriggeredFlag = false; + } else { + if (!flushTriggeredFlag) { + flush(); + flushTriggeredFlag = true; + } } - return res; + inactivityFireEventPtr->reset(); + journalTimer.add(inactivityFireEventPtr); } +const iores +JournalImpl::handleInactivityTimer(const iores r) throw (jexception) +{ + writeActivityFlag = true; + return r; +} + Modified: store/trunk/cpp/lib/JournalImpl.h =================================================================== --- store/trunk/cpp/lib/JournalImpl.h 2007-11-05 22:23:04 UTC (rev 1238) +++ store/trunk/cpp/lib/JournalImpl.h 2007-11-05 22:51:24 UTC (rev 1239) @@ -36,21 +36,50 @@ namespace rhm { namespace bdbstore { - class FireEvent; + class JournalImpl; + class InactivityFireEvent : public virtual qpid::broker::TimerTaskA + { + JournalImpl* parent; + + public: + InactivityFireEvent(JournalImpl* p, const qpid::sys::Duration timeout): + qpid::broker::TimerTaskA(timeout), parent(p) {} + virtual ~InactivityFireEvent() {} + void fire(); + inline void cancel() { parent=0; } + }; + + class GetEventsFireEvent : public virtual qpid::broker::TimerTaskA + { + JournalImpl* parent; + + public: + GetEventsFireEvent(JournalImpl* p, const qpid::sys::Duration timeout): + qpid::broker::TimerTaskA(timeout), parent(p) {} + virtual ~GetEventsFireEvent() {} + void fire(); + inline void cancel() { parent=0; } + }; + class JournalImpl : public journal::jcntl { private: - static qpid::broker::TimerA journalGetEventsTimer; - bool timerSetFlag; - qpid::broker::TimerTaskA::intrusive_ptr fireEventPtr; - + static qpid::broker::TimerA journalTimer; + bool getEventsTimerSetFlag; + qpid::broker::TimerTaskA::intrusive_ptr getEventsFireEventsPtr; + + bool writeActivityFlag; + bool flushTriggeredFlag; + qpid::broker::TimerTaskA::intrusive_ptr inactivityFireEventPtr; + public: JournalImpl(const std::string& journalId, const std::string& journalDirectory, const std::string& journalBaseFilename, - const qpid::sys::Duration timeout); + const qpid::sys::Duration getEventsTimeout, + const qpid::sys::Duration flushTimeout); ~JournalImpl(); void recover(std::deque<journal::data_tok*>* rd_dtokl, const journal::aio_cb rd_cb, @@ -65,30 +94,43 @@ &aio_wr_callback, prep_tx_list, queue_id); } + // Overrides for write inactivity timer + const journal::iores enqueue_data_record(const void* const data_buff, + const size_t tot_data_len, const size_t this_data_len, journal::data_tok* dtokp, + const bool transient = false) throw (journal::jexception); + + const journal::iores enqueue_txn_data_record(const void* const data_buff, + const size_t tot_data_len, const size_t this_data_len, journal::data_tok* dtokp, + const std::string& xid, const bool transient = false) + throw (journal::jexception); + + const journal::iores dequeue_data_record(journal::data_tok* const dtokp) + throw (journal::jexception); + + const journal::iores dequeue_txn_data_record(journal::data_tok* const dtokp, + const std::string& xid) throw (journal::jexception); + + const journal::iores txn_abort(journal::data_tok* const dtokp, const std::string& xid) + throw (journal::jexception); + + const journal::iores txn_commit(journal::data_tok* const dtokp, const std::string& xid) + throw (journal::jexception); + + void stop(bool block_till_aio_cmpl = false) throw (journal::jexception); + + // Overrides for get_events timer void flush() throw (journal::jexception); // TimerTask callback - void fire(); - + void getEventsFire(); + void flushFire(); + private: - const journal::iores handleEventTimer(const journal::iores res); - }; - - class FireEvent : public virtual qpid::broker::TimerTaskA - { - JournalImpl* parent; + const journal::iores handleInactivityTimer(const journal::iores r) + throw (journal::jexception); + }; // class JournalImpl - public: - FireEvent(JournalImpl* p, const qpid::sys::Duration timeout): - qpid::broker::TimerTaskA(timeout), parent(p) {} - virtual ~FireEvent() {} - inline void fire() { if (parent) parent->fire(); unref(); } - inline void cancel() { cancelled=true; parent=0; } - }; + } // namespace bdbstore +} // namespace rhm - - - } // namespace bdbstore - } // namespace rhm - #endif Modified: store/trunk/cpp/lib/TxnCtxt.h =================================================================== --- store/trunk/cpp/lib/TxnCtxt.h 2007-11-05 22:23:04 UTC (rev 1238) +++ store/trunk/cpp/lib/TxnCtxt.h 2007-11-05 22:51:24 UTC (rev 1239) @@ -76,9 +76,8 @@ jc->txn_abort(dtokp.get(), getXid()); } } catch (rhm::journal::jexception& e) { - std::string str; //std::cout << "Error commit" << e << std::endl; - THROW_STORE_EXCEPTION("Error commit" + e.to_string(str)); + THROW_STORE_EXCEPTION("Error commit" + e.to_string()); } } @@ -117,9 +116,8 @@ jc->get_wr_events(); } }catch (rhm::journal::jexception& e) { - std::string str; //std::cout << " TxnCtxt: Error sync 3" << e << std::flush; - THROW_STORE_EXCEPTION("Error sync" + e.to_string(str)); + THROW_STORE_EXCEPTION("Error sync" + e.to_string()); } } firstloop = false; Modified: store/trunk/cpp/lib/jrnl/jcntl.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/jcntl.cpp 2007-11-05 22:23:04 UTC (rev 1238) +++ store/trunk/cpp/lib/jrnl/jcntl.cpp 2007-11-05 22:51:24 UTC (rev 1239) @@ -72,7 +72,7 @@ { if (_init_flag && !_stop_flag) try { stop(true); } - catch (jexception e) { std::cerr << e << std::endl; } + catch (const jexception& e) { std::cerr << e << std::endl; } if (_datafh) { for (u_int32_t i=0; i<JRNL_NUM_FILES; i++) @@ -304,6 +304,16 @@ } } +void +jcntl::flush() throw (jexception) +{ + if (!_init_flag) + return; + if (_readonly_flag) + throw jexception(jerrno::JERR_JCNTL_READONLY, "jcntl", "flush"); + _wmgr.flush(); +} + // Private functions void @@ -372,7 +382,7 @@ rd._ffid = ji.get_start_file(); rd._empty = false; } - catch (jexception& e) + catch (const jexception& e) { if (e.err_code() != jerrno::JERR_JINF_JDATEMPTY) throw e; @@ -441,7 +451,11 @@ { // If the enqueue is part of a pending txn, it will not yet be in emap try { _emap.lock(dr.deq_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 e; + } dr.get_xid(&xidp); assert(xidp != NULL); std::string xid((char*)xidp, dr.xid_size()); @@ -455,7 +469,7 @@ _emap.get_remove_fid(dr.deq_rid()); rd._enq_cnt_list[fid]--; } - catch (jexception& e) + catch (const jexception& e) { if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw e; @@ -485,7 +499,7 @@ if (!itr->_enq_flag) _emap.unlock(itr->_drid); } - catch(jexception e) + catch(const jexception& e) { if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw e; Modified: store/trunk/cpp/lib/jrnl/jcntl.hpp =================================================================== --- store/trunk/cpp/lib/jrnl/jcntl.hpp 2007-11-05 22:23:04 UTC (rev 1238) +++ store/trunk/cpp/lib/jrnl/jcntl.hpp 2007-11-05 22:51:24 UTC (rev 1239) @@ -531,7 +531,7 @@ /** * \brief Force a flush of the write page cache, creating a single AIO write operation. */ - inline void flush() throw (jexception) { _wmgr.flush(); } + void flush() throw (jexception); inline const u_int32_t get_enq_cnt() const { return _emap.size(); } inline const u_int32_t get_wr_outstanding_aio_dblks() const Modified: store/trunk/cpp/lib/jrnl/jexception.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/jexception.cpp 2007-11-05 22:23:04 UTC (rev 1238) +++ store/trunk/cpp/lib/jrnl/jexception.cpp 2007-11-05 22:51:24 UTC (rev 1239) @@ -83,8 +83,8 @@ jexception::~jexception() throw () {} -const std::string& -jexception::to_string(std::string& str) const +const std::string +jexception::to_string() const { std::stringstream ss; if (!_throwing_class.empty() && !_throwing_fn.empty()) @@ -96,28 +96,27 @@ ss << " " << err_msg; if (!_additional_info.empty()) ss << " (" << _additional_info << ")"; - return str.assign(ss.str()); + return ss.str(); } const char* jexception::what() const throw () { - return _additional_info.c_str(); +// return _additional_info.c_str(); + return to_string().c_str(); } std::ostream& operator<<(std::ostream& os, const jexception& je) { - std::string str; - os << je.to_string(str); + os << je.to_string(); return os; } std::ostream& operator<<(std::ostream& os, const jexception* jePtr) { - std::string str; - os << jePtr->to_string(str); + os << jePtr->to_string(); return os; } Modified: store/trunk/cpp/lib/jrnl/jexception.hpp =================================================================== --- store/trunk/cpp/lib/jrnl/jexception.hpp 2007-11-05 22:23:04 UTC (rev 1238) +++ store/trunk/cpp/lib/jrnl/jexception.hpp 2007-11-05 22:51:24 UTC (rev 1239) @@ -72,7 +72,7 @@ jexception(const u_int32_t err_code, const std::string& additional_info, const std::string& throwing_class, const std::string& throwing_fn); virtual ~jexception() throw (); - const std::string& to_string(std::string& str) const; + const std::string to_string() const; virtual const char* what() const throw (); // override std::exception::what() inline const u_int32_t err_code() const { return _err_code; } Modified: store/trunk/cpp/tests/jrnl/JournalSystemTests.cpp =================================================================== --- store/trunk/cpp/tests/jrnl/JournalSystemTests.cpp 2007-11-05 22:23:04 UTC (rev 1238) +++ store/trunk/cpp/tests/jrnl/JournalSystemTests.cpp 2007-11-05 22:51:24 UTC (rev 1239) @@ -1410,8 +1410,7 @@ catch (rhm::journal::jexception e) { t.finalize(); - std::string s; - CPPUNIT_FAIL(e.to_string(s)); + CPPUNIT_FAIL(e.to_string()); } t.finalize(); }