Commit 6a2d8a83 authored by Paras Garg's avatar Paras Garg

Added local changes to git

parent cfb4eeec
*.o
.vscode/
.build/*
server
*.txt
\ No newline at end of file
SRCS := $(shell ls src/)
SRC_DIR := src
OBJS := $(SRCS:.cpp=.o)
BUILD_DIR := .build
OBJS := $(addprefix $(BUILD_DIR)/, $(OBJS))
CXX = g++
CXXFLAGS += -O3 -Wall -std=c++17 -I header
CXXFLAGS += -g
LIBS += -libverbs
LIBS += -lrdmacm
LIBS += -pthread
LIBS += -lrocksdb
Target := server
.phony = clean
$(BUILD_DIR)/%.o: $(SRC_DIR)/%.cpp
$(CXX) -o $@ $(CXXFLAGS) -c $< $(LIBS)
$(Target) : $(OBJS) | $(BUILD_DIR)
$(CXX) -o $@ $^ $(LIBS)
$(BUILD_DIR) :
mkdir -p $@
\ No newline at end of file
#ifndef __ConQueue__
#define __ConQueue__
#include <queue>
#include <mutex>
#include <condition_variable>
#include <set>
#include <string>
class Comparator
{
public:
inline bool operator()(const struct ibv_wc* c1,const struct ibv_wc* c2)
const
{
struct SalRequest *req1 = (struct SalRequest *)c1->wr_id;
struct SalRequest *req2 = (struct SalRequest *)c2->wr_id;
if(req1->keySize != req2->keySize)
return true;
char* key1 = (char*)req1+SalRequestHeaderSize;
char* key2 = (char*)req2+SalRequestHeaderSize;
for(int i=0;i<req1->keySize;i++)
{
if(key1[i]!=key2[i])
return true;
}
return false;
}
};
class ConcurrentQueue
{
private:
std::queue<struct ibv_wc *> queue1;
std::queue<struct ibv_wc *> queue2;
std::mutex queueMutex;
std::set<struct ibv_wc *, Comparator> runningRequests;
std::condition_variable queueCv;
public:
void push(struct ibv_wc *const &data)
{
std::unique_lock<std::mutex> lock(queueMutex);
queue1.push(data);
std::cout<<data<<std::endl;
lock.unlock();
queueCv.notify_one();
}
bool empty()
{
std::unique_lock<std::mutex> lock(queueMutex);
return queue1.empty() && queue2.empty();
}
struct ibv_wc *try_pop()
{
struct ibv_wc *value = NULL;
std::unique_lock<std::mutex> lock(queueMutex);
if (queue2.empty())
{
queueCv.wait(lock, [&]
{ return queue1.size() > 0; });
value = queue1.front();
queue1.pop();
}
else
{
value = queue2.front();
queue2.pop();
}
if (value->opcode != IBV_WC_RECV)
{
return value;
}
std::cout<<"value "<<value<<std::endl;
if (runningRequests.empty())
{
runningRequests.insert(value);
return value
;
}
auto it = runningRequests.find(value);
if (it != runningRequests.end())
{
queue2.push(value);
std::cout<<"found putting in 2"<<std::endl;
return NULL;
}
return value;
}
void removeFromSet(struct ibv_wc* data)
{
std::unique_lock<std::mutex> lock(queueMutex);
std::cout<<"removing"<<data<<std::endl;
runningRequests.erase(data);
}
void
wait_and_pop(struct ibv_wc *&popped_value)
{
std::unique_lock<std::mutex> lock(queueMutex);
queueCv.wait(lock, [&]
{ return queue1.size() > 0; });
popped_value = queue1.front();
queue1.pop();
}
};
#endif
/*
template <typename Data>
class ConcurrentQueue
{
private:
std::queue1<Data> queue;
std::queue2<Data> queue;
std::mutex queueMutex;
std::condition_variable queueCv;
public:
void push(Data const &data)
{
std::unique_lock<std::mutex> lock(queueMutex);
queue1.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 (queue2.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
#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 *_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
#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
#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
#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
#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
#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);
new std::thread(&RdmaRepCqProcessor::processRepEvent, this,data);
}
//_executor->dispatchRepCqEvents(wc_array, ret);
}
}
void processRepEvent(struct ibv_wc* data)
{
std::cout<<"procesing Replication request"<<std::endl;
}
void close()
{
_stop = true;
if (_compQueueThread != NULL)
_compQueueThread->join();
}
};
#endif
#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
#include <rdma/rdma_cma.h>
#include <rdma/rdma_verbs.h>
#include <stdint.h>
#include <iostream>
#include <thread>
#include <unordered_map>
#ifndef __RDMASALCQPROCESSOR__
#define __RDMASALCQPROCESSOR__
#include "Executor.hpp"
class RdmaSalCqProcessor
{
public:
struct ibv_comp_channel *_compChannel{NULL};
struct ibv_cq *_completionQueue{NULL};
std::thread *_compQueueThread{NULL};
bool _stop{false};
Executor *_executor{NULL};
RdmaSalCqProcessor(Executor *ex, ibv_context *verbs, int compQueueSize);
struct ibv_cq *getCq();
void start();
void processCQEvents();
void close();
};
#endif
#ifndef __RdmaInvEndpoint__
#define __RdmaInvEndpoint__
#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 "MessageFormats.hpp"
#include <rocksdb/db.h>
class RdmaSalEndpoint : public RdmaEndpoint
{
public:
rocksdb::DB *_db;
RdmaSalEndpoint(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);
void processDelete(struct SalRequest *);
void processGet(struct SalRequest *req);
void processPut(struct SalRequest *req);
int sendMessage(const char *buffer, uint32_t size);
void close();
};
#endif
\ No newline at end of file
#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>
#ifndef __RDMASERVERENDPOINTGROUP__
#define __RDMASERVERENDPOINTGROUP__
#include "RdmaReplicationEndpoint.hpp"
#include "RdmaSalEndpoint.hpp"
#include "RdmaEndpointGroup.hpp"
#include "CqEventData.hpp"
#include "Executor.hpp"
#include "RdmaCmProcessor.hpp"
#include "RdmaSalCqProcessor.hpp"
#include "RdmaRepCqProcessor.hpp"
class RdmaServerEndpointGroup : public RdmaEndpointGroup
{
/*
* Variables to maintain Group state
*/
static int CONN_STATE_INITIALIZED;
static int CONN_STATE_BINDED;
static int CONN_STATE_CONNECTED;
static int CONN_STATE_CLOSED;
struct rdma_cm_id *_cm_id{NULL};
RdmaCmProcessor *_cmProcessor{NULL};
RdmaSalCqProcessor *_salCqProcessor{NULL};
RdmaRepCqProcessor *_repCqProcessor{NULL};
Executor *_executor;
/*
* variables to maintain queue state
*/
int _sendQueueSize{0};
int _recvQueueSize{0};
int _compQueueSize{0};
int _sendMsgSize{0};
int _recvMsgSize{0};
rocksdb::DB *_db;
mutable std::shared_mutex _salMutex;
mutable std::shared_mutex _repMutex;
void clientClose();
public:
RdmaServerEndpointGroup(int sendQueueSize, int recvQueueSize, int compQueueSize, int sendMsgSize, int recvMsgSize);
// void setExecutor(Executor *executor);
void bind(const char *ip, const char *port, int backlog);
struct ibv_cq *createSalCq(struct rdma_cm_id *id);
struct ibv_cq *createRepCq(struct rdma_cm_id *id);
void dispatchRepCqEvents(ibv_wc wc[], int size);
void dispatchSalCqEvents(ibv_wc wc[], int size);
void setExecutor(Executor *ex);
void startCmProcessor(bool newThread);
void processCmEvent(struct rdma_cm_event *event);
void createEpCmEvent(struct rdma_cm_event *event);
void close();
};
#endif
\ No newline at end of file
#ifndef __Runnable__
#define __Runnable__
#include <iostream>
class Runnable
{
public:
virtual void run() = 0;
};
#endif
\ No newline at end of file
#ifndef __TaskThread__
#define __TaskThread__
#include "Runnable.hpp"
#include "CqEventData.hpp"
#include "RdmaEndpointGroup.hpp"
#include "ConcurrentQueue.hpp"
#include <pthread.h>
#include <iostream>
#include <queue>
class TaskThread
{
private:
ConcurrentQueue *_taskQueue;
bool _stop{false};
int _id;
pthread_t thread;
RdmaEndpointGroup *_group;
public:
TaskThread(int id, int cpu, ConcurrentQueue *, RdmaEndpointGroup *group);
TaskThread(int id, ConcurrentQueue *, RdmaEndpointGroup *group);
void replicateSalRequest(char *salRequest,uint32_t size);
static void *run(void *object);
void stop();
void processEvent(struct ibv_wc *data);
~TaskThread();
};
#endif
\ No newline at end of file
// https://www.rdmamojo.com/2013/03/09/ibv_get_cq_event/
// https://man7.org/linux/man-pages/man3/ibv_poll_cq.3.html
// https://man7.org/linux/man-pages/man3/ibv_get_cq_event.3.
//https://docs.microsoft.com/en-us/cpp/cpp/delegating-constructors?view=msvc-170
https://www.toptal.com/c-plus-plus/c-plus-plus-understanding-compilation
/*
cpu_set_t cpuset;
CPU_ZERO(&cpuset);
CPU_SET(i, &cpuset);
int rc = pthread_setaffinity_np(threads[i].native_handle(),
sizeof(cpu_set_t), &cpuset);
if (rc != 0) {
std::cerr << "Error calling pthread_setaffinity_np: " << rc << "\n";
}
*/
\ No newline at end of file
#include "Executor.hpp"
Executor::Executor(int size, RdmaEndpointGroup *group)
: _size(size), _group(group)
{
_taskQueue = new ConcurrentQueue();
_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)