Merge pull request #7 from logicalclocks/HOPSWORKS-779

[HOPSWORKS] [ePipe] Fix a race condition while reading from ndb
logicalclocks · Nov 5, 2018 · 9c4db07 · 9c4db07
2 parents e9dae3d + 8312264
commit 9c4db07
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 4 deletions.
diff --git a/include/ConcurrentPriorityQueue.h b/include/ConcurrentPriorityQueue.h
@@ -32,7 +32,7 @@ class ConcurrentPriorityQueue {
 public:
   ConcurrentPriorityQueue();
   void push(Data data);
-  void top(Data& result);
+  void pop(Data& result);
   void pop();
   void wait_and_pop(Data &result);
   bool empty();
@@ -68,9 +68,10 @@ void ConcurrentPriorityQueue<Data, DataCompartor>::wait_and_pop(Data& result) {
 }
 
 template<typename Data, typename DataCompartor>
-void ConcurrentPriorityQueue<Data, DataCompartor>::top(Data& result) {
+void ConcurrentPriorityQueue<Data, DataCompartor>::pop(Data& result) {
   boost::mutex::scoped_lock lock(mLock);
   result = mQueue.top();
+  mQueue.pop();
 }
 
 template<typename Data, typename DataCompartor>

diff --git a/include/NdbDataReaders.h b/include/NdbDataReaders.h
@@ -120,13 +120,13 @@ template<typename Data, typename Conn, typename Keys>
 void NdbDataReaders<Data, Conn, Keys>::processWaiting() {
   while(!mWaitingOutQueue->empty()){
     Bulk<Keys> out;
-    mWaitingOutQueue->top(out);
+    mWaitingOutQueue->pop(out);
     if(out.mProcessingIndex == mLastSent + 1){
       LOG_INFO("publish enriched events with index ["  << out.mProcessingIndex << "] to Elastic");
       mElasticSearch->addData(out);
       mLastSent++;
-      mWaitingOutQueue->pop();
     }else{
+      mWaitingOutQueue->push(out);
       break;
     }
   }