Attachment 857888 Details for Bug 1060202 – Proposed patch

[patch] Proposed patch

bz1060202.patch (text/plain), 10.79 KB, created by Pavel Moravec on 2014-01-31 15:13:23 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Pavel Moravec

Created: 2014-01-31 15:13:23 UTC

Size: 10.79 KB

patch

obsolete

>Index: cpp/src/qpid/legacystore/MessageStoreImpl.cpp
>===================================================================
>--- cpp/src/qpid/legacystore/MessageStoreImpl.cpp	(revision 1562753)
>+++ cpp/src/qpid/legacystore/MessageStoreImpl.cpp	(working copy)
>@@ -713,7 +713,7 @@
>             tpcc->prepare(tplStorePtr.get());
> 
>             qpid::broker::RecoverableTransaction::shared_ptr dtx;
>-            if (!incomplTplTxnFlag) dtx = registry.recoverTransaction(xid, txn);
>+            if (!incomplTplTxnFlag) dtx = registry.recoverTransaction(xid, txn, broker->getOptions().dtxDefaultTimeout);
>             if (pt.enqueues.get()) {
>                 for (LockedMappings::iterator j = pt.enqueues->begin(); j != pt.enqueues->end(); j++) {
>                     tpcc->addXidRecord(queues[j->first]->getExternalQueueStore());
>Index: cpp/src/qpid/linearstore/MessageStoreImpl.cpp
>===================================================================
>--- cpp/src/qpid/linearstore/MessageStoreImpl.cpp	(revision 1562753)
>+++ cpp/src/qpid/linearstore/MessageStoreImpl.cpp	(working copy)
>@@ -650,7 +650,7 @@
>             tpcc->prepare(tplStorePtr.get());
> 
>             qpid::broker::RecoverableTransaction::shared_ptr dtx;
>-            if (!incomplTplTxnFlag) dtx = registry_.recoverTransaction(xid, txn);
>+            if (!incomplTplTxnFlag) dtx = registry_.recoverTransaction(xid, txn, broker->getOptions().dtxDefaultTimeout);
>             if (pt.enqueues.get()) {
>                 for (LockedMappings::iterator j = pt.enqueues->begin(); j != pt.enqueues->end(); j++) {
>                     tpcc->addXidRecord(queues[j->first]->getExternalQueueStore());
>Index: cpp/src/qpid/store/ms-clfs/MSSqlClfsProvider.cpp
>===================================================================
>--- cpp/src/qpid/store/ms-clfs/MSSqlClfsProvider.cpp	(revision 1562753)
>+++ cpp/src/qpid/store/ms-clfs/MSSqlClfsProvider.cpp	(working copy)
>@@ -1021,7 +1021,7 @@
>     for (i = preparedMap.begin(); i != preparedMap.end(); ++i) {
>         std::auto_ptr<TPCTransactionContext> ctx(new TPCTransactionContext(i->second));
>         std::auto_ptr<qpid::broker::TPCTransactionContext> brokerCtx(ctx);
>-        dtxMap[i->first] = recoverer.recoverTransaction(i->first, brokerCtx);
>+        dtxMap[i->first] = recoverer.recoverTransaction(i->first, brokerCtx, store->getBroker()->getOptions().dtxDefaultTimeout);
>     }
> }
> 
>Index: cpp/src/qpid/store/ms-sql/MSSqlProvider.cpp
>===================================================================
>--- cpp/src/qpid/store/ms-sql/MSSqlProvider.cpp	(revision 1562753)
>+++ cpp/src/qpid/store/ms-sql/MSSqlProvider.cpp	(working copy)
>@@ -1169,7 +1169,7 @@
>                                                                         *iXid));
>             tx->setPrepared();
>             std::auto_ptr<qpid::broker::TPCTransactionContext> tc(tx);
>-            dtxMap[*iXid] = recoverer.recoverTransaction(*iXid, tc);
>+            dtxMap[*iXid] = recoverer.recoverTransaction(*iXid, tc, store->getBroker()->getOptions().dtxDefaultTimeout);
>         }
>     }
>     catch(_com_error &e) {
>Index: cpp/src/qpid/broker/Broker.h
>===================================================================
>--- cpp/src/qpid/broker/Broker.h	(revision 1562753)
>+++ cpp/src/qpid/broker/Broker.h	(working copy)
>@@ -119,6 +119,7 @@
>         bool timestampRcvMsgs;
>         sys::Duration linkMaintenanceInterval;
>         sys::Duration linkHeartbeatInterval;
>+        uint32_t dtxDefaultTimeout;
>         uint32_t maxNegotiateTime;  // Max time in ms for connection with no negotiation
>         std::string fedTag;
> 
>Index: cpp/src/qpid/broker/RecoveryManagerImpl.h
>===================================================================
>--- cpp/src/qpid/broker/RecoveryManagerImpl.h	(revision 1562753)
>+++ cpp/src/qpid/broker/RecoveryManagerImpl.h	(working copy)
>@@ -52,7 +52,8 @@
>         RecoverableQueue::shared_ptr recoverQueue(framing::Buffer& buffer);
>         RecoverableMessage::shared_ptr recoverMessage(framing::Buffer& buffer);
>         RecoverableTransaction::shared_ptr recoverTransaction(const std::string& xid, 
>-                                                              std::auto_ptr<TPCTransactionContext> txn);
>+                                                              std::auto_ptr<TPCTransactionContext> txn,
>+                                                              uint32_t secs);
>         RecoverableConfig::shared_ptr recoverConfig(framing::Buffer& buffer);
>         void recoveryComplete();
>     };
>Index: cpp/src/qpid/broker/DtxManager.cpp
>===================================================================
>--- cpp/src/qpid/broker/DtxManager.cpp	(revision 1562753)
>+++ cpp/src/qpid/broker/DtxManager.cpp	(working copy)
>@@ -66,9 +66,9 @@
> 
> DtxManager::~DtxManager() {}
> 
>-void DtxManager::start(const std::string& xid, boost::intrusive_ptr<DtxBuffer> ops)
>+void DtxManager::start(const std::string& xid, boost::intrusive_ptr<DtxBuffer> ops, uint32_t secs)
> {
>-    createWork(xid)->add(ops);
>+    createWork(xid, secs)->add(ops);
> }
> 
> void DtxManager::join(const std::string& xid, boost::intrusive_ptr<DtxBuffer> ops)
>@@ -76,9 +76,9 @@
>     getWork(xid)->add(ops);
> }
> 
>-void DtxManager::recover(const std::string& xid, std::auto_ptr<TPCTransactionContext> txn, boost::intrusive_ptr<DtxBuffer> ops)
>+void DtxManager::recover(const std::string& xid, std::auto_ptr<TPCTransactionContext> txn, boost::intrusive_ptr<DtxBuffer> ops, uint32_t secs)
> {
>-    createWork(xid)->recover(txn, ops);
>+    createWork(xid, secs)->recover(txn, ops);
> }
> 
> bool DtxManager::prepare(const std::string& xid)
>@@ -143,15 +143,18 @@
>     }
> }
> 
>-DtxWorkRecord* DtxManager::createWork(const std::string& xid)
>+DtxWorkRecord* DtxManager::createWork(const std::string& xid, uint32_t secs)
> {
>     Mutex::ScopedLock locker(lock);
>     WorkMap::iterator i = work.find(xid);
>     if (i != work.end()) {
>         throw NotAllowedException(QPID_MSG("Xid " << convert(xid) << " is already known (use 'join' to add work to an existing xid)"));
>     } else {
>-        std::string ncxid = xid; // Work around const correctness problems in ptr_map.
>-        return ptr_map_ptr(work.insert(ncxid, new DtxWorkRecord(ncxid, store)).first);
>+        std::string ncxid = xid; // Work around const correctness problems in ncxid
>+        DtxWorkRecord* dtxWorkRecord = new DtxWorkRecord(xid, store);
>+        work.insert(ncxid, dtxWorkRecord);
>+        setTimeout(xid, secs);
>+        return dtxWorkRecord;
>     }
> }
> 
>Index: cpp/src/qpid/broker/RecoveryManagerImpl.cpp
>===================================================================
>--- cpp/src/qpid/broker/RecoveryManagerImpl.cpp	(revision 1562753)
>+++ cpp/src/qpid/broker/RecoveryManagerImpl.cpp	(working copy)
>@@ -127,10 +127,10 @@
> }
> 
> RecoverableTransaction::shared_ptr RecoveryManagerImpl::recoverTransaction(const std::string& xid,
>-                                                                           std::auto_ptr<TPCTransactionContext> txn)
>+                                                                           std::auto_ptr<TPCTransactionContext> txn, uint32_t secs)
> {
>     boost::intrusive_ptr<DtxBuffer> buffer(new DtxBuffer());
>-    dtxMgr.recover(xid, txn, buffer);
>+    dtxMgr.recover(xid, txn, buffer, secs);
>     return RecoverableTransaction::shared_ptr(new RecoverableTransactionImpl(buffer));
> }
> 
>Index: cpp/src/qpid/broker/Broker.cpp
>===================================================================
>--- cpp/src/qpid/broker/Broker.cpp	(revision 1562753)
>+++ cpp/src/qpid/broker/Broker.cpp	(working copy)
>@@ -148,6 +148,7 @@
>     timestampRcvMsgs(false),    // set the 0.10 timestamp delivery property
>     linkMaintenanceInterval(2*sys::TIME_SEC),
>     linkHeartbeatInterval(120*sys::TIME_SEC),
>+    dtxDefaultTimeout(3600),
>     maxNegotiateTime(10000)     // 10s
> {
>     int c = sys::SystemInfo::concurrency();
>@@ -192,6 +193,7 @@
>          "Interval to check link health and re-connect  if need be")
>         ("link-heartbeat-interval", optValue(linkHeartbeatInterval, "SECONDS"),
>          "Heartbeat interval for a federation link")
>+        ("dtx-default-timeout", optValue(dtxDefaultTimeout, "SECONDS"), "Default timeout for DTX transaction before aborting it")
>         ("max-negotiate-time", optValue(maxNegotiateTime, "MILLISECONDS"), "Maximum time a connection can take to send the initial protocol negotiation")
>         ("federation-tag", optValue(fedTag, "NAME"), "Override the federation tag")
>         ;
>Index: cpp/src/qpid/broker/DtxManager.h
>===================================================================
>--- cpp/src/qpid/broker/DtxManager.h	(revision 1562753)
>+++ cpp/src/qpid/broker/DtxManager.h	(working copy)
>@@ -46,14 +46,14 @@
>     qpid::sys::Timer* timer;
> 
>     void remove(const std::string& xid);
>-    DtxWorkRecord* createWork(const std::string& xid);
>+    DtxWorkRecord* createWork(const std::string& xid, uint32_t secs);
> 
> public:
>     DtxManager(sys::Timer&);
>     ~DtxManager();
>-    void start(const std::string& xid, boost::intrusive_ptr<DtxBuffer> work);
>+    void start(const std::string& xid, boost::intrusive_ptr<DtxBuffer> work, uint32_t secs);
>     void join(const std::string& xid, boost::intrusive_ptr<DtxBuffer> work);
>-    void recover(const std::string& xid, std::auto_ptr<TPCTransactionContext> txn, boost::intrusive_ptr<DtxBuffer> work);
>+    void recover(const std::string& xid, std::auto_ptr<TPCTransactionContext> txn, boost::intrusive_ptr<DtxBuffer> work, uint32_t secs);
>     bool prepare(const std::string& xid);
>     bool commit(const std::string& xid, bool onePhase);
>     void rollback(const std::string& xid);
>Index: cpp/src/qpid/broker/RecoveryManager.h
>===================================================================
>--- cpp/src/qpid/broker/RecoveryManager.h	(revision 1562753)
>+++ cpp/src/qpid/broker/RecoveryManager.h	(working copy)
>@@ -39,7 +39,8 @@
>     virtual RecoverableQueue::shared_ptr recoverQueue(framing::Buffer& buffer) = 0;
>     virtual RecoverableMessage::shared_ptr recoverMessage(framing::Buffer& buffer) = 0;
>     virtual RecoverableTransaction::shared_ptr recoverTransaction(const std::string& xid, 
>-                                                                  std::auto_ptr<TPCTransactionContext> txn) = 0;
>+                                                                  std::auto_ptr<TPCTransactionContext> txn,
>+                                                                  uint32_t secs) = 0;
>     virtual RecoverableConfig::shared_ptr recoverConfig(framing::Buffer& buffer) = 0;
> 
>     virtual void recoveryComplete() = 0;
>Index: cpp/src/qpid/broker/SemanticState.cpp
>===================================================================
>--- cpp/src/qpid/broker/SemanticState.cpp	(revision 1562753)
>+++ cpp/src/qpid/broker/SemanticState.cpp	(working copy)
>@@ -227,7 +227,7 @@
>     if (join) {
>         mgr.join(xid, dtxBuffer);
>     } else {
>-        mgr.start(xid, dtxBuffer);
>+        mgr.start(xid, dtxBuffer, session.getBroker().getOptions().dtxDefaultTimeout);
>     }
> }
>

Actions: View | Diff

Attachments on bug 1060202: 857827 | 857888