add flash option to create-cluster

src/IStorage.h

@@ -33,11 +33,10 @@ public:
     virtual bool enumerate(callback fn) const = 0;
     virtual size_t count() const = 0;
 
-    virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem) {
+    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) {
-            bool fOverwrite = (rgfOverwrite != nullptr) ? rgfOverwrite[ielem] : false;
-            insert(rgkeys[ielem], rgcbkeys[ielem], rgvals[ielem], rgcbvals[ielem], fOverwrite);
+            insert(rgkeys[ielem], rgcbkeys[ielem], rgvals[ielem], rgcbvals[ielem], false);
         }
         endWriteBatch();
     }

src/StorageCache.cpp

@@ -114,14 +114,13 @@ void StorageCache::insert(sds key, const void *data, size_t cbdata, bool fOverwr
 }
 
 long _dictKeyIndex(dict *d, const void *key, uint64_t hash, dictEntry **existing);
-void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem)
+void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem)
 {
     std::vector<dictEntry*> vechashes;
     if (m_pdict != nullptr) {
         vechashes.reserve(celem);
 
         for (size_t ielem = 0; ielem < celem; ++ielem) {
-            if (rgfOverwrite != nullptr && rgfOverwrite[ielem]) continue;
             dictEntry *de = (dictEntry*)zmalloc(sizeof(dictEntry));
             de->key = (void*)dictGenHashFunction(rgkeys[ielem], (int)rgcbkeys[ielem]);
             de->v.u64 = 1;
@@ -153,7 +152,7 @@ void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, si
     }
     ul.unlock();
 
-    m_spstorage->bulkInsert(rgkeys, rgcbkeys, rgvals, rgcbvals, rgfOverwrite, celem);
+    m_spstorage->bulkInsert(rgkeys, rgcbkeys, rgvals, rgcbvals, celem);
 
     bulkInsertsInProgress--;
 }

src/StorageCache.h

@@ -41,7 +41,7 @@ public:
     void clear(void(callback)(void*));
     void clearAsync();
     void insert(sds key, const void *data, size_t cbdata, bool fOverwrite);
-    void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem);
+    void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem);
     void retrieve(sds key, IStorage::callbackSingle fn) const;
     bool erase(sds key);
     void emergencyFreeCache();

src/db.cpp

@@ -1931,6 +1931,7 @@ void setExpire(client *c, redisDb *db, robj *key, expireEntry &&e)
  * is associated with this key (i.e. the key is non volatile) */
 expireEntry *redisDbPersistentDataSnapshot::getExpire(const char *key) {
     /* No expire? return ASAP */
+    std::unique_lock<fastlock> ul(g_expireLock);
     if (expireSize() == 0)
         return nullptr;
 
@@ -2727,8 +2728,11 @@ void redisDbPersistentData::clear(void(callback)(void*))
         m_cnewKeysPending = 0;
         m_fAllChanged++;
     }
+    {
+    std::unique_lock<fastlock> ul(g_expireLock);
     delete m_setexpire;
     m_setexpire = new (MALLOC_LOCAL) expireset();
+    }
     if (m_spstorage != nullptr)
         m_spstorage->clear(callback);
     dictEmpty(m_pdictTombstone,callback);
@@ -2861,7 +2865,8 @@ LNotFound:
             {
                 dictAdd(m_pdict, sdsNewKey, o);
                 o->SetFExpires(spexpire != nullptr);
-
+                
+                std::unique_lock<fastlock> ul(g_expireLock);
                 if (spexpire != nullptr)
                 {
                     auto itr = m_setexpire->find(sdsKey);
@@ -2884,6 +2889,7 @@ LNotFound:
     if (*pde != nullptr && dictGetVal(*pde) != nullptr)
     {
         robj *o = (robj*)dictGetVal(*pde);
+        std::unique_lock<fastlock> ul(g_expireLock);
         serverAssert(o->FExpires() == (m_setexpire->find(sdsKey) != m_setexpire->end()));
     }
 }
@@ -2908,13 +2914,15 @@ void redisDbPersistentData::storeDatabase()
     dictReleaseIterator(di);
 }
 
-/* static */ sds redisDbPersistentData::serializeChange(redisDbPersistentData *db, const char *key)
+/* static */ void redisDbPersistentData::serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const char *key, bool fUpdate)
 {
     auto itr = db->find_cached_threadsafe(key);
     if (itr == nullptr)
-        return nullptr;
+        return;
     robj *o = itr.val();
-    return serializeStoredObjectAndExpire(db, (const char*) itr.key(), o);
+    sds temp = serializeStoredObjectAndExpire(db, (const char*) itr.key(), o);
+    storage->insert((sds)key, temp, sdslen(temp), fUpdate);
+    sdsfree(temp);
 }
 
 bool redisDbPersistentData::processChanges(bool fSnapshot)
@@ -2957,30 +2965,10 @@ bool redisDbPersistentData::processChanges(bool fSnapshot)
             {
                 dictIterator *di = dictGetIterator(m_dictChanged);
                 dictEntry *de;
-                std::vector<char*> veckeys;
-                std::vector<size_t> veccbkeys;
-                std::vector<char*> vecvals;
-                std::vector<size_t> veccbvals;
-                std::vector<char> vecoverwrite;
-                veckeys.reserve(dictSize(m_dictChanged));
-                veccbkeys.reserve(dictSize(m_dictChanged));
-                vecvals.reserve(dictSize(m_dictChanged));
-                veccbvals.reserve(dictSize(m_dictChanged));
-                vecoverwrite.reserve(dictSize(m_dictChanged));
                 while ((de = dictNext(di)) != nullptr)
                 {
-                    sds val = serializeChange(this, (const char*)dictGetKey(de));
-                    if (val != nullptr) {
-                        veckeys.push_back((char*)dictGetKey(de));
-                        veccbkeys.push_back(sdslen((sds)dictGetKey(de)));
-                        vecvals.push_back(val);
-                        veccbvals.push_back(sdslen(val));
-                        vecoverwrite.push_back((bool)dictGetVal(de));
-                    }
+                    serializeAndStoreChange(m_spstorage.get(), this, (const char*)dictGetKey(de), (bool)dictGetVal(de));
                 }
-                m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), vecoverwrite.data(), veckeys.size());
-                for (auto val : vecvals)
-                    sdsfree(val);
                 dictReleaseIterator(di);
             }
         }
@@ -3014,7 +3002,7 @@ void redisDbPersistentData::processChangesAsync(std::atomic<int> &pendingJobs)
             vecvals.push_back(temp);
             veccbvals.push_back(sdslen(temp));
         }
-        m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), nullptr, veckeys.size());
+        m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), veckeys.size());
         for (auto val : vecvals)
             sdsfree(val);
         dictReleaseIterator(di);
@@ -3033,7 +3021,7 @@ void redisDbPersistentData::bulkDirectStorageInsert(char **rgKeys, size_t *rgcbK
         }
         aeReleaseLock();
     }
-    m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, nullptr, celem);
+    m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, celem);
 }
 
 void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot **psnapshotFree)
@@ -3042,30 +3030,10 @@ void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot **
     {
         dictIterator *di = dictGetIterator(m_dictChangedStorageFlush);
         dictEntry *de;
-        std::vector<char*> veckeys;
-        std::vector<size_t> veccbkeys;
-        std::vector<char*> vecvals;
-        std::vector<size_t> veccbvals;
-        std::vector<char> vecoverwrite;
-        veckeys.reserve(dictSize(m_dictChanged));
-        veccbkeys.reserve(dictSize(m_dictChanged));
-        vecvals.reserve(dictSize(m_dictChanged));
-        veccbvals.reserve(dictSize(m_dictChanged));
-        vecoverwrite.resize(dictSize(m_dictChanged));
         while ((de = dictNext(di)) != nullptr)
         {
-            sds val = serializeChange((redisDbPersistentData*)m_pdbSnapshotStorageFlush, (const char*)dictGetKey(de));
-            if (val != nullptr) {
-                veckeys.push_back((char*)dictGetKey(de));
-                veccbkeys.push_back(sdslen((sds)dictGetKey(de)));
-                vecvals.push_back(val);
-                veccbvals.push_back(sdslen(val));
-                vecoverwrite.push_back((bool)dictGetVal(de));
-            }
+            serializeAndStoreChange(m_spstorage.get(), (redisDbPersistentData*)m_pdbSnapshotStorageFlush, (const char*)dictGetKey(de), (bool)dictGetVal(de));
         }
-        m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), vecoverwrite.data(), veckeys.size());
-        for (auto val : vecvals)
-            sdsfree(val);
         dictReleaseIterator(di);
         dictRelease(m_dictChangedStorageFlush);
         m_dictChangedStorageFlush = nullptr;
@@ -3294,6 +3262,7 @@ std::unique_ptr<expireEntry> deserializeExpire(sds key, const char *str, size_t
 
 sds serializeStoredObjectAndExpire(redisDbPersistentData *db, const char *key, robj_roptr o)
 {
+    std::unique_lock<fastlock> ul(g_expireLock);
     auto itrExpire = db->setexpire()->find(key);
     const expireEntry *pexpire = nullptr;
     if (itrExpire != db->setexpire()->end())
@@ -3418,6 +3387,7 @@ void redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
                     dictAdd(m_pdict, sharedKey, o);
                     o->SetFExpires(spexpire != nullptr);
 
+                    std::unique_lock<fastlock> ul(g_expireLock);
                     if (spexpire != nullptr)
                     {
                         auto itr = m_setexpire->find(sharedKey);

src/expire.cpp

@@ -481,6 +481,7 @@ void expireSlaveKeys(void) {
     if (slaveKeysWithExpire == NULL ||
         dictSize(slaveKeysWithExpire) == 0) return;
 
+    std::unique_lock<fastlock> ul(g_expireLock);
     int cycles = 0, noexpire = 0;
     mstime_t start = mstime();
     while(1) {

src/object.cpp

@@ -1668,6 +1668,7 @@ robj *deserializeStoredObjectCore(const void *data, size_t cb)
 robj *deserializeStoredObject(const redisDbPersistentData *db, const char *key, const void *data, size_t cb)
 {
     robj *o = deserializeStoredObjectCore(data, cb);
+    std::unique_lock<fastlock> ul(g_expireLock);
     o->SetFExpires(db->setexpire()->exists(key));
     return o;
 }

src/server.h

@@ -1217,7 +1217,7 @@ protected:
     uint64_t m_mvccCheckpoint = 0;
 
 private:
-    static sds serializeChange(redisDbPersistentData *db, const char *key);
+    static void serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const char *key, bool fUpdate);
 
     void ensure(const char *key);
     void ensure(const char *key, dictEntry **de);

src/storage/rocksdb.cpp

@@ -43,9 +43,8 @@ void RocksDBStorageProvider::insert(const char *key, size_t cchKey, void *data,
         ++m_count;
 }
 
-void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem)
+void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem)
 {
-    size_t coverwrites = 0;
     if (celem >= 16384) {
         rocksdb::Options options = DefaultRocksDBOptions();
         rocksdb::SstFileWriter sst_file_writer(rocksdb::EnvOptions(), options, options.comparator);
@@ -93,15 +92,8 @@ void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **
         m_spdb->Write(WriteOptions(), spbatch.get());
     }
 
-    if (rgfOverwrite != nullptr) {
-        for (size_t ielem = 0; ielem < celem; ++ielem) {
-            if (rgfOverwrite[ielem])
-                ++coverwrites;
-        }
-    }
-
     std::unique_lock<fastlock> l(m_lock);
-    m_count += celem - coverwrites;
+    m_count += celem;
 }
 
 bool RocksDBStorageProvider::erase(const char *key, size_t cchKey)

src/storage/rocksdb.h

@@ -38,7 +38,7 @@ public:
     virtual void beginWriteBatch() override;
     virtual void endWriteBatch() override;
 
-    virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem) override;
+    virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem) override;
 
     virtual void batch_lock() override;
     virtual void batch_unlock() override;

utils/create-cluster/create-cluster

@@ -26,7 +26,11 @@ then
     while [ $((PORT < ENDPORT)) != "0" ]; do
         PORT=$((PORT+1))
         echo "Starting $PORT"
-        $BIN_PATH/keydb-server --port $PORT  --protected-mode $PROTECTED_MODE --cluster-enabled yes --cluster-config-file nodes-${PORT}.conf --cluster-node-timeout $TIMEOUT --appendonly yes --appendfilename appendonly-${PORT}.aof --dbfilename dump-${PORT}.rdb --logfile ${PORT}.log --daemonize yes ${ADDITIONAL_OPTIONS}
+        if [ "$2" == "flash" ]
+        then
+            ADDITIONAL_OPTIONS="--save \"\" \"\" \"\" --semi-ordered-set-bucket-size 8 --client-output-buffer-limit replica 1 1 0 --maxmemory 100000000 --storage-provider flash ./$PORT.flash"
+        fi
+        $BIN_PATH/keydb-server --server-threads 4 --port $PORT  --protected-mode $PROTECTED_MODE --cluster-enabled yes --cluster-config-file nodes-${PORT}.conf --cluster-node-timeout $TIMEOUT --dbfilename dump-${PORT}.rdb --logfile ${PORT}.log --daemonize yes ${ADDITIONAL_OPTIONS}
     done
     exit 0
 fi
@@ -97,6 +101,8 @@ then
     rm -rf appendonly*.aof
     rm -rf dump*.rdb
     rm -rf nodes*.conf
+    rm -rf *.flash
+    rm -rf temp*.rdb
     exit 0
 fi