Merge branch 'unstable' into RELEASE_6

.github/workflows/ci.yml

@@ -10,6 +10,7 @@ jobs:
     - uses: actions/checkout@v1
     - name: make
       run: |
+          sudo apt-get update
           sudo apt-get -y install uuid-dev libcurl4-openssl-dev
           make BUILD_TLS=yes -j2
     - name: gen-cert
@@ -34,6 +35,7 @@ jobs:
     - uses: actions/checkout@v1
     - name: make -j2
       run: | 
+        sudo apt-get update
         sudo apt-get -y install uuid-dev libcurl4-openssl-dev
         make -j2
 
@@ -50,6 +52,7 @@ jobs:
     - uses: actions/checkout@v1
     - name: make
       run: |
+          sudo apt-get update
           sudo apt-get -y install uuid-dev libcurl4-openssl-dev
           make MALLOC=libc -j2
 

.github/workflows/endurance.yml

@@ -17,6 +17,6 @@ jobs:
           make -j8
     - name: test-multithread (5X)
       run: |
-        sudo apt-get install -y tcl8.5
+        sudo apt-get install -y tcl tcl-tls
         ./runtest --loopn 5 --config server-threads 3 --clients 5 --endurance
 

README.md

@@ -1,6 +1,5 @@
 ![Current Release](https://img.shields.io/github/release/JohnSully/KeyDB.svg)
 ![CI](https://github.com/JohnSully/KeyDB/workflows/CI/badge.svg?branch=unstable)
-[![Join the chat at https://gitter.im/KeyDB/community](https://badges.gitter.im/KeyDB/community.svg)](https://gitter.im/KeyDB/community?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 [![StackShare](http://img.shields.io/badge/tech-stack-0690fa.svg?style=flat)](https://stackshare.io/eq-alpha-technology-inc/eq-alpha-technology-inc)
 
 ##### New! Want to extend KeyDB with Javascript?  Try [ModJS](https://github.com/JohnSully/ModJS)

TLS.md

@@ -56,8 +56,6 @@ Note that unlike Redis, KeyDB fully supports multithreading of TLS connections.
 To-Do List
 ----------
 
-- [ ] Add session caching support. Check if/how it's handled by clients to
-  assess how useful/important it is.
 - [ ] redis-benchmark support. The current implementation is a mix of using
   hiredis for parsing and basic networking (establishing connections), but
   directly manipulating sockets for most actions. This will need to be cleaned

keydb.conf

@@ -199,6 +199,22 @@ tcp-keepalive 300
 #
 # tls-prefer-server-ciphers yes
 
+# By default, TLS session caching is enabled to allow faster and less expensive
+# reconnections by clients that support it. Use the following directive to disable
+# caching.
+#
+# tls-session-caching no
+
+# Change the default number of TLS sessions cached. A zero value sets the cache
+# to unlimited size. The default size is 20480.
+#
+# tls-session-cache-size 5000
+
+# Change the default timeout of cached TLS sessions. The default timeout is 300
+# seconds.
+#
+# tls-session-cache-timeout 60
+
 ################################# GENERAL #####################################
 
 # By default KeyDB does not run as a daemon. Use 'yes' if you need it.
@@ -401,6 +417,20 @@ dir ./
 #
 replica-serve-stale-data yes
 
+# Active Replicas will allow read only data access while loading remote RDBs
+# provided they are permitted to serve stale data.  As an option you may also
+# permit them to accept write commands.  This is an EXPERIMENTAL feature and
+# may result in commands not being fully synchronized
+#
+# allow-write-during-load no
+
+# You can modify the number of masters necessary to form a replica quorum when
+# multi-master is enabled and replica-serve-stale-data is "no".  By default 
+# this is set to -1 which implies the number of known masters (e.g. those
+# you added with replicaof)
+#
+# replica-quorum -1
+
 # You can configure a replica instance to accept writes or not. Writing against
 # a replica instance may be useful to store some ephemeral data (because data
 # written on a replica will be easily deleted after resync with the master) but

src/ae.cpp

@@ -258,9 +258,11 @@ int aeCreateRemoteFileEvent(aeEventLoop *eventLoop, int fd, int mask,
 
     if (fSynchronous)
     {
+        {
         std::unique_lock<std::mutex> ulock(cmd.pctl->mutexcv, std::adopt_lock);
         cmd.pctl->cv.wait(ulock);
         ret = cmd.pctl->rval;
+        }
         delete cmd.pctl;
     }
 
@@ -300,7 +302,7 @@ int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fSynch
     cmd.fLock = fLock;
     if (fSynchronous)
     {
-        cmd.pctl = new (MALLOC_LOCAL) aeCommandControl();
+        cmd.pctl = new (MALLOC_LOCAL) aeCommandControl;
         cmd.pctl->mutexcv.lock();
     }
 
@@ -311,9 +313,11 @@ int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fSynch
     int ret = AE_OK;
     if (fSynchronous)
     {
+        {
         std::unique_lock<std::mutex> ulock(cmd.pctl->mutexcv, std::adopt_lock);
         cmd.pctl->cv.wait(ulock);
         ret = cmd.pctl->rval;
+        }
         delete cmd.pctl;
     }
     return ret;

src/aelocker.h

@@ -9,11 +9,14 @@ class AeLocker
     {
     }
 
-    void arm(client *c) // if a client is passed, then the client is already locked
+    void arm(client *c, bool fIfNeeded = false) // if a client is passed, then the client is already locked
     {
         if (m_fArmed)
             return;
 
+        if (fIfNeeded && aeThreadOwnsLock())
+            return;
+
         serverAssertDebug(!GlobalLocksAcquired());
 
         if (c != nullptr)

src/aof.cpp

@@ -1426,7 +1426,8 @@ int rewriteAppendOnlyFileRio(rio *aof) {
         /* Iterate this DB writing every entry */
         while((de = dictNext(di)) != NULL) {
             sds keystr;
-            robj key, *o;
+            redisObjectStack key;
+            robj *o = nullptr;
 
             keystr = (sds)dictGetKey(de);
             o = (robj*)dictGetVal(de);

src/config.cpp

@@ -2180,7 +2180,7 @@ static int updateTlsCfg(char *val, char *prev, const char **err) {
     UNUSED(prev);
     UNUSED(err);
     if (tlsConfigure(&g_pserver->tls_ctx_config) == C_ERR) {
-        *err = "Unable to configure tls-cert-file. Check server logs.";
+        *err = "Unable to update TLS configuration. Check server logs.";
         return 0;
     }
     return 1;
@@ -2190,6 +2190,12 @@ static int updateTlsCfgBool(int val, int prev, const char **err) {
     UNUSED(prev);
     return updateTlsCfg(NULL, NULL, err);
 }
+
+static int updateTlsCfgInt(long long val, long long prev, const char **err) {
+    UNUSED(val);
+    UNUSED(prev);
+    return updateTlsCfg(NULL, NULL, err);
+}
 #endif  /* USE_OPENSSL */
 
 int fDummy = false;
@@ -2230,6 +2236,8 @@ standardConfig configs[] = {
     createBoolConfig("appendonly", NULL, MODIFIABLE_CONFIG, g_pserver->aof_enabled, 0, NULL, updateAppendonly),
     createBoolConfig("cluster-allow-reads-when-down", NULL, MODIFIABLE_CONFIG, g_pserver->cluster_allow_reads_when_down, 0, NULL, NULL),
     createBoolConfig("multi-master-no-forward", NULL, MODIFIABLE_CONFIG, cserver.multimaster_no_forward, 0, validateMultiMasterNoForward, NULL),
+    createBoolConfig("allow-write-during-load", NULL, MODIFIABLE_CONFIG, g_pserver->fWriteDuringActiveLoad, 0, NULL, NULL),
+    createBoolConfig("io-threads-do-reads", NULL, IMMUTABLE_CONFIG, fDummy, 0, NULL, NULL),
 
     /* String Configs */
     createStringConfig("aclfile", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->acl_filename, "", NULL, NULL),
@@ -2287,6 +2295,7 @@ standardConfig configs[] = {
     createIntConfig("min-replicas-to-write", "min-slaves-to-write", MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->repl_min_slaves_to_write, 0, INTEGER_CONFIG, NULL, updateGoodSlaves),
     createIntConfig("min-replicas-max-lag", "min-slaves-max-lag", MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->repl_min_slaves_max_lag, 10, INTEGER_CONFIG, NULL, updateGoodSlaves),
     createIntConfig("min-clients-per-thread", NULL, MODIFIABLE_CONFIG, 0, 400, cserver.thread_min_client_threshold, 50, INTEGER_CONFIG, NULL, NULL),
+    createIntConfig("replica-quorum", NULL, MODIFIABLE_CONFIG, -1, INT_MAX, g_pserver->repl_quorum, -1, INTEGER_CONFIG, NULL, NULL),
     /* Unsigned int configs */
     createUIntConfig("maxclients", NULL, MODIFIABLE_CONFIG, 1, UINT_MAX, g_pserver->maxclients, 10000, INTEGER_CONFIG, NULL, updateMaxclients),
 
@@ -2324,10 +2333,13 @@ standardConfig configs[] = {
 
 #ifdef USE_OPENSSL
     createIntConfig("tls-port", NULL, IMMUTABLE_CONFIG, 0, 65535, g_pserver->tls_port, 0, INTEGER_CONFIG, NULL, NULL), /* TCP port. */
+    createIntConfig("tls-session-cache-size", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->tls_ctx_config.session_cache_size, 20*1024, INTEGER_CONFIG, NULL, updateTlsCfgInt),
+    createIntConfig("tls-session-cache-timeout", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->tls_ctx_config.session_cache_timeout, 300, INTEGER_CONFIG, NULL, updateTlsCfgInt),
     createBoolConfig("tls-cluster", NULL, MODIFIABLE_CONFIG, g_pserver->tls_cluster, 0, NULL, NULL),
     createBoolConfig("tls-replication", NULL, MODIFIABLE_CONFIG, g_pserver->tls_replication, 0, NULL, NULL),
     createBoolConfig("tls-auth-clients", NULL, MODIFIABLE_CONFIG, g_pserver->tls_auth_clients, 1, NULL, NULL),
     createBoolConfig("tls-prefer-server-ciphers", NULL, MODIFIABLE_CONFIG, g_pserver->tls_ctx_config.prefer_server_ciphers, 0, NULL, updateTlsCfgBool),
+    createBoolConfig("tls-session-caching", NULL, MODIFIABLE_CONFIG, g_pserver->tls_ctx_config.session_caching, 1, NULL, updateTlsCfgBool),
     createStringConfig("tls-cert-file", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->tls_ctx_config.cert_file, NULL, NULL, updateTlsCfg),
     createStringConfig("tls-key-file", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->tls_ctx_config.key_file, NULL, NULL, updateTlsCfg),
     createStringConfig("tls-dh-params-file", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->tls_ctx_config.dh_params_file, NULL, NULL, updateTlsCfg),

src/db.cpp

@@ -93,7 +93,7 @@ static robj *lookupKey(redisDb *db, robj *key, int flags) {
         updateDbValAccess(de, flags);
 
         if (flags & LOOKUP_UPDATEMVCC) {
-            val->mvcc_tstamp = getMvccTstamp();
+            setMvccTstamp(val, getMvccTstamp());
         }
         return val;
     } else {
@@ -206,7 +206,9 @@ int dbAddCore(redisDb *db, robj *key, robj *val) {
     serverAssert(!val->FExpires());
     sds copy = sdsdup(szFromObj(key));
     int retval = dictAdd(db->pdict, copy, val);
-    val->mvcc_tstamp = key->mvcc_tstamp = getMvccTstamp();
+    uint64_t mvcc = getMvccTstamp();
+    setMvccTstamp(key, mvcc);
+    setMvccTstamp(val, mvcc);
 
     if (retval == DICT_OK)
     {
@@ -256,7 +258,7 @@ void dbOverwriteCore(redisDb *db, dictEntry *de, robj *key, robj *val, bool fUpd
     if (fUpdateMvcc) {
         if (val->getrefcount(std::memory_order_relaxed) == OBJ_SHARED_REFCOUNT)
             val = dupStringObject(val);
-        val->mvcc_tstamp = getMvccTstamp();
+        setMvccTstamp(val, getMvccTstamp());
     }
 
     dictSetVal(db->pdict, de, val);
@@ -291,12 +293,12 @@ int dbMerge(redisDb *db, robj *key, robj *val, int fReplace)
             return (dbAddCore(db, key, val) == DICT_OK);
 
         robj *old = (robj*)dictGetVal(de);
-        if (old->mvcc_tstamp <= val->mvcc_tstamp)
+        if (mvccFromObj(old) <= mvccFromObj(val))
         {
             dbOverwriteCore(db, de, key, val, false, true);
             return true;
         }
-        
+
         return false;
     }
     else
@@ -1393,7 +1395,7 @@ void setExpire(client *c, redisDb *db, robj *key, robj *subkey, long long when)
         db->setexpire->insert(e);
     }
 
-    int writable_slave = listLength(g_pserver->masters) && g_pserver->repl_slave_ro == 0;
+    int writable_slave = listLength(g_pserver->masters) && g_pserver->repl_slave_ro == 0 && !g_pserver->fActiveReplica;
     if (c && writable_slave && !(c->flags & CLIENT_MASTER))
         rememberSlaveKeyWithExpire(db,key);
 }
@@ -1430,7 +1432,7 @@ void setExpire(client *c, redisDb *db, robj *key, expireEntry &&e)
     ((robj*)dictGetVal(kde))->SetFExpires(true);
 
 
-    int writable_slave = listLength(g_pserver->masters) && g_pserver->repl_slave_ro == 0;
+    int writable_slave = listLength(g_pserver->masters) && g_pserver->repl_slave_ro == 0 && !g_pserver->fActiveReplica;
     if (c && writable_slave && !(c->flags & CLIENT_MASTER))
         rememberSlaveKeyWithExpire(db,key);
 }
@@ -1486,7 +1488,6 @@ void propagateExpire(redisDb *db, robj *key, int lazy) {
 void propagateSubkeyExpire(redisDb *db, int type, robj *key, robj *subkey)
 {
     robj *argv[3];
-    robj objT;
     redisCommand *cmd = nullptr;
     switch (type)
     {

src/defrag.cpp

@@ -55,7 +55,8 @@ bool replaceSateliteOSetKeyPtr(expireset &set, sds oldkey, sds newkey);
  * returns NULL in case the allocatoin wasn't moved.
  * when it returns a non-null value, the old pointer was already released
  * and should NOT be accessed. */
-void* activeDefragAlloc(void *ptr) {
+template<typename TPTR>
+TPTR* activeDefragAlloc(TPTR *ptr) {
     size_t size;
     void *newptr;
     if(!je_get_defrag_hint(ptr)) {
@@ -70,7 +71,14 @@ void* activeDefragAlloc(void *ptr) {
     newptr = zmalloc_no_tcache(size);
     memcpy(newptr, ptr, size);
     zfree_no_tcache(ptr);
-    return newptr;
+    return (TPTR*)newptr;
+}
+
+template<>
+robj* activeDefragAlloc(robj *o) {
+    void *pvSrc = allocPtrFromObj(o);
+    void *pvDst = activeDefragAlloc(pvSrc);
+    return objFromAllocPtr(pvDst);
 }
 
 /*Defrag helper for sds strings