keydb instead of KeyDB

* add s3 config

* missing include

* ignore s3 config in config test

* Update introspection.tcl

* Update config.cpp

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/childinfo.cpp

@@ -42,6 +42,7 @@ typedef struct {
  * RDB / AOF saving process from the child to the parent (for instance
  * the amount of copy on write memory used) */
 void openChildInfoPipe(void) {
+    serverAssert(g_pserver->child_info_pipe[0] == -1);
     if (pipe(g_pserver->child_info_pipe) == -1) {
         /* On error our two file descriptors should be still set to -1,
          * but we call anyway closeChildInfoPipe() since can't hurt. */

src/config.cpp

@@ -35,6 +35,7 @@
 
 #include <fcntl.h>
 #include <sys/stat.h>
+#include <sys/wait.h>
 #ifdef __linux__
 #include <sys/sysinfo.h>
 #endif
@@ -2456,6 +2457,32 @@ static int isValidAOFfilename(char *val, const char **err) {
     return 1;
 }
 
+static int isValidS3Bucket(char *s3bucket, const char **err) {
+    int status = EXIT_FAILURE;
+    pid_t pid = fork();
+    if (pid < 0)
+    {
+        *err = "couldn't fork to call aws cli";
+        return 0;
+    }
+
+    if (pid == 0)
+    {
+        execlp("aws", "aws", "s3", "ls", s3bucket);
+        exit(EXIT_FAILURE);
+    }
+    else 
+    {
+        waitpid(pid, &status, 0);
+    }
+
+    if (status != EXIT_SUCCESS) {
+        *err = "could not access s3 bucket";
+        return 0;
+    }
+    return 1;
+}
+
 /* Validate specified string is a valid proc-title-template */
 static int isValidProcTitleTemplate(char *val, const char **err) {
     if (!validateProcTitleTemplate(val)) {
@@ -2803,6 +2830,7 @@ standardConfig configs[] = {
     createStringConfig("cluster-announce-ip", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->cluster_announce_ip, NULL, NULL, NULL),
     createStringConfig("syslog-ident", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->syslog_ident, "redis", NULL, NULL),
     createStringConfig("dbfilename", NULL, MODIFIABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->rdb_filename, CONFIG_DEFAULT_RDB_FILENAME, isValidDBfilename, NULL),
+    createStringConfig("db-s3-object", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->rdb_s3bucketpath, NULL, isValidS3Bucket, NULL),
     createStringConfig("appendfilename", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->aof_filename, "appendonly.aof", isValidAOFfilename, NULL),
     createStringConfig("server_cpulist", NULL, IMMUTABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->server_cpulist, NULL, NULL, NULL),
     createStringConfig("bio_cpulist", NULL, IMMUTABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->bio_cpulist, NULL, NULL, NULL),

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));
     }
 }
@@ -3002,8 +3005,16 @@ 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)
 {
+    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, celem);
 }
 

src/rdb.cpp

@@ -1490,7 +1490,7 @@ int rdbSaveFile(char *filename, const redisDbPersistentDataSnapshot **rgpdb, rdb
     rio rdb;
     int error = 0;
 
-    snprintf(tmpfile,sizeof(tmpfile),"temp-%d-%d.rdb", getpid(), g_pserver->rdbThreadVars.tmpfileNum);
+    getTempFileName(tmpfile, g_pserver->rdbThreadVars.tmpfileNum);
     fp = fopen(tmpfile,"w");
     if (!fp) {
         char *cwdp = getcwd(cwd,MAXPATHLEN);
@@ -1660,12 +1660,14 @@ int launchRdbSaveThread(pthread_t &child, rdbSaveInfo *rsi)
         pthread_attr_t tattr;
         pthread_attr_init(&tattr);
         pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB
+        openChildInfoPipe();
         if (pthread_create(&child, &tattr, rdbSaveThread, args)) {
             pthread_attr_destroy(&tattr);
             for (int idb = 0; idb < cserver.dbnum; ++idb)
                 g_pserver->db[idb]->endSnapshot(args->rgpdb[idb]);
             args->~rdbSaveThreadArgs();
             zfree(args);
+            closeChildInfoPipe();
             return C_ERR;
         }
         pthread_attr_destroy(&tattr);
@@ -1683,13 +1685,11 @@ int rdbSaveBackground(rdbSaveInfo *rsi) {
 
     g_pserver->dirty_before_bgsave = g_pserver->dirty;
     g_pserver->lastbgsave_try = time(NULL);
-    openChildInfoPipe();
 
     start = ustime();
     latencyStartMonitor(g_pserver->rdb_save_latency);
 
     if (launchRdbSaveThread(child, rsi) != C_OK) {
-        closeChildInfoPipe();
         g_pserver->lastbgsave_status = C_ERR;
         serverLog(LL_WARNING,"Can't save in background: fork: %s",
             strerror(errno));

src/rdb.h

@@ -151,6 +151,7 @@ int rdbLoad(rdbSaveInfo *rsi, int rdbflags);
 int rdbLoadFile(const char *filename, rdbSaveInfo *rsi, int rdbflags);
 int rdbSaveBackground(rdbSaveInfo *rsi);
 int rdbSaveToSlavesSockets(rdbSaveInfo *rsi);
+void getTempFileName(char tmpfile[], int tmpfileNum);
 void rdbRemoveTempFile(pid_t childpid, int from_signal);
 int rdbSave(const redisDbPersistentDataSnapshot **rgpdb, rdbSaveInfo *rsi);
 int rdbSaveFile(char *filename, const redisDbPersistentDataSnapshot **rgpdb, rdbSaveInfo *rsi);

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;
 
@@ -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) {

tests/unit/introspection.tcl

@@ -162,15 +162,16 @@ start_server {tags {"introspection"}} {
             aof_rewrite_cpulist
             time-thread-priority
             bgsave_cpulist
-	        storage-cache-mode
-	        storage-provider-options
-	        use-fork
+	    storage-cache-mode
+	    storage-provider-options
+	    use-fork
             multi-master
             active-replica
             bind
             set-proc-title
             repl-backlog-disk-reserve
 	    tls-allowlist
+            db-s3-object
         }
 
         if {!$::tls} {

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}