rhmessaging commits: r2307 - mgmt/trunk/cumin/python/cumin.
by rhmessaging-commits＠lists.jboss.org
Author: eallen
Date: 2008-08-14 14:44:43 -0400 (Thu, 14 Aug 2008)
New Revision: 2307
Modified:
mgmt/trunk/cumin/python/cumin/widgets.py
Log:
Making FilteredCheckboxIdColumn more extensible by using a callback instead of field name and values
Modified: mgmt/trunk/cumin/python/cumin/widgets.py
===================================================================
--- mgmt/trunk/cumin/python/cumin/widgets.py 2008-08-14 18:36:13 UTC (rev 2306)
+++ mgmt/trunk/cumin/python/cumin/widgets.py 2008-08-14 18:44:43 UTC (rev 2307)
@@ -577,16 +577,16 @@
return self.column.ids.path
class FilteredCheckboxIdColumn(CheckboxIdColumn):
- def __init__(self, app, name, form, col_name, skip_list):
+ def __init__(self, app, name, form, callback=None):
super(FilteredCheckboxIdColumn, self).__init__(app, name, form)
- # list of names to skip putting a checkbox by
- self.__skip_list = skip_list
- self.__col_name = col_name
+ # call back that returns True if the checkbox is to be disabled
+ self.__callback = callback
def do_render(self, session, data):
+ disabled = self.__callback and self.__callback(session, data) or False
return super(FilteredCheckboxIdColumn, self).do_render(session, data,
- disabled=data[self.__col_name] in self.__skip_list)
+ disabled=disabled)
class NameField(StringField):
def __init__(self, app, name, form):
17 years, 8 months
- 1
- 0
rhmessaging commits: r2306 - in store/branches/mrg-1.0/cpp/lib: jrnl and 1 other directory.
by rhmessaging-commits＠lists.jboss.org
Author: kpvdr
Date: 2008-08-14 14:36:13 -0400 (Thu, 14 Aug 2008)
New Revision: 2306
Modified:
store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp
store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h
store/branches/mrg-1.0/cpp/lib/JournalImpl.h
store/branches/mrg-1.0/cpp/lib/jrnl/jcntl.cpp
store/branches/mrg-1.0/cpp/lib/jrnl/jcntl.hpp
store/branches/mrg-1.0/cpp/lib/jrnl/rmgr.cpp
store/branches/mrg-1.0/cpp/lib/jrnl/rmgr.hpp
store/branches/mrg-1.0/cpp/lib/jrnl/wmgr.cpp
Log:
Backport of trunk checkin r.2305: Fix for BZ458656 "List of xids from dtx.recover is not as expected". The call to BdbMessageStore::collectPreparedXids() was using the prepared list obtained at recover, but this call needs to provide a current list, no matter when called. Added ability to ignore transactions when reading, and created a subclass of JournalImpl (called TplJournalImpl) which implements the ignore as standard practice when calling read. This enables reads of pending transactions from this list. Now BdbMessageStore::collectPreparedXids() re-reads the TPL before returning the XID list.
Modified: store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp
===================================================================
--- store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp 2008-08-14 17:49:28 UTC (rev 2305)
+++ store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp 2008-08-14 18:36:13 UTC (rev 2306)
@@ -243,7 +243,7 @@
open(mappingDb, txn.get(), "mappings.db", true);
open(bindingDb, txn.get(), "bindings.db", true);
open(generalDb, txn.get(), "general.db", false);
- tplStorePtr.reset(new JournalImpl("TplStore", getTplBaseDir(), "tpl", defJournalGetEventsTimeout, defJournalFlushTimeout));
+ tplStorePtr.reset(new TplJournalImpl("TplStore", getTplBaseDir(), "tpl", defJournalGetEventsTimeout, defJournalFlushTimeout));
txn.commit();
} catch (const journal::jexception& e) {
txn.abort();
@@ -910,77 +910,84 @@
return count;
}
-void BdbMessageStore::recoverTplStore(TplRecoverMap& tplMap)
+void BdbMessageStore::readTplStore()
{
+ tplRecoverMap.clear();
+ journal::txn_map& tmap = tplStorePtr->get_txn_map();
+ DataTokenImpl dtokp;
+ void* dbuff = NULL; size_t dbuffSize = 0;
+ void* xidbuff = NULL; size_t xidbuffSize = 0;
+ bool transientFlag = false;
+ bool externalFlag = false;
+ bool done = false;
+ try {
+ unsigned aio_sleep_cnt = 0;
+ while (!done) {
+ dtokp.reset();
+ dtokp.set_wstate(DataTokenImpl::ENQ);
+ switch (tplStorePtr->read_data_record(&dbuff, dbuffSize, &xidbuff, xidbuffSize, transientFlag, externalFlag, &dtokp)) {
+ case rhm::journal::RHM_IORES_SUCCESS: {
+ // Every TPL record contains both data and an XID
+ assert(dbuffSize>0);
+ assert(xidbuffSize>0);
+ std::string xid(static_cast<const char*>(xidbuff), xidbuffSize);
+ bool is2PC = *(static_cast<char*>(dbuff)) != 0;
+
+ // Check transaction details; add to recover map
+ journal::txn_data_list txnList = tmap.get_tdata_list(xid);
+ unsigned enqCnt = 0;
+ unsigned deqCnt = 0;
+ u_int64_t rid = 0;
+
+ // Assume commit (roll forward) in cases where only prepare has been called - ie only enqueue record exists.
+ // Note: will apply to both 1PC and 2PC transactions.
+ bool commitFlag = true;
+
+ for (journal::tdl_itr j = txnList.begin(); j<txnList.end(); j++) {
+ if (j->_enq_flag) {
+ rid = j->_rid;
+ enqCnt++;
+ } else {
+ commitFlag = j->_commit_flag;
+ deqCnt++;
+ }
+ }
+ assert(enqCnt == 1);
+ assert(deqCnt <= 1);
+ tplRecoverMap.insert(TplRecoverMapPair(xid, TplRecoverStruct(rid, deqCnt == 1, commitFlag, is2PC)));
+
+ ::free(xidbuff);
+ aio_sleep_cnt = 0;
+ break;
+ }
+ case rhm::journal::RHM_IORES_PAGE_AIOWAIT:
+ if (++aio_sleep_cnt > MAX_AIO_SLEEPS)
+ THROW_STORE_EXCEPTION("Timeout waiting for AIO in BdbMessageStore::recoverTplStore()");
+ ::usleep(AIO_SLEEP_TIME);
+ break;
+ case rhm::journal::RHM_IORES_EMPTY:
+ done = true;
+ break; // done with all messages. (add call in jrnl to test that _emap is empty.)
+ default:
+ assert("Store Error: Unexpected msg state");
+ } // switch
+ }
+ } catch (const journal::jexception& e) {
+ THROW_STORE_EXCEPTION(std::string("TPL recoverTplStore() failed: ") + e.what());
+ }
+}
+
+void BdbMessageStore::recoverTplStore()
+{
if (journal::jdir::exists(tplStorePtr->jrnl_dir() + tplStorePtr->base_filename() + ".jinf")) {
u_int64_t thisHighestRid;
tplStorePtr->recover(tplNumJrnlFiles, tplJrnlFsizeSblks, tplWCachePgSizeSblks, tplWCacheNumPages, 0, thisHighestRid, 0);
- journal::txn_map& tmap = tplStorePtr->get_txn_map();
if (thisHighestRid > highestRid)
highestRid = thisHighestRid;
- DataTokenImpl dtokp;
- void* dbuff = NULL; size_t dbuffSize = 0;
- void* xidbuff = NULL; size_t xidbuffSize = 0;
- bool transientFlag = false;
- bool externalFlag = false;
- bool done = false;
- try {
- unsigned aio_sleep_cnt = 0;
- while (!done) {
- dtokp.reset();
- dtokp.set_wstate(DataTokenImpl::ENQ);
- switch (tplStorePtr->read_data_record(&dbuff, dbuffSize, &xidbuff, xidbuffSize, transientFlag, externalFlag, &dtokp)) {
- case rhm::journal::RHM_IORES_SUCCESS: {
- // Every TPL record contains both data and an XID
- assert(dbuffSize>0);
- assert(xidbuffSize>0);
- std::string xid(static_cast<const char*>(xidbuff), xidbuffSize);
- bool is2PC = *(static_cast<char*>(dbuff)) != 0;
+ // Load tplRecoverMap by reading the TPL store
+ readTplStore();
- // Check transaction details; add to recover map
- journal::txn_data_list txnList = tmap.get_tdata_list(xid);
- unsigned enqCnt = 0;
- unsigned deqCnt = 0;
- u_int64_t rid = 0;
-
- // Assume commit (roll forward) in cases where only prepare has been called - ie only enqueue record exists.
- // Note: will apply to both 1PC and 2PC transactions.
- bool commitFlag = true;
-
- for (journal::tdl_itr j = txnList.begin(); j<txnList.end(); j++) {
- if (j->_enq_flag) {
- rid = j->_rid;
- enqCnt++;
- } else {
- commitFlag = j->_commit_flag;
- deqCnt++;
- }
- }
- assert(enqCnt == 1);
- assert(deqCnt <= 1);
- tplMap.insert(TplRecoverMapPair(xid, TplRecoverStruct(rid, deqCnt == 1, commitFlag, is2PC)));
-
- ::free(xidbuff);
- aio_sleep_cnt = 0;
- break;
- }
- case rhm::journal::RHM_IORES_PAGE_AIOWAIT:
- if (++aio_sleep_cnt > MAX_AIO_SLEEPS)
- THROW_STORE_EXCEPTION("Timeout waiting for AIO in BdbMessageStore::recoverTplStore()");
- ::usleep(AIO_SLEEP_TIME);
- break;
- case rhm::journal::RHM_IORES_EMPTY:
- done = true;
- break; // done with all messages. (add call in jrnl to test that _emap is empty.)
- default:
- assert("Store Error: Unexpected msg state");
- } // switch
- }
- } catch (const journal::jexception& e) {
- THROW_STORE_EXCEPTION(std::string("TPL recoverTplStore() failed: ") + e.what());
- }
-
tplStorePtr->recover_complete(); // start journal.
}
}
@@ -988,7 +995,7 @@
void BdbMessageStore::recoverLockedMappings(txn_list& txns)
{
if (!tplStorePtr->is_ready())
- recoverTplStore(tplRecoverMap);
+ recoverTplStore();
// Abort unprepaired xids and populate the locked maps
for (TplRecoverMapCitr i = tplRecoverMap.begin(); i != tplRecoverMap.end(); i++) {
@@ -1002,8 +1009,12 @@
void BdbMessageStore::collectPreparedXids(std::set<std::string>& xids)
{
- if (!tplStorePtr->is_ready())
- recoverTplStore(tplRecoverMap);
+ if (tplStorePtr->is_ready()) {
+ tplStorePtr->read_reset();
+ readTplStore();
+ } else {
+ recoverTplStore();
+ }
for (TplRecoverMapCitr i = tplRecoverMap.begin(); i != tplRecoverMap.end(); i++) {
// Discard all txns that are to be rolled forward/back and 1PC transactions
if (!i->second.deq_flag && i->second.tpc_flag)
Modified: store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h
===================================================================
--- store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h 2008-08-14 17:49:28 UTC (rev 2305)
+++ store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h 2008-08-14 18:36:13 UTC (rev 2306)
@@ -93,7 +93,7 @@
Db generalDb;
// Pointer to Transaction Prepared List (TPL) journal instance
- boost::shared_ptr<JournalImpl> tplStorePtr;
+ boost::shared_ptr<TplJournalImpl> tplStorePtr;
TplRecoverMap tplRecoverMap;
IdSequence queueIdSequence;
@@ -158,7 +158,8 @@
queue_index& index,
txn_list& locked,
message_index& prepared);
- void recoverTplStore(TplRecoverMap& tplMap);
+ void readTplStore();
+ void recoverTplStore();
void recoverLockedMappings(txn_list& txns);
TxnCtxt* check(qpid::broker::TransactionContext* ctxt);
void store(const qpid::broker::PersistableQueue* queue,
Modified: store/branches/mrg-1.0/cpp/lib/JournalImpl.h
===================================================================
--- store/branches/mrg-1.0/cpp/lib/JournalImpl.h 2008-08-14 17:49:28 UTC (rev 2305)
+++ store/branches/mrg-1.0/cpp/lib/JournalImpl.h 2008-08-14 18:36:13 UTC (rev 2306)
@@ -207,6 +207,28 @@
}
}; // class JournalImpl
+ class TplJournalImpl : public JournalImpl
+ {
+ public:
+ TplJournalImpl(const std::string& journalId,
+ const std::string& journalDirectory,
+ const std::string& journalBaseFilename,
+ const qpid::sys::Duration getEventsTimeout,
+ const qpid::sys::Duration flushTimeout) :
+ JournalImpl(journalId, journalDirectory, journalBaseFilename, getEventsTimeout, flushTimeout)
+ {}
+
+ virtual ~TplJournalImpl() {}
+
+ // Special version of read_data_record that ignores transactions - needed when reading the TPL
+ inline journal::iores read_data_record(void** const datapp, std::size_t& dsize,
+ void** const xidpp, std::size_t& xidsize, bool& transient, bool& external,
+ journal::data_tok* const dtokp) {
+ return JournalImpl::read_data_record(datapp, dsize, xidpp, xidsize, transient, external, dtokp, true);
+ }
+ inline void read_reset() { _rmgr.invalidate(); }
+ };
+
} // namespace bdbstore
} // namespace rhm
Modified: store/branches/mrg-1.0/cpp/lib/jrnl/jcntl.cpp
===================================================================
--- store/branches/mrg-1.0/cpp/lib/jrnl/jcntl.cpp 2008-08-14 17:49:28 UTC (rev 2305)
+++ store/branches/mrg-1.0/cpp/lib/jrnl/jcntl.cpp 2008-08-14 18:36:13 UTC (rev 2306)
@@ -316,14 +316,14 @@
#define RCINVALID_SLEEP_TIME_MS 5
iores
jcntl::read_data_record(void** const datapp, std::size_t& dsize, void** const xidpp, std::size_t& xidsize,
- bool& transient, bool& external, data_tok* const dtokp)
+ bool& transient, bool& external, data_tok* const dtokp, bool ignore_pending_txns)
{
check_rstatus("read_data");
unsigned cnt = 0;
iores res;
do
{
- res = _rmgr.read(datapp, dsize, xidpp, xidsize, transient, external, dtokp);
+ res = _rmgr.read(datapp, dsize, xidpp, xidsize, transient, external, dtokp, ignore_pending_txns);
if (res == RHM_IORES_RCINVALID)
{
get_wr_events(); // check for outstanding write events
Modified: store/branches/mrg-1.0/cpp/lib/jrnl/jcntl.hpp
===================================================================
--- store/branches/mrg-1.0/cpp/lib/jrnl/jcntl.hpp 2008-08-14 17:49:28 UTC (rev 2305)
+++ store/branches/mrg-1.0/cpp/lib/jrnl/jcntl.hpp 2008-08-14 18:36:13 UTC (rev 2306)
@@ -415,7 +415,8 @@
* \exception TODO
*/
iores read_data_record(void** const datapp, std::size_t& dsize, void** const xidpp,
- std::size_t& xidsize, bool& transient, bool& external, data_tok* const dtokp);
+ std::size_t& xidsize, bool& transient, bool& external, data_tok* const dtokp,
+ bool ignore_pending_txns = false);
/**
* \brief Dequeues (marks as no longer needed) data record in journal.
Modified: store/branches/mrg-1.0/cpp/lib/jrnl/rmgr.cpp
===================================================================
--- store/branches/mrg-1.0/cpp/lib/jrnl/rmgr.cpp 2008-08-14 17:49:28 UTC (rev 2305)
+++ store/branches/mrg-1.0/cpp/lib/jrnl/rmgr.cpp 2008-08-14 18:36:13 UTC (rev 2306)
@@ -90,7 +90,7 @@
iores
rmgr::read(void** const datapp, std::size_t& dsize, void** const xidpp, std::size_t& xidsize,
- bool& transient, bool& external, data_tok* dtokp)
+ bool& transient, bool& external, data_tok* dtokp, bool ignore_pending_txns)
{
iores res = pre_read_check(dtokp);
if (res != RHM_IORES_SUCCESS)
@@ -164,8 +164,9 @@
}
catch (const jexception& e)
{
+ bool enforce_txns = !_jc->is_read_only() && !ignore_pending_txns;
// Block read for transactionally locked record (only when not recovering)
- if (e.err_code() == jerrno::JERR_MAP_LOCKED && !_jc->is_read_only())
+ if (e.err_code() == jerrno::JERR_MAP_LOCKED && enforce_txns)
return RHM_IORES_TXPENDING;
// (Recover mode only) Ok, not in emap - now search tmap, if present then read
@@ -184,7 +185,7 @@
is_enq = ditr->_drid == _hdr._rid;
}
}
- if (!_jc->is_read_only() && is_enq)
+ if (enforce_txns && is_enq)
return RHM_IORES_TXPENDING;
}
#endif
Modified: store/branches/mrg-1.0/cpp/lib/jrnl/rmgr.hpp
===================================================================
--- store/branches/mrg-1.0/cpp/lib/jrnl/rmgr.hpp 2008-08-14 17:49:28 UTC (rev 2305)
+++ store/branches/mrg-1.0/cpp/lib/jrnl/rmgr.hpp 2008-08-14 18:36:13 UTC (rev 2306)
@@ -77,7 +77,8 @@
void initialize(const rd_aio_cb rd_cb);
iores read(void** const datapp, std::size_t& dsize, void** const xidpp,
- std::size_t& xidsize, bool& transient, bool& external, data_tok* dtokp);
+ std::size_t& xidsize, bool& transient, bool& external, data_tok* dtokp,
+ bool ignore_pending_txns);
u_int32_t get_events(page_state state = AIO_COMPLETE);
void recover_complete();
inline bool is_valid() const {return _valid; }
Modified: store/branches/mrg-1.0/cpp/lib/jrnl/wmgr.cpp
===================================================================
--- store/branches/mrg-1.0/cpp/lib/jrnl/wmgr.cpp 2008-08-14 17:49:28 UTC (rev 2305)
+++ store/branches/mrg-1.0/cpp/lib/jrnl/wmgr.cpp 2008-08-14 18:36:13 UTC (rev 2306)
@@ -834,7 +834,6 @@
#endif
break;
case data_tok::COMMIT_SUBM:
-//std::cout << " * commit-ret, queue=" << _jc->id() << " xid=" << dtokp->xid() << std::endl << std::flush;
dtokl.push_back(dtokp);
tot_data_toks++;
dtokp->set_wstate(data_tok::COMMITTED);
17 years, 8 months
- 1
- 0
rhmessaging commits: r2305 - in store/trunk/cpp/lib: jrnl and 1 other directory.
by rhmessaging-commits＠lists.jboss.org
Author: kpvdr
Date: 2008-08-14 13:49:28 -0400 (Thu, 14 Aug 2008)
New Revision: 2305
Modified:
store/trunk/cpp/lib/BdbMessageStore.cpp
store/trunk/cpp/lib/BdbMessageStore.h
store/trunk/cpp/lib/JournalImpl.h
store/trunk/cpp/lib/jrnl/jcntl.cpp
store/trunk/cpp/lib/jrnl/jcntl.hpp
store/trunk/cpp/lib/jrnl/rmgr.cpp
store/trunk/cpp/lib/jrnl/rmgr.hpp
store/trunk/cpp/lib/jrnl/txn_map.hpp
Log:
Fix for 458656 "List of xids from dtx.recover is not as expected". The call to BdbMessageStore::collectPreparedXids() was using the prepared list obtained at recover, but this call needs to provide a current list, no matter when called. Added ability to ignore transactions when reading, and created a subclass of JournalImpl (called TplJournalImpl) which implements the ignore as standard practice when calling read. This enables reads of pending transactions from this list. Now BdbMessageStore::collectPreparedXids() re-reads the TPL before returning the XID list.
Modified: store/trunk/cpp/lib/BdbMessageStore.cpp
===================================================================
--- store/trunk/cpp/lib/BdbMessageStore.cpp 2008-08-14 13:44:04 UTC (rev 2304)
+++ store/trunk/cpp/lib/BdbMessageStore.cpp 2008-08-14 17:49:28 UTC (rev 2305)
@@ -249,7 +249,7 @@
open(mappingDb, txn.get(), "mappings.db", true);
open(bindingDb, txn.get(), "bindings.db", true);
open(generalDb, txn.get(), "general.db", false);
- tplStorePtr.reset(new JournalImpl("TplStore", getTplBaseDir(), "tpl", defJournalGetEventsTimeout, defJournalFlushTimeout));
+ tplStorePtr.reset(new TplJournalImpl("TplStore", getTplBaseDir(), "tpl", defJournalGetEventsTimeout, defJournalFlushTimeout));
txn.commit();
} catch (const journal::jexception& e) {
txn.abort();
@@ -930,77 +930,84 @@
return count;
}
-void BdbMessageStore::recoverTplStore(TplRecoverMap& tplMap)
+void BdbMessageStore::readTplStore()
{
+ tplRecoverMap.clear();
+ journal::txn_map& tmap = tplStorePtr->get_txn_map();
+ DataTokenImpl dtokp;
+ void* dbuff = NULL; size_t dbuffSize = 0;
+ void* xidbuff = NULL; size_t xidbuffSize = 0;
+ bool transientFlag = false;
+ bool externalFlag = false;
+ bool done = false;
+ try {
+ unsigned aio_sleep_cnt = 0;
+ while (!done) {
+ dtokp.reset();
+ dtokp.set_wstate(DataTokenImpl::ENQ);
+ switch (tplStorePtr->read_data_record(&dbuff, dbuffSize, &xidbuff, xidbuffSize, transientFlag, externalFlag, &dtokp)) {
+ case rhm::journal::RHM_IORES_SUCCESS: {
+ // Every TPL record contains both data and an XID
+ assert(dbuffSize>0);
+ assert(xidbuffSize>0);
+ std::string xid(static_cast<const char*>(xidbuff), xidbuffSize);
+ bool is2PC = *(static_cast<char*>(dbuff)) != 0;
+
+ // Check transaction details; add to recover map
+ journal::txn_data_list txnList = tmap.get_tdata_list(xid);
+ unsigned enqCnt = 0;
+ unsigned deqCnt = 0;
+ u_int64_t rid = 0;
+
+ // Assume commit (roll forward) in cases where only prepare has been called - ie only enqueue record exists.
+ // Note: will apply to both 1PC and 2PC transactions.
+ bool commitFlag = true;
+
+ for (journal::tdl_itr j = txnList.begin(); j<txnList.end(); j++) {
+ if (j->_enq_flag) {
+ rid = j->_rid;
+ enqCnt++;
+ } else {
+ commitFlag = j->_commit_flag;
+ deqCnt++;
+ }
+ }
+ assert(enqCnt == 1);
+ assert(deqCnt <= 1);
+ tplRecoverMap.insert(TplRecoverMapPair(xid, TplRecoverStruct(rid, deqCnt == 1, commitFlag, is2PC)));
+
+ ::free(xidbuff);
+ aio_sleep_cnt = 0;
+ break;
+ }
+ case rhm::journal::RHM_IORES_PAGE_AIOWAIT:
+ if (++aio_sleep_cnt > MAX_AIO_SLEEPS)
+ THROW_STORE_EXCEPTION("Timeout waiting for AIO in BdbMessageStore::recoverTplStore()");
+ ::usleep(AIO_SLEEP_TIME);
+ break;
+ case rhm::journal::RHM_IORES_EMPTY:
+ done = true;
+ break; // done with all messages. (add call in jrnl to test that _emap is empty.)
+ default:
+ assert("Store Error: Unexpected msg state");
+ } // switch
+ }
+ } catch (const journal::jexception& e) {
+ THROW_STORE_EXCEPTION(std::string("TPL recoverTplStore() failed: ") + e.what());
+ }
+}
+
+void BdbMessageStore::recoverTplStore()
+{
if (journal::jdir::exists(tplStorePtr->jrnl_dir() + tplStorePtr->base_filename() + ".jinf")) {
u_int64_t thisHighestRid;
tplStorePtr->recover(tplNumJrnlFiles, tplJrnlFsizeSblks, tplWCachePgSizeSblks, tplWCacheNumPages, 0, thisHighestRid, 0);
- journal::txn_map& tmap = tplStorePtr->get_txn_map();
if (thisHighestRid > highestRid)
highestRid = thisHighestRid;
- DataTokenImpl dtokp;
- void* dbuff = NULL; size_t dbuffSize = 0;
- void* xidbuff = NULL; size_t xidbuffSize = 0;
- bool transientFlag = false;
- bool externalFlag = false;
- bool done = false;
- try {
- unsigned aio_sleep_cnt = 0;
- while (!done) {
- dtokp.reset();
- dtokp.set_wstate(DataTokenImpl::ENQ);
- switch (tplStorePtr->read_data_record(&dbuff, dbuffSize, &xidbuff, xidbuffSize, transientFlag, externalFlag, &dtokp)) {
- case rhm::journal::RHM_IORES_SUCCESS: {
- // Every TPL record contains both data and an XID
- assert(dbuffSize>0);
- assert(xidbuffSize>0);
- std::string xid(static_cast<const char*>(xidbuff), xidbuffSize);
- bool is2PC = *(static_cast<char*>(dbuff)) != 0;
+ // Load tplRecoverMap by reading the TPL store
+ readTplStore();
- // Check transaction details; add to recover map
- journal::txn_data_list txnList = tmap.get_tdata_list(xid);
- unsigned enqCnt = 0;
- unsigned deqCnt = 0;
- u_int64_t rid = 0;
-
- // Assume commit (roll forward) in cases where only prepare has been called - ie only enqueue record exists.
- // Note: will apply to both 1PC and 2PC transactions.
- bool commitFlag = true;
-
- for (journal::tdl_itr j = txnList.begin(); j<txnList.end(); j++) {
- if (j->_enq_flag) {
- rid = j->_rid;
- enqCnt++;
- } else {
- commitFlag = j->_commit_flag;
- deqCnt++;
- }
- }
- assert(enqCnt == 1);
- assert(deqCnt <= 1);
- tplMap.insert(TplRecoverMapPair(xid, TplRecoverStruct(rid, deqCnt == 1, commitFlag, is2PC)));
-
- ::free(xidbuff);
- aio_sleep_cnt = 0;
- break;
- }
- case rhm::journal::RHM_IORES_PAGE_AIOWAIT:
- if (++aio_sleep_cnt > MAX_AIO_SLEEPS)
- THROW_STORE_EXCEPTION("Timeout waiting for AIO in BdbMessageStore::recoverTplStore()");
- ::usleep(AIO_SLEEP_TIME);
- break;
- case rhm::journal::RHM_IORES_EMPTY:
- done = true;
- break; // done with all messages. (add call in jrnl to test that _emap is empty.)
- default:
- assert("Store Error: Unexpected msg state");
- } // switch
- }
- } catch (const journal::jexception& e) {
- THROW_STORE_EXCEPTION(std::string("TPL recoverTplStore() failed: ") + e.what());
- }
-
tplStorePtr->recover_complete(); // start journal.
}
}
@@ -1008,7 +1015,7 @@
void BdbMessageStore::recoverLockedMappings(txn_list& txns)
{
if (!tplStorePtr->is_ready())
- recoverTplStore(tplRecoverMap);
+ recoverTplStore();
// Abort unprepaired xids and populate the locked maps
for (TplRecoverMapCitr i = tplRecoverMap.begin(); i != tplRecoverMap.end(); i++) {
@@ -1022,8 +1029,12 @@
void BdbMessageStore::collectPreparedXids(std::set<std::string>& xids)
{
- if (!tplStorePtr->is_ready())
- recoverTplStore(tplRecoverMap);
+ if (tplStorePtr->is_ready()) {
+ tplStorePtr->read_reset();
+ readTplStore();
+ } else {
+ recoverTplStore();
+ }
for (TplRecoverMapCitr i = tplRecoverMap.begin(); i != tplRecoverMap.end(); i++) {
// Discard all txns that are to be rolled forward/back and 1PC transactions
if (!i->second.deq_flag && i->second.tpc_flag)
Modified: store/trunk/cpp/lib/BdbMessageStore.h
===================================================================
--- store/trunk/cpp/lib/BdbMessageStore.h 2008-08-14 13:44:04 UTC (rev 2304)
+++ store/trunk/cpp/lib/BdbMessageStore.h 2008-08-14 17:49:28 UTC (rev 2305)
@@ -93,7 +93,7 @@
Db generalDb;
// Pointer to Transaction Prepared List (TPL) journal instance
- boost::shared_ptr<JournalImpl> tplStorePtr;
+ boost::shared_ptr<TplJournalImpl> tplStorePtr;
TplRecoverMap tplRecoverMap;
IdSequence queueIdSequence;
@@ -158,7 +158,8 @@
queue_index& index,
txn_list& locked,
message_index& prepared);
- void recoverTplStore(TplRecoverMap& tplMap);
+ void readTplStore();
+ void recoverTplStore();
void recoverLockedMappings(txn_list& txns);
TxnCtxt* check(qpid::broker::TransactionContext* ctxt);
void store(const qpid::broker::PersistableQueue* queue,
Modified: store/trunk/cpp/lib/JournalImpl.h
===================================================================
--- store/trunk/cpp/lib/JournalImpl.h 2008-08-14 13:44:04 UTC (rev 2304)
+++ store/trunk/cpp/lib/JournalImpl.h 2008-08-14 17:49:28 UTC (rev 2305)
@@ -75,7 +75,7 @@
boost::intrusive_ptr<qpid::broker::TimerTask> getEventsFireEventsPtr;
pthread_mutex_t _getf_mutex; // getEventsTimerSetFlag mutex
- u_int64_t lastReadRid; // rid of last read msg
+ u_int64_t lastReadRid; // rid of last read msg for loadMsgContent()
bool writeActivityFlag;
bool flushTriggeredFlag;
@@ -207,6 +207,28 @@
}
}; // class JournalImpl
+ class TplJournalImpl : public JournalImpl
+ {
+ public:
+ TplJournalImpl(const std::string& journalId,
+ const std::string& journalDirectory,
+ const std::string& journalBaseFilename,
+ const qpid::sys::Duration getEventsTimeout,
+ const qpid::sys::Duration flushTimeout) :
+ JournalImpl(journalId, journalDirectory, journalBaseFilename, getEventsTimeout, flushTimeout)
+ {}
+
+ virtual ~TplJournalImpl() {}
+
+ // Special version of read_data_record that ignores transactions - needed when reading the TPL
+ inline journal::iores read_data_record(void** const datapp, std::size_t& dsize,
+ void** const xidpp, std::size_t& xidsize, bool& transient, bool& external,
+ journal::data_tok* const dtokp) {
+ return JournalImpl::read_data_record(datapp, dsize, xidpp, xidsize, transient, external, dtokp, true);
+ }
+ inline void read_reset() { _rmgr.invalidate(); }
+ };
+
} // namespace bdbstore
} // namespace rhm
Modified: store/trunk/cpp/lib/jrnl/jcntl.cpp
===================================================================
--- store/trunk/cpp/lib/jrnl/jcntl.cpp 2008-08-14 13:44:04 UTC (rev 2304)
+++ store/trunk/cpp/lib/jrnl/jcntl.cpp 2008-08-14 17:49:28 UTC (rev 2305)
@@ -310,14 +310,14 @@
#define RCINVALID_SLEEP_TIME_MS 5
iores
jcntl::read_data_record(void** const datapp, std::size_t& dsize, void** const xidpp, std::size_t& xidsize,
- bool& transient, bool& external, data_tok* const dtokp)
+ bool& transient, bool& external, data_tok* const dtokp, bool ignore_pending_txns)
{
check_rstatus("read_data");
unsigned cnt = 0;
iores res;
do
{
- res = _rmgr.read(datapp, dsize, xidpp, xidsize, transient, external, dtokp);
+ res = _rmgr.read(datapp, dsize, xidpp, xidsize, transient, external, dtokp, ignore_pending_txns);
if (res == RHM_IORES_RCINVALID)
{
get_wr_events(); // check for outstanding write events
Modified: store/trunk/cpp/lib/jrnl/jcntl.hpp
===================================================================
--- store/trunk/cpp/lib/jrnl/jcntl.hpp 2008-08-14 13:44:04 UTC (rev 2304)
+++ store/trunk/cpp/lib/jrnl/jcntl.hpp 2008-08-14 17:49:28 UTC (rev 2305)
@@ -415,7 +415,8 @@
* \exception TODO
*/
iores read_data_record(void** const datapp, std::size_t& dsize, void** const xidpp,
- std::size_t& xidsize, bool& transient, bool& external, data_tok* const dtokp);
+ std::size_t& xidsize, bool& transient, bool& external, data_tok* const dtokp,
+ bool ignore_pending_txns = false);
/**
* \brief Dequeues (marks as no longer needed) data record in journal.
Modified: store/trunk/cpp/lib/jrnl/rmgr.cpp
===================================================================
--- store/trunk/cpp/lib/jrnl/rmgr.cpp 2008-08-14 13:44:04 UTC (rev 2304)
+++ store/trunk/cpp/lib/jrnl/rmgr.cpp 2008-08-14 17:49:28 UTC (rev 2305)
@@ -90,7 +90,7 @@
iores
rmgr::read(void** const datapp, std::size_t& dsize, void** const xidpp, std::size_t& xidsize,
- bool& transient, bool& external, data_tok* dtokp)
+ bool& transient, bool& external, data_tok* dtokp, bool ignore_pending_txns)
{
iores res = pre_read_check(dtokp);
if (res != RHM_IORES_SUCCESS)
@@ -161,8 +161,9 @@
}
catch (const jexception& e)
{
+ bool enforce_txns = !_jc->is_read_only() && !ignore_pending_txns;
// Block read for transactionally locked record (only when not recovering)
- if (e.err_code() == jerrno::JERR_MAP_LOCKED && !_jc->is_read_only())
+ if (e.err_code() == jerrno::JERR_MAP_LOCKED && enforce_txns)
return RHM_IORES_TXPENDING;
// (Recover mode only) Ok, not in emap - now search tmap, if present then read
@@ -181,7 +182,7 @@
is_enq = ditr->_drid == _hdr._rid;
}
}
- if (!_jc->is_read_only() && is_enq)
+ if (enforce_txns && is_enq)
return RHM_IORES_TXPENDING;
}
Modified: store/trunk/cpp/lib/jrnl/rmgr.hpp
===================================================================
--- store/trunk/cpp/lib/jrnl/rmgr.hpp 2008-08-14 13:44:04 UTC (rev 2304)
+++ store/trunk/cpp/lib/jrnl/rmgr.hpp 2008-08-14 17:49:28 UTC (rev 2305)
@@ -77,7 +77,8 @@
void initialize(const rd_aio_cb rd_cb);
iores read(void** const datapp, std::size_t& dsize, void** const xidpp,
- std::size_t& xidsize, bool& transient, bool& external, data_tok* dtokp);
+ std::size_t& xidsize, bool& transient, bool& external, data_tok* dtokp,
+ bool ignore_pending_txns);
u_int32_t get_events(page_state state = AIO_COMPLETE);
void recover_complete();
inline bool is_valid() const {return _valid; }
Modified: store/trunk/cpp/lib/jrnl/txn_map.hpp
===================================================================
--- store/trunk/cpp/lib/jrnl/txn_map.hpp 2008-08-14 13:44:04 UTC (rev 2304)
+++ store/trunk/cpp/lib/jrnl/txn_map.hpp 2008-08-14 17:49:28 UTC (rev 2305)
@@ -102,9 +102,9 @@
* <pre>
* key data
*
- * xid1 --- vector< [ rid, drid, fid, enq_flag, aio_compl ] >
- * xid2 --- vector< [ rid, drid, fid, enq_flag, aio_compl ] >
- * xid3 --- vector< [ rid, drid, fid, enq_flag, aio_compl ] >
+ * xid1 --- vector< [ rid, drid, fid, enq_flag, commit_flag, aio_compl ] >
+ * xid2 --- vector< [ rid, drid, fid, enq_flag, commit_flag, aio_compl ] >
+ * xid3 --- vector< [ rid, drid, fid, enq_flag, commit_flag, aio_compl ] >
* ...
* </pre>
*/
17 years, 8 months
- 1
- 0
rhmessaging commits: r2304 - mgmt/trunk/cumin/python/cumin.
by rhmessaging-commits＠lists.jboss.org
Author: eallen
Date: 2008-08-14 09:44:04 -0400 (Thu, 14 Aug 2008)
New Revision: 2304
Modified:
mgmt/trunk/cumin/python/cumin/brokerlink.py
mgmt/trunk/cumin/python/cumin/brokerlink.strings
Log:
Added a state switch to the Add Route page to show Active/All exchanges
Modified: mgmt/trunk/cumin/python/cumin/brokerlink.py
===================================================================
--- mgmt/trunk/cumin/python/cumin/brokerlink.py 2008-08-14 13:43:12 UTC (rev 2303)
+++ mgmt/trunk/cumin/python/cumin/brokerlink.py 2008-08-14 13:44:04 UTC (rev 2304)
@@ -1,4 +1,4 @@
-from cumin.binding import ExchangeKeysField
+from cumin.binding import ExchangeKeysField, ExchangeState
from cumin.exchange import ExchangeInfo
from formats import *
from model import *
@@ -280,6 +280,13 @@
self.exchange_tmpl = Template(self, "exchange_html")
+ self.state = ExchangeState(app, "phase")
+ self.add_child(self.state)
+
+ self.phase = Parameter(app, "phase")
+ self.add_parameter(self.phase)
+ form.add_form_parameter(self.phase)
+
def get(self, session):
return self.exchange_param.get(session)
@@ -304,6 +311,9 @@
def render_checked(self, session, exchange):
return self.exchange_param.get(session) == str(exchange.id) and "checked=\"checked\"" or ""
+ def render_phase_path(self, session, *args):
+ return self.phase.path
+
def render_exchanges(self, session, *args):
#reg = self.frame.frame.frame.get_object(session)
@@ -314,7 +324,8 @@
writer = Writer()
for exchange in sortedExchanges:
- if ExchangeInfo.is_builtin(exchange) or is_active(exchange):
+ if ExchangeInfo.is_builtin(exchange) or \
+ not (self.state.is_active(session) and not is_active(exchange)):
if not exchange.name in ["qpid.management", ""]:
if not self.get(session):
self.exchange_param.set(session, str(exchange.id))
@@ -322,6 +333,11 @@
return writer.to_string()
+ def pre_process(self, session, *args):
+ phase = self.phase.get(session)
+ if phase:
+ self.state.set(session, phase)
+
class BridgeAdd(CuminFieldForm):
def __init__(self, app, name):
super(BridgeAdd, self).__init__(app, name)
@@ -349,7 +365,10 @@
def render_title(self, session, link):
return "Add Route to '%s:%d'" % (link.host, link.port)
-
+
+ def pre_process(self, session, *args):
+ self.__exchange.pre_process(session, *args)
+
def process_cancel(self, session, *args):
self.__key.set(session, "")
self.__tag.set(session, "")
Modified: mgmt/trunk/cumin/python/cumin/brokerlink.strings
===================================================================
--- mgmt/trunk/cumin/python/cumin/brokerlink.strings 2008-08-14 13:43:12 UTC (rev 2303)
+++ mgmt/trunk/cumin/python/cumin/brokerlink.strings 2008-08-14 13:44:04 UTC (rev 2304)
@@ -131,6 +131,7 @@
<div class="head">{title}</div>
<div class="body">
<div class="field multiple">
+ <div class="rfloat">{phase}</div>
<div class="title">Source Broker</div>
<div class="inputs"><div class="input_prompt">Address</div><input type="text" name="{broker_name_path}" value="{broker_name_value}" tabindex="100" size="32"/>{broker_name_error}</div>
<div class="optional">
@@ -144,7 +145,6 @@
</div>
{fields}
</div>
- {form_error}
<div class="foot">
{help}
{submit}
@@ -163,6 +163,7 @@
</div>
[ExchangeRadioField.html]
+ <div class="rfloat">{phase}</div> <input id="phase_state" type="hidden" name="{phase_path}" value="" />
<span class="legend">Choose an Exchange</span>
<fieldset>
{exchanges}
17 years, 8 months
- 1
- 0
rhmessaging commits: r2303 - mgmt/trunk/cumin/python/cumin.
by rhmessaging-commits＠lists.jboss.org
Author: eallen
Date: 2008-08-14 09:43:12 -0400 (Thu, 14 Aug 2008)
New Revision: 2303
Modified:
mgmt/trunk/cumin/python/cumin/binding.py
mgmt/trunk/cumin/python/cumin/queue.py
mgmt/trunk/cumin/python/cumin/widgets.py
Log:
Added a pre_process method to handle setting up state switches that do a form submit
Modified: mgmt/trunk/cumin/python/cumin/binding.py
===================================================================
--- mgmt/trunk/cumin/python/cumin/binding.py 2008-08-14 12:38:25 UTC (rev 2302)
+++ mgmt/trunk/cumin/python/cumin/binding.py 2008-08-14 13:43:12 UTC (rev 2303)
@@ -281,11 +281,6 @@
self.add_child(self.state)
def render_phase_path(self, session, vhost):
- # the hidden input phase MUST be rendered before the stateswitch
- phase = self.phase.get(session)
- if phase:
- self.state.set(session, phase)
-
return self.phase.path
def get_args(self, session):
@@ -331,6 +326,11 @@
return writer.to_string()
+ def pre_process(self, session):
+ phase = self.phase.get(session)
+ if phase:
+ self.state.set(session, phase)
+
def post_process(self, session):
self.dict_param.clear()
Modified: mgmt/trunk/cumin/python/cumin/queue.py
===================================================================
--- mgmt/trunk/cumin/python/cumin/queue.py 2008-08-14 12:38:25 UTC (rev 2302)
+++ mgmt/trunk/cumin/python/cumin/queue.py 2008-08-14 13:43:12 UTC (rev 2303)
@@ -349,15 +349,18 @@
self.bindings = ExchangeKeysField(app, "bindings", self)
self.add_field(self.bindings)
+ def pre_process(self, session):
+ self.bindings.pre_process(session)
+
+ def post_process(self, session):
+ self.bindings.post_process(session)
+
def validate(self, session, queue_name):
super_error = super(QueueForm, self).validate(session)
(errors, form_binding_info) = self.bindings.get_binding_errors(session, queue_name)
return (errors or super_error, form_binding_info)
class QueueAdd(QueueForm):
- def post_process(self, session):
- self.bindings.post_process(session)
-
def process_cancel(self, session):
branch = session.branch()
self.frame.show_view(branch)
@@ -528,6 +531,9 @@
return "<ul class=\"errors\" style=\"margin:0; float:left;\"><li>%s</li></ul>" % \
"</li><li>".join(errors["no_exchanges"])
+ def pre_process(self, session):
+ self.bindings.pre_process(session)
+
def post_process(self, session):
self.bindings.post_process(session)
Modified: mgmt/trunk/cumin/python/cumin/widgets.py
===================================================================
--- mgmt/trunk/cumin/python/cumin/widgets.py 2008-08-14 12:38:25 UTC (rev 2302)
+++ mgmt/trunk/cumin/python/cumin/widgets.py 2008-08-14 13:43:12 UTC (rev 2303)
@@ -146,6 +146,8 @@
def do_process(self, session, *args):
self.page.set_modal(session, True)
+ self.pre_process(session, *args)
+
if self.__cancel.get(session):
self.__cancel.set(session, False)
@@ -171,6 +173,9 @@
def post_process(self, session, *args):
pass
+ def pre_process(self, session, *args):
+ pass
+
def render_cancel_content(self, session, *args):
return "Cancel"
17 years, 8 months
- 1
- 0
rhmessaging commits: r2302 - mgmt/trunk/cumin/resources.
by rhmessaging-commits＠lists.jboss.org
Author: eallen
Date: 2008-08-14 08:38:25 -0400 (Thu, 14 Aug 2008)
New Revision: 2302
Modified:
mgmt/trunk/cumin/resources/help.html
Log:
Added main.broker.peer.bridgeadd.help for Add Route page
Modified: mgmt/trunk/cumin/resources/help.html
===================================================================
--- mgmt/trunk/cumin/resources/help.html 2008-08-13 22:23:34 UTC (rev 2301)
+++ mgmt/trunk/cumin/resources/help.html 2008-08-14 12:38:25 UTC (rev 2302)
@@ -157,6 +157,25 @@
</li>
</ul>
+ <div class="titlepage">
+ <a name="main.broker.peer.bridgeadd.help"/>
+ <h3 class="title">Adding a Route to a Broker Link</h3>
+ </div>
+ <p>
+ Adds a Route to an existing Broker Link. Messages are exchanged with a linked broker based on Routes. A Route requires an exchange and a routing key.
+ </p>
+ <p>
+ Optionally, a tag and exclude list can be supplied.
+ </p>
+
+<ul class="docnav">
+ <li class="up">
+ <a accesskey="u" href="#top">
+ <strong>Help Index</strong>
+ </a>
+ </li>
+</ul>
+
</div><!-- section -->
<div style="height:1000px;"><!-- --></div>
17 years, 8 months
- 1
- 0
rhmessaging commits: r2301 - mgmt/trunk/cumin/python/cumin.
by rhmessaging-commits＠lists.jboss.org
Author: eallen
Date: 2008-08-13 18:23:34 -0400 (Wed, 13 Aug 2008)
New Revision: 2301
Modified:
mgmt/trunk/cumin/python/cumin/brokerlink.py
mgmt/trunk/cumin/python/cumin/brokerlink.strings
mgmt/trunk/cumin/python/cumin/model.py
Log:
Added "Add Route" to Route List
Modified: mgmt/trunk/cumin/python/cumin/brokerlink.py
===================================================================
--- mgmt/trunk/cumin/python/cumin/brokerlink.py 2008-08-13 22:20:39 UTC (rev 2300)
+++ mgmt/trunk/cumin/python/cumin/brokerlink.py 2008-08-13 22:23:34 UTC (rev 2301)
@@ -1,14 +1,16 @@
+from cumin.binding import ExchangeKeysField
+from cumin.exchange import ExchangeInfo
+from formats import *
+from model import *
+from parameters import *
+from stat import *
+from util import *
+from widgets import *
from wooly import *
-from wooly.widgets import *
from wooly.forms import *
from wooly.resources import *
+from wooly.widgets import *
-from stat import *
-from model import *
-from widgets import *
-from parameters import *
-from formats import *
-from util import *
strings = StringCatalog(__file__)
@@ -127,10 +129,17 @@
remove = LinkRemove(app, "remove")
self.add_mode(remove)
self.set_remove_mode(remove)
+
+ self.__bridge_add = BridgeAdd(app, "bridgeadd")
+ self.add_mode(self.__bridge_add)
def render_title(self, session, peer):
return super(PeerFrame, self).render_title(session, peer)
+ def show_bridge_add(self, session):
+ self.page.set_current_frame(session, self.__bridge_add)
+ return self.show_mode(session, self.__bridge_add)
+
class LinkRemove(CuminConfirmForm):
def get_args(self, session):
return self.frame.get_args(session)
@@ -215,6 +224,11 @@
start, end = self.get_bounds(session)
return peer.bridges[start:end]
+ def render_add_bridge_url(self, session, route):
+ branch = session.branch()
+ self.frame.show_bridge_add(branch)
+ return branch.marshal()
+
def render_item_destination(self, session, route):
return route.link.managedBroker
@@ -255,6 +269,122 @@
def render_title(self, session, exchange):
return "Messages Received, Routed, and Dropped"
+class ExchangeRadioField(RadioField):
+ def __init__(self, app, name, form):
+ super(ExchangeRadioField, self).__init__(app, name, form)
+
+ self.exchange_param = Parameter(app, "exchange_param")
+ self.exchange_param.default = ""
+ self.add_parameter(self.exchange_param)
+ form.add_form_parameter(self.exchange_param)
+
+ self.exchange_tmpl = Template(self, "exchange_html")
+
+ def get(self, session):
+ return self.exchange_param.get(session)
+
+ def set(self, session, value):
+ self.exchange_param.set(session, value)
+
+ def render_title(self, session):
+ return "Choose an Exchange"
+
+ def render_exchange_name_path(self, session, exchange):
+ return self.exchange_param.path
+
+ def render_exchange_name(self, session, exchange):
+ return exchange.name or "Default"
+
+ def render_exchange_value(self, session, exchange):
+ return str(exchange.id)
+
+ def render_exchange_id(self, session, exchange):
+ return ".".join((self.path, str(exchange.id)))
+
+ def render_checked(self, session, exchange):
+ return self.exchange_param.get(session) == str(exchange.id) and "checked=\"checked\"" or ""
+
+ def render_exchanges(self, session, *args):
+
+ #reg = self.frame.frame.frame.get_object(session)
+ broker = self.get_parent_named("broker")
+ reg = broker.get_object(session)
+ vhost = reg.getDefaultVhost()
+ sortedExchanges = sorted_by(vhost.exchanges)
+
+ writer = Writer()
+ for exchange in sortedExchanges:
+ if ExchangeInfo.is_builtin(exchange) or is_active(exchange):
+ if not exchange.name in ["qpid.management", ""]:
+ if not self.get(session):
+ self.exchange_param.set(session, str(exchange.id))
+ self.exchange_tmpl.render(writer, session, exchange)
+
+ return writer.to_string()
+
+class BridgeAdd(CuminFieldForm):
+ def __init__(self, app, name):
+ super(BridgeAdd, self).__init__(app, name)
+
+ self.__exchange = ExchangeRadioField(app, "exchange", self)
+ self.add_field(self.__exchange)
+
+ self.__key = TextField(app, "key", self)
+ self.__key.set_title("Routing Key")
+ self.add_field(self.__key)
+
+ self.__tag = TextField(app, "tag", self)
+ self.__tag.set_required(False)
+ self.__tag.set_title("Tag")
+ self.add_field(self.__tag)
+
+ self.__excludes = TextField(app, "excludes", self)
+ self.__excludes.set_required(False)
+ self.__excludes.set_title("Excludes")
+ self.add_field(self.__excludes)
+
+
+ def get_args(self, session):
+ return self.frame.get_args(session)
+
+ def render_title(self, session, link):
+ return "Add Route to '%s:%d'" % (link.host, link.port)
+
+ def process_cancel(self, session, *args):
+ self.__key.set(session, "")
+ self.__tag.set(session, "")
+ self.__excludes.set(session, "")
+ self.__exchange.set(session, "")
+
+ branch = session.branch()
+ self.frame.show_view(branch)
+ self.page.set_redirect_url(session, branch.marshal())
+
+ def process_submit(self, session, link):
+ errors = self.validate(session)
+ if errors:
+ pass
+ else:
+ key = self.__key.get(session)
+ tag = self.__tag.get(session)
+ excludes = self.__excludes.get(session)
+ exchange_id = self.__exchange.get(session)
+ exchange = Exchange.get(int(exchange_id))
+ durable = exchange.durable
+
+ args = {"durable": durable,
+ "exchange": exchange.name,
+ "key": key,
+ "tag": tag,
+ "excludes": excludes
+ }
+
+ action = self.app.model.link.bridge
+ action.invoke(link, args)
+
+ self.process_cancel(session, link)
+
+
class BrokerLinkAdd(CuminFieldForm):
def __init__(self, app, name):
super(BrokerLinkAdd, self).__init__(app, name)
Modified: mgmt/trunk/cumin/python/cumin/brokerlink.strings
===================================================================
--- mgmt/trunk/cumin/python/cumin/brokerlink.strings 2008-08-13 22:20:39 UTC (rev 2300)
+++ mgmt/trunk/cumin/python/cumin/brokerlink.strings 2008-08-13 22:23:34 UTC (rev 2301)
@@ -48,6 +48,10 @@
</form>
[PeerRouteSet.html]
+ <ul class="actions">
+ <li><a class="nav" href="{add_bridge_url}">Add Route</a></li>
+ </ul>
+
<div class="rfloat">{page}</div>
<ul class="radiotabs"> </ul>
@@ -152,3 +156,15 @@
wooly.doc().elembyid("{id}").node.elements[0].focus();
</script>
+[ExchangeRadioField.exchange_html]
+ <div class="field">
+ <input id="{exchange_id}" type="radio" name="{exchange_name_path}" {checked} value="{exchange_value}"/>
+ <label for="{exchange_id}">{exchange_name}</label>
+ </div>
+
+[ExchangeRadioField.html]
+ <span class="legend">Choose an Exchange</span>
+ <fieldset>
+ {exchanges}
+ </fieldset>
+
Modified: mgmt/trunk/cumin/python/cumin/model.py
===================================================================
--- mgmt/trunk/cumin/python/cumin/model.py 2008-08-13 22:20:39 UTC (rev 2300)
+++ mgmt/trunk/cumin/python/cumin/model.py 2008-08-13 22:23:34 UTC (rev 2301)
@@ -1209,6 +1209,9 @@
action = self.Close(self, "close")
action.summary = True
+ action = self.Bridge(self, "bridge")
+ action.summary = True
+
def get_title(self, session):
return "Broker Link"
@@ -1219,6 +1222,32 @@
frame = self.cumin_model.show_main(session)
return frame.show_broker(session, link.vhost.broker.registration)
+ class Bridge(CuminAction):
+ def show(self, session, link):
+ frame = self.cumin_class.show_object(session, link)
+ frame = frame.show_peer(session, link)
+ return frame.show_bridge_add(session)
+
+ def get_title(self, session):
+ return "Add Route"
+
+ def get_verb(self, session):
+ return "Add Route"
+
+ def do_invoke(self, link, args, completion):
+ durable = args["durable"]
+ src = args["exchange"]
+ dest = args["exchange"] # see qpid-route
+ key = args["key"]
+ tag = args["tag"]
+ excludes = args["excludes"]
+ srcIsQueue = 0
+ srcIsLocal = 0
+
+ link.bridge(self.cumin_model.data, completion,
+ durable, src, dest, key,
+ tag, excludes, srcIsQueue, srcIsLocal)
+
class Close(CuminAction):
def show(self, session, link):
frame = self.cumin_class.show_object(session, link)
17 years, 8 months
- 1
- 0
rhmessaging commits: r2300 - mgmt/trunk/cumin/python/cumin.
by rhmessaging-commits＠lists.jboss.org
Author: eallen
Date: 2008-08-13 18:20:39 -0400 (Wed, 13 Aug 2008)
New Revision: 2300
Modified:
mgmt/trunk/cumin/python/cumin/widgets.py
Log:
Added generic TextField
Implemented logic to use the required attribute of NameField
Modified: mgmt/trunk/cumin/python/cumin/widgets.py
===================================================================
--- mgmt/trunk/cumin/python/cumin/widgets.py 2008-08-13 22:04:16 UTC (rev 2299)
+++ mgmt/trunk/cumin/python/cumin/widgets.py 2008-08-13 22:20:39 UTC (rev 2300)
@@ -594,10 +594,13 @@
def render_title(self, session):
return "Name"
+ def set_required(self, required):
+ self.required = required
+
def do_validate(self, session, errors):
name = self.get(session)
- if name == "":
+ if name == "" and self.required:
errors.append(MissingValueError())
else:
for char in self.illegal_chars:
@@ -610,6 +613,18 @@
break
+class TextField(NameField):
+ def __init__(self, app, name, form):
+ super(TextField, self).__init__(app, name, form)
+
+ self.__title = "Title"
+
+ def set_title(self, title):
+ self.__title = title
+
+ def render_title(self, session):
+ return self.__title
+
class ExchangeNameField(NameField):
def __init__(self, app, name, form):
super(ExchangeNameField, self).__init__(app, name, form)
17 years, 8 months
- 1
- 0
rhmessaging commits: r2299 - mgmt/trunk/cumin/python/cumin.
by rhmessaging-commits＠lists.jboss.org
Author: eallen
Date: 2008-08-13 18:04:16 -0400 (Wed, 13 Aug 2008)
New Revision: 2299
Modified:
mgmt/trunk/cumin/python/cumin/queue.py
Log:
Removed unused method
Modified: mgmt/trunk/cumin/python/cumin/queue.py
===================================================================
--- mgmt/trunk/cumin/python/cumin/queue.py 2008-08-13 20:27:30 UTC (rev 2298)
+++ mgmt/trunk/cumin/python/cumin/queue.py 2008-08-13 22:04:16 UTC (rev 2299)
@@ -566,12 +566,6 @@
queue = self.frame.get_object(session)
return "Add Binding to the Queue '%s'" % queue.name
- def find_match_value(self, binding_info, this_exchange, match_info):
- for m_info in binding_info[this_exchange]:
- if m_info.startswith(match_info):
- if m_info.endswith("nv"):
- return binding_info[this_exchange][m_info]
-
class Errors(Attribute):
def get_default(self, session):
return dict()
17 years, 8 months
- 1
- 0
rhmessaging commits: r2298 - in store/trunk/cpp: lib/jrnl and 3 other directories.
by rhmessaging-commits＠lists.jboss.org
Author: kpvdr
Date: 2008-08-13 16:27:30 -0400 (Wed, 13 Aug 2008)
New Revision: 2298
Modified:
store/trunk/cpp/lib/BdbMessageStore.cpp
store/trunk/cpp/lib/BdbMessageStore.h
store/trunk/cpp/lib/JournalImpl.cpp
store/trunk/cpp/lib/PreparedTransaction.h
store/trunk/cpp/lib/TxnCtxt.h
store/trunk/cpp/lib/jrnl/enq_map.cpp
store/trunk/cpp/lib/jrnl/enq_map.hpp
store/trunk/cpp/lib/jrnl/jcntl.cpp
store/trunk/cpp/lib/jrnl/jcntl.hpp
store/trunk/cpp/lib/jrnl/jerrno.cpp
store/trunk/cpp/lib/jrnl/jerrno.hpp
store/trunk/cpp/lib/jrnl/wmgr.cpp
store/trunk/cpp/tests/jrnl/_st_basic.cpp
store/trunk/cpp/tests/jrnl/_st_basic_txn.cpp
store/trunk/cpp/tests/jrnl/jtt/jfile_chk.py
store/trunk/cpp/tests/persistence.py
Log:
Backport from store-1.0 branch: Fix for BZ458053: r.2279 "txtest failures when broker killed during transfer phase". Modified message recovery to correctly predict outcome of to-be-rolled-forward/back transactions. Access to jcntl::_emap was required for this, so some accessers were added to class jcntl. Includes fix to python file check program jfile_chk.py which incorrectly detected owi on last-to-first file transition and message content overflowed. r.2281 (minor), r.2297 Additional fixes for BZ458053 "txtest failures when broker killed during transfer phase".
Modified: store/trunk/cpp/lib/BdbMessageStore.cpp
===================================================================
--- store/trunk/cpp/lib/BdbMessageStore.cpp 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/lib/BdbMessageStore.cpp 2008-08-13 20:27:30 UTC (rev 2298)
@@ -52,10 +52,12 @@
BdbMessageStore::TplRecoverStruct::TplRecoverStruct(const u_int64_t _rid,
const bool _deq_flag,
- const bool _commit_flag) :
+ const bool _commit_flag,
+ const bool _tpc_flag) :
rid(_rid),
deq_flag(_deq_flag),
- commit_flag(_commit_flag)
+ commit_flag(_commit_flag),
+ tpc_flag(_tpc_flag)
{}
BdbMessageStore::BdbMessageStore(const char* envpath) :
@@ -361,7 +363,7 @@
// TODO: Is this mutex necessary?
qpid::sys::Mutex::ScopedLock sl(jrnlCreateLock);
jQueue = new JournalImpl(queue.getName(), getJrnlDir(queue),
- string("JournalData"), defJournalGetEventsTimeout,
+ std::string("JournalData"), defJournalGetEventsTimeout,
defJournalFlushTimeout);
}
@@ -535,49 +537,69 @@
//recover transactions:
for (txn_list::iterator i = prepared.begin(); i != prepared.end(); i++) {
- TPCTxnCtxt* tpcc = new TPCTxnCtxt(i->xid, &messageIdSequence);
- std::auto_ptr<TPCTransactionContext> txn(tpcc);
-
- tpcc->prepare(tplStorePtr.get());
if (mgmtObject != 0) {
mgmtObject->inc_tplTransactionDepth();
mgmtObject->inc_tplTxnPrepares();
- }
+ }
+ std::string xid = i->xid;
+
// Restore data token state in TxnCtxt
- TplRecoverMapCitr citr = tplRecoverMap.find(i->xid);
+ TplRecoverMapCitr citr = tplRecoverMap.find(xid);
if (citr == tplRecoverMap.end()) THROW_STORE_EXCEPTION("XID not found in tplRecoverMap");
- tpcc->recoverDtok(citr->second.rid, i->xid);
// If a record is found that is dequeued but not committed/aborted from tplStore, then a complete() call
// was interrupted part way through committing/aborting the impacted queues. Complete this process.
- bool incomplTxnFlag = citr->second.deq_flag;
+ bool incomplTplTxnFlag = citr->second.deq_flag;
- RecoverableTransaction::shared_ptr dtx;
- if (!incomplTxnFlag) dtx = registry.recoverTransaction(i->xid, txn);
- if (i->enqueues.get()) {
- for (LockedMappings::iterator j = i->enqueues->begin(); j != i->enqueues->end(); j++) {
- tpcc->addXidRecord(queues[j->first]->getExternalQueueStore());
- if (!incomplTxnFlag) dtx->enqueue(queues[j->first], messages[j->second]);
+ if (citr->second.tpc_flag) {
+ // Dtx (2PC) transaction
+ TPCTxnCtxt* tpcc = new TPCTxnCtxt(xid, &messageIdSequence);
+ std::auto_ptr<TPCTransactionContext> txn(tpcc);
+ tpcc->recoverDtok(citr->second.rid, xid);
+ tpcc->prepare(tplStorePtr.get());
+
+ RecoverableTransaction::shared_ptr dtx;
+ if (!incomplTplTxnFlag) dtx = registry.recoverTransaction(xid, txn);
+ if (i->enqueues.get()) {
+ for (LockedMappings::iterator j = i->enqueues->begin(); j != i->enqueues->end(); j++) {
+ tpcc->addXidRecord(queues[j->first]->getExternalQueueStore());
+ if (!incomplTplTxnFlag) dtx->enqueue(queues[j->first], messages[j->second]);
+ }
}
- }
- if (i->dequeues.get()) {
- for (LockedMappings::iterator j = i->dequeues->begin(); j != i->dequeues->end(); j++) {
- tpcc->addXidRecord(queues[j->first]->getExternalQueueStore());
- if (!incomplTxnFlag) dtx->dequeue(queues[j->first], messages[j->second]);
+ if (i->dequeues.get()) {
+ for (LockedMappings::iterator j = i->dequeues->begin(); j != i->dequeues->end(); j++) {
+ tpcc->addXidRecord(queues[j->first]->getExternalQueueStore());
+ if (!incomplTplTxnFlag) dtx->dequeue(queues[j->first], messages[j->second]);
+ }
}
- }
- if (incomplTxnFlag) {
- tpcc->complete(citr->second.commit_flag);
+ if (incomplTplTxnFlag) {
+ tpcc->complete(citr->second.commit_flag);
+ }
} else {
- if (mgmtObject != 0) {
- mgmtObject->inc_tplTransactionDepth();
- mgmtObject->inc_tplTxnPrepares();
- }
+ // Local (1PC) transaction
+ boost::shared_ptr<TxnCtxt> opcc(new TxnCtxt(xid, &messageIdSequence));
+ opcc->recoverDtok(citr->second.rid, xid);
+ opcc->prepare(tplStorePtr.get());
+
+ if (i->enqueues.get()) {
+ for (LockedMappings::iterator j = i->enqueues->begin(); j != i->enqueues->end(); j++) {
+ opcc->addXidRecord(queues[j->first]->getExternalQueueStore());
+ }
+ }
+ if (i->dequeues.get()) {
+ for (LockedMappings::iterator j = i->dequeues->begin(); j != i->dequeues->end(); j++) {
+ opcc->addXidRecord(queues[j->first]->getExternalQueueStore());
+ }
+ }
+ if (incomplTplTxnFlag) {
+ opcc->complete(citr->second.commit_flag);
+ } else {
+ completed(*opcc.get(), citr->second.commit_flag);
+ }
}
}
-
registry.recoveryComplete();
}
@@ -606,7 +628,7 @@
JournalImpl* jQueue = 0;
{
qpid::sys::Mutex::ScopedLock sl(jrnlCreateLock);
- jQueue = new JournalImpl(queueName, getJrnlDir(queueName), string("JournalData"), defJournalGetEventsTimeout, defJournalFlushTimeout);
+ jQueue = new JournalImpl(queueName, getJrnlDir(queueName), std::string("JournalData"), defJournalGetEventsTimeout, defJournalFlushTimeout);
}
queue->setExternalQueueStore(dynamic_cast<ExternalQueueStore*>(jQueue));
@@ -675,8 +697,8 @@
THROW_STORE_EXCEPTION("Not enough data for binding");
}
uint64_t queueId = buffer.getLongLong();
- string queueName;
- string routingkey;
+ std::string queueName;
+ std::string routingkey;
FieldTable args;
buffer.getShortString(queueName);
buffer.getShortString(routingkey);
@@ -774,15 +796,39 @@
}
PreparedTransaction::list::iterator i = PreparedTransaction::getLockedPreparedTransaction(prepared, queue->getPersistenceId(), dtokp.rid());
- if (i == prepared.end()) { // not locked
+ if (i == prepared.end()) { // not in prepared list
queue->recover(msg);
} else {
- TplRecoverMapCitr citr = tplRecoverMap.find(i->xid);
+ u_int64_t rid = dtokp.rid();
+ std::string xid(i->xid);
+ TplRecoverMapCitr citr = tplRecoverMap.find(xid);
if (citr == tplRecoverMap.end()) THROW_STORE_EXCEPTION("XID not found in tplRecoverMap");
- if (citr->second.deq_flag) { // deq present in prepared list, this xid is part of incomplete txn commit/abort
- if (citr->second.commit_flag) queue->recover(msg); // treat as non-tx, roll forward (else aborted, throw away)
+
+ // deq present in prepared list, this xid is part of incomplete txn commit/abort
+ // or this is a 1PC txn that must be rolled forward
+ if (citr->second.deq_flag || !citr->second.tpc_flag) {
+ if (jc->is_enqueued(rid, true)) {
+ // Enqueue is non-tx, dequeue tx
+ assert(jc->is_locked(rid)); // This record MUST be locked by a txn dequeue
+ if (!citr->second.commit_flag) {
+ queue->recover(msg); // recover message in abort case only
+ }
+ } else {
+ // Enqueue and/or dequeue tx
+ journal::txn_map& tmap = jc->get_txn_map();
+ journal::txn_data_list txnList = tmap.get_tdata_list(xid);
+ bool enq = false;
+ bool deq = false;
+ for (journal::tdl_itr j = txnList.begin(); j<txnList.end(); j++) {
+ if (j->_enq_flag && j->_rid == rid) enq = true;
+ else if (!j->_enq_flag && j->_drid == rid) deq = true;
+ }
+ if (enq && !deq && citr->second.commit_flag) {
+ queue->recover(msg); // recover txn message in commit case only
+ }
+ }
} else {
- messages[dtokp.rid()] = msg;
+ messages[rid] = msg;
}
}
@@ -845,7 +891,7 @@
txn.commit();
} catch (const DbException& e) {
txn.abort();
- THROW_STORE_EXCEPTION("Unexpected BDB error in BdbMessageStore::getExternMessage(): " + string(e.what()));
+ THROW_STORE_EXCEPTION("Unexpected BDB error in BdbMessageStore::getExternMessage(): " + std::string(e.what()));
} catch (...) {
txn.abort();
throw;
@@ -884,54 +930,85 @@
return count;
}
-void BdbMessageStore::recoverTplStore()
+void BdbMessageStore::recoverTplStore(TplRecoverMap& tplMap)
{
if (journal::jdir::exists(tplStorePtr->jrnl_dir() + tplStorePtr->base_filename() + ".jinf")) {
u_int64_t thisHighestRid;
tplStorePtr->recover(tplNumJrnlFiles, tplJrnlFsizeSblks, tplWCachePgSizeSblks, tplWCacheNumPages, 0, thisHighestRid, 0);
+ journal::txn_map& tmap = tplStorePtr->get_txn_map();
if (thisHighestRid > highestRid)
highestRid = thisHighestRid;
- tplStorePtr->recover_complete(); // start journal.
- }
-}
-void BdbMessageStore::getTplRecoverMap(TplRecoverMap& tplMap)
-{
- if (tplStorePtr.get()) {
- if (!tplStorePtr->is_ready())
- recoverTplStore();
+ DataTokenImpl dtokp;
+ void* dbuff = NULL; size_t dbuffSize = 0;
+ void* xidbuff = NULL; size_t xidbuffSize = 0;
+ bool transientFlag = false;
+ bool externalFlag = false;
+ bool done = false;
+ try {
+ unsigned aio_sleep_cnt = 0;
+ while (!done) {
+ dtokp.reset();
+ dtokp.set_wstate(DataTokenImpl::ENQ);
+ switch (tplStorePtr->read_data_record(&dbuff, dbuffSize, &xidbuff, xidbuffSize, transientFlag, externalFlag, &dtokp)) {
+ case rhm::journal::RHM_IORES_SUCCESS: {
+ // Every TPL record contains both data and an XID
+ assert(dbuffSize>0);
+ assert(xidbuffSize>0);
+ std::string xid(static_cast<const char*>(xidbuff), xidbuffSize);
+ bool is2PC = *(static_cast<char*>(dbuff)) != 0;
- // TODO: The journal will return a const txn_map and the txn_map will support
- // const operations at some point. Using non-const txn_map this way is ugly...
- journal::txn_map& tmap = tplStorePtr->get_txn_map();
- std::vector<std::string> xidList;
- tmap.xid_list(xidList);
- for (std::vector<std::string>::const_iterator i = xidList.begin(); i<xidList.end(); i++) {
- journal::txn_data_list txnList = tmap.get_tdata_list(*i);
- unsigned enqCnt = 0;
- unsigned deqCnt = 0;
- u_int64_t rid = 0;
- bool commitFlag = false;
- for (journal::tdl_itr j = txnList.begin(); j<txnList.end(); j++) {
- if (j->_enq_flag) {
- rid = j->_rid;
- enqCnt++;
- } else {
- commitFlag = j->_commit_flag;
- deqCnt++;
- }
+ // Check transaction details; add to recover map
+ journal::txn_data_list txnList = tmap.get_tdata_list(xid);
+ unsigned enqCnt = 0;
+ unsigned deqCnt = 0;
+ u_int64_t rid = 0;
+
+ // Assume commit (roll forward) in cases where only prepare has been called - ie only enqueue record exists.
+ // Note: will apply to both 1PC and 2PC transactions.
+ bool commitFlag = true;
+
+ for (journal::tdl_itr j = txnList.begin(); j<txnList.end(); j++) {
+ if (j->_enq_flag) {
+ rid = j->_rid;
+ enqCnt++;
+ } else {
+ commitFlag = j->_commit_flag;
+ deqCnt++;
+ }
+ }
+ assert(enqCnt == 1);
+ assert(deqCnt <= 1);
+ tplMap.insert(TplRecoverMapPair(xid, TplRecoverStruct(rid, deqCnt == 1, commitFlag, is2PC)));
+
+ ::free(xidbuff);
+ aio_sleep_cnt = 0;
+ break;
+ }
+ case rhm::journal::RHM_IORES_PAGE_AIOWAIT:
+ if (++aio_sleep_cnt > MAX_AIO_SLEEPS)
+ THROW_STORE_EXCEPTION("Timeout waiting for AIO in BdbMessageStore::recoverTplStore()");
+ ::usleep(AIO_SLEEP_TIME);
+ break;
+ case rhm::journal::RHM_IORES_EMPTY:
+ done = true;
+ break; // done with all messages. (add call in jrnl to test that _emap is empty.)
+ default:
+ assert("Store Error: Unexpected msg state");
+ } // switch
}
- assert(enqCnt == 1);
- assert(deqCnt <= 1);
- tplMap.insert(TplRecoverMapPair(*i, TplRecoverStruct(rid, deqCnt == 1, commitFlag)));
+ } catch (const journal::jexception& e) {
+ THROW_STORE_EXCEPTION(std::string("TPL recoverTplStore() failed: ") + e.what());
}
+
+ tplStorePtr->recover_complete(); // start journal.
}
}
void BdbMessageStore::recoverLockedMappings(txn_list& txns)
{
if (!tplStorePtr->is_ready())
- getTplRecoverMap(tplRecoverMap);
+ recoverTplStore(tplRecoverMap);
// Abort unprepaired xids and populate the locked maps
for (TplRecoverMapCitr i = tplRecoverMap.begin(); i != tplRecoverMap.end(); i++) {
@@ -943,12 +1020,13 @@
}
}
-void BdbMessageStore::collectPreparedXids(std::set<string>& xids)
+void BdbMessageStore::collectPreparedXids(std::set<std::string>& xids)
{
if (!tplStorePtr->is_ready())
- getTplRecoverMap(tplRecoverMap);
+ recoverTplStore(tplRecoverMap);
for (TplRecoverMapCitr i = tplRecoverMap.begin(); i != tplRecoverMap.end(); i++) {
- if (!i->second.deq_flag) // Discard all txns that are to be rolled forward/back
+ // Discard all txns that are to be rolled forward/back and 1PC transactions
+ if (!i->second.deq_flag && i->second.tpc_flag)
xids.insert(i->first);
}
}
@@ -1021,7 +1099,7 @@
try {
int status = db.get(txn, &key, &peek, 0);
if (status != DB_BUFFER_SMALL) {
- THROW_STORE_EXCEPTION("Unexpected status code when determining record length: " + string(DbEnv::strerror(status)));
+ THROW_STORE_EXCEPTION("Unexpected status code when determining record length: " + std::string(DbEnv::strerror(status)));
}
} catch (const DbMemoryException& expected) {
//api doc indicates may throw exception instead of status = DB_BUFFER_SMALL;
@@ -1257,7 +1335,7 @@
ddtokp->set_rid(messageIdSequence.next());
ddtokp->set_dequeue_rid(msg->getPersistenceId());
ddtokp->set_wstate(DataTokenImpl::ENQ);
- string tid;
+ std::string tid;
if (ctxt) {
TxnCtxt* txn = check(ctxt);
tid = txn->getXid();
@@ -1369,11 +1447,16 @@
try {
chkTplStoreInit(); // Late initialize (if needed)
+ // This sync is requred to ensure multi-queue atomicity - ie all txn data
+ // must hit the disk on *all* queues before the TPL prepare (enq) is written.
+ ctxt->sync();
+
ctxt->incrDtokRef();
DataTokenImpl* dtokp = ctxt->getDtok();
dtokp->set_external_rid(true);
dtokp->set_rid(messageIdSequence.next());
- tplStorePtr->enqueue_txn_data_record(0, 0, 0, dtokp, ctxt->getXid(), false);
+ char tpcFlag = static_cast<char>(ctxt->isTPC());
+ tplStorePtr->enqueue_txn_data_record(&tpcFlag, sizeof(char), sizeof(char), dtokp, ctxt->getXid(), false);
ctxt->prepare(tplStorePtr.get());
// make sure all the data is written to disk before returning
ctxt->sync();
@@ -1447,7 +1530,7 @@
} else if (status == DB_NOTFOUND) {
return false;
} else {
- THROW_STORE_EXCEPTION("Deletion failed: " + string(DbEnv::strerror(status)));
+ THROW_STORE_EXCEPTION("Deletion failed: " + std::string(DbEnv::strerror(status)));
}
}
@@ -1527,33 +1610,33 @@
}
}
-string BdbMessageStore::getJrnlBaseDir()
+std::string BdbMessageStore::getJrnlBaseDir()
{
std::stringstream dir;
dir << storeDir << "/rhm/jrnl/" ;
return dir.str();
}
-string BdbMessageStore::getBdbBaseDir()
+std::string BdbMessageStore::getBdbBaseDir()
{
std::stringstream dir;
dir << storeDir << "/rhm/dat/" ;
return dir.str();
}
-string BdbMessageStore::getTplBaseDir()
+std::string BdbMessageStore::getTplBaseDir()
{
std::stringstream dir;
dir << storeDir << "/rhm/tpl/" ;
return dir.str();
}
-string BdbMessageStore::getJrnlDir(const qpid::broker::PersistableQueue& queue) //for exmaple /var/rhm/ + queueDir/
+std::string BdbMessageStore::getJrnlDir(const qpid::broker::PersistableQueue& queue) //for exmaple /var/rhm/ + queueDir/
{
return getJrnlDir(queue.getName().c_str());
}
-string BdbMessageStore::getJrnlDir(const char* queueName) //for exmaple /var/rhm/ + queueDir/
+std::string BdbMessageStore::getJrnlDir(const char* queueName) //for exmaple /var/rhm/ + queueDir/
{
std::stringstream dir;
dir << getJrnlBaseDir() << std::hex << std::setfill('0') << std::setw(4);
Modified: store/trunk/cpp/lib/BdbMessageStore.h
===================================================================
--- store/trunk/cpp/lib/BdbMessageStore.h 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/lib/BdbMessageStore.h 2008-08-13 20:27:30 UTC (rev 2298)
@@ -63,10 +63,11 @@
// Structs for Transaction Recover List (TPL) recover state
struct TplRecoverStruct {
- u_int64_t rid;
+ u_int64_t rid; // rid of TPL record
bool deq_flag;
bool commit_flag;
- TplRecoverStruct(const u_int64_t _rid, const bool _deq_flag, const bool _commit_flag);
+ bool tpc_flag;
+ TplRecoverStruct(const u_int64_t _rid, const bool _deq_flag, const bool _commit_flag, const bool _tpc_flag);
};
typedef TplRecoverStruct TplRecover;
typedef std::pair<std::string, TplRecover> TplRecoverMapPair;
@@ -157,8 +158,7 @@
queue_index& index,
txn_list& locked,
message_index& prepared);
- void recoverTplStore();
- void getTplRecoverMap(TplRecoverMap& tplMap);
+ void recoverTplStore(TplRecoverMap& tplMap);
void recoverLockedMappings(txn_list& txns);
TxnCtxt* check(qpid::broker::TransactionContext* ctxt);
void store(const qpid::broker::PersistableQueue* queue,
Modified: store/trunk/cpp/lib/JournalImpl.cpp
===================================================================
--- store/trunk/cpp/lib/JournalImpl.cpp 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/lib/JournalImpl.cpp 2008-08-13 20:27:30 UTC (rev 2298)
@@ -214,7 +214,7 @@
}
}
std::ostringstream oss2;
- oss2 << "Recover phase I complete; highest rid found = 0x" << std::hex << highest_rid;
+ oss2 << "Recover phase 1 complete; highest rid found = 0x" << std::hex << highest_rid;
oss2 << std::dec << "; emap.size=" << _emap.size() << "; tmap.size=" << _tmap.size();
oss2 << "; journal now read-only.";
log(LOG_DEBUG, oss2.str());
@@ -232,7 +232,7 @@
JournalImpl::recover_complete()
{
jcntl::recover_complete();
- log(LOG_DEBUG, "Recover phase II complete; journal now writable.");
+ log(LOG_DEBUG, "Recover phase 2 complete; journal now writable.");
}
#define MAX_AIO_SLEEPS 1000 // 10 sec
Modified: store/trunk/cpp/lib/PreparedTransaction.h
===================================================================
--- store/trunk/cpp/lib/PreparedTransaction.h 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/lib/PreparedTransaction.h 2008-08-13 20:27:30 UTC (rev 2298)
@@ -47,6 +47,7 @@
void add(queue_id queue, message_id message);
bool isLocked(queue_id queue, message_id message);
+ std::size_t size() { return locked.size(); }
iterator begin() { return locked.begin(); }
iterator end() { return locked.end(); }
Modified: store/trunk/cpp/lib/TxnCtxt.h
===================================================================
--- store/trunk/cpp/lib/TxnCtxt.h 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/lib/TxnCtxt.h 2008-08-13 20:27:30 UTC (rev 2298)
@@ -86,7 +86,7 @@
try {
if (commit) {
jc->txn_commit(dtokp.get(), getXid());
- if (isTPC()) sync();
+ sync();
} else {
jc->txn_abort(dtokp.get(), getXid());
}
@@ -106,6 +106,8 @@
}
}
+ TxnCtxt(std::string _tid, IdSequence* _loggedtx) : loggedtx(_loggedtx), dtokp(new DataTokenImpl), preparedXidStorePtr(0), tid(_tid), txn(0) {}
+
/**
* Call to make sure all the data for this txn is written to safe store
*
Modified: store/trunk/cpp/lib/jrnl/enq_map.cpp
===================================================================
--- store/trunk/cpp/lib/jrnl/enq_map.cpp 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/lib/jrnl/enq_map.cpp 2008-08-13 20:27:30 UTC (rev 2298)
@@ -130,16 +130,13 @@
}
bool
-enq_map::is_enqueued(const u_int64_t rid)
+enq_map::is_enqueued(const u_int64_t rid, bool ignore_lock)
{
- emap_itr itr;
- {
- slock s(&_mutex);
- itr = _map.find(rid);
- }
+ slock s(&_mutex);
+ emap_itr itr = _map.find(rid);
if (itr == _map.end()) // not found in map
return false;
- if (itr->second.second) // locked
+ if (!ignore_lock && itr->second.second) // locked
return false;
return true;
}
Modified: store/trunk/cpp/lib/jrnl/enq_map.hpp
===================================================================
--- store/trunk/cpp/lib/jrnl/enq_map.hpp 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/lib/jrnl/enq_map.hpp 2008-08-13 20:27:30 UTC (rev 2298)
@@ -95,7 +95,7 @@
void insert_fid(const u_int64_t rid, const u_int16_t fid, const bool locked);
u_int16_t get_fid(const u_int64_t rid);
u_int16_t get_remove_fid(const u_int64_t rid, const bool txn_flag = false);
- bool is_enqueued(const u_int64_t rid);
+ bool is_enqueued(const u_int64_t rid, bool ignore_lock = false);
void lock(const u_int64_t rid);
void unlock(const u_int64_t rid);
bool is_locked(const u_int64_t rid);
Modified: store/trunk/cpp/lib/jrnl/jcntl.cpp
===================================================================
--- store/trunk/cpp/lib/jrnl/jcntl.cpp 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/lib/jrnl/jcntl.cpp 2008-08-13 20:27:30 UTC (rev 2298)
@@ -634,8 +634,20 @@
for (std::vector<std::string>::iterator itr = xid_list.begin(); itr != xid_list.end(); itr++)
{
std::vector<std::string>::const_iterator pitr = std::find(prep_txn_list_ptr->begin(), prep_txn_list_ptr->end(), *itr);
- if (pitr == prep_txn_list_ptr->end())
- _tmap.get_remove_tdata_list(*itr);
+ if (pitr == prep_txn_list_ptr->end()) // not found in prepared list
+ {
+ txn_data_list tdl = _tmap.get_remove_tdata_list(*itr);
+ // Unlock any affected enqueues in emap
+ for (tdl_itr i=tdl.begin(); i<tdl.end(); i++)
+ {
+ if (i->_enq_flag) // enq op - decrement enqueue count
+ rd._enq_cnt_list[i->_fid]--;
+ else if (_emap.is_enqueued(i->_drid, true)) // deq op - unlock enq record
+ _emap.unlock(i->_drid);
+ }
+ // Write abort record to disk
+ rcvr_write_abort(rd, *itr);
+ }
}
}
}
@@ -951,7 +963,7 @@
ofsp.write((const char*)buff, JRNL_DBLK_SIZE);
assert(!ofsp.fail());
std::ostringstream oss;
- oss << "Wrote filler record at offs=0x" << std::hex << file_pos << std::dec;
+ oss << "Recover phase write: Wrote filler record at offs=0x" << std::hex << file_pos << std::dec;
this->log(LOG_NOTICE, oss.str());
file_pos = ofsp.tellp();
}
@@ -961,5 +973,99 @@
}
}
+// TODO - FIXME - Unify the recover and normal aio write methods.
+// Normally, writes are not performed during recover mode (journal is in read-only
+// mode) so initialization of the aio write controllers is deferred until recover
+// is complete. Currenlty because journal is still in recover mode when
+// rcvr_write_abort() is called, normal writes are not possible, so std::ofstream
+// writes are used instead. Lots of logic duplication!
+void
+jcntl::rcvr_write_abort(rcvdat& rd, std::string xid)
+{
+ const u_int32_t sblk_size = JRNL_DBLK_SIZE * JRNL_SBLK_SIZE;
+ const u_int32_t amagic = RHM_JDAT_TXA_MAGIC;
+ const u_int32_t xmagic = RHM_JDAT_EMPTY_MAGIC;
+
+ // Check last record ends on sblk boundary
+ assert(rd._eo % sblk_size == 0);
+
+ // Find fid and posn to write
+ u_int16_t fid = rd._lfid;
+ std::streampos file_pos = rd._eo;
+ if (rd._eo/sblk_size >= _jfsize_sblks) // file full, use next file
+ {
+ if (++fid >= _num_jfiles)
+ {
+ fid = 0;
+ rd._owi = !rd._owi;
+ }
+ file_pos = 0;
+ }
+
+ // Prepare a buffer of at least 1 sblock
+ u_int32_t abort_dblks = txn_rec::size_dblks(sizeof(txn_hdr) + xid.size());
+ std::size_t buffsize = abort_dblks < JRNL_SBLK_SIZE ? JRNL_SBLK_SIZE : abort_dblks;
+ void* buff = std::malloc(buffsize * JRNL_DBLK_SIZE);
+ assert(buff != 0);
+
+ // Initialize file stream
+ std::ostringstream fn;
+ fn << _jdir.dirname() << "/" << _base_filename << ".";
+ fn << std::hex << std::setfill('0') << std::setw(4) << fid << "." << JRNL_DATA_EXTENSION;
+ std::ofstream ofs(fn.str().c_str(),
+ std::ios_base::in | std::ios_base::out | std::ios_base::binary);
+ if (!ofs.good())
+ throw jexception(jerrno::JERR__FILEIO, fn.str(), "jcntl", "rcvr_write_abort");
+ if (file_pos)
+ ofs.seekp(file_pos);
+ else
+ {
+ // New file, write new file header
+ file_hdr fh(RHM_JDAT_FILE_MAGIC, RHM_JDAT_VERSION, ++rd._h_rid, fid, sblk_size, rd._owi, true);
+ ofs.write((const char*)&fh, sizeof(file_hdr));
+ assert(!ofs.fail());
+ // fill remainder of sblk with fill char
+ std::memset((char*)buff, RHM_CLEAN_CHAR, sblk_size - sizeof(file_hdr));
+ ofs.write((const char*)buff, sblk_size - sizeof(file_hdr));
+ assert(!ofs.fail());
+ // log write action
+ std::ostringstream oss;
+ oss << "Recover phase write: File header in fid=" << fid << " at offs=0x0 for txn abort record";
+ this->log(LOG_NOTICE, oss.str());
+ file_pos = ofs.tellp();
+ }
+
+ // Write abort record
+ txn_rec ar(amagic, ++rd._h_rid, xid.data(), xid.size(), rd._owi);
+ u_int32_t res = ar.encode(buff, 0, abort_dblks);
+ assert(res == abort_dblks);
+ ofs.write((const char*)buff, JRNL_DBLK_SIZE);
+ assert(!ofs.fail());
+ // log write action
+ std::ostringstream oss;
+ oss << "Recover phase write: Aborted unprepared transaction xid=" << xid << " at offs=0x" << std::hex << file_pos << std::dec;
+ this->log(LOG_NOTICE, oss.str());
+ file_pos = ofs.tellp();
+
+ // Prepare filler record
+ std::memcpy(buff, (void*)&xmagic, sizeof(xmagic));
+ std::memset((char*)buff + sizeof(xmagic), RHM_CLEAN_CHAR, JRNL_DBLK_SIZE - sizeof(xmagic));
+ // Write filler as many times as needed to get to next sblk boundary
+ while (file_pos % sblk_size)
+ {
+ ofs.write((const char*)buff, JRNL_DBLK_SIZE);
+ assert(!ofs.fail());
+ std::ostringstream oss;
+ oss << "Recover phase write: Wrote filler record at offs=0x" << std::hex << file_pos << std::dec;
+ this->log(LOG_NOTICE, oss.str());
+ file_pos = ofs.tellp();
+ }
+ rd._eo = file_pos;
+
+ // Clean up
+ ofs.close();
+ std::free(buff);
+}
+
} // namespace journal
} // namespace rhm
Modified: store/trunk/cpp/lib/jrnl/jcntl.hpp
===================================================================
--- store/trunk/cpp/lib/jrnl/jcntl.hpp 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/lib/jrnl/jcntl.hpp 2008-08-13 20:27:30 UTC (rev 2298)
@@ -557,7 +557,15 @@
* false if the rid is transactionally enqueued and is not committed, or if it is
* locked (i.e. transactionally dequeued, but the dequeue has not been committed).
*/
- inline bool is_enqueued(const u_int64_t rid) { return _emap.is_enqueued(rid); }
+ inline bool is_enqueued(const u_int64_t rid, bool ignore_lock = false)
+ { return _emap.is_enqueued(rid, ignore_lock); }
+ inline bool is_locked(const u_int64_t rid)
+ { if (_emap.is_enqueued(rid, true)) return _emap.is_locked(rid); return false; }
+ inline void enq_rid_list(std::vector<u_int64_t>& rids) { _emap.rid_list(rids); }
+ inline void enq_xid_list(std::vector<std::string>& xids) { _tmap.xid_list(xids); }
+ inline u_int32_t get_open_txn_cnt() const { return _tmap.size(); }
+ // TODO Make this a const, but txn_map must support const first.
+ inline txn_map& get_txn_map() { return _tmap; }
/**
* \brief Check if the journal is stopped.
@@ -606,10 +614,6 @@
inline u_int32_t jfsize_sblks() const { return _jfsize_sblks; }
- inline u_int32_t get_open_txn_cnt() const { return _tmap.size(); }
- // TODO Make this a const, but txn_map must support const first.
- inline txn_map& get_txn_map() { return _tmap; }
-
// Logging
virtual void log(log_level level, const std::string& log_stmt) const;
virtual void log(log_level level, const char* const log_stmt) const;
@@ -668,6 +672,8 @@
std::streampos& read_pos);
void check_journal_alignment(const u_int16_t fid, std::streampos& rec_offset);
+
+ void rcvr_write_abort(rcvdat& rd, std::string xid);
};
} // namespace journal
Modified: store/trunk/cpp/lib/jrnl/jerrno.cpp
===================================================================
--- store/trunk/cpp/lib/jrnl/jerrno.cpp 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/lib/jrnl/jerrno.cpp 2008-08-13 20:27:30 UTC (rev 2298)
@@ -94,6 +94,7 @@
const u_int32_t jerrno::JERR_WMGR_BADDTOKSTATE = 0x0802;
const u_int32_t jerrno::JERR_WMGR_ENQDISCONT = 0x0803;
const u_int32_t jerrno::JERR_WMGR_DEQDISCONT = 0x0804;
+const u_int32_t jerrno::JERR_WMGR_DEQRIDNOTENQ = 0x0805;
// class rmgr
const u_int32_t jerrno::JERR_RMGR_UNKNOWNMAGIC = 0x0900;
@@ -187,6 +188,7 @@
"Enqueued new dtok when previous enqueue returned partly completed (state ENQ_PART).";
_err_map[JERR_WMGR_DEQDISCONT] = "JERR_WMGR_DEQDISCONT: "
"Dequeued new dtok when previous dequeue returned partly completed (state DEQ_PART).";
+ _err_map[JERR_WMGR_DEQRIDNOTENQ] = "JERR_WMGR_DEQRIDNOTENQ: Dequeue rid is not enqueued.";
// class rmgr
_err_map[JERR_RMGR_UNKNOWNMAGIC] = "JERR_RMGR_UNKNOWNMAGIC: Found record with unknown magic.";
Modified: store/trunk/cpp/lib/jrnl/jerrno.hpp
===================================================================
--- store/trunk/cpp/lib/jrnl/jerrno.hpp 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/lib/jrnl/jerrno.hpp 2008-08-13 20:27:30 UTC (rev 2298)
@@ -112,6 +112,7 @@
static const u_int32_t JERR_WMGR_BADDTOKSTATE; ///< Data token in illegal state.
static const u_int32_t JERR_WMGR_ENQDISCONT; ///< Enq. new dtok when previous part compl.
static const u_int32_t JERR_WMGR_DEQDISCONT; ///< Deq. new dtok when previous part compl.
+ static const u_int32_t JERR_WMGR_DEQRIDNOTENQ; ///< Deq. rid not enqueued
// class rmgr
static const u_int32_t JERR_RMGR_UNKNOWNMAGIC; ///< Found record with unknown magic
Modified: store/trunk/cpp/lib/jrnl/wmgr.cpp
===================================================================
--- store/trunk/cpp/lib/jrnl/wmgr.cpp 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/lib/jrnl/wmgr.cpp 2008-08-13 20:27:30 UTC (rev 2298)
@@ -1000,13 +1000,34 @@
wmgr::dequeue_check(const std::string& xid, const u_int64_t drid)
{
// First check emap
- try { _emap.get_fid(drid); }
+ bool found = false;
+ try
+ {
+ _emap.get_fid(drid);
+ found = true;
+ }
catch(const jexception& e)
{
if (e.err_code() != jerrno::JERR_MAP_NOTFOUND)
throw;
- _tmap.get_data(xid, drid); // not in emap, try tmap
+ if (xid.size())
+ try
+ {
+ _tmap.get_data(xid, drid); // not in emap, try tmap
+ found = true;
+ }
+ catch (const jexception& e)
+ {
+ if (e.err_code() != jerrno::JERR_MAP_NOTFOUND)
+ throw;
+ }
}
+ if (!found)
+ {
+ std::ostringstream oss;
+ oss << "jrnl=" << _jc->id() << " drid=0x" << std::hex << drid;
+ throw jexception(jerrno::JERR_WMGR_DEQRIDNOTENQ, oss.str(), "wmgr", "dequeue_check");
+ }
}
void
Modified: store/trunk/cpp/tests/jrnl/_st_basic.cpp
===================================================================
--- store/trunk/cpp/tests/jrnl/_st_basic.cpp 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/tests/jrnl/_st_basic.cpp 2008-08-13 20:27:30 UTC (rev 2298)
@@ -408,7 +408,7 @@
u_int64_t rid = enq_msg(jc, 0, create_msg(msg, 0, MSG_SIZE), false);
deq_msg(jc, rid);
try{ deq_msg(jc, rid); BOOST_ERROR("Did not throw exception on second dequeue."); }
- catch (const jexception& e){ BOOST_CHECK_EQUAL(e.err_code(), jerrno::JERR_MAP_NOTFOUND); }
+ catch (const jexception& e){ BOOST_CHECK_EQUAL(e.err_code(), jerrno::JERR_WMGR_DEQRIDNOTENQ); }
rid = enq_msg(jc, 1, create_msg(msg, 1, MSG_SIZE), false);
deq_msg(jc, rid);
}
Modified: store/trunk/cpp/tests/jrnl/_st_basic_txn.cpp
===================================================================
--- store/trunk/cpp/tests/jrnl/_st_basic_txn.cpp 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/tests/jrnl/_st_basic_txn.cpp 2008-08-13 20:27:30 UTC (rev 2298)
@@ -87,9 +87,9 @@
try
{
deq_msg(jc, m);
- BOOST_ERROR("Expected dequeue to fail with exception JERR_MAP_NOTFOUND.");
+ BOOST_ERROR("Expected dequeue to fail with exception JERR_WMGR_DEQRIDNOTENQ.");
}
- catch (const jexception& e) { if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw; }
+ catch (const jexception& e) { if (e.err_code() != jerrno::JERR_WMGR_DEQRIDNOTENQ) throw; }
}
}
catch(const exception& e) { BOOST_FAIL(e.what()); }
@@ -138,9 +138,9 @@
try
{
deq_msg(jc, 3*m);
- BOOST_ERROR("Expected dequeue to fail with exception JERR_MAP_NOTFOUND.");
+ BOOST_ERROR("Expected dequeue to fail with exception JERR_WMGR_DEQRIDNOTENQ.");
}
- catch (const jexception& e) { if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw; }
+ catch (const jexception& e) { if (e.err_code() != jerrno::JERR_WMGR_DEQRIDNOTENQ) throw; }
}
}
catch(const exception& e) { BOOST_FAIL(e.what()); }
Modified: store/trunk/cpp/tests/jrnl/jtt/jfile_chk.py
===================================================================
--- store/trunk/cpp/tests/jrnl/jtt/jfile_chk.py 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/tests/jrnl/jtt/jfile_chk.py 2008-08-13 20:27:30 UTC (rev 2298)
@@ -445,6 +445,7 @@
self.first_rec = False
self.last_file = False
self.last_rid = -1
+ self.fhdr_owi_at_msg_start = None
self.proc_args(argv)
self.proc_csv()
@@ -474,6 +475,7 @@
stop = True;
else:
self.rec_cnt += 1
+ self.fhdr_owi_at_msg_start = self.fhdr.owi()
if self.first_rec:
if self.fhdr.fro != hdr.foffs:
raise Exception('File header first record offset mismatch: fro=0x%08x; rec_offs=0x%08x' % (self.fhdr.fro, hdr.foffs))
@@ -577,7 +579,8 @@
for rec in self.tmap[hdr.xid]:
if isinstance(rec[1], DeqHdr):
if self.emap[rec[1].deq_rid] != None:
- self.emap[rec[1].deq_rid][2] = False # Unlock enq record
+ t = self.emap[rec[1].deq_rid]
+ self.emap[rec[1].deq_rid] = (t[0], t[1], False) # Unlock enq record
del self.tmap[hdr.xid]
if len(mismatched_rids) > 0:
warn = ' (WARNING: transactional dequeues not found in enqueue map; rids=%s)' % mismatched_rids
@@ -656,7 +659,7 @@
return self.file_num
def check_owi(self, hdr):
- return self.fhdr.owi() == hdr.owi()
+ return self.fhdr_owi_at_msg_start == hdr.owi()
def check_rid(self, hdr):
if self.last_rid != -1 and hdr.rid <= self.last_rid:
Modified: store/trunk/cpp/tests/persistence.py
===================================================================
--- store/trunk/cpp/tests/persistence.py 2008-08-13 19:24:20 UTC (rev 2297)
+++ store/trunk/cpp/tests/persistence.py 2008-08-13 20:27:30 UTC (rev 2298)
@@ -231,7 +231,7 @@
if txc.global_id not in ids:
self.fail("Recovered xids not as expected. missing: %s" % (txc))
if txd.global_id not in ids:
- self.fail("Recovered xids not as expected. missing: %s" % (txc))
+ self.fail("Recovered xids not as expected. missing: %s" % (txd))
self.assertEqual(2, len(xids))
17 years, 8 months
- 1
- 0