Configured rocksdb

code/cppfol/header/RdmaServerEndpointGroup.hpp

@@ -49,6 +49,7 @@ class RdmaServerEndpointGroup
     int _sendMsgSize{0};
     int _recvMsgSize{0};
     rocksdb::DB *_db;
+    rocksdb::Options _options;
 
     mutable std::shared_mutex _salMutex;
     mutable std::shared_mutex _repMutex;
@@ -60,7 +61,7 @@ public:
     std::unordered_map<uint32_t, RdmaRepEndpoint*> *_qpRepEndpointMap;
 
     RdmaServerEndpointGroup(int sendQueueSize, int recvQueueSize, int compQueueSize,
-                            int sendMsgSize, int recvMsgSize, rocksdb::DB *db);
+                            int sendMsgSize, int recvMsgSize, rocksdb::DB *db,rocksdb::Options &options);
     // void setExecutor(Executor *executor);
     void bind(const char *ip, const char *port, int backlog);
 

code/cppfol/readme.md

+https://www.confluent.io/blog/how-to-tune-rocksdb-kafka-streams-state-stores-performance/
+
 // 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.

code/cppfol/src/RdmaServerEndpointGroup.cpp

@@ -6,9 +6,14 @@ int RdmaServerEndpointGroup::CONN_STATE_CONNECTED = 4;
 int RdmaServerEndpointGroup::CONN_STATE_CLOSED = 5;
 
 RdmaServerEndpointGroup::RdmaServerEndpointGroup(int sendQueueSize, int recvQueueSize, int compQueueSize,
-                                                 int sendMsgSize, int recvMsgSize, rocksdb::DB *db)
-    : _sendQueueSize(sendQueueSize), _recvQueueSize(recvQueueSize), _compQueueSize(compQueueSize),
-      _sendMsgSize(sendMsgSize), _recvMsgSize(recvMsgSize), _db(db)
+                                                 int sendMsgSize, int recvMsgSize, rocksdb::DB *db, rocksdb::Options &options)
+    : _sendQueueSize(sendQueueSize),
+      _recvQueueSize(recvQueueSize),
+      _compQueueSize(compQueueSize),
+      _sendMsgSize(sendMsgSize),
+      _recvMsgSize(recvMsgSize),
+      _db(db),
+      _options(options)
 {
     CPPLog::LOG_INFO("SalEndpointGroup : Step 1 creating event channel");
     _eventChannel = rdma_create_event_channel();
@@ -207,6 +212,7 @@ void RdmaServerEndpointGroup::processCmEvent(struct rdma_cm_event *event)
             {
                 _qpSalEndpointMap->erase(qp);
                 delete ((RdmaSalEndpoint *)event->id->context);
+                 std::cout<<_options.statistics->ToString();
             }
             auto it2 = _qpRepEndpointMap->find(qp);
             if (it2 != _qpRepEndpointMap->end())

code/cppfol/src/fol.cpp

@@ -5,20 +5,22 @@
 #include "Executor.hpp"
 #include "Properties.hpp"
 #include "rocksdb/db.h"
+#include "rocksdb/memtablerep.h"
+#include "rocksdb/table.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/options.h"
 
-int connectToRepServer(Properties &prop, RdmaRepEndpointGroup* repGroup)
+int connectToRepServer(Properties &prop, RdmaRepEndpointGroup *repGroup)
 {
     int followers = stoi(prop.getValue("FOLLOWERS"));
-    std::cout<<"followers "<<followers<<"\n";
-    RdmaClientRepEndpoint* clientEPs[followers];
-    for(int i = 0 ;i< followers ;i++)
+    std::cout << "followers " << followers << "\n";
+    RdmaClientRepEndpoint *clientEPs[followers];
+    for (int i = 0; i < followers; i++)
     {
         clientEPs[i] = repGroup->createEndpoint();
-        std::string ip = prop.getValue("FOLLOWER"+std::to_string(i+1)+"_IP");
-        std::string port = prop.getValue("FOLLOWER"+std::to_string(i+1)+"_PORT");
-        std::cout<<"Connecting to follower "<<ip<<":"<<port<<"\n";
+        std::string ip = prop.getValue("FOLLOWER" + std::to_string(i + 1) + "_IP");
+        std::string port = prop.getValue("FOLLOWER" + std::to_string(i + 1) + "_PORT");
+        std::cout << "Connecting to follower " << ip << ":" << port << "\n";
         clientEPs[i]->connect(ip.c_str(), port.c_str(), "fol");
     }
     return 0;
@@ -37,12 +39,35 @@ int main()
     std::string serverPort = prop.getValue("SERVER_PORT");
     int executorPoolSize = stoi(prop.getValue("EXECUTOR_POOL_SIZE"));
     int enableLogging = stoi(prop.getValue("ENABLE_LOGGING"));
+    int maxFlushBackGroundJobs = stoi(prop.getValue("ROCKS_MAX_FLUSH_BACKGROUND_JOBS"));
+    int maxCompactionBackGroundJobs = stoi(prop.getValue("ROCKS_MAX_COMPACTION_BACKGROUND_JOBS"));
+    int writeBufferSize = stoi(prop.getValue("ROCKS_WRITE_BUFFER_SIZE"));
+    int lruCacheSize = stoi(prop.getValue("ROCKS_CACHE_SIZE"));
     if (enableLogging == 0)
         CPPLog::Logger::getInstance()->updateLogLevel(CPPLog::DISABLE_LOG);
 
     rocksdb::DB *db;
     rocksdb::Options options;
+    options.statistics = rocksdb::CreateDBStatistics();
+    if (lruCacheSize != -1)
+    {
+        std::shared_ptr<rocksdb::Cache> cache = rocksdb::NewLRUCache(8 << 20);
+        ROCKSDB_NAMESPACE::BlockBasedTableOptions table_options;
+        table_options.block_cache = cache;
+        auto table_factory = rocksdb::NewBlockBasedTableFactory(table_options);
+        options.table_factory.reset(table_factory);
+    }
+
+    if (writeBufferSize != -1)
+        options.write_buffer_size = writeBufferSize << 20;
     options.create_if_missing = true;
+    // setting flush threads
+    if (maxFlushBackGroundJobs != -1)
+        options.env->SetBackgroundThreads(maxFlushBackGroundJobs, rocksdb::Env::Priority::HIGH);
+    // setting compaction threads
+    if (maxCompactionBackGroundJobs != -1)
+        options.env->SetBackgroundThreads(maxCompactionBackGroundJobs, rocksdb::Env::Priority::LOW);
+        
     // open a database with a name which corresponds to a file system directory
     rocksdb::Status status = rocksdb::DB::Open(options, dbpath, &db);
     if (!status.ok())
@@ -51,11 +76,11 @@ int main()
         exit(1);
     }
     CPPLog::LOG_ALWAYS("Rocks started\n\n");
-
+    std::cout << "Backgroundtthreads" << options.env->GetBackgroundThreads();
     Executor *executor = new Executor(executorPoolSize);
 
     RdmaRepEndpointGroup *repgroup = new RdmaRepEndpointGroup(sendQS, recvQS, compQS, sendMS, recvMS, 1, 100000, db);
-    RdmaServerEndpointGroup *sgroup = new RdmaServerEndpointGroup(sendQS, recvQS, compQS, sendMS, recvMS, db);
+    RdmaServerEndpointGroup *sgroup = new RdmaServerEndpointGroup(sendQS, recvQS, compQS, sendMS, recvMS, db, options);
 
     executor->createThreads(repgroup->_qpRepEndpointMap, sgroup->_qpRepEndpointMap, sgroup->_qpSalEndpointMap);
     repgroup->setExecutor(executor);