Merge branch 'main' into fast_flash

* add s3 config

* missing include

* ignore s3 config in config test

* Update introspection.tcl

* Update config.cpp

src/IStorage.h

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

src/SnapshotPayloadParseState.cpp

@@ -140,7 +140,7 @@ void SnapshotPayloadParseState::flushQueuedKeys() {
     auto &insertsInFlightTmp = insertsInFlight; // C++ GRRRRRRRRRRRRRRRR, we don't want to capute "this" because that's dangerous
     if (current_database < cserver.dbnum) {
         g_pserver->asyncworkqueue->AddWorkFunction([idb, vecqueuedKeys = std::move(this->vecqueuedKeys), vecqueuedKeysCb = std::move(this->vecqueuedKeysCb), vecqueuedVals = std::move(this->vecqueuedVals), vecqueuedValsCb = std::move(this->vecqueuedValsCb), &insertsInFlightTmp, pallocator = m_spallocator.release()]() mutable {
-            g_pserver->db[idb]->bulkStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
+            g_pserver->db[idb]->bulkDirectStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
             --insertsInFlightTmp;
             delete pallocator;
         });

src/StorageCache.cpp

@@ -114,13 +114,14 @@ 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, size_t celem)
+void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, 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;
@@ -152,7 +153,7 @@ void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, si
     }
     ul.unlock();
 
-    m_spstorage->bulkInsert(rgkeys, rgcbkeys, rgvals, rgcbvals, celem);
+    m_spstorage->bulkInsert(rgkeys, rgcbkeys, rgvals, rgcbvals, rgfOverwrite, 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, size_t celem);
+    void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem);
     void retrieve(sds key, IStorage::callbackSingle fn) const;
     bool erase(sds key);
     void emergencyFreeCache();

src/db.cpp

@@ -2607,12 +2607,15 @@ void clusterStorageLoadCallback(const char *rgchkey, size_t cch, void *)
     slotToKeyUpdateKeyCore(rgchkey, cch, true /*add*/);
 }
 
-void moduleLoadCallback(const char * rgchKey, size_t cchKey, void *data) {
-    if (g_pserver->cluster_enabled) {
-        clusterStorageLoadCallback(rgchKey, cchKey, data);
-    }
-    robj *keyobj = createEmbeddedStringObject(rgchKey, cchKey);
-    moduleNotifyKeyspaceEvent(NOTIFY_LOADED, "loaded", keyobj, *(int *)data);
+void moduleLoadCallback(const char * rgchKey, size_t, void *data) {
+    redisObjectStack keyobj;
+    initStaticStringObject(keyobj, const_cast<char *>(rgchKey));
+    moduleNotifyKeyspaceEvent(NOTIFY_LOADED, "loaded", &keyobj, *(int *)data);
+}
+
+void moduleClusterLoadCallback(const char * rgchKey, size_t cchKey, void *data) {
+    clusterStorageLoadCallback(rgchKey, cchKey, data);
+    moduleLoadCallback(rgchKey, cchKey, data);
 }
 
 void redisDb::initialize(int id)
@@ -2633,7 +2636,7 @@ void redisDb::storageProviderInitialize()
 {
     if (g_pserver->m_pstorageFactory != nullptr)
     {
-        IStorageFactory::key_load_iterator itr = moduleLoadCallback;
+        IStorageFactory::key_load_iterator itr = g_pserver->cluster_enabled ? moduleClusterLoadCallback : moduleLoadCallback;
         this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, &id));
     }
 }
@@ -2905,15 +2908,13 @@ void redisDbPersistentData::storeDatabase()
     dictReleaseIterator(di);
 }
 
-/* static */ void redisDbPersistentData::serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const char *key, bool fUpdate)
+/* static */ sds redisDbPersistentData::serializeChange(redisDbPersistentData *db, const char *key)
 {
     auto itr = db->find_cached_threadsafe(key);
     if (itr == nullptr)
-        return;
+        return nullptr;
     robj *o = itr.val();
-    sds temp = serializeStoredObjectAndExpire(db, (const char*) itr.key(), o);
-    storage->insert((sds)key, temp, sdslen(temp), fUpdate);
-    sdsfree(temp);
+    return serializeStoredObjectAndExpire(db, (const char*) itr.key(), o);
 }
 
 bool redisDbPersistentData::processChanges(bool fSnapshot)
@@ -2956,10 +2957,30 @@ 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)
                 {
-                    serializeAndStoreChange(m_spstorage.get(), this, (const char*)dictGetKey(de), (bool)dictGetVal(de));
+                    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));
+                    }
                 }
+                m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), vecoverwrite.data(), veckeys.size());
+                for (auto val : vecvals)
+                    sdsfree(val);
                 dictReleaseIterator(di);
             }
         }
@@ -2993,7 +3014,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(), veckeys.size());
+        m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), nullptr, veckeys.size());
         for (auto val : vecvals)
             sdsfree(val);
         dictReleaseIterator(di);
@@ -3002,9 +3023,17 @@ void redisDbPersistentData::processChangesAsync(std::atomic<int> &pendingJobs)
     });
 }
 
-void redisDbPersistentData::bulkStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem)
+/* This function is to bulk insert directly to storage provider bypassing in memory, assumes rgKeys and rgVals are not sds strings */
+void redisDbPersistentData::bulkDirectStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem)
 {
-    m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, celem);
+    if (g_pserver->cluster_enabled) {
+        aeAcquireLock();
+        for (size_t i = 0; i < celem; i++) {
+            slotToKeyUpdateKeyCore(rgKeys[i], rgcbKeys[i], 1);
+        }
+        aeReleaseLock();
+    }
+    m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, nullptr, celem);
 }
 
 void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot **psnapshotFree)
@@ -3013,10 +3042,30 @@ 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)
         {
-            serializeAndStoreChange(m_spstorage.get(), (redisDbPersistentData*)m_pdbSnapshotStorageFlush, (const char*)dictGetKey(de), (bool)dictGetVal(de));
+            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));
+            }
         }
+        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;

src/replication.cpp

@@ -3797,9 +3797,11 @@ int connectWithMaster(redisMaster *mi) {
     connSetPrivateData(mi->repl_transfer_s, mi);
     if (connConnect(mi->repl_transfer_s, mi->masterhost, mi->masterport,
                 NET_FIRST_BIND_ADDR, syncWithMaster) == C_ERR) {
-        int sev = g_pserver->enable_multimaster ? LL_NOTICE : LL_WARNING;   // with multimaster its not unheard of to intentiallionall have downed masters
-        serverLog(sev,"Unable to connect to MASTER: %s",
-                connGetLastError(mi->repl_transfer_s));
+        const char *err = "Unknown Error";
+        if (mi->repl_transfer_s->last_errno != 0)
+            err = connGetLastError(mi->repl_transfer_s);
+        int sev = g_pserver->enable_multimaster ? LL_NOTICE : LL_WARNING;   // with multimaster its not unheard of to intentionally have downed masters
+        serverLog(sev, "Unable to connect to MASTER: %s", err);
         connClose(mi->repl_transfer_s);
         mi->repl_transfer_s = NULL;
         return C_ERR;

src/scripting.cpp

@@ -1264,6 +1264,9 @@ void scriptingInit(int setup) {
 
     /* Finally set the table as 'redis' global var. */
     lua_setglobal(lua,"redis");
+    /* Set table as 'keydb' global var as well */
+    lua_getglobal(lua,"redis");
+    lua_setglobal(lua,"keydb");
 
     /* Replace math.random and math.randomseed with our implementations. */
     lua_getglobal(lua,"math");

src/server.h

@@ -1209,7 +1209,7 @@ public:
     bool FSnapshot() const { return m_spdbSnapshotHOLDER != nullptr; }
 
     std::unique_ptr<const StorageCache> CloneStorageCache() { return std::unique_ptr<const StorageCache>(m_spstorage->clone()); }
-    void bulkStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem);
+    void bulkDirectStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem);
 
     dict_iter find_cached_threadsafe(const char *key) const;
 
@@ -1217,7 +1217,7 @@ protected:
     uint64_t m_mvccCheckpoint = 0;
 
 private:
-    static void serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const char *key, bool fUpdate);
+    static sds serializeChange(redisDbPersistentData *db, const char *key);
 
     void ensure(const char *key);
     void ensure(const char *key, dictEntry **de);
@@ -1370,7 +1370,7 @@ struct redisDb : public redisDbPersistentDataSnapshot
     using redisDbPersistentData::FRehashing;
     using redisDbPersistentData::FTrackingChanges;
     using redisDbPersistentData::CloneStorageCache;
-    using redisDbPersistentData::bulkStorageInsert;
+    using redisDbPersistentData::bulkDirectStorageInsert;
 
 public:
     const redisDbPersistentDataSnapshot *createSnapshot(uint64_t mvccCheckpoint, bool fOptional) {

src/storage/rocksdb.cpp

@@ -43,8 +43,9 @@ 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, size_t celem)
+void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem)
 {
+    size_t coverwrites = 0;
     if (celem >= 16384) {
         rocksdb::Options options = DefaultRocksDBOptions();
         rocksdb::SstFileWriter sst_file_writer(rocksdb::EnvOptions(), options, options.comparator);
@@ -92,8 +93,15 @@ 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;
+    m_count += celem - coverwrites;
 }
 
 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, size_t celem) override;
+    virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem) override;
 
     virtual void batch_lock() override;
     virtual void batch_unlock() override;

tests/unit/scripting.tcl

@@ -43,10 +43,15 @@ start_server {tags {"scripting"}} {
         r eval {return redis.call('get',KEYS[1])} 1 mykey
     } {myval}
 
-    test { EVAL - keys command works? } {
+    test {EVAL - keys command works? } {
         r eval {return redis.call('keys', 'test')} 0
     }
 
+    test {EVAL - KeyDB global works } {
+        r eval {return keydb.call('get', KEYS[1])} 1 mykey
+        assert_equal [r eval {return redis.call('get',KEYS[1])} 1 mykey] [r eval {return keydb.call('get', KEYS[1])} 1 mykey]
+    }
+
     test {EVALSHA - Can we call a SHA1 if already defined?} {
         r evalsha fd758d1589d044dd850a6f05d52f2eefd27f033f 1 mykey
     } {myval}