Fix issue #300

No code change

.github/workflows/endurance.yml

@@ -2,7 +2,7 @@ name: Endurance
 
 on:
   schedule:
-    - cron: '30 * * * *'
+    - cron: '0 * * * *'
 
 jobs:
 

pkg/deb/master_changelog

@@ -1,3 +1,34 @@
+keydb (6:6.0.18-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * 6.0.18 - This release contains a number of improvements to stability and performance. 
+  * Additional tests were added with improvements to existing test reliability
+  * Merged Redis 6.0.10 into KeyDB
+  * Issues resolved in this release: #214, #222, #238, #257, #273, #276, #285
+
+ -- Ben Schermel <ben@eqalpha.com>  Fri, 26 Mar 2021 16:00:00 +0000
+
+keydb (6:6.0.16-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * 6.0.16 - This release contains significant performance improvements as well as fixes to the following issues: #234, #236, #233, #207, #229, #231
+  * Reduced memory consumption when Active Replication is not used
+  * Reduced CPU consumption
+  * Improved TLS performance
+  * replica-quorum config for multi-master.Intended to be used with "replica-serve-stale-data no", if at least N replicas are connected we will consider our master link up and serve data.
+  * Subkey expires performance improvement for big sets
+
+ -- Ben Schermel <ben@eqalpha.com>  Mon, 28 Sep 2020 16:00:00 +0000
+
+keydb (6:6.0.13-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * 6.0.13 New features added: KEYDB.MEXISTS, REPLICAOF REMOVE, multi-master-no-forward configuration option
+  * KEYDB.MEXISTS - return exactly which keys exist in the database not just the count
+  * REPLICAOF REMOVE - Remove specific masters in a multi master setup
+  * multi-master-no-forward configuration option - Reduce network traffic in mesh topology multimaster setups
+  * Redis 6.0.5 feature parity
+  * Fixed Issues: 194 Bad directive or wrong number of arguments error on 'lazyfree-lazy-user-del no'; 209 Timeout using modules; 210 Databases are not merged on multi-master sync; 211 rpm packages now signed; 212 rpm file permissions updated
+
+ -- Ben Schermel <ben@eqalpha.com>  Mon, 13 Jul 2020 8:00:00 +0000
+
 keydb (6:6.0.9-1distribution_placeholder) codename_placeholder; urgency=medium
 
   * 6.0.9 Addressing issues 187 - cpu lockup with subkey expire, 190 - missing sentinel binary in keydb-tools. 

pkg/rpm/generate_rpms.sh

@@ -6,6 +6,9 @@ version=$(grep KEYDB_REAL_VERSION $DIR/../../src/version.h | awk '{ printf $3 }'
 release=1 # by default this will always be 1 for keydb version structure. If build release version needs to be update you can modify here
 arch=$(uname -m)
 dist=el$(rpm -q --queryformat '%{VERSION}' centos-release | cut -d. -f1)
+if [[ "$dist" == "elpackage centos-release is not installed" ]]; then
+        dist=el$(rpm -q --queryformat '%{VERSION}' centos-linux-release  | cut -d. -f1)
+fi
 
 if [[ "$arch" != "aarch64" ]] && [[ "$arch" != "x86_64" ]]; then
 	echo "This script is only valid and tested for aarch64 and x86_64 architectures. You are trying to use: $arch"

src/ae.cpp

@@ -842,15 +842,9 @@ void aeSetAfterSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *aftersleep,
     eventLoop->aftersleepFlags = flags;
 }
 
-thread_local spin_worker tl_worker = nullptr;
-void setAeLockSetThreadSpinWorker(spin_worker worker)
-{
-    tl_worker = worker;
-}
-
 void aeAcquireLock()
 {
-    g_lock.lock(tl_worker);
+    g_lock.lock();
 }
 
 int aeTryAcquireLock(int fWeak)

src/ae.h

@@ -164,7 +164,6 @@ aeEventLoop *aeGetCurrentEventLoop();
 int aeResizeSetSize(aeEventLoop *eventLoop, int setsize);
 void aeSetDontWait(aeEventLoop *eventLoop, int noWait);
 
-void setAeLockSetThreadSpinWorker(spin_worker worker);
 void aeAcquireLock();
 int aeTryAcquireLock(int fWeak);
 void aeReleaseLock();

src/dict.cpp

@@ -127,7 +127,6 @@ int _dictInit(dict *d, dictType *type,
     d->privdata = privDataPtr;
     d->rehashidx = -1;
     d->iterators = 0;
-    d->asyncdata = nullptr;
     return DICT_OK;
 }
 
@@ -189,13 +188,13 @@ int dictExpand(dict *d, unsigned long size)
 int dictRehash(dict *d, int n) {
     int empty_visits = n*10; /* Max number of empty buckets to visit. */
     if (!dictIsRehashing(d)) return 0;
-    if (d->asyncdata) return 0;
 
     while(n-- && d->ht[0].used != 0) {
         dictEntry *de, *nextde;
 
         /* Note that rehashidx can't overflow as we are sure there are more
          * elements because ht[0].used != 0 */
+        assert(d->ht[0].size > (unsigned long)d->rehashidx);
         while(d->ht[0].table[d->rehashidx] == NULL) {
             d->rehashidx++;
             if (--empty_visits == 0) return 1;
@@ -231,144 +230,6 @@ int dictRehash(dict *d, int n) {
     return 1;
 }
 
-dictAsyncRehashCtl *dictRehashAsyncStart(dict *d, int buckets) {
-    if (!dictIsRehashing(d)) return 0;
-
-    d->asyncdata = new dictAsyncRehashCtl(d, d->asyncdata);
-
-    int empty_visits = buckets * 10;
-
-    while (d->asyncdata->queue.size() < (size_t)buckets && d->rehashidx < d->ht[0].size) {
-        dictEntry *de;
-
-        /* Note that rehashidx can't overflow as we are sure there are more
-         * elements because ht[0].used != 0 */
-        while(d->ht[0].table[d->rehashidx] == NULL) {
-            d->rehashidx++;
-            if (--empty_visits == 0) goto LDone;
-            if (d->rehashidx >= d->ht[0].size) goto LDone;
-        }
-
-        de = d->ht[0].table[d->rehashidx];
-        // We have to queue every node in the bucket, even if we go over our target size
-        while (de != nullptr) {
-            d->asyncdata->queue.emplace_back(de);
-            de = de->next;
-        }
-        d->rehashidx++;
-    }
-
-LDone:
-    if (d->asyncdata->queue.empty()) {
-        // We didn't find any work to do (can happen if there is a large gap in the hash table)
-        auto asyncT = d->asyncdata;
-        d->asyncdata = d->asyncdata->next;
-        delete asyncT;
-        return nullptr;
-    }
-    return d->asyncdata;
-}
-
-void dictRehashAsync(dictAsyncRehashCtl *ctl) {
-    for (size_t idx = ctl->hashIdx; idx < ctl->queue.size(); ++idx) {
-        auto &wi = ctl->queue[idx];
-        wi.hash = dictHashKey(ctl->dict, dictGetKey(wi.de));
-    }
-    ctl->hashIdx = ctl->queue.size();
-    ctl->done = true;
-}
-
-bool dictRehashSomeAsync(dictAsyncRehashCtl *ctl, size_t hashes) {
-    size_t max = std::min(ctl->hashIdx + hashes, ctl->queue.size());
-    for (; ctl->hashIdx < max; ++ctl->hashIdx) {
-        auto &wi = ctl->queue[ctl->hashIdx];
-        wi.hash = dictHashKey(ctl->dict, dictGetKey(wi.de));
-    }
-
-    if (ctl->hashIdx == ctl->queue.size()) ctl->done = true;
-    return ctl->hashIdx < ctl->queue.size();
-}
-
-void dictCompleteRehashAsync(dictAsyncRehashCtl *ctl, bool fFree) {
-    dict *d = ctl->dict;
-    assert(ctl->done);
-    
-    // Unlink ourselves
-    bool fUnlinked = false;
-    dictAsyncRehashCtl **pprev = &d->asyncdata;
-
-    while (*pprev != nullptr) {
-        if (*pprev == ctl) {
-            *pprev = ctl->next;
-            fUnlinked = true;
-            break;
-        }
-        pprev = &((*pprev)->next);
-    }
-
-    if (fUnlinked) {
-        if (ctl->next != nullptr && ctl->deGCList != nullptr) {
-            // An older work item may depend on our free list, so pass our free list to them
-            dictEntry **deGC = &ctl->next->deGCList;
-            while (*deGC != nullptr) deGC = &((*deGC)->next);
-            *deGC = ctl->deGCList;
-            ctl->deGCList = nullptr;
-        }
-    }
-
-    if (fUnlinked && !ctl->release) {
-        if (d->ht[0].table != nullptr) {    // can be null if we're cleared during the rehash
-            for (auto &wi : ctl->queue) {
-                // We need to remove it from the source hash table, and store it in the dest.
-                //  Note that it may have been deleted in the meantime and therefore not exist.
-                //  In this case it will be in the garbage collection list
-                
-                dictEntry **pdePrev = &d->ht[0].table[wi.hash & d->ht[0].sizemask];
-                while (*pdePrev != nullptr && *pdePrev != wi.de) {
-                    pdePrev = &((*pdePrev)->next);
-                }
-                if (*pdePrev != nullptr) {  // The element may be NULL if its in the GC list
-                    assert(*pdePrev == wi.de);
-                    *pdePrev = wi.de->next;
-                    // Now link it to the dest hash table
-                    wi.de->next = d->ht[1].table[wi.hash & d->ht[1].sizemask];
-                    d->ht[1].table[wi.hash & d->ht[1].sizemask] = wi.de;
-                    d->ht[0].used--;
-                    d->ht[1].used++;
-                }
-            }
-        }
-
-        /* Check if we already rehashed the whole table... */
-        if (d->ht[0].used == 0 && d->asyncdata == nullptr) {
-            zfree(d->ht[0].table);
-            d->ht[0] = d->ht[1];
-            _dictReset(&d->ht[1]);
-            d->rehashidx = -1;
-        }
-    }
-
-    if (fFree) {
-        while (ctl->deGCList != nullptr) {
-            auto next = ctl->deGCList->next;
-            dictFreeKey(d, ctl->deGCList);
-            dictFreeVal(d, ctl->deGCList);
-            zfree(ctl->deGCList);
-            ctl->deGCList = next;
-        }
-
-        // Was the dictionary free'd while we were in flight?
-        if (ctl->release) {
-            if (d->asyncdata != nullptr)
-                d->asyncdata->release = true;
-            else
-                dictRelease(d);
-        }
-
-        delete ctl;
-    }
-}
-
 long long timeInMilliseconds(void) {
     struct timeval tv;
 
@@ -524,14 +385,9 @@ static dictEntry *dictGenericDelete(dict *d, const void *key, int nofree) {
                 else
                     d->ht[table].table[idx] = he->next;
                 if (!nofree) {
-                    if (table == 0 && d->asyncdata != nullptr && idx < d->rehashidx) {
-                        he->next = d->asyncdata->deGCList;
-                        d->asyncdata->deGCList = he->next;
-                    } else {
-                        dictFreeKey(d, he);
-                        dictFreeVal(d, he);
-                        zfree(he);
-                    }
+                    dictFreeKey(d, he);
+                    dictFreeVal(d, he);
+                    zfree(he);
                 }
                 d->ht[table].used--;
                 return he;
@@ -579,14 +435,9 @@ dictEntry *dictUnlink(dict *ht, const void *key) {
  * to dictUnlink(). It's safe to call this function with 'he' = NULL. */
 void dictFreeUnlinkedEntry(dict *d, dictEntry *he) {
     if (he == NULL) return;
-    if (d->asyncdata) {
-        he->next = d->asyncdata->deGCList;
-        d->asyncdata->deGCList = he;
-    } else { 
-        dictFreeKey(d, he);
-        dictFreeVal(d, he);
-        zfree(he);
-    }
+    dictFreeKey(d, he);
+    dictFreeVal(d, he);
+    zfree(he);
 }
 
 /* Destroy an entire dictionary */
@@ -602,14 +453,9 @@ int _dictClear(dict *d, dictht *ht, void(callback)(void *)) {
         if ((he = ht->table[i]) == NULL) continue;
         while(he) {
             nextHe = he->next;
-            if (d->asyncdata && i < d->rehashidx) {
-                he->next = d->asyncdata->deGCList;
-                d->asyncdata->deGCList = he;
-            } else {
-                dictFreeKey(d, he);
-                dictFreeVal(d, he);
-                zfree(he);
-            }
+            dictFreeKey(d, he);
+            dictFreeVal(d, he);
+            zfree(he);
             ht->used--;
             he = nextHe;
         }
@@ -624,10 +470,6 @@ int _dictClear(dict *d, dictht *ht, void(callback)(void *)) {
 /* Clear & Release the hash table */
 void dictRelease(dict *d)
 {
-    if (d->asyncdata) {
-        d->asyncdata->release = true;
-        return;
-    }
     _dictClear(d,&d->ht[0],NULL);
     _dictClear(d,&d->ht[1],NULL);
     zfree(d);

src/dict.h

@@ -36,8 +36,6 @@
 #include <stdint.h>
 
 #ifdef __cplusplus
-#include <vector>
-#include <atomic>
 extern "C" {
 #endif
 
@@ -83,40 +81,12 @@ typedef struct dictht {
     unsigned long used;
 } dictht;
 
-#ifdef __cplusplus
-struct dictAsyncRehashCtl {
-    struct workItem {
-        dictEntry *de;
-        uint64_t hash;
-        workItem(dictEntry *de) {
-            this->de = de;
-        }
-    };
-
-    static const int c_targetQueueSize = 512;
-    dictEntry *deGCList = nullptr;
-    struct dict *dict = nullptr;
-    std::vector<workItem> queue;
-    size_t hashIdx = 0;
-    bool release = false;
-    dictAsyncRehashCtl *next = nullptr;
-    std::atomic<bool> done { false };
-
-    dictAsyncRehashCtl(struct dict *d, dictAsyncRehashCtl *next) : dict(d), next(next) {
-        queue.reserve(c_targetQueueSize);
-    }
-};
-#else
-struct  dictAsyncRehashCtl;
-#endif
-
 typedef struct dict {
     dictType *type;
     void *privdata;
     dictht ht[2];
     long rehashidx; /* rehashing not in progress if rehashidx == -1 */
     unsigned long iterators; /* number of iterators currently running */
-    dictAsyncRehashCtl *asyncdata;
 } dict;
 
 /* If safe is set to 1 this is a safe iterator, that means, you can call
@@ -220,12 +190,6 @@ unsigned long dictScan(dict *d, unsigned long v, dictScanFunction *fn, dictScanB
 uint64_t dictGetHash(dict *d, const void *key);
 dictEntry **dictFindEntryRefByPtrAndHash(dict *d, const void *oldptr, uint64_t hash);
 
-/* Async Rehash Functions */
-dictAsyncRehashCtl *dictRehashAsyncStart(dict *d, int buckets = dictAsyncRehashCtl::c_targetQueueSize);
-void dictRehashAsync(dictAsyncRehashCtl *ctl);
-bool dictRehashSomeAsync(dictAsyncRehashCtl *ctl, size_t hashes);
-void dictCompleteRehashAsync(dictAsyncRehashCtl *ctl, bool fFree);
-
 /* Hash table types */
 extern dictType dictTypeHeapStringCopyKey;
 extern dictType dictTypeHeapStrings;

src/fastlock.cpp

@@ -338,7 +338,7 @@ extern "C" void fastlock_init(struct fastlock *lock, const char *name)
 }
 
 #ifndef ASM_SPINLOCK
-extern "C" void fastlock_lock(struct fastlock *lock, spin_worker worker)
+extern "C" void fastlock_lock(struct fastlock *lock)
 {
     int pidOwner;
     __atomic_load(&lock->m_pidOwner, &pidOwner, __ATOMIC_ACQUIRE);
@@ -356,32 +356,20 @@ extern "C" void fastlock_lock(struct fastlock *lock, spin_worker worker)
     __atomic_load(&g_fHighCpuPressure, &fHighPressure, __ATOMIC_RELAXED);
     unsigned loopLimit = fHighPressure ? 0x10000 : 0x100000;
 
-    if (worker != nullptr) {
-        for (;;) {
-            __atomic_load(&lock->m_ticket.u, &ticketT.u, __ATOMIC_ACQUIRE);
-            if ((ticketT.u & 0xffff) == myticket)
-                break;
-            if (!worker())
-                goto LNormalLoop;
-        }
-    } else {
-LNormalLoop:
-        for (;;)
-        {
-            __atomic_load(&lock->m_ticket.u, &ticketT.u, __ATOMIC_ACQUIRE);
-            if ((ticketT.u & 0xffff) == myticket)
-                break;
+    for (;;)
+    {
+        __atomic_load(&lock->m_ticket.u, &ticketT.u, __ATOMIC_ACQUIRE);
+        if ((ticketT.u & 0xffff) == myticket)
+            break;
 
 #if defined(__i386__) || defined(__amd64__)
-            __asm__ __volatile__ ("pause");
+        __asm__ __volatile__ ("pause");
 #elif defined(__aarch64__)
-            __asm__ __volatile__ ("yield");
+        __asm__ __volatile__ ("yield");
 #endif
-
-            if ((++cloops % loopLimit) == 0)
-            {
-                fastlock_sleep(lock, tid, ticketT.u, myticket);
-            }
+        if ((++cloops % loopLimit) == 0)
+        {
+            fastlock_sleep(lock, tid, ticketT.u, myticket);
         }
     }
 

src/fastlock.h

@@ -6,16 +6,10 @@
 extern "C" {
 #endif
 
-typedef int (*spin_worker)();
-
 /* Begin C API */
 struct fastlock;
 void fastlock_init(struct fastlock *lock, const char *name);
-#ifdef __cplusplus
-void fastlock_lock(struct fastlock *lock, spin_worker worker = nullptr);
-#else
-void fastlock_lock(struct fastlock *lock, spin_worker worker);
-#endif
+void fastlock_lock(struct fastlock *lock);
 int fastlock_trylock(struct fastlock *lock, int fWeak);
 void fastlock_unlock(struct fastlock *lock);
 void fastlock_free(struct fastlock *lock);
@@ -62,9 +56,9 @@ struct fastlock
         fastlock_init(this, name);
     }
 
-    inline void lock(spin_worker worker = nullptr)
+    inline void lock()
     {
-        fastlock_lock(this, worker);
+        fastlock_lock(this);
     }
 
     inline bool try_lock(bool fWeak = false)

src/fastlock_x64.asm

@@ -22,22 +22,14 @@ fastlock_lock:
 	# [rdi+64] ...
 	#	uint16_t active
 	#	uint16_t avail
-	#
-	# RSI points to a spin function to call, or NULL
 	
 	# First get our TID and put it in ecx
-	sub rsp, 24				# We only use 16 bytes, but we also need the stack aligned
-	.cfi_adjust_cfa_offset 24
-	mov [rsp], rdi			# we need our struct pointer (also balance the stack for the call)
-	mov [rsp+8], rsi		# backup the spin  function
-	
+	push rdi                # we need our struct pointer (also balance the stack for the call)
+	.cfi_adjust_cfa_offset 8
 	call gettid             # get our thread ID (TLS is nasty in ASM so don't bother inlining)
 	mov esi, eax            # back it up in esi
-
-	mov rdi, [rsp]          # Restore spin struct
-	mov r8, [rsp+8]         # restore the function (in a different register)
-	add rsp, 24
-	.cfi_adjust_cfa_offset -24
+	pop rdi                 # get our pointer back
+	.cfi_adjust_cfa_offset -8
 
 	cmp [rdi], esi          # Is the TID we got back the owner of the lock?
 	je .LLocked             # Don't spin in that case
@@ -55,8 +47,6 @@ fastlock_lock:
 	# ax now contains the ticket
 	# OK Start the wait loop
 	xor ecx, ecx
-	test r8, r8
-	jnz .LLoopFunction
 .ALIGN 16
 .LLoop:
 	mov edx, [rdi+64]
@@ -93,40 +83,6 @@ fastlock_lock:
 	mov [rdi], esi          # lock->m_pidOwner = gettid()
 	inc dword ptr [rdi+4]   # lock->m_depth++
 	ret
-
-.LLoopFunction:
-	sub rsp, 40
-	.cfi_adjust_cfa_offset 40
-	xor ecx, ecx
-	mov [rsp], rcx
-	mov [rsp+8], r8
-	mov [rsp+16], rdi
-	mov [rsp+24], rsi
-	mov [rsp+32], eax
-.LLoopFunctionCore:
-	mov edx, [rdi+64]
-	cmp dx, ax
-	je .LExitLoopFunction
-	mov r8, [rsp+8]
-	call r8
-	test eax, eax
-	jz .LExitLoopFunctionForNormal
-	mov eax, [rsp+32]    # restore clobbered eax
-	mov rdi, [rsp+16]
-	jmp .LLoopFunctionCore
-.LExitLoopFunction:
-	mov rsi, [rsp+24]
-	add rsp, 40
-	.cfi_adjust_cfa_offset -40
-	jmp .LLocked
-.LExitLoopFunctionForNormal:
-	xor ecx, ecx
-	mov rdi, [rsp+16]
-	mov rsi, [rsp+24]
-	mov eax, [rsp+32]
-	add rsp, 40
-	.cfi_adjust_cfa_offset -40
-	jmp .LLoop
 .cfi_endproc
 
 .ALIGN 16

src/networking.cpp

@@ -2315,6 +2315,8 @@ void processInputBuffer(client *c, int callFlags) {
         /* Immediately abort if the client is in the middle of something. */
         if (c->flags & CLIENT_BLOCKED) break;
 
+        if (c->flags & CLIENT_EXECUTING_COMMAND) break;
+
         /* Don't process input from the master while there is a busy script
          * condition on the replica. We want just to accumulate the replication
          * stream (instead of replying -BUSY like we do with other clients) and
@@ -2349,13 +2351,16 @@ void processInputBuffer(client *c, int callFlags) {
         if (c->argc == 0) {
             resetClient(c);
         } else {
+            c->flags |= CLIENT_EXECUTING_COMMAND;
             /* We are finally ready to execute the command. */
             if (processCommandAndResetClient(c, callFlags) == C_ERR) {
                 /* If the client is no longer valid, we avoid exiting this
                  * loop and trimming the client buffer later. So we return
                  * ASAP in that case. */
+                c->flags &= ~CLIENT_EXECUTING_COMMAND;
                 return;
             }
+            c->flags &= ~CLIENT_EXECUTING_COMMAND;
         }
     }
 
@@ -3486,16 +3491,6 @@ void processEventsWhileBlocked(int iel) {
     locker.arm(nullptr);
     locker.release();
 
-    // Try to complete any async rehashes (this would normally happen in dbCron, but that won't run here)
-    for (int idb = 0; idb < cserver.dbnum; ++idb) {
-        redisDb *db = &g_pserver->db[idb];
-        while (db->dict->asyncdata != nullptr) {
-            if (!db->dict->asyncdata->done)
-                break;
-            dictCompleteRehashAsync(db->dict->asyncdata, false /*fFree*/);
-        }
-    }
-
     // Restore it so the calling code is not confused
     if (fReplBacklog) {
         g_pserver->repl_batch_idxStart = g_pserver->repl_backlog_idx;

src/server.cpp

@@ -1508,12 +1508,13 @@ void tryResizeHashTables(int dbid) {
  * table will use two tables for a long time. So we try to use 1 millisecond
  * of CPU time at every call of this function to perform some rehashing.
  *
- * The function returns the number of rehashes if some rehashing was performed, otherwise 0
+ * The function returns 1 if some rehashing was performed, otherwise 0
  * is returned. */
 int incrementallyRehash(int dbid) {
     /* Keys dictionary */
     if (dictIsRehashing(g_pserver->db[dbid].dict)) {
-        return dictRehashMilliseconds(g_pserver->db[dbid].dict,1);
+        dictRehashMilliseconds(g_pserver->db[dbid].dict,1);
+        return 1; /* already used our millisecond for this loop... */
     }
     return 0;
 }
@@ -1771,33 +1772,23 @@ bool expireOwnKeys()
     return false;
 }
 
-int hash_spin_worker() {
-    auto ctl = serverTL->rehashCtl;
-    return dictRehashSomeAsync(ctl, 1);
-}
-
 /* This function handles 'background' operations we are required to do
  * incrementally in Redis databases, such as active key expiring, resizing,
  * rehashing. */
-void databasesCron(bool fMainThread) {
-    serverAssert(GlobalLocksAcquired());
-    static int rehashes_per_ms = 0;
-    static int async_rehashes = 0;
-    if (fMainThread) {
-        /* Expire keys by random sampling. Not required for slaves
-        * as master will synthesize DELs for us. */
-        if (g_pserver->active_expire_enabled) {
-            if (expireOwnKeys()) {
-                activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);
-            } else {
-                expireSlaveKeys();
-            }
+void databasesCron(void) {
+    /* Expire keys by random sampling. Not required for slaves
+     * as master will synthesize DELs for us. */
+    if (g_pserver->active_expire_enabled) {
+        if (expireOwnKeys()) {
+            activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);
+        } else {
+            expireSlaveKeys();
         }
-
-        /* Defrag keys gradually. */
-        activeDefragCycle();
     }
 
+    /* Defrag keys gradually. */
+    activeDefragCycle();
+
     /* Perform hash tables rehashing if needed, but only if there are no
      * other processes saving the DB on disk. Otherwise rehashing is bad
      * as will cause a lot of copy-on-write of memory pages. */
@@ -1813,69 +1804,28 @@ void databasesCron(bool fMainThread) {
         /* Don't test more DBs than we have. */
         if (dbs_per_call > cserver.dbnum) dbs_per_call = cserver.dbnum;
 
-        if (fMainThread) {
-            /* Resize */
-            for (j = 0; j < dbs_per_call; j++) {
-                tryResizeHashTables(resize_db % cserver.dbnum);
-                resize_db++;
-            }
+        /* Resize */
+        for (j = 0; j < dbs_per_call; j++) {
+            tryResizeHashTables(resize_db % cserver.dbnum);
+            resize_db++;
         }
 
         /* Rehash */
         if (g_pserver->activerehashing) {
             for (j = 0; j < dbs_per_call; j++) {
-                if (serverTL->rehashCtl != nullptr) {
-                    if (dictRehashSomeAsync(serverTL->rehashCtl, 5)) {
-                        break;
-                    } else {
-                        dictCompleteRehashAsync(serverTL->rehashCtl, true /*fFree*/);
-                        serverTL->rehashCtl = nullptr;
-                    }
-                }
-
-                serverAssert(serverTL->rehashCtl == nullptr);
-                /* Are we async rehashing? And if so is it time to re-calibrate? */
-                /* The recalibration limit is a prime number to ensure balancing across threads */
-                if (rehashes_per_ms > 0 && async_rehashes < 131 && cserver.active_defrag_enabled) {
-                    serverTL->rehashCtl = dictRehashAsyncStart(g_pserver->db[rehash_db].dict, rehashes_per_ms);
-                    ++async_rehashes;
-                }
-                if (serverTL->rehashCtl)
-                    break;
-                
-                // Before starting anything new, can we end the rehash of a blocked thread?
-                if (g_pserver->db[rehash_db].dict->asyncdata != nullptr) {
-                    auto asyncdata = g_pserver->db[rehash_db].dict->asyncdata;
-                    if (asyncdata->done) {
-                        dictCompleteRehashAsync(asyncdata, false /*fFree*/);    // Don't free because we don't own the pointer
-                        serverAssert(g_pserver->db[rehash_db].dict->asyncdata != asyncdata);
-                        break;  // completion can be expensive, don't do anything else
-                    }
-                }
-
-                rehashes_per_ms = incrementallyRehash(rehash_db);
-                async_rehashes = 0;
-                if (rehashes_per_ms > 0) {
+                int work_done = incrementallyRehash(rehash_db);
+                if (work_done) {
                     /* If the function did some work, stop here, we'll do
-                    * more at the next cron loop. */
-                    if (!cserver.active_defrag_enabled) {
-                        serverLog(LL_VERBOSE, "Calibrated rehashes per ms: %d", rehashes_per_ms);
-                    }
+                     * more at the next cron loop. */
                     break;
-                } else if (g_pserver->db[rehash_db].dict->asyncdata == nullptr) {
-                    /* If this db didn't need rehash and we have none in flight, we'll try the next one. */
+                } else {
+                    /* If this db didn't need rehash, we'll try the next one. */
                     rehash_db++;
                     rehash_db %= cserver.dbnum;
                 }
             }
         }
     }
-
-    if (serverTL->rehashCtl) {
-        setAeLockSetThreadSpinWorker(hash_spin_worker);
-    } else {
-        setAeLockSetThreadSpinWorker(nullptr);
-    }
 }
 
 /* We take a cached value of the unix time in the global state because with
@@ -2115,7 +2065,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     clientsCron(IDX_EVENT_LOOP_MAIN);
 
     /* Handle background operations on Redis databases. */
-    databasesCron(true /* fMainThread */);
+    databasesCron();
 
     /* Start a scheduled AOF rewrite if this was requested by the user while
      * a BGSAVE was in progress. */
@@ -2265,9 +2215,6 @@ int serverCronLite(struct aeEventLoop *eventLoop, long long id, void *clientData
         processUnblockedClients(iel);
     }
 
-    /* Handle background operations on Redis databases. */
-    databasesCron(false /* fMainThread */);
-
     /* Unpause clients if enough time has elapsed */
     unpauseClientsIfNecessary();
     

src/server.h

@@ -441,10 +441,8 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
 #define CLIENT_PENDING_READ (1<<29) /* The client has pending reads and was put
                                        in the list of clients we can read
                                        from. */
-#define CLIENT_PENDING_COMMAND (1<<30) /* Used in threaded I/O to signal after
-                                          we return single threaded that the
-                                          client has already pending commands
-                                          to be executed. */
+#define CLIENT_EXECUTING_COMMAND (1<<30) /* Used to handle reentrency cases in processCommandWhileBlocked 
+                                            to ensure we don't process a client already executing */
 #define CLIENT_TRACKING (1ULL<<31) /* Client enabled keys tracking in order to
                                    perform client side caching. */
 #define CLIENT_TRACKING_BROKEN_REDIR (1ULL<<32) /* Target client is invalid. */
@@ -1393,7 +1391,6 @@ struct redisServerThreadVars {
     long unsigned commandsExecuted = 0;
     bool fRetrySetAofEvent = false;
     std::vector<client*> vecclientsProcess;
-    dictAsyncRehashCtl *rehashCtl = nullptr;
 };
 
 struct redisMaster {

src/version.h

@@ -1,4 +1,4 @@
-#define KEYDB_REAL_VERSION "255.255.255"
-#define KEYDB_VERSION_NUM 0x00ffffff
+#define KEYDB_REAL_VERSION "6.0.18"
+#define KEYDB_VERSION_NUM 0x00060012
 extern const char *KEYDB_SET_VERSION;   // Unlike real version, this can be overriden by the config
 

tests/unit/scripting.tcl

@@ -430,6 +430,21 @@ start_server {tags {"scripting"}} {
         set res
     } {102}
 
+    test {EVAL with pipelined command (No crash)} {
+        r flushall
+        r config set lua-time-limit 1
+        set rd [redis_deferring_client]
+        $rd eval {for i=1,1000000 do redis.call('set', i, 'sdfdsfd') end} 0
+        $rd set testkey foo
+        $rd get testkey
+        after 1200
+        catch {r echo "foo"} err
+        assert_match {BUSY*} $err
+        $rd read
+        $rd close
+    }
+
+
     test {EVAL timeout from AOF} {
         # generate a long running script that is propagated to the AOF as script
         # make sure that the script times out during loading