commitChanges async api

src/IStorage.h

@@ -7,6 +7,13 @@
 #define METADATA_DB_IDENTIFIER "c299fde0-6d42-4ec4-b939-34f680ffe39f"
 
 struct StorageToken {
+    enum class TokenType {
+        SingleRead,
+        SingleWrite,
+        Delete,
+        BatchWrite,
+    };
+    TokenType type;
     std::unordered_set<struct client *> setc;
     struct redisDbPersistentData *db;
     virtual ~StorageToken() {}
@@ -46,6 +53,9 @@ class IStorage
     virtual StorageToken *begin_retrieve(struct aeEventLoop *, aePostFunctionTokenProc, sds *, size_t) {return nullptr;};
     virtual void complete_retrieve(StorageToken * /*tok*/, callbackSingle /*fn*/) {};
 
+    virtual StorageToken* begin_endWriteBatch(struct aeEventLoop *, aePostFunctionTokenProc*) {} // NOP
+    virtual void complete_endWriteBatch(StorageToken * /*tok*/) {};
+
     virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem) {
         beginWriteBatch();
         for (size_t ielem = 0; ielem < celem; ++ielem) {

src/StorageCache.cpp

@@ -233,4 +233,5 @@ void StorageCache::emergencyFreeCache() {
             dictRelease(d);
         });
     }
-}
+}
+

src/StorageCache.h

@@ -45,6 +45,8 @@ class StorageCache
     void retrieve(sds key, IStorage::callbackSingle fn) const;
     StorageToken *begin_retrieve(struct aeEventLoop *el, aePostFunctionTokenProc proc, sds *rgkey, size_t ckey);
     void complete_retrieve(StorageToken *tok, IStorage::callbackSingle fn);
+    StorageToken* begin_endWriteBatch(struct aeEventLoop *el, aePostFunctionTokenProc* proc) {m_spstorage->begin_endWriteBatch(el,proc);} // NOP
+    void complete_endWriteBatch(StorageToken *tok) {m_spstorage->complete_endWriteBatch(tok);};
     bool erase(sds key);
     void emergencyFreeCache();
     bool keycacheIsEnabled() const { return m_pdict != nullptr; }

src/db.cpp

@@ -3072,6 +3072,8 @@ void redisDbPersistentData::bulkDirectStorageInsert(char **rgKeys, size_t *rgcbK
 
 void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot **psnapshotFree)
 {
+
+    std::unordered_set<client *> setcBlocked;
     if (m_pdbSnapshotStorageFlush)
     {
         dictIterator *di = dictGetIterator(m_dictChangedStorageFlush);
@@ -3086,8 +3088,22 @@ void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot **
         *psnapshotFree = m_pdbSnapshotStorageFlush;
         m_pdbSnapshotStorageFlush = nullptr;
     }
+
     if (m_spstorage != nullptr)
-        m_spstorage->endWriteBatch();
+    {
+        auto tok = m_spstorage->begin_endWriteBatch(serverTL->el, storageLoadCallback);
+        if (tok != nullptr)
+        {
+            for (client *c : setcBlocked) //need to check how to push client to blocked list
+            {
+                if (!(c->flags & CLIENT_BLOCKED))
+                    blockClient(c, BLOCKED_STORAGE);
+            }
+           // tok->setc = std::move(setcBlocked);
+            tok->db = this;
+            tok->type = StorageToken::TokenType::BatchWrite;
+        }
+    }
 }
 
 redisDbPersistentData::~redisDbPersistentData()

src/storage/rocksdb.cpp

@@ -255,6 +255,21 @@ void RocksDBStorageProvider::endWriteBatch()
     m_lock.unlock();
 }
 
+StorageToken* RocksDBStorageProvider::begin_endWriteBatch(struct aeEventLoop *el, aePostFunctionTokenProc* callback)
+{
+    StorageToken *tok = new StorageToken();
+    auto pbatch = m_spbatch.get();
+    (*m_pfactory->m_wqueue)->AddWorkFunction([this, el,callback,tok,&pbatch]{
+        m_spdb->Write(WriteOptions(),pbatch);
+        aePostFunction(el,callback,tok);
+    });
+}
+
+void RocksDBStorageProvider::complete_endWriteBatch(StorageToken* tok){
+   // m_spbatch = nullptr;
+    m_lock.unlock();
+}
+
 void RocksDBStorageProvider::batch_lock()
 {
     m_lock.lock();

src/storage/rocksdb.h

@@ -42,6 +42,8 @@ class RocksDBStorageProvider : public IStorage
 
     virtual void beginWriteBatch() override;
     virtual void endWriteBatch() override;
+    virtual StorageToken* begin_endWriteBatch(struct aeEventLoop *el, aePostFunctionTokenProc* proc);
+    virtual void complete_endWriteBatch(StorageToken *tok);
 
     virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem) override;