Merge remote-tracking branch 'internal/main'

Author: msotheeswaran-sc

build.yaml

@@ -21,13 +21,13 @@ machamp:
         parent: make-build
         # https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
         builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
-        command: ./runtest --clients 4 --verbose 
+        command: ./runtest --clients 4 --verbose --tls
       cluster-test:
         type: cmd
         parent: make-build
         # https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
         builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
-        command: ./runtest-cluster 
+        command: ./runtest-cluster --tls
       sentinel-test:
         type: cmd
         parent: make-build

src/SnapshotPayloadParseState.cpp

@@ -136,14 +136,33 @@ void SnapshotPayloadParseState::flushQueuedKeys() {
     int idb = current_database;
     serverAssert(vecqueuedKeys.size() == vecqueuedVals.size());
     auto sizePrev = vecqueuedKeys.size();
-    (*insertsInFlight)++;
-    std::weak_ptr<std::atomic<int>> 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]->bulkDirectStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
-            (*(insertsInFlightTmp.lock()))--;
-            delete pallocator;
-        });
+        if (g_pserver->m_pstorageFactory) {
+            (*insertsInFlight)++;
+            std::weak_ptr<std::atomic<int>> insertsInFlightTmp = insertsInFlight; // C++ GRRRRRRRRRRRRRRRR, we don't want to capute "this" because that's dangerous
+            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]->bulkDirectStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
+                (*(insertsInFlightTmp.lock()))--;
+                delete pallocator;
+            });
+        } else {
+            for (size_t ival = 0; ival < vecqueuedKeys.size(); ++ival) {
+                size_t offset = 0;
+                auto spexpire = deserializeExpire(vecqueuedVals[ival], vecqueuedValsCb[ival], &offset);    
+                auto o = deserializeStoredObject(vecqueuedVals[ival] + offset, vecqueuedValsCb[ival] - offset);
+                sds sdsKey = sdsnewlen(vecqueuedKeys[ival], -static_cast<ssize_t>(vecqueuedKeysCb[ival]));
+                if (dbMerge(g_pserver->db[idb], sdsKey, o, false)) {
+                    if (spexpire != nullptr)
+                        g_pserver->db[idb]->setExpire(sdsKey, std::move(*spexpire));
+                } else {
+                    sdsfree(sdsKey);
+                }
+            }
+            vecqueuedKeys.clear();
+            vecqueuedKeysCb.clear();
+            vecqueuedVals.clear();
+            vecqueuedValsCb.clear();
+        }
     } else {
         // else drop the data
         vecqueuedKeys.clear();

src/config.cpp

@@ -2983,6 +2983,7 @@ standardConfig configs[] = {
     createIntConfig("overload-protect-percent", NULL, MODIFIABLE_CONFIG, 0, 200, g_pserver->overload_protect_threshold, 0, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("force-eviction-percent", NULL, MODIFIABLE_CONFIG, 0, 100, g_pserver->force_eviction_percent, 0, INTEGER_CONFIG, NULL, NULL),
     createBoolConfig("enable-async-rehash", NULL, MODIFIABLE_CONFIG, g_pserver->enable_async_rehash, 1, NULL, NULL),
+    createBoolConfig("enable-keydb-fastsync", NULL, MODIFIABLE_CONFIG, g_pserver->fEnableFastSync, 0, NULL, NULL),
 
 #ifdef USE_OPENSSL
     createIntConfig("tls-port", NULL, MODIFIABLE_CONFIG, 0, 65535, g_pserver->tls_port, 0, INTEGER_CONFIG, NULL, updateTLSPort), /* TCP port. */

src/db.cpp

@@ -56,7 +56,6 @@ int expireIfNeeded(redisDb *db, robj *key, robj *o);
 void slotToKeyUpdateKeyCore(const char *key, size_t keylen, int add);
 
 std::unique_ptr<expireEntry> deserializeExpire(const char *str, size_t cch, size_t *poffset);
-sds serializeStoredObjectAndExpire(robj_roptr o);
 
 dictType dictChangeDescType {
     dictSdsHash,                /* hash function */

src/networking.cpp

@@ -163,7 +163,6 @@ client *createClient(connection *conn, int iel) {
     c->multibulklen = 0;
     c->bulklen = -1;
     c->sentlen = 0;
-    c->sentlenAsync = 0;
     c->flags = 0;
     c->fPendingAsyncWrite = FALSE;
     c->fPendingAsyncWriteHandler = FALSE;

src/redis-cli.c

@@ -2524,7 +2524,10 @@ static int clusterManagerExecTransaction(clusterManagerNode *node,
 
 static int clusterManagerNodeConnect(clusterManagerNode *node) {
     if (node->context) redisFree(node->context);
-    node->context = redisConnect(node->ip, node->port);
+    struct timeval tv;
+    tv.tv_sec = config.cluster_manager_command.timeout / 1000;
+    tv.tv_usec = (config.cluster_manager_command.timeout % 1000) * 1000;
+    node->context = redisConnectWithTimeout(node->ip, node->port, tv);
     if (!node->context->err && config.tls) {
         const char *err = NULL;
         if (cliSecureConnection(node->context, config.sslconfig, &err) == REDIS_ERR && err) {
@@ -5626,7 +5629,10 @@ static int clusterManagerCommandImport(int argc, char **argv) {
     char *reply_err = NULL;
     redisReply *src_reply = NULL;
     // Connect to the source node.
-    redisContext *src_ctx = redisConnect(src_ip, src_port);
+    struct timeval tv;
+    tv.tv_sec = config.cluster_manager_command.timeout / 1000;
+    tv.tv_usec = (config.cluster_manager_command.timeout % 1000) * 1000;
+    redisContext *src_ctx = redisConnectWithTimeout(src_ip, src_port, tv);
     if (src_ctx->err) {
         success = 0;
         fprintf(stderr,"Could not connect to KeyDB at %s:%d: %s.\n", src_ip,