Added common cqevents list to wait for replication

code/cppfol/header/CqEventData.hpp

@@ -11,7 +11,7 @@ class CQEventData
     CQEventData(int repRequired,RdmaSalEndpoint* ep)
     : _repRequired(repRequired),_ep(ep)
     {
-        std::cout<<"created\n";
+        std::cout<<"created "<<_repRequired<<"\n";
     }
     ~CQEventData()
     {

code/cppfol/header/Executor.hpp

@@ -13,6 +13,7 @@ class Executor
     std::vector<TaskThread *> _taskThreads{NULL};
     ConcurrentQueue *_taskQueue{NULL};
     std::unordered_map<uint32_t,std::shared_ptr<CQEventData>> cqEvents;
+    std::mutex _cqEventsMutex;
 
 public:
     Executor(int size);

code/cppfol/header/TaskThread.hpp

@@ -26,19 +26,21 @@ private:
     std::unordered_map<uint32_t, RdmaRepEndpoint *> *_serverRepMap;
     std::unordered_map<uint32_t, RdmaSalEndpoint *> *_salMap;
     std::unordered_map<uint32_t, std::shared_ptr<CQEventData>> &_cqEvents;
+    std::mutex &_cqEventsMutex;
 
 public:
     TaskThread(int id, int cpu, ConcurrentQueue *,
                std::unordered_map<uint32_t, RdmaRepEndpoint *> *clientRepMap,
                std::unordered_map<uint32_t, RdmaRepEndpoint *> *serverRepMap,
                std::unordered_map<uint32_t, RdmaSalEndpoint *> *salMap,
-               std::unordered_map<uint32_t, std::shared_ptr<CQEventData>> &cqEvents);
+               std::unordered_map<uint32_t, std::shared_ptr<CQEventData>> &cqEvents,
+               std::mutex &_cqEventsMutex);
 
     void replicateSalRequest(char *salRequest, uint32_t size);
     void sendInvalidation(char *salRequest);
     static void *run(void *object);
     void stop();
-    void processEvent(RdmaSalEndpoint *ep, struct ibv_wc *data);
+    void processSalEvent(RdmaSalEndpoint *ep, struct ibv_wc *data);
     void processRepEvent(RdmaRepEndpoint *ep, struct ibv_wc *data);
     void processReplicationDone(uint32_t reqid, MessageType type, bool localWrite);
     ~TaskThread();

code/cppfol/src/Executor.cpp

@@ -12,7 +12,7 @@ void Executor::createThreads(std::unordered_map<uint32_t, RdmaRepEndpoint *> *cl
 {
     for (int i = 0; i < _size; i++)
     {
-        TaskThread *thread = new TaskThread(i, i, _taskQueue, clientRepMap,serverRepMap, salMap,cqEvents);
+        TaskThread *thread = new TaskThread(i, i, _taskQueue, clientRepMap,serverRepMap, salMap,cqEvents,_cqEventsMutex);
         _taskThreads.push_back(thread);
     }
 }

code/cppfol/src/TaskThread.cpp

@@ -4,8 +4,9 @@ TaskThread::TaskThread(int id, int cpu, ConcurrentQueue *taskqueue,
                        std::unordered_map<uint32_t, RdmaRepEndpoint *> *clientRepMap,
                        std::unordered_map<uint32_t, RdmaRepEndpoint *> *serverRepMap,
                        std::unordered_map<uint32_t, RdmaSalEndpoint *> *salMap,
-                       std::unordered_map<uint32_t, std::shared_ptr<CQEventData>> &cqEvents)
-    : _id(id), _clientRepMap(clientRepMap), _serverRepMap(serverRepMap), _salMap(salMap), _cqEvents(cqEvents)
+                       std::unordered_map<uint32_t, std::shared_ptr<CQEventData>> &cqEvents,
+                       std::mutex &cqEventsMutex)
+    : _id(id), _clientRepMap(clientRepMap), _serverRepMap(serverRepMap), _salMap(salMap), _cqEvents(cqEvents), _cqEventsMutex(cqEventsMutex)
 {
     _taskQueue = taskqueue;
     if (pthread_create(&thread, NULL, &TaskThread::run, this))
@@ -94,7 +95,7 @@ inline void *TaskThread::run(void *object)
                 continue;
             }
             else if (salEp != nullptr)
-                thread->processEvent(salEp, data);
+                thread->processSalEvent(salEp, data);
             else if (repEp != nullptr)
                 thread->processRepEvent(repEp, data);
             thread->_taskQueue->removeFromSet(data);
@@ -104,7 +105,7 @@ inline void *TaskThread::run(void *object)
     return NULL;
 }
 
-void TaskThread::processEvent(RdmaSalEndpoint *ep, struct ibv_wc *data)
+void TaskThread::processSalEvent(RdmaSalEndpoint *ep, struct ibv_wc *data)
 {
     std::cout << "processing sal event" << data->opcode << "\n";
     /* sal Request*/
@@ -128,34 +129,42 @@ void TaskThread::processEvent(RdmaSalEndpoint *ep, struct ibv_wc *data)
          * We need to wait for replication and invalidation to be done for delete and put request,hence we are putting data in map
          */
         rocksdb::Status s;
-        switch (req->type)
+        if (req->type == MessageType::GET)
         {
-        case MessageType::GET:
             ep->processGet(req);
-            break;
-        case MessageType::DELETE:
+        }
+        else if (req->type == MessageType::DELETE)
+        {
+            std::unique_lock<std::mutex> lock(_cqEventsMutex);
             _cqEvents[id] = cqevent;
+            lock.unlock();
             replicateSalRequest(buffer, data->byte_len);
             sendInvalidation(buffer);
             s = ep->_db->Delete(rocksdb::WriteOptions(), {(char *)req + MessageHeaderSize, req->keySize});
-            processReplicationDone(req->id, MessageType::FAILURE, true);
-            break;
-        case MessageType::PUT:
-            std::cout << "id " << id << " " << _cqEvents.size() << '\n';
+            if (s.ok())
+                processReplicationDone(req->id, MessageType::SUCCESS, true);
+            else
+                processReplicationDone(req->id, MessageType::FAILURE, true);
+        }
+        else if (req->type == MessageType::PUT)
+        {
+            std::unique_lock<std::mutex> lock(_cqEventsMutex);
             _cqEvents[id] = cqevent;
-            std::cout << "size " << _cqEvents.size() << "\n";
+            lock.unlock();
             replicateSalRequest(buffer, data->byte_len);
             sendInvalidation(buffer);
             s = ep->_db->Put(rocksdb::WriteOptions(), {(char *)req + MessageHeaderSize, req->keySize},
                              {(char *)req + MessageHeaderSize + req->keySize, req->valueSize});
-            ;
-            processReplicationDone(req->id, MessageType::FAILURE, true);
-            break;
-        default:
+            if (s.ok())
+                processReplicationDone(req->id, MessageType::SUCCESS, true);
+            else
+                processReplicationDone(req->id, MessageType::FAILURE, true);
+        }
+        else
+        {
             std::ostringstream ss;
             ss << "SalRequest invalid req type" << data->opcode;
             CPPLog::LOG_ERROR(ss);
-            break;
         }
         delete[] buffer;
     }
@@ -166,8 +175,6 @@ void TaskThread::processEvent(RdmaSalEndpoint *ep, struct ibv_wc *data)
         CPPLog::LOG_INFO(ss);
         break;
     }
-    std::cout << "size "
-              << " " << _cqEvents.size() << '\n';
 }
 
 void TaskThread::processRepEvent(RdmaRepEndpoint *ep, struct ibv_wc *data)
@@ -223,8 +230,9 @@ void TaskThread::processRepEvent(RdmaRepEndpoint *ep, struct ibv_wc *data)
 
 void TaskThread::processReplicationDone(uint32_t id, MessageType type, bool localWrite)
 {
+    std::unique_lock<std::mutex> cqEventlock(_cqEventsMutex);
     auto it = _cqEvents.find(id);
-    // std::cout<<"size " <<_cqEvents.size()<<"\n";
+    std::cout << "size " << _cqEvents.size() << "\n";
     if (it == _cqEvents.end())
     {
         std::ostringstream ss;
@@ -233,13 +241,18 @@ void TaskThread::processReplicationDone(uint32_t id, MessageType type, bool loca
         return;
     }
     // std::cout<<"rep done\n";
+    // For local write we  do not increase repdone count
     if (!localWrite)
         ++(it->second->_repDone);
+    /* if there is no rep event then this condition will be false(0<0) and we can send response back to client*/
     if (it->second->_repDone < it->second->_repRequired)
     {
         // std::cout<<"event 2\n";
         return;
     }
+    _cqEvents.erase(it);
+    cqEventlock.unlock();
+
     RdmaSalEndpoint *ep = it->second->_ep;
     char *sendBuf = nullptr;
     std::unique_lock<std::mutex> lock(ep->_sendBuffersM);
@@ -277,7 +290,6 @@ void TaskThread::replicateSalRequest(char *req, uint32_t size)
     {
         serverRepIt->second->sendMessage(req, size);
     }
-
 }
 void TaskThread::sendInvalidation(char *req)
 {