Memory lock in raft

src/kvstore/CMakeLists.txt

@@ -6,7 +6,6 @@ nebula_add_library(
     PartManager.cpp
     NebulaStore.cpp
     RocksEngineConfig.cpp
-    LogEncoder.cpp
     NebulaSnapshotManager.cpp
     RateLimiter.cpp
     plugins/elasticsearch/ESListener.cpp

src/kvstore/Common.h

@@ -270,6 +270,16 @@ inline rocksdb::Slice toSlice(const folly::StringPiece& str) {
 using KVMap = std::unordered_map<std::string, std::string>;
 using KVArrayIterator = std::vector<KV>::const_iterator;
 
+class MergeableAtomicOpResult {
+ public:
+  nebula::cpp2::ErrorCode code;
+  std::string batch;  // batched result, like before.
+  std::list<std::string> readSet;
+  std::list<std::string> writeSet;
+};
+
+using MergeableAtomicOp = folly::Function<MergeableAtomicOpResult(void)>;
+
 }  // namespace kvstore
 }  // namespace nebula
 

src/kvstore/KVStore.h

@@ -312,7 +312,7 @@ class KVStore {
    */
   virtual void asyncAtomicOp(GraphSpaceID spaceId,
                              PartitionID partId,
-                             raftex::AtomicOp op,
+                             MergeableAtomicOp op,
                              KVCallback cb) = 0;
 
   /**

src/kvstore/NebulaStore.cpp

@@ -918,7 +918,7 @@ void NebulaStore::asyncRemoveRange(GraphSpaceID spaceId,
 
 void NebulaStore::asyncAtomicOp(GraphSpaceID spaceId,
                                 PartitionID partId,
-                                raftex::AtomicOp op,
+                                MergeableAtomicOp op,
                                 KVCallback cb) {
   auto ret = part(spaceId, partId);
   if (!ok(ret)) {

src/kvstore/NebulaStore.h

@@ -415,7 +415,7 @@ class NebulaStore : public KVStore, public Handler {
    */
   void asyncAtomicOp(GraphSpaceID spaceId,
                      PartitionID partId,
-                     raftex::AtomicOp op,
+                     MergeableAtomicOp op,
                      KVCallback cb) override;
 
   /**

src/kvstore/Part.cpp

@@ -115,7 +115,7 @@ void Part::sync(KVCallback cb) {
       [callback = std::move(cb)](nebula::cpp2::ErrorCode code) mutable { callback(code); });
 }
 
-void Part::asyncAtomicOp(raftex::AtomicOp op, KVCallback cb) {
+void Part::asyncAtomicOp(MergeableAtomicOp op, KVCallback cb) {
   atomicOpAsync(std::move(op))
       .thenValue(
           [callback = std::move(cb)](nebula::cpp2::ErrorCode code) mutable { callback(code); });

src/kvstore/Part.h

@@ -124,7 +124,7 @@ class Part : public raftex::RaftPart {
    * @param op Atomic operation
    * @param cb Callback when has a result
    */
-  void asyncAtomicOp(raftex::AtomicOp op, KVCallback cb);
+  void asyncAtomicOp(MergeableAtomicOp op, KVCallback cb);
 
   /**
    * @brief Add a raft learner asynchronously by adding raft log

src/kvstore/raftex/CMakeLists.txt

@@ -5,6 +5,7 @@ nebula_add_library(
     RaftexService.cpp
     Host.cpp
     SnapshotManager.cpp
+    ../LogEncoder.cpp
 )
 
 nebula_add_subdirectory(test)