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

Author: cctrieloff Date: 2007-10-16 08:29:00 -0400 (Tue, 16 Oct 2007) New Revision: 1079 Modified: store/trunk/cpp/lib/BdbMessageStore.cpp store/trunk/cpp/lib/BdbMessageStore.h store/trunk/cpp/lib/JournalImpl.cpp store/trunk/cpp/lib/TxnCtxt.h store/trunk/cpp/lib/jrnl/jcntl.cpp store/trunk/cpp/lib/jrnl/rmgr.cpp store/trunk/cpp/lib/jrnl/txn_map.cpp store/trunk/cpp/lib/jrnl/txn_map.hpp store/trunk/cpp/lib/jrnl/wmgr.cpp store/trunk/cpp/tests/TwoPhaseCommitTest.cpp Log: - Txn fixes - TPC fixes - All unit tests now pass - still has sys tests issues, so jrnl not enabled yet Modified: store/trunk/cpp/lib/BdbMessageStore.cpp =================================================================== --- store/trunk/cpp/lib/BdbMessageStore.cpp 2007-10-15 21:13:51 UTC (rev 1078) +++ store/trunk/cpp/lib/BdbMessageStore.cpp 2007-10-16 12:29:00 UTC (rev 1079) @@ -262,15 +262,18 @@ } //recover transactions: for (txn_list::iterator i = prepared.begin(); i != prepared.end(); i++) { - RecoverableTransaction::shared_ptr dtx = - registry.recoverTransaction(i->xid, std::auto_ptr<TPCTransactionContext>(new TPCTxnCtxt(i->xid, &messageIdSequence))); + + TPCTxnCtxt* tpcc = new TPCTxnCtxt(i->xid, &messageIdSequence); + RecoverableTransaction::shared_ptr dtx = registry.recoverTransaction(i->xid, std::auto_ptr<TPCTransactionContext>(tpcc)); if (i->enqueues.get()) { for (LockedMappings::iterator j = i->enqueues->begin(); j != i->enqueues->end(); j++) { + tpcc->addXidRecord(queues[j->first]->getExternalQueueStore()); 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()); dtx->dequeue(queues[j->first], messages[j->second]); } } @@ -427,7 +430,8 @@ Buffer contentBuff(data + contentOffset, contentSize); msg->decodeContent(contentBuff); } - if (xidbuffSize > 0 && PreparedTransaction::isLocked(locked, queue->getPersistenceId(), dtokp.rid()) ) { + + if (PreparedTransaction::isLocked(locked, queue->getPersistenceId(), dtokp.rid()) ) { prepared[dtokp.rid()] = msg; } else { queue->recover(msg); @@ -450,8 +454,6 @@ break; case rhm::journal::RHM_IORES_EMPTY: read = false; - // inline const u_int32_t get_enq_cnt() const { return _emap.size(); } - assert (jc->get_enq_cnt() == msg_count); break; // done with all messages. ((add call in jrnl to test that _emap is empty. default: assert( "Store Error: Unexpected msg state"); @@ -558,7 +560,7 @@ for (std::set<string>::iterator i = known.begin(); i != known.end(); i++) { if (prepared.find(*i) == prepared.end()) { TPCTxnCtxt txn(*i, NULL); - completed(txn, dequeueXidDb, enqueueXidDb); + completed(txn, dequeueXidDb, enqueueXidDb, false); } } readLockedMappings(enqueueXidDb, enqueues); @@ -758,7 +760,7 @@ if (usingJrnl()){ // add queue* to the txn map.. - if (ctxt) txn->addXidRecord(queue); + if (ctxt) txn->addXidRecord(queue.getExternalQueueStore()); }else{ msg.enqueueComplete(); // set enqueued for ack put(mappingDb, txn->get(), key, value); @@ -876,7 +878,7 @@ if (usingJrnl()){ // add queue* to the txn map.. - if (ctxt) txn->addXidRecord(queue); + if (ctxt) txn->addXidRecord(queue.getExternalQueueStore()); async_dequeue(ctxt, msg, queue); } else if (txn->isTPC()) { @@ -997,7 +999,7 @@ } } -void BdbMessageStore::completed(TPCTxnCtxt& txn, Db& discard, Db& apply) +void BdbMessageStore::completed(TPCTxnCtxt& txn, Db& discard, Db& apply, bool commit) { if (!txn.get()) txn.begin(env); @@ -1022,7 +1024,7 @@ } prepareXidDb.del(txn.get(), &key, 0); - txn.commit(); + txn.complete(commit); } catch (std::exception& e) { std::cout << "Error completing xid " << txn.getXid() << ": " << e.what() << std::endl; txn.abort(); @@ -1061,7 +1063,7 @@ Dbt value(&dummy, sizeof(dummy)); // make sure all the data is written to disk before returning - txn->sync(); + txn->sync(); prepareXidDb.put(txn->get(), &key, &value, 0); txn->commit(); @@ -1069,14 +1071,14 @@ txn->abort(); throw e; } - } void BdbMessageStore::commit(TransactionContext& ctxt) { - TxnCtxt* txn(check(&ctxt)); +std::cout << " commit1" << std::flush; + TxnCtxt* txn(check(&ctxt)); if (txn->isTPC()) { - completed(*dynamic_cast<TPCTxnCtxt*>(txn), enqueueXidDb, dequeueXidDb); + completed(*dynamic_cast<TPCTxnCtxt*>(txn), enqueueXidDb, dequeueXidDb, true); } else { txn->commit(); } @@ -1086,7 +1088,7 @@ { TxnCtxt* txn(check(&ctxt)); if (txn->isTPC()) { - completed(*dynamic_cast<TPCTxnCtxt*>(txn), dequeueXidDb, enqueueXidDb); + completed(*dynamic_cast<TPCTxnCtxt*>(txn), dequeueXidDb, enqueueXidDb, false); } else { txn->abort(); } Modified: store/trunk/cpp/lib/BdbMessageStore.h =================================================================== --- store/trunk/cpp/lib/BdbMessageStore.h 2007-10-15 21:13:51 UTC (rev 1078) +++ store/trunk/cpp/lib/BdbMessageStore.h 2007-10-16 12:29:00 UTC (rev 1079) @@ -100,7 +100,7 @@ bool deleteIfUnused(DbTxn* txn, Dbt& messageId); void destroy(Db& db, const qpid::broker::Persistable& p); bool create(Db& db, IdSequence& seq, const qpid::broker::Persistable& p); - void completed(TPCTxnCtxt& txn, Db& discard, Db& apply); + void completed(TPCTxnCtxt& txn, Db& discard, Db& apply, bool commit); void record2pcOp(Db& db, TPCTxnCtxt& txn, u_int64_t messageId, u_int64_t queueId); u_int64_t getRecordSize(Db& db, Dbt& key); Modified: store/trunk/cpp/lib/JournalImpl.cpp =================================================================== --- store/trunk/cpp/lib/JournalImpl.cpp 2007-10-15 21:13:51 UTC (rev 1078) +++ store/trunk/cpp/lib/JournalImpl.cpp 2007-10-16 12:29:00 UTC (rev 1079) @@ -58,7 +58,7 @@ if (tdl_itr->_enq_flag) { // enqueue op i->enqueues->add(queue_id, tdl_itr->_rid); } else { // dequeue op - i->dequeues->add(queue_id, tdl_itr->_rid); + i->dequeues->add(queue_id, tdl_itr->_drid); } } } Modified: store/trunk/cpp/lib/TxnCtxt.h =================================================================== --- store/trunk/cpp/lib/TxnCtxt.h 2007-10-15 21:13:51 UTC (rev 1078) +++ store/trunk/cpp/lib/TxnCtxt.h 2007-10-16 12:29:00 UTC (rev 1079) @@ -43,8 +43,8 @@ class TxnCtxt : public qpid::broker::TransactionContext { -private: - typedef std::set<const qpid::broker::PersistableQueue*> ipqdef; +protected: + typedef std::set<qpid::broker::ExternalQueueStore*> ipqdef; ipqdef impactedQueues; // list of Queues used in the txn static unsigned int count; mutable qpid::sys::Mutex Lock; @@ -62,7 +62,7 @@ void completeTXN(bool commit){ for (TxnCtxt::ipqdef::iterator i = impactedQueues.begin(); i != impactedQueues.end(); i++) { - JournalImpl* jc = static_cast<JournalImpl*>((*i)->getExternalQueueStore()); + JournalImpl* jc = static_cast<JournalImpl*>(*i); if (jc && loggedtx) { /* if using journal */ DataTokenImpl* dtokp = new DataTokenImpl; dtokp->set_rid(loggedtx->next()); @@ -70,11 +70,12 @@ if (commit) jc->txn_commit(dtokp, getXid()); else + { jc->txn_abort(dtokp, getXid()); - + } } catch (rhm::journal::jexception& e) { std::string str; -std::cout << "Error commit" << e << std::endl; +//std::cout << "Error commit" << e << std::endl; delete dtokp; THROW_STORE_EXCEPTION("Error commit" + e.to_string(str)); } @@ -103,7 +104,7 @@ if (!firstloop) ::usleep(AIO_SLEEP_TIME); // move this into the get events call aiolib.. allWritten = true; for (TxnCtxt::ipqdef::iterator i = impactedQueues.begin(); i != impactedQueues.end(); i++) { - JournalImpl* jc = static_cast<JournalImpl*>((*i)->getExternalQueueStore()); + JournalImpl* jc = static_cast<JournalImpl*>(*i); try { @@ -116,8 +117,7 @@ } }catch (rhm::journal::jexception& e) { std::string str; -std::cout << "Error sync" << e << std::endl; - +//std::cout << " TxnCtxt: Error sync 3" << e << std::flush; THROW_STORE_EXCEPTION("Error sync" + e.to_string(str)); } } @@ -134,8 +134,8 @@ virtual const std::string& getXid() { return tid; } void deleteXidRecord(){ impactedQueues.clear(); } - void addXidRecord(const qpid::broker::PersistableQueue& queue){ - impactedQueues.insert(&queue); } + void addXidRecord(qpid::broker::ExternalQueueStore* queue){ + impactedQueues.insert(queue); } }; @@ -146,6 +146,11 @@ TPCTxnCtxt(const std::string& _xid, IdSequence* _loggedtx) : TxnCtxt(_loggedtx), xid(_xid) {} virtual bool isTPC() { return true; } virtual const std::string& getXid() { return xid; } + // commit the BDB abort, abort commit the jnrl + void commit(){ txn->commit(0); txn = 0; } + void abort(){ txn->abort(); txn = 0; } + void complete(bool commit){ + txn->commit(0); completeTXN(commit); txn = 0; } }; }} Modified: store/trunk/cpp/lib/jrnl/jcntl.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/jcntl.cpp 2007-10-15 21:13:51 UTC (rev 1078) +++ store/trunk/cpp/lib/jrnl/jcntl.cpp 2007-10-16 12:29:00 UTC (rev 1079) @@ -375,7 +375,7 @@ std::vector<std::string>::const_iterator pitr = std::find(prep_txn_list.begin(), prep_txn_list.end(), *itr); if (pitr == prep_txn_list.end()) _tmap.get_remove_tdata_list(*itr); - } + } } } @@ -392,22 +392,19 @@ { case RHM_JDAT_ENQ_MAGIC: { -std::cout << " e" << h._rid << std::flush; enq_rec er; while (!done) { -std::cout << "*" << std::flush; done = er.rcv_decode(h, ifsp, cum_size_read); jfile_cycle(fid, ifsp, rd, false); } rd._enq_cnt_list[fid]++; if (er.xid_size()) { -std::cout << "$" << std::flush; er.get_xid(&xidp); assert(xidp != NULL); std::string xid((char*)xidp, er.xid_size()); - _tmap.insert_txn_data(xid, txn_data(h._rid, fid, true)); + _tmap.insert_txn_data(xid, txn_data(h._rid, 0, fid, true)); free(xidp); } else @@ -418,24 +415,21 @@ break; case RHM_JDAT_DEQ_MAGIC: { -std::cout << " d" << h._rid << std::flush; deq_rec dr; while (!done) { -std::cout << "*" << std::flush; done = dr.rcv_decode(h, ifsp, cum_size_read); jfile_cycle(fid, ifsp, rd, false); } if (dr.xid_size()) { -std::cout << "$" << std::flush; // If the enqueue is part of a pending txn, it will not yet be in emap - try { _emap.lock(h._rid); } + try { _emap.lock(dr.deq_rid()); } catch(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()); - _tmap.insert_txn_data(xid, txn_data(dr.deq_rid(), fid, false)); + _tmap.insert_txn_data(xid, txn_data(dr.rid(), dr.deq_rid(), fid, false)); free(xidp); } else @@ -457,12 +451,10 @@ break; case RHM_JDAT_TXA_MAGIC: { -std::cout << " a" << h._rid << std::flush; txn_rec ar; while (!done) { -std::cout << "*" << std::flush; - done = ar.rcv_decode(h, ifsp, cum_size_read); + done = ar.rcv_decode(h, ifsp, cum_size_read); jfile_cycle(fid, ifsp, rd, false); } // Delete this txn from tmap, unlock any locked records in emap @@ -472,14 +464,18 @@ txn_data_list tdl = _tmap.get_remove_tdata_list(xid); for (tdl_itr itr = tdl.begin(); itr != tdl.end(); itr++) { - try { _emap.unlock(itr->_rid); } + try + { + if (!itr->_enq_flag) + _emap.unlock(itr->_drid); + } catch(jexception e) { if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw e; } if (itr->_enq_flag) - _wrfc.decr_enqcnt(itr->_fid); + rd._enq_cnt_list[fid]--; } free(xidp); if (rd._h_rid < h._rid) @@ -488,11 +484,9 @@ break; case RHM_JDAT_TXC_MAGIC: { -std::cout << " c" << h._rid << std::flush; txn_rec cr; while (!done) { -std::cout << "*" << std::flush; done = cr.rcv_decode(h, ifsp, cum_size_read); jfile_cycle(fid, ifsp, rd, false); } @@ -500,17 +494,15 @@ cr.get_xid(&xidp); assert(xidp != NULL); std::string xid((char*)xidp, cr.xid_size()); -std::cout << "@" << std::flush; txn_data_list tdl = _tmap.get_remove_tdata_list(xid); for (tdl_itr itr = tdl.begin(); itr != tdl.end(); itr++) { -std::cout << " enq_flag=" << itr->_enq_flag << std::flush; if (itr->_enq_flag) // txn enqueue _emap.insert_fid(itr->_rid, itr->_fid); else // txn dequeue { - u_int16_t fid = _emap.get_remove_fid(itr->_rid, true); - _wrfc.decr_enqcnt(fid); + u_int16_t fid = _emap.get_remove_fid(itr->_drid, true); + rd._enq_cnt_list[fid]--; } } @@ -521,18 +513,15 @@ break; case RHM_JDAT_EMPTY_MAGIC: { -std::cout << " x" << std::flush; u_int32_t rec_dblks = jrec::size_dblks(sizeof(hdr)); ifsp->ignore(rec_dblks * JRNL_DBLK_SIZE - sizeof(hdr)); } break; case 0: -std::cout << " z" << std::flush; rd._lfid = fid; rd._eo = ifsp->tellg(); return false; default: -std::cout << " ?" << std::flush; std::stringstream ss; ss << std::hex << std::setfill('0') << "Magic=0x" << std::setw(8) << h._magic; throw jexception(jerrno::JERR_JCNTL_UNKNOWNMAGIC, ss.str(), "jcntl", @@ -589,7 +578,6 @@ void jcntl::aio_wr_callback(jcntl* journal, u_int32_t num_dtoks) { - //kpvdr TODO -- this list needs to be mutexed...??? // need to delete the dtok's std::deque<rhm::journal::data_tok*> this_dtok_list(journal->_aio_wr_cmpl_dtok_list.begin(), @@ -601,28 +589,23 @@ data_tok*& dtokp = this_dtok_list.front(); if (!journal->is_stopped() && dtokp->getSourceMessage()) { - data_tok::write_state st = dtokp->wstate(); - if (st == data_tok::ENQ) - { -//std::cout << "----- enqueueComplete rid=" << dtokp->rid() << std::endl; - - dtokp->getSourceMessage()->enqueueComplete(); - /// cct --- if TPC work out what to do !!! - } - else if (dtokp->wstate() == data_tok::DEQ) + switch (dtokp->wstate()) { -//std::cout << "----- dequeueComplete rid=" << dtokp->rid() << std::endl; - - dtokp->getSourceMessage()->dequeueComplete(); - - if ( dtokp->getSourceMessage()->isDequeueComplete() ) // clear id after last dequeue - dtokp->getSourceMessage()->setPersistenceId(0); + case data_tok::ENQ: + dtokp->getSourceMessage()->enqueueComplete(); + break; + case data_tok::DEQ: + dtokp->getSourceMessage()->dequeueComplete(); + if ( dtokp->getSourceMessage()->isDequeueComplete() ) // clear id after last dequeue + dtokp->getSourceMessage()->setPersistenceId(0); + break; + default: + ; } } this_dtok_list.pop_front(); delete dtokp; } - } void Modified: store/trunk/cpp/lib/jrnl/rmgr.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/rmgr.cpp 2007-10-15 21:13:51 UTC (rev 1078) +++ store/trunk/cpp/lib/jrnl/rmgr.cpp 2007-10-16 12:29:00 UTC (rev 1079) @@ -32,6 +32,7 @@ #include <jrnl/rmgr.hpp> +#include <jrnl/jcntl.hpp> #include <assert.h> #include <cerrno> #include <sstream> @@ -287,8 +288,24 @@ } catch (jexception& e) { - if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) + if (e.err_code() == jerrno::JERR_MAP_LOCKED && !_jc->is_read_only()) throw e; + + // Ok, not in emap, now search tmap for recover + if (_jc->is_read_only()) + { + std::vector<std::string> xid_list; + _tmap.xid_list(xid_list); + for (std::vector<std::string>::iterator itr = xid_list.begin(); itr != xid_list.end() && !is_enq; itr++) + { + txn_data_list tx_list = _tmap.get_tdata_list(*itr); + for (tdl_itr ditr = tx_list.begin(); ditr != tx_list.end() && !is_enq; ditr++) + { + if (ditr->_rid == _hdr._rid) + is_enq = true; + } + } + } //std::cout << "-nf" << std::flush; } #endif @@ -299,7 +316,7 @@ // Is this locked by a pending dequeue transaction? try { - if (_emap.is_locked(_hdr._rid)) + if (_emap.is_locked(_hdr._rid) && !_jc->is_read_only()) return RHM_IORES_TXPENDING; } catch (jexception e) Modified: store/trunk/cpp/lib/jrnl/txn_map.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/txn_map.cpp 2007-10-15 21:13:51 UTC (rev 1078) +++ store/trunk/cpp/lib/jrnl/txn_map.cpp 2007-10-16 12:29:00 UTC (rev 1079) @@ -36,13 +36,17 @@ #include <sstream> #include <jrnl/jerrno.hpp> +#include <iostream> // for debug + namespace rhm { namespace journal { -txn_data_struct::txn_data_struct(const u_int64_t rid, const u_int16_t fid, const bool enq_flag): +txn_data_struct::txn_data_struct(const u_int64_t rid, const u_int64_t drid, const u_int16_t fid, + const bool enq_flag): _rid(rid), + _drid(drid), _fid(fid), _enq_flag(enq_flag), _aio_compl(false) @@ -139,10 +143,11 @@ ss << std::hex << "xid=\"" << xid << "\""; throw jexception(jerrno::JERR_MAP_NOTFOUND, ss.str(), "txn_map", "is_txn_synced"); } - txn_data_list list = itr->second; +//std::cout << " its: found XID" << std::flush; bool is_synced = true; - for (tdl_itr litr = list.begin(); litr < list.end(); litr++) + for (tdl_itr litr = itr->second.begin(); litr < itr->second.end(); litr++) { +//std::cout << " rid=" << litr->_rid << " aioc=" << litr->_aio_compl << std::flush; if (!litr->_aio_compl) { is_synced = false; @@ -154,7 +159,7 @@ } const bool -txn_map::set_aio_compl(const std::string& xid, const u_int64_t rid) +txn_map::set_aio_compl(const std::string& xid, const u_int64_t rid) throw (jexception) { bool ok = true; bool found = false; @@ -164,13 +169,19 @@ ok = false; else { - txn_data_list list = itr->second; - for (tdl_itr litr = list.begin(); litr < list.end(); litr++) +//std::cout << " sac: found XID" << std::flush; +// txn_data_list list = itr->second; + for (tdl_itr litr = itr->second.begin(); litr < itr->second.end(); litr++) { if (litr->_rid == rid) { +// txn_data_struct t(litr->_rid, litr->_drid, litr->_fid, litr->_enq_flag); +// t._aio_compl = true; +// itr->second.erase(litr); +// itr->second.push_back(t); found = true; litr->_aio_compl = true; +//std::cout << " rid=" << rid << " aioc=" << litr->_aio_compl << " ptr=" << std::flush; break; } } Modified: store/trunk/cpp/lib/jrnl/txn_map.hpp =================================================================== --- store/trunk/cpp/lib/jrnl/txn_map.hpp 2007-10-15 21:13:51 UTC (rev 1078) +++ store/trunk/cpp/lib/jrnl/txn_map.hpp 2007-10-16 12:29:00 UTC (rev 1079) @@ -54,10 +54,11 @@ struct txn_data_struct { u_int64_t _rid; ///< Record id for this operation + u_int64_t _drid; ///< Dequeue record id for this operation u_int16_t _fid; ///< File id, to be used when transferring to emap on commit bool _enq_flag; ///< If true, enq op, otherwise deq op bool _aio_compl; ///< Initially false, set to true when AIO returns - txn_data_struct(const u_int64_t rid, const u_int16_t fid, const bool enq_flag); + txn_data_struct(const u_int64_t rid, const u_int64_t drid, const u_int16_t fid, const bool enq_flag); }; typedef txn_data_struct txn_data; typedef std::vector<txn_data> txn_data_list; @@ -82,7 +83,7 @@ const txn_data_list get_remove_tdata_list(const std::string& xid) throw (jexception); const u_int32_t get_rid_count(const std::string& xid) throw (jexception); const bool is_txn_synced(const std::string& xid) throw (jexception); - const bool set_aio_compl(const std::string& xid, const u_int64_t rid); + const bool set_aio_compl(const std::string& xid, const u_int64_t rid) throw (jexception); inline void clear() { _map.clear(); } inline const bool empty() const { return _map.empty(); } inline const u_int16_t size() const { return (u_int16_t)_map.size(); } Modified: store/trunk/cpp/lib/jrnl/wmgr.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/wmgr.cpp 2007-10-15 21:13:51 UTC (rev 1078) +++ store/trunk/cpp/lib/jrnl/wmgr.cpp 2007-10-16 12:29:00 UTC (rev 1079) @@ -167,7 +167,7 @@ if (xid_len) // If part of transaction, add to transaction map { std::string xid((char*)xid_ptr, xid_len); - _tmap.insert_txn_data(xid, txn_data(rid, dtokp->fid(), true)); + _tmap.insert_txn_data(xid, txn_data(rid, 0, dtokp->fid(), true)); } else _emap.insert_fid(rid, dtokp->fid()); @@ -307,10 +307,10 @@ if (xid_len) // If part of transaction, add to transaction map { // If the enqueue is part of a pending txn, it will not yet be in emap - try { _emap.lock(rid); } + try { _emap.lock(dequeue_rid); } catch(jexception e) { if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw e; } std::string xid((char*)xid_ptr, xid_len); - _tmap.insert_txn_data(xid, txn_data(dequeue_rid, dtokp->fid(), false)); + _tmap.insert_txn_data(xid, txn_data(rid, dequeue_rid, dtokp->fid(), false)); } else { @@ -444,7 +444,11 @@ txn_data_list tdl = _tmap.get_remove_tdata_list(xid); for (tdl_itr itr = tdl.begin(); itr != tdl.end(); itr++) { - try { _emap.unlock(itr->_rid); } + try + { + if (!itr->_enq_flag) + _emap.unlock(itr->_drid); + } catch(jexception e) { if (e.err_code() != jerrno::JERR_MAP_NOTFOUND) throw e; } if (itr->_enq_flag) _wrfc.decr_enqcnt(itr->_fid); @@ -587,7 +591,7 @@ _emap.insert_fid(itr->_rid, itr->_fid); else // txn dequeue { - u_int16_t fid = _emap.get_remove_fid(itr->_rid, true); + u_int16_t fid = _emap.get_remove_fid(itr->_drid, true); _wrfc.decr_enqcnt(fid); } } Modified: store/trunk/cpp/tests/TwoPhaseCommitTest.cpp =================================================================== --- store/trunk/cpp/tests/TwoPhaseCommitTest.cpp 2007-10-15 21:13:51 UTC (rev 1078) +++ store/trunk/cpp/tests/TwoPhaseCommitTest.cpp 2007-10-16 12:29:00 UTC (rev 1079) @@ -41,7 +41,6 @@ class TwoPhaseCommitTest : public CppUnit::TestCase { CPPUNIT_TEST_SUITE(TwoPhaseCommitTest); - CPPUNIT_TEST(testCommitSwap); CPPUNIT_TEST(testPrepareAndAbortSwap); CPPUNIT_TEST(testAbortNoPrepareSwap); @@ -72,9 +71,10 @@ { TwoPhaseCommitTest* const test; const string messageId; + Message::shared_ptr msg; public: Swap(TwoPhaseCommitTest* const test_, const string& messageId_): test(test_), messageId(messageId_) {} - void init(){ test->deliver(messageId, test->queueA); } + void init(){ msg = test->deliver(messageId, test->queueA); } void run(TPCTransactionContext* txn) { test->swap(txn); } void check(bool committed) { test->swapCheck(committed, messageId); } }; @@ -82,13 +82,16 @@ class Enqueue : public Strategy { TwoPhaseCommitTest* const test; + Message::shared_ptr msg1; + Message::shared_ptr msg2; + Message::shared_ptr msg3; public: Enqueue(TwoPhaseCommitTest* const test_): test(test_) {} void init() {} void run(TPCTransactionContext* txn) { - test->enqueue(txn, "Enqueue1"); - test->enqueue(txn, "Enqueue2"); - test->enqueue(txn, "Enqueue3"); + msg1 = test->enqueue(txn, "Enqueue1"); + msg2 = test->enqueue(txn, "Enqueue2"); + msg3 = test->enqueue(txn, "Enqueue3"); } void check(bool committed) { if (committed) { @@ -103,12 +106,15 @@ class Dequeue : public Strategy { TwoPhaseCommitTest* const test; + Message::shared_ptr msg1; + Message::shared_ptr msg2; + Message::shared_ptr msg3; public: Dequeue(TwoPhaseCommitTest* const test_): test(test_) {} void init() { - test->deliver("Dequeue1", test->queueA); - test->deliver("Dequeue2", test->queueA); - test->deliver("Dequeue3", test->queueA); + msg1 = test->deliver("Dequeue1", test->queueA); + msg2 = test->deliver("Dequeue2", test->queueA); + msg3 = test->deliver("Dequeue3", test->queueA); } void run(TPCTransactionContext* txn) { test->dequeue(txn); @@ -132,7 +138,10 @@ QueueRegistry queues; Queue::shared_ptr queueA; Queue::shared_ptr queueB; - + Message::shared_ptr msg1; + Message::shared_ptr msg2; + Message::shared_ptr msg4; + public: TwoPhaseCommitTest() : nameA("queueA"), nameB("queueB") {} @@ -214,7 +223,6 @@ std::auto_ptr<TPCTransactionContext> txn(store->begin("my-xid")); swap.run(txn.get()); store->prepare(*txn); - restart(); //check that the message is not available from either queue @@ -261,29 +269,31 @@ void swap(TPCTransactionContext* txn) { - Message::shared_ptr msg = queueA->dequeue().payload;//just dequeues in memory + msg1 = queueA->dequeue().payload;//just dequeues in memory //move the message from one queue to the other as part of a //distributed transaction - queueB->enqueue(txn, msg);//note: need to enqueue it first to avoid message being deleted - queueA->dequeue(txn, msg); + queueB->enqueue(txn, msg1);//note: need to enqueue it first to avoid message being deleted + queueA->dequeue(txn, msg1); } void dequeue(TPCTransactionContext* txn) { - Message::shared_ptr msg = queueA->dequeue().payload;//just dequeues in memory - queueA->dequeue(txn, msg); + msg2 = queueA->dequeue().payload;//just dequeues in memory + queueA->dequeue(txn, msg2); } - void enqueue(TPCTransactionContext* txn, const string& msgid) + Message::shared_ptr enqueue(TPCTransactionContext* txn, const string& msgid) { Message::shared_ptr msg = createMessage(msgid); queueA->enqueue(txn, msg); + return msg; } - void deliver(const string& msgid, Queue::shared_ptr& queue) + Message::shared_ptr deliver(const string& msgid, Queue::shared_ptr& queue) { - Message::shared_ptr msg = createMessage(msgid); - queue->deliver(msg); + msg4 = createMessage(msgid); + queue->deliver(msg4); + return msg4; } void setup()