QPID-3858: WIP - fixed a few logic errors and race conditions in the consume and transactional code

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/asyncstore@1354149 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 22 insertions, 33 deletions

diff --git a/cpp/src/qpid/asyncStore/TxnHandleImpl.cpp b/cpp/src/qpid/asyncStore/TxnHandleImpl.cpp
index af3a8f01cf..c5371f161c 100644
--- a/cpp/src/qpid/asyncStore/TxnHandleImpl.cpp
+++ b/cpp/src/qpid/asyncStore/TxnHandleImpl.cpp
@@ -110,7 +110,10 @@ void
 TxnHandleImpl::createLocalXid()
 {
     uuid_t uuid;
+
+    // TODO: This call might not be thread safe - Valgrind's helgrind tool emits warnings for this:
     ::uuid_generate_random(uuid);
+
     char uuidStr[37]; // 36-char uuid + trailing '\0'
     ::uuid_unparse(uuid, uuidStr);
     m_xid.assign(uuidStr);
diff --git a/cpp/src/tests/storePerftools/asyncPerf/DeliveryRecord.cpp b/cpp/src/tests/storePerftools/asyncPerf/DeliveryRecord.cpp
index f89ba22b2a..b7250ecf40 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/DeliveryRecord.cpp
+++ b/cpp/src/tests/storePerftools/asyncPerf/DeliveryRecord.cpp
@@ -54,19 +54,6 @@ DeliveryRecord::accept()
     return isRedundant();
 }
 
-/*
-bool
-DeliveryRecord::accept(qpid::broker::TxnHandle& txn)
-{
-    if (!m_ended) {
-        m_queuedMessage.getQueue()->dequeue(txn, m_queuedMessage);
-        m_accepted = true;
-        setEnded();
-    }
-    return isRedundant();
-}
-*/
-
 bool
 DeliveryRecord::isAccepted() const
 {
@@ -102,9 +89,7 @@ DeliveryRecord::dequeue(qpid::broker::TxnHandle& txn)
 void
 DeliveryRecord::committed() const
 {
-//std::cout << "DeliveryRecord::committed()" << std::endl << std::flush;
-    m_msgConsumer.dequeueComplete();
-    //m_queuedMessage.getQueue()->dequeueCommitted(m_queuedMessage);
+    m_msgConsumer.commitComplete();
 }
 
 QueuedMessage
diff --git a/cpp/src/tests/storePerftools/asyncPerf/DeliveryRecord.h b/cpp/src/tests/storePerftools/asyncPerf/DeliveryRecord.h
index ea8eba0468..427cf846f0 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/DeliveryRecord.h
+++ b/cpp/src/tests/storePerftools/asyncPerf/DeliveryRecord.h
@@ -44,7 +44,6 @@ public:
                    bool accepted);
     virtual ~DeliveryRecord();
     bool accept();
-//    bool accept(qpid::broker::TxnHandle& txn);
     bool isAccepted() const;
     bool setEnded();
     bool isEnded() const;
diff --git a/cpp/src/tests/storePerftools/asyncPerf/MessageConsumer.cpp b/cpp/src/tests/storePerftools/asyncPerf/MessageConsumer.cpp
index 3ac6867ce1..4a2bc2bf0c 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/MessageConsumer.cpp
+++ b/cpp/src/tests/storePerftools/asyncPerf/MessageConsumer.cpp
@@ -53,17 +53,12 @@ MessageConsumer::~MessageConsumer()
 void
 MessageConsumer::record(boost::shared_ptr<DeliveryRecord> dr)
 {
-    // TODO: May need a lock?
     m_unacked.push_back(dr);
 }
 
 void
-MessageConsumer::dequeueComplete()
-{
-//std::cout << "MessageConsumer::dequeueComplete()" << std::endl << std::flush;
-    // TODO: May need a lock
-    //++m_numMsgs;
-}
+MessageConsumer::commitComplete()
+{}
 
 void*
 MessageConsumer::runConsumers()
@@ -75,8 +70,10 @@ MessageConsumer::runConsumers()
         tb = new qpid::broker::TxnBuffer(m_resultQueue);
     }
 
+    uint32_t msgsPerConsumer = m_perfTestParams.m_numEnqThreadsPerQueue * m_perfTestParams.m_numMsgs /
+                               m_perfTestParams.m_numDeqThreadsPerQueue;
     uint32_t numMsgs = 0UL;
-    while (numMsgs < m_perfTestParams.m_numMsgs) {
+    while (numMsgs < msgsPerConsumer) {
         if (m_queue->dispatch(*this)) {
             ++numMsgs;
             if (useTxns) {
@@ -101,7 +98,6 @@ MessageConsumer::runConsumers()
                     (*i)->accept();
                 }
                 m_unacked.clear();
-                //++numMsgs;
             }
         } else {
             ::usleep(1000); // TODO - replace this poller with condition variable
diff --git a/cpp/src/tests/storePerftools/asyncPerf/MessageConsumer.h b/cpp/src/tests/storePerftools/asyncPerf/MessageConsumer.h
index e733990cf7..b110520889 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/MessageConsumer.h
+++ b/cpp/src/tests/storePerftools/asyncPerf/MessageConsumer.h
@@ -52,7 +52,7 @@ public:
                     boost::shared_ptr<SimpleQueue> queue);
     virtual ~MessageConsumer();
     void record(boost::shared_ptr<DeliveryRecord> dr);
-    void dequeueComplete();
+    void commitComplete();
 
     void* runConsumers();
     static void* startConsumers(void* ptr);
diff --git a/cpp/src/tests/storePerftools/asyncPerf/PerfTest.cpp b/cpp/src/tests/storePerftools/asyncPerf/PerfTest.cpp
index e3fdd1c44d..4d145d321d 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/PerfTest.cpp
+++ b/cpp/src/tests/storePerftools/asyncPerf/PerfTest.cpp
@@ -80,16 +80,23 @@ PerfTest::run()
         tests::storePerftools::common::ScopedTimer st(m_testResult);
 
         for (uint16_t q = 0; q < m_testOpts.m_numQueues; q++) {
-            boost::shared_ptr<MessageProducer> mp(new MessageProducer(m_testOpts, m_msgData, m_store, m_resultQueue, m_queueList[q]));
-            m_producers.push_back(mp);
             for (uint16_t t = 0; t < m_testOpts.m_numEnqThreadsPerQueue; t++) { // TODO - replace with qpid threads
+                boost::shared_ptr<MessageProducer> mp(new MessageProducer(m_testOpts,
+                                                                          m_msgData,
+                                                                          m_store,
+                                                                          m_resultQueue,
+                                                                          m_queueList[q]));
+                m_producers.push_back(mp);
                 boost::shared_ptr<tests::storePerftools::common::Thread> tp(new tests::storePerftools::common::Thread(mp->startProducers,
                                                                                                                       reinterpret_cast<void*>(mp.get())));
                 threads.push_back(tp);
             }
-            boost::shared_ptr<MessageConsumer> mc(new MessageConsumer(m_testOpts, m_store, m_resultQueue, m_queueList[q]));
-            m_consumers.push_back(mc);
             for (uint16_t dt = 0; dt < m_testOpts.m_numDeqThreadsPerQueue; ++dt) { // TODO - replace with qpid threads
+                boost::shared_ptr<MessageConsumer> mc(new MessageConsumer(m_testOpts,
+                                                                          m_store,
+                                                                          m_resultQueue,
+                                                                          m_queueList[q]));
+                m_consumers.push_back(mc);
                 boost::shared_ptr<tests::storePerftools::common::Thread> tp(new tests::storePerftools::common::Thread(mc->startConsumers,
                                                                                                                       reinterpret_cast<void*>(mc.get())));
                 threads.push_back(tp);
diff --git a/cpp/src/tests/storePerftools/asyncPerf/TxnAccept.cpp b/cpp/src/tests/storePerftools/asyncPerf/TxnAccept.cpp
index 7e737ed21a..3c5b99b5d5 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/TxnAccept.cpp
+++ b/cpp/src/tests/storePerftools/asyncPerf/TxnAccept.cpp
@@ -30,7 +30,7 @@ namespace storePerftools {
 namespace asyncPerf {
 
 TxnAccept::TxnAccept(std::deque<boost::shared_ptr<DeliveryRecord> >& ops) :
-        m_ops(ops.begin(), ops.end())
+        m_ops(ops)
 {}
 
 TxnAccept::~TxnAccept()
@@ -63,7 +63,6 @@ TxnAccept::commit()  throw()
             (*i)->committed();
             (*i)->setEnded();
         }
-        //m_ops.clear();
     } catch (const std::exception& e) {
         std::cerr << "TxnAccept: Failed to commit transaction: " << e.what() << std::endl;
     } catch(...) {