Same Executor model for sal and client

.gitignore

+*.o
+.vscode/
\ No newline at end of file

ConcurrentQueue.hpp

+#ifndef __ConQueue__
+#define __ConQueue__
+#include <queue>
+#include <mutex>
+#include <condition_variable>
+
+template <typename Data>
+class ConcurrentQueue
+{
+private:
+    std::queue<Data> queue;
+    std::mutex queueMutex;
+    std::condition_variable queueCv;
+
+public:
+    void push(Data const &data)
+    {
+        std::unique_lock<std::mutex> lock(queueMutex);
+        queue.push(data);
+        lock.unlock();
+        queueCv.notify_one();
+    }
+
+    bool empty()
+    {
+        std::unique_lock<std::mutex> lock(queueMutex);
+        return queue.empty();
+    }
+
+    bool try_pop(Data &popped_value)
+    {
+        std::unique_lock<std::mutex> lock(queueMutex);
+        if (queue.empty())
+        {
+            return false;
+        }
+
+        popped_value = queue.front();
+        queue.pop();
+        return true;
+    }
+
+    void wait_and_pop(Data &popped_value)
+    {
+        std::unique_lock<std::mutex> lock(queueMutex);
+        queueCv.wait(lock, [&]{return queue.size() > 0;});
+        popped_value = queue.front();
+        queue.pop();
+    }
+};
+#endif
\ No newline at end of file

CqEventData.hpp

+#ifndef __CQEVENTDATA__
+
+#define __CQEVENTDATA__
+#include <rdma/rdma_verbs.h>
+
+class RdmaEndpoint;
+
+class CqEventData
+{
+public:
+    uint64_t _wr_id;
+    enum ibv_wc_opcode _opcode;
+    uint32_t _byte_len;
+    RdmaEndpoint *_endpoint;
+    CqEventData(uint64_t wr_id, enum ibv_wc_opcode opcode, uint32_t byte_len, RdmaEndpoint *endpoint)
+        : _wr_id(wr_id), _opcode(opcode), _byte_len(byte_len), _endpoint(endpoint)
+    {
+    }
+};
+
+#endif
\ No newline at end of file

Executor.cpp

+#include "Executor.hpp"
+Executor::Executor(int size, RdmaEndpointGroup *group)
+    : _size(size), _group(group)
+{
+    _taskQueue = new ConcurrentQueue<struct ibv_wc *>();
+    _taskThreads = new std::vector<TaskThread *>();
+    _taskThreads->reserve(size);
+    for (int i = 0; i < _size; i++)
+    {
+        TaskThread *thread = new TaskThread(i, _taskQueue, _group);
+        _taskThreads->push_back(thread);
+    }
+}
+ void Executor::submit(struct ibv_wc *task)
+{
+    _taskQueue->push(task);
+}
+void Executor::getTask()
+{
+}
\ No newline at end of file

Executor.hpp

+#ifndef __Executor__
+#define __Executor__
+#include <vector>
+
+#include "CqEventData.hpp"
+#include "RdmaEndpointGroup.hpp"
+#include "ConcurrentQueue.hpp"
+#include "TaskThread.hpp"
+class Executor
+{
+    int _size{0};
+    std::vector<TaskThread *> *_taskThreads{NULL};
+    ConcurrentQueue<ibv_wc *> *_taskQueue{NULL};
+    RdmaEndpointGroup *_group;
+
+public:
+    Executor(int size, RdmaEndpointGroup *group);
+    void submit(struct ibv_wc *task);
+    void getTask();
+};
+//long affinities[]
+#endif
\ No newline at end of file

Makefile

+
+all: clean server
+
+CXXFLAGS += -O3
+LIBS += -libverbs
+LIBS += -lrdmacm
+LIBS += -pthread
+LIBS += -lrocksdb
+
+object : 
+	g++ -std=c++17 -c  *.cpp $(LIBS)
+
+
+server1: object
+	g++ -std=c++17 -o  server -ggdb *.o $(LIBS)
+
+server: object
+	g++ -std=c++17 -o server -ggdb Server.o Executor.o TaskThread.o RdmaServerEndpointGroup.o RdmaReplicationEndpoint.o RdmaSalEndpoint.o RdmaEndpoint.o RdmaCmProcessor.o RdmaCqProcessor.o $(LIBS)
+.PHONY:
+clean:
+	rm -f *.o *.gch server

MessageFormats.hpp

+#ifndef __MessageFormats__
+#define __MessageFormats__
+enum RequestType
+{
+    GET,
+    PUT,
+    DELETE,
+    INVALIDATE
+};
+
+
+struct __attribute__ ((__packed__)) SalRequest
+{
+    uint32_t id;
+    enum RequestType type;
+    uint32_t keySize;
+    uint32_t valueSize;
+};
+
+struct __attribute__ ((__packed__)) SalResponse
+{
+    //private:
+    uint32_t  id;
+    enum RequestType type;
+    //public:
+    uint32_t size;
+};
+
+struct __attribute__ ((__packed__)) InvRequest
+{
+    //private:
+    uint32_t  id;
+    enum RequestType type;
+    //public:
+    uint32_t keySize;
+    
+};
+
+
+static uint32_t SUCCESS  = 0;
+static uint32_t FAILURE = 1;
+
+static int32_t SalRequestHeaderSize = sizeof(SalRequest);
+static int32_t SalResponseSize = sizeof(SalResponse);
+static uint32_t InvRequestHeaderSize = sizeof(InvRequest);
+#endif
\ No newline at end of file

RdmaCmProcessor.cpp

+#include "RdmaCmProcessor.hpp"
+#include <iostream>
+
+RdmaCmProcessor::RdmaCmProcessor(RdmaEndpointGroup *group)
+    : _endpointGroup(group)
+{
+    std::cout << "CMProcessor : Step 1 creating event channel" << std::endl;
+    _eventChannel = rdma_create_event_channel();
+    if (_eventChannel == NULL)
+    {
+        std::cout << "CMProcesor : error creating event channel";
+    }
+}
+
+struct rdma_cm_id *RdmaCmProcessor::createId()
+{
+    struct rdma_cm_id *id = NULL;
+    int ret = rdma_create_id(_eventChannel, &id, NULL, RDMA_PS_TCP);
+    if (ret == -1)
+        std::cout << "CMProcesor : rdma_create_id failed" << std::endl;
+    return id;
+}
+
+void RdmaCmProcessor::processCmEvent()
+{
+    int ret;
+    struct rdma_cm_event *event;
+    std::cout << "CMProcessor : starting cm processing thread" << std::endl;
+    while (!_stop)
+    {
+        ret = rdma_get_cm_event(_eventChannel, &event);
+        if (ret)
+        {
+            std::cout << "CMProcesor : rdma_get_cm_event failed" << std::endl;
+            continue;
+        }
+        _endpointGroup->processCmEvent(event);
+        ret = rdma_ack_cm_event(event);
+        if (ret)
+        {
+            std::cout << "CMProcesor : rdma_ack_cm_event failed";
+        }
+    }
+}
+
+void RdmaCmProcessor::start(bool newThread)
+{
+    if (newThread)
+        _cmEventThread = new std::thread(&RdmaCmProcessor::processCmEvent, this);
+    else
+        processCmEvent();
+}
+
+void RdmaCmProcessor::close()
+{
+    _stop = true;
+    if (_cmEventThread != NULL)
+        _cmEventThread->join();
+
+    rdma_destroy_event_channel(_eventChannel);
+}

RdmaCmProcessor.hpp

+#include <rdma/rdma_cma.h>
+#include <rdma/rdma_verbs.h>
+#include <stdint.h>
+#include <thread>
+#include <iostream>
+
+#ifndef __RDMACMPROCESSOR__
+#define __RDMACMPROCESSOR__
+
+#include "RdmaEndpointGroup.hpp"
+
+class RdmaCmProcessor
+{
+
+    struct rdma_event_channel *_eventChannel{NULL};
+    std::thread *_cmEventThread{NULL};
+    RdmaEndpointGroup *_endpointGroup{NULL};
+    bool _stop{false};
+
+public:
+    RdmaCmProcessor(RdmaEndpointGroup *group);
+    struct rdma_cm_id *createId();
+    void processCmEvent();
+    void start(bool newThread);
+    void close();
+};
+#endif
\ No newline at end of file

RdmaEndpoint.cpp

+#include "RdmaEndpoint.hpp"
+int RdmaEndpoint::CONN_STATE_INITIALIZED = 1;
+int RdmaEndpoint::CONN_STATE_RESOURCES_ALLOCATED = 2;
+int RdmaEndpoint::CONN_STATE_CONNECTED = 3;
+int RdmaEndpoint::CONN_STATE_CLOSED = 4;
+
+
+RdmaEndpoint::RdmaEndpoint(struct rdma_cm_id *id, struct ibv_cq *completionQueue, int sendQueueSize,
+                                       int recvQueueSize, int sendMsgSize, int recvMsgSize)
+    : _cm_id(id), _completionQueue(completionQueue), _sendQueueSize(sendQueueSize),
+      _recvQueueSize(recvQueueSize), _sendMsgSize(sendMsgSize), _recvMsgSize(recvMsgSize)
+{
+    _state = CONN_STATE_INITIALIZED;
+    _sendBuffers = new boost::lockfree::queue<void*>(_sendMsgSize);
+}
+void RdmaEndpoint::createResources()
+{
+    if (_state != CONN_STATE_INITIALIZED)
+    {
+        std::cout << "RdmaEndpoint : createResource invalid satte" << std::endl;
+    }
+
+    _protectionDomain = ibv_alloc_pd(_cm_id->verbs);
+    if (_protectionDomain == NULL)
+    {
+        std::cout << "RdmaEndpoint : ibv_alloc_pd failed " << std::endl;
+        return;
+    }
+
+    struct ibv_qp_init_attr qp_init_attr;
+    memset(&qp_init_attr, 0, sizeof(qp_init_attr));
+    //This is used to set endpoint address with qp
+    qp_init_attr.qp_context = (void *)this;
+    // if not set 0, all work requests submitted to SQ will always generate a Work Completion
+    qp_init_attr.sq_sig_all = 1;
+    // completion queue can be shared or you can use distinct completion queues.
+    qp_init_attr.send_cq = _completionQueue;
+    qp_init_attr.recv_cq = _completionQueue;
+    qp_init_attr.qp_type = IBV_QPT_RC;
+    // increase if you want to keep more send work requests in the SQ.
+    qp_init_attr.cap.max_send_wr = _sendQueueSize;
+    // increase if you want to keep more receive work requests in the RQ.
+    qp_init_attr.cap.max_recv_wr = _recvQueueSize;
+    // increase if you allow send work requests to have multiple scatter gather entry (SGE).
+    qp_init_attr.cap.max_send_sge = 1;
+    // increase if you allow receive work requests to have multiple scatter gather entry (SGE).
+    qp_init_attr.cap.max_recv_sge = 1;
+
+    int ret = rdma_create_qp(_cm_id, _protectionDomain, &qp_init_attr);
+    if (ret)
+    {
+        std::cout << "RdmaEndpoint : ibv_create_cq failed\n";
+    }
+    if (_cm_id->pd == NULL)
+    {
+        std::cout << "RdmaEndpoint : pd not set" << std::endl;
+        _cm_id->pd = _protectionDomain;
+    }
+    _sendBuff = (char*)malloc(_sendMsgSize * _sendQueueSize);
+    if (_sendBuff == NULL)
+        std::cout << "RdmaEndpoint :  sendBuff malloc failed" << std::endl;
+    _recvBuff = (char*)malloc(_recvMsgSize * _recvQueueSize);
+
+    _sendMr = rdma_reg_write(_cm_id, reinterpret_cast<void *>(_sendBuff), _sendMsgSize * _sendQueueSize);
+    if (_sendMr == NULL)
+        std::cout << "RdmaEndpoint :  sendMr reg failed" << std::endl;
+
+    if (_recvBuff == NULL)
+        std::cout << "RdmaEndpoint :  recvBuff malloc failed" << std::endl;
+    _recvMr = rdma_reg_read(_cm_id, reinterpret_cast<void *>(_recvBuff), _recvMsgSize * _recvQueueSize);
+    if (_recvMr == NULL)
+        std::cout << "RdmaEndpoint :  recvMr reg failed" << std::endl;
+
+    for (int i = 0; i < _recvQueueSize; i++)
+    {
+        char *const location = _recvBuff + i * _recvMsgSize;
+        rdma_post_recv(_cm_id, reinterpret_cast<void *>(location), reinterpret_cast<void *>(location),
+                       _recvMsgSize, _recvMr);
+    }
+    
+    for (int i = 0; i < _sendQueueSize; i++)
+    {
+        void* const location  = _sendBuff + i * _sendMsgSize;
+        _sendBuffers->push(location);
+    }
+
+    _state = CONN_STATE_RESOURCES_ALLOCATED;
+}
+
+
+void RdmaEndpoint::processCmEvent(struct rdma_cm_event *event)
+{
+    std::cout << "RdmaEndpoint : Event " << rdma_event_str(event->event) << std::endl;
+    if (event->event == RDMA_CM_EVENT_CONNECT_REQUEST)
+    {
+        std::cout << "RdmaEndpoint : Connect request";
+    }
+    else if (event->event == RDMA_CM_EVENT_ESTABLISHED)
+    {
+        if (_state != CONN_STATE_RESOURCES_ALLOCATED)
+        {
+            std::cout << "RdmaEndpoint : EstablishedEvent invalid state " << std::endl;
+        }
+        std::cout << "RdmaEndpoint : step 6 Connected" << std::endl;
+        _state = CONN_STATE_CONNECTED;
+    }
+    else if (event->event == RDMA_CM_EVENT_DISCONNECTED)
+    {
+        std::cout << "RdmaEndpoint : step 7 disconnected" << std::endl;
+        clientClose();
+    }
+}
+
+void RdmaEndpoint::clientClose()
+{
+    if (_state != CONN_STATE_CONNECTED)
+    {
+        std::cout << "RdmaEndpoint :  clientClose invalid state" << std::endl;
+        return;
+    }
+    std::cout<<"RdmaEndpoint : closing connection qp "<<_cm_id->qp->qp_num<<  std::endl;
+    int ret;
+    ret = rdma_disconnect(_cm_id);
+    if (ret)
+    {
+        std::cout << "RdmaEndpoint : rdma_disconnect failed" << std::endl;
+    }
+    
+    ret = rdma_dereg_mr(_sendMr);
+    if (ret)
+    {
+        std::cout << "RdmaEndpoint : rdma_dereg_mr send failed" << std::endl;
+    }
+    free(_sendBuff);
+    
+    ret = rdma_dereg_mr(_recvMr);
+    if (ret)
+    {
+        std::cout << "RdmaEndpoint : rdma_dereg_mr recv failed" << std::endl;
+    }
+    free(_recvBuff);
+    
+    rdma_destroy_qp(_cm_id);
+
+    std::cout<<"des qp"<<std::endl;
+   // ret = rdma_destroy_id(_cm_id);
+
+    std::cout<<"des mr"<<std::endl;
+    if (ret)
+    {
+        std::cout << "RdmaEndpoint : rdma_destroy_id failed" << std::endl;
+    }
+    _state = CONN_STATE_CLOSED;
+    std::cout<<"closed"<<std::endl;
+}

RdmaEndpoint.hpp

+#ifndef __RDMAENDPOINT__
+#define __RDMAENDPOINT__
+#include <iostream>
+#include <boost/lockfree/queue.hpp>
+#include <rdma/rdma_cma.h>
+#include <rdma/rdma_verbs.h>
+#include <stdint.h>
+#include <thread>
+#include <netdb.h>
+#include <arpa/inet.h>
+#include <map>
+#include <vector>
+#include <mutex>
+#include <shared_mutex>
+#include "CqEventData.hpp"
+
+class RdmaEndpoint
+{
+public:
+    static int CONN_STATE_INITIALIZED;
+    static int CONN_STATE_RESOURCES_ALLOCATED;
+    static int CONN_STATE_CONNECTED;
+    static int CONN_STATE_CLOSED;
+
+    struct rdma_cm_id *_cm_id{NULL};
+    struct ibv_cq *_completionQueue{NULL};
+    struct ibv_pd *_protectionDomain{NULL};
+
+    int _sendQueueSize{0};
+    int _recvQueueSize{0};
+    int _sendMsgSize{0};
+    int _recvMsgSize{0};
+    int _state{0};
+
+    char *_sendBuff{NULL};
+    char *_recvBuff{NULL};
+    struct ibv_mr *_sendMr{NULL};
+    struct ibv_mr *_recvMr{NULL};
+    boost::lockfree::queue<void *> *_sendBuffers{NULL};
+
+    RdmaEndpoint(struct rdma_cm_id *id, struct ibv_cq *completionQueue, int sendQueueSize,
+                 int recvQueueSize, int sendMsgSize, int recvMsgSize);
+
+    void createResources();
+    void processCmEvent(struct rdma_cm_event *event);
+    void clientClose();
+    virtual void processSendCompletion(struct ibv_wc *data) = 0;
+    virtual void processRecvCompletion(struct ibv_wc *data) = 0;
+};
+#endif
\ No newline at end of file

RdmaEndpointGroup.hpp

+#ifndef __RDMA_ENDPOINT_GROUP__
+#define __RDMA_ENDPOINT_GROUP__
+#include <rdma/rdma_cma.h>
+#include <rdma/rdma_verbs.h>
+#include <stdint.h>
+#include <thread>
+#include <netdb.h>
+#include <arpa/inet.h>
+#include <map>
+#include <vector>
+#include <mutex>
+#include <shared_mutex>
+#include "RdmaSalEndpoint.hpp"
+#include "RdmaReplicationEndpoint.hpp"
+
+class RdmaEndpointGroup
+{
+public:
+    std::vector<RdmaSalEndpoint *> *_salEps{NULL};
+    std::vector<RdmaReplicationEndpoint *> *_repEps{NULL};
+    std::unordered_map<uint32_t, RdmaReplicationEndpoint *> *_qpRepEndpointMap{NULL};
+    std::unordered_map<uint32_t, RdmaSalEndpoint *> *_qpSalEndpointMap{NULL};
+
+    virtual void processCmEvent(struct rdma_cm_event *event) = 0;
+};
+#endif
\ No newline at end of file

RdmaRepCqProcessor.cpp

+#include "RdmaRepCqProcessor.hpp"
\ No newline at end of file

RdmaRepCqProcessor.hpp

+#include <rdma/rdma_cma.h>
+#include <rdma/rdma_verbs.h>
+#include <stdint.h>
+#include <iostream>
+#include <thread>
+#include <unordered_map>
+
+#ifndef __RDMAREPCQPROCESSOR__
+#define __RDMAREPCQPROCESSOR__
+
+#include "Executor.hpp"
+
+class RdmaRepCqProcessor
+{
+public:
+	struct ibv_comp_channel *_compChannel{NULL};
+	struct ibv_cq *_completionQueue{NULL};
+	std::thread *_compQueueThread{NULL};
+	bool _stop{false};
+	Executor *_executor{NULL};
+
+	RdmaRepCqProcessor(Executor *ex, ibv_context *verbs, int compQueueSize)
+		: _executor(ex)
+	{
+		_compChannel = ibv_create_comp_channel(verbs);
+		if (_compChannel == NULL)
+		{
+			std::cout << "CqProcessr : ibv_create_comp_channel failed\n";
+			return;
+		}
+		_completionQueue = ibv_create_cq(verbs, compQueueSize, NULL, _compChannel, 0);
+		if (_completionQueue == NULL)
+		{
+			std::cout << "CqProcessr : ibv_create_cq failed" << std::endl;
+			return;
+		}
+		int ret = ibv_req_notify_cq(_completionQueue, 0);
+		if (ret)
+		{
+			std::cout << "CqProcessr : ibv_req_notify_cq failed\n";
+		}
+	}
+	struct ibv_cq *getCq()
+	{
+		return _completionQueue;
+	}
+	void start()
+	{
+		std::cout << "CqProcessr : starting process CQ events" << std::endl;
+		_compQueueThread = new std::thread(&RdmaRepCqProcessor::processCQEvents, this);
+	}
+	void processCQEvents()
+	{
+		int ret = 0;
+		struct ibv_cq *cq;
+		void *context;
+		const int nevent = 10;
+		struct ibv_wc *wc_array = new struct ibv_wc[nevent];
+		while (!_stop)
+		{
+			ret = ibv_get_cq_event(_compChannel, &cq, &context);
+			if (ret == -1)
+			{
+				std::cout << "CqProcessr : ibv_get_cq_event failed\n";
+				close();
+			}
+			ibv_ack_cq_events(cq, 1);
+			ret = ibv_req_notify_cq(_completionQueue, 0);
+			if (ret)
+			{
+				std::cout << "CqProcessr : ibv_req_notify_cq failed\n";
+				close();
+			}
+			ret = ibv_poll_cq(cq, nevent, wc_array);
+			if (ret < 0)
+			{
+				std::cout << "CqProcessr : ibv_poll_cq failed\n";
+				close();
+			}
+			if (ret == 0)
+				continue;
+			for (int i = 0; i < ret; i++)
+			{
+				struct ibv_wc *data = new struct ibv_wc(wc_array[i]);
+				data->vendor_err = 1;
+				_executor->submit(data);
+			}
+
+			//_executor->dispatchRepCqEvents(wc_array, ret);
+		}
+	}
+
+	void close()
+	{
+		_stop = true;
+		if (_compQueueThread != NULL)
+			_compQueueThread->join();
+	}
+};
+
+#endif

RdmaReplicationEndpoint.cpp

+#include "RdmaReplicationEndpoint.hpp"
+
+RdmaReplicationEndpoint::RdmaReplicationEndpoint(struct rdma_cm_id *id, struct ibv_cq *completionQueue, int sendQueueSize,
+                                                 int recvQueueSize, int sendMsgSize, int recvMsgSize,rocksdb::DB *db)
+    : RdmaEndpoint(id, completionQueue, sendQueueSize, recvQueueSize, sendMsgSize, recvMsgSize)
+    ,_db(db)
+{
+}
+
+void RdmaReplicationEndpoint::processSendCompletion(struct ibv_wc *data)
+{
+    std::cout << "send completion\n";
+    _sendBuffers->push((void *)data->wr_id);
+}
+void RdmaReplicationEndpoint::processRecvCompletion(struct ibv_wc *data)
+{
+    std::cout << "recv completion\n";
+    std::cout << "recieve" << (char *)(data->wr_id) << "\n";
+    char* request = new char[data->byte_len];
+    memcpy(request, (void *)data->wr_id,data->byte_len);
+    rdma_post_recv(_cm_id, (void *)data->wr_id, (void *)data->wr_id, _recvMsgSize, _recvMr);
+}
+
+int RdmaReplicationEndpoint::sendMessage(const char *buffer, uint32_t size)
+{
+    if (size > _sendMsgSize)
+        return -1;
+    void* sendBuffer = nullptr;
+    _sendBuffers->pop(sendBuffer);
+    if (sendBuffer == nullptr)
+        return -1;
+    memcpy(sendBuffer, buffer, size);
+    return rdma_post_send(_cm_id, sendBuffer, sendBuffer, size, _sendMr, 0);
+}

RdmaReplicationEndpoint.hpp

+#ifndef __RDMASERVERENDPOINT__
+#define __RDMASERVERENDPOINT__
+
+#include <rdma/rdma_cma.h>
+#include <rdma/rdma_verbs.h>
+#include <stdint.h>
+#include <errno.h>
+#include <iostream>
+#include <boost/lockfree/queue.hpp>
+#include "RdmaEndpoint.hpp"
+#include "CqEventData.hpp"
+#include <rocksdb/db.h>
+
+class RdmaReplicationEndpoint : public RdmaEndpoint
+{
+
+    rocksdb::DB *_db;
+    std::atomic<uint64_t> _requestId{12};
+    
+public:
+    RdmaReplicationEndpoint(struct rdma_cm_id *id, struct ibv_cq *completionQueue, int sendQueueSize,
+                       int recvQueueSize, int sendMsgSize, int recvMsgSize,rocksdb::DB *_db);
+
+    void processCqEvent(struct ibv_wc wc);
+    void processSendCompletion(struct ibv_wc* data);
+    void processRecvCompletion(struct ibv_wc* data);
+    int sendMessage(const char *buffer, uint32_t size);
+    void close();
+};
+#endif
\ No newline at end of file

RdmaSalCqProcessor.cpp

+#include "RdmaSalCqProcessor.hpp"
+RdmaSalCqProcessor::RdmaSalCqProcessor(Executor *ex, ibv_context *verbs, int compQueueSize)
+        : _executor(ex)
+    {
+        _compChannel = ibv_create_comp_channel(verbs);
+        if (_compChannel == NULL)
+        {
+            std::cout << "SalCqProcessr : ibv_create_comp_channel failed\n";
+            return;
+        }
+        _completionQueue = ibv_create_cq(verbs, compQueueSize, NULL, _compChannel, 0);
+        if (_completionQueue == NULL)
+        {
+            std::cout << "SalCqProcessr : ibv_create_cq failed" << std::endl;
+            return;
+        }
+        int ret = ibv_req_notify_cq(_completionQueue, 0);
+        if (ret)
+        {
+            std::cout << "SalCqProcessr : ibv_req_notify_cq failed\n";
+        }
+    }
+    struct ibv_cq *RdmaSalCqProcessor::getCq()
+    {
+        return _completionQueue;
+    }
+    void RdmaSalCqProcessor::start()
+    {
+        std::cout << "SalCqProcessr : starting process CQ events" << std::endl;
+        _compQueueThread = new std::thread(&RdmaSalCqProcessor::processCQEvents, this);
+    }
+    void RdmaSalCqProcessor::processCQEvents()
+    {
+        int ret = 0;
+        struct ibv_cq *cq;
+        void *context;
+        const int nevent = 10;
+        struct ibv_wc wc_array[nevent];
+        while (!_stop)
+        {
+            /*
+            * get_CQ_event is a blocking call and it wait save some cpu cycles but.
+            * it might not be that efficient compared to polling
+            */
+            ret = ibv_get_cq_event(_compChannel, &cq, &context);