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

[rhmessaging-commits at lists.jboss.org]

Author: kpvdr
Date: 2008-07-23 15:00:33 -0400 (Wed, 23 Jul 2008)
New Revision: 2211

Modified:
   store/trunk/cpp/lib/BdbMessageStore.cpp
   store/trunk/cpp/lib/TxnCtxt.h
   store/trunk/cpp/lib/jrnl/jcntl.cpp
Log:
Small enhancements and optimizations for transactional performance

Modified: store/trunk/cpp/lib/BdbMessageStore.cpp
===================================================================
--- store/trunk/cpp/lib/BdbMessageStore.cpp	2008-07-23 18:46:28 UTC (rev 2210)
+++ store/trunk/cpp/lib/BdbMessageStore.cpp	2008-07-23 19:00:33 UTC (rev 2211)
@@ -536,7 +536,8 @@
         // was interrupted part way through committing/aborting the impacted queues. Complete this process.
         bool incomplTxnFlag = citr->second.deq_flag;
 
-        RecoverableTransaction::shared_ptr dtx = registry.recoverTransaction(i->xid, txn);
+        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());
@@ -915,7 +916,8 @@
     if (!tplStorePtr->is_ready())
         getTplRecoverMap(tplRecoverMap);
     for (TplRecoverMapCitr i = tplRecoverMap.begin(); i != tplRecoverMap.end(); i++) {
-        xids.insert(i->first);
+        if (!i->second.deq_flag) // Discard all txns that are to be rolled forward/back
+            xids.insert(i->first);
     }
 }
 

Modified: store/trunk/cpp/lib/TxnCtxt.h
===================================================================
--- store/trunk/cpp/lib/TxnCtxt.h	2008-07-23 18:46:28 UTC (rev 2210)
+++ store/trunk/cpp/lib/TxnCtxt.h	2008-07-23 19:00:33 UTC (rev 2211)
@@ -86,7 +86,7 @@
             try {
                 if (commit) {
                     jc->txn_commit(dtokp.get(), getXid());
-                    jc->flush(true);
+                    if (isTPC()) sync();
                 } else {
                     jc->txn_abort(dtokp.get(), getXid());
                 }
@@ -114,8 +114,8 @@
 
     virtual ~TxnCtxt() { if(txn) abort(); }
 
-#define MAX_SYNC_SLEEPS 1000 // ~1 second
-#define SYNC_SLEEP_TIME 1000 // 1 milisecond
+#define MAX_SYNC_SLEEPS 5000 // ~1 second
+#define SYNC_SLEEP_TIME 200 // 0.2 ms
 
     void sync() {
         bool allWritten = false;

Modified: store/trunk/cpp/lib/jrnl/jcntl.cpp
===================================================================
--- store/trunk/cpp/lib/jrnl/jcntl.cpp	2008-07-23 18:46:28 UTC (rev 2210)
+++ store/trunk/cpp/lib/jrnl/jcntl.cpp	2008-07-23 19:00:33 UTC (rev 2211)
@@ -510,8 +510,8 @@
     ji.write();
 }
 
-#define AIO_CMPL_SLEEP      1000    // 1 ms
-#define MAX_AIO_CMPL_SLEEPS 10000   // Total: 10 sec
+#define AIO_CMPL_SLEEP      200    // 0.2 ms
+#define MAX_AIO_CMPL_SLEEPS 50000   // Total: 10 sec
 
 void
 jcntl::aio_cmpl_wait()