[rhmessaging-commits] rhmessaging commits: r2234 - store/branches/mrg-1.0/cpp/lib.

[rhmessaging-commits at lists.jboss.org]

Author: kpvdr
Date: 2008-07-30 11:20:43 -0400 (Wed, 30 Jul 2008)
New Revision: 2234

Modified:
   store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp
   store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h
Log:
Backport of trunk r.2233: Proposed fix for BZ457239: "Qpidd sometimes deadlocks on queue_declare from Python". BDB is not always thread safe for certain operations. Added a lock on queue create (the cause of this bug), added local transaction to BdbMessageStore::getRecordSize() (used by BdbMessageStore::appendContent()) and BdbMessageStore::getExternMessage() which were found by inspection to be vulnerable for similar reasons.

Modified: store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp
===================================================================
--- store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp	2008-07-30 15:00:18 UTC (rev 2233)
+++ store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp	2008-07-30 15:20:43 UTC (rev 2234)
@@ -359,6 +359,7 @@
         THROW_STORE_EXCEPTION(std::string("Queue ") + queue.getName() + ": create() failed: " + e.what());
     }
     try {
+        qpid::sys::Mutex::ScopedLock sl(bdbQueueLock);
         if (!create(queueDb, queueIdSequence, queue)) {
             THROW_STORE_EXCEPTION("Queue already exists: " + queue.getName());
         }
@@ -776,23 +777,35 @@
                                                                  uint64_t messageId,
                                                                  unsigned& headerSize)
 {
+    RecoverableMessage::shared_ptr ret;
     Dbt key (&messageId, sizeof(messageId));
     size_t preamble_length = sizeof(u_int32_t); /*header size*/
 
     BufferValue value(preamble_length, 0);
     value.buffer.record();
-    if (messageDb.get(0, &key, &value, 0) == DB_NOTFOUND) {
-        THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!");
-    }
-    //read header only to begin with
-    headerSize = value.buffer.getLong();
 
-    BufferValue header(headerSize, preamble_length);
-    if (messageDb.get(0, &key, &header, 0) == DB_NOTFOUND) {
-        THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!");
+    TxnCtxt txn;
+    txn.begin(env);
+    try {
+        if (messageDb.get(txn.get(), &key, &value, 0) == DB_NOTFOUND) {
+            txn.abort();
+            THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!");
+        }
+        //read header only to begin with
+        headerSize = value.buffer.getLong();
+
+        BufferValue header(headerSize, preamble_length);
+        if (messageDb.get(txn.get(), &key, &header, 0) == DB_NOTFOUND) {
+            txn.abort();
+            THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!");
+        }
+        ret = recovery.recoverMessage(header.buffer);
+        txn.commit();
+    } catch (const DbException& e) {
+        txn.abort();
+        THROW_STORE_EXCEPTION("Unexpected BDB error in BdbMessageStore::getExternMessage(): " + string(e.what()));
     }
-
-    return recovery.recoverMessage(header.buffer);
+    return ret;
 }
 
 int BdbMessageStore::enqueueMessage(TxnCtxt& txn,

Modified: store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h
===================================================================
--- store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h	2008-07-30 15:00:18 UTC (rev 2233)
+++ store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h	2008-07-30 15:20:43 UTC (rev 2234)
@@ -115,6 +115,7 @@
     static qpid::sys::Duration defJournalFlushTimeout;
     qpid::management::Store::shared_ptr mgmtObject;
     qpid::sys::Mutex jrnlCreateLock;
+    qpid::sys::Mutex bdbQueueLock;
 
     // Parameter validation and calculation
     static u_int16_t chkJrnlNumFilesParam(const u_int16_t param,