check for musl in logging.tcl as backtrace() is not available with it

update rpm script for centos8

Dockerfile

@@ -0,0 +1,11 @@
+FROM alpine
+
+WORKDIR /repo
+
+RUN apk add --no-cache coreutils gcc linux-headers make musl-dev util-linux-dev openssl-dev curl-dev g++ bash git perl tcl
+
+COPY . .
+
+RUN make distclean && make -j
+
+CMD ["./runtest"]

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"

runtest-moduleapi

@@ -29,4 +29,6 @@ $TCLSH tests/test_helper.tcl \
 --single unit/moduleapi/blockedclient \
 --single unit/moduleapi/moduleloadsave \
 --single unit/moduleapi/getkeys \
+--single unit/moduleapi/timers \
+--config server-threads 3 \
 "${@}"

src/Makefile

@@ -48,8 +48,8 @@ endif
 USEASM?=true
 
 ifneq ($(strip $(SANITIZE)),)
-	CFLAGS+= -fsanitize=$(SANITIZE) -DSANITIZE
-	CXXFLAGS+= -fsanitize=$(SANITIZE) -DSANITIZE
+	CFLAGS+= -fsanitize=$(SANITIZE) -DSANITIZE  -fno-omit-frame-pointer
+	CXXFLAGS+= -fsanitize=$(SANITIZE) -DSANITIZE  -fno-omit-frame-pointer
 	LDFLAGS+= -fsanitize=$(SANITIZE)
 	MALLOC=libc
 	USEASM=false

src/ae.cpp

@@ -109,13 +109,6 @@ enum class AE_ASYNC_OP
     CreateFileEvent,
 };
 
-struct aeCommandControl
-{
-    std::condition_variable cv;
-    std::atomic<int> rval;
-    std::mutex mutexcv;
-};
-
 struct aeCommand
 {
     AE_ASYNC_OP op;
@@ -128,7 +121,6 @@ struct aeCommand
         std::function<void()> *pfn;
     };
     void *clientData;
-    aeCommandControl *pctl;
 };
 
 void aeProcessCmd(aeEventLoop *eventLoop, int fd, void *, int )
@@ -149,19 +141,7 @@ void aeProcessCmd(aeEventLoop *eventLoop, int fd, void *, int )
             break;
 
         case AE_ASYNC_OP::CreateFileEvent:
-        {
-            if (cmd.pctl != nullptr)
-            {
-                cmd.pctl->mutexcv.lock();
-                std::atomic_store(&cmd.pctl->rval, aeCreateFileEvent(eventLoop, cmd.fd, cmd.mask, cmd.fproc, cmd.clientData));
-                cmd.pctl->cv.notify_all();
-                cmd.pctl->mutexcv.unlock();
-            }
-            else
-            {
-                aeCreateFileEvent(eventLoop, cmd.fd, cmd.mask, cmd.fproc, cmd.clientData);
-            }
-        }
+            aeCreateFileEvent(eventLoop, cmd.fd, cmd.mask, cmd.fproc, cmd.clientData);
             break;
 
         case AE_ASYNC_OP::PostFunction:
@@ -175,19 +155,11 @@ void aeProcessCmd(aeEventLoop *eventLoop, int fd, void *, int )
 
         case AE_ASYNC_OP::PostCppFunction:
         {
-            if (cmd.pctl != nullptr)
-                cmd.pctl->mutexcv.lock();
-            
             std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
             if (cmd.fLock)
                 ulock.lock();
             (*cmd.pfn)();
-            
-            if (cmd.pctl != nullptr)
-            {
-                cmd.pctl->cv.notify_all();
-                cmd.pctl->mutexcv.unlock();
-            }
+
             delete cmd.pfn;
         }
             break;
@@ -226,7 +198,7 @@ ssize_t safe_write(int fd, const void *pv, size_t cb)
 }
 
 int aeCreateRemoteFileEvent(aeEventLoop *eventLoop, int fd, int mask,
-        aeFileProc *proc, void *clientData, int fSynchronous)
+        aeFileProc *proc, void *clientData)
 {
     if (eventLoop == g_eventLoopThisThread)
         return aeCreateFileEvent(eventLoop, fd, mask, proc, clientData);
@@ -239,13 +211,7 @@ int aeCreateRemoteFileEvent(aeEventLoop *eventLoop, int fd, int mask,
     cmd.mask = mask;
     cmd.fproc = proc;
     cmd.clientData = clientData;
-    cmd.pctl = nullptr;
     cmd.fLock = true;
-    if (fSynchronous)
-    {
-        cmd.pctl = new (MALLOC_LOCAL) aeCommandControl();
-        cmd.pctl->mutexcv.lock();
-    }
 
     auto size = safe_write(eventLoop->fdCmdWrite, &cmd, sizeof(cmd));
     if (size != sizeof(cmd))
@@ -254,16 +220,6 @@ int aeCreateRemoteFileEvent(aeEventLoop *eventLoop, int fd, int mask,
         serverAssert(errno == EAGAIN);
         ret = AE_ERR;
     }
-    
-    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;
 }
@@ -286,7 +242,7 @@ int aePostFunction(aeEventLoop *eventLoop, aePostFunctionProc *proc, void *arg)
     return AE_OK;
 }
 
-int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fSynchronous, bool fLock, bool fForceQueue)
+int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fLock, bool fForceQueue)
 {
     if (eventLoop == g_eventLoopThisThread && !fForceQueue)
     {
@@ -297,13 +253,7 @@ int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fSynch
     aeCommand cmd = {};
     cmd.op = AE_ASYNC_OP::PostCppFunction;
     cmd.pfn = new (MALLOC_LOCAL) std::function<void()>(fn);
-    cmd.pctl = nullptr;
     cmd.fLock = fLock;
-    if (fSynchronous)
-    {
-        cmd.pctl = new (MALLOC_LOCAL) aeCommandControl;
-        cmd.pctl->mutexcv.lock();
-    }
 
     auto size = write(eventLoop->fdCmdWrite, &cmd, sizeof(cmd));
     if (!(!size || size == sizeof(cmd))) {
@@ -314,17 +264,7 @@ int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fSynch
     if (size == 0)
         return AE_ERR;
 
-    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;
+    return AE_OK;
 }
 
 aeEventLoop *aeCreateEventLoop(int setsize) {
@@ -902,9 +842,15 @@ 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();
+    g_lock.lock(tl_worker);
 }
 
 int aeTryAcquireLock(int fWeak)

src/ae.h

@@ -135,7 +135,7 @@ aeEventLoop *aeCreateEventLoop(int setsize);
 int aePostFunction(aeEventLoop *eventLoop, aePostFunctionProc *proc, void *arg);
 #ifdef __cplusplus
 }   // EXTERN C
-int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fSynchronous = false, bool fLock = true, bool fForceQueue = false);
+int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fLock = true, bool fForceQueue = false);
 extern "C" {
 #endif
 void aeDeleteEventLoop(aeEventLoop *eventLoop);
@@ -144,7 +144,7 @@ int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
         aeFileProc *proc, void *clientData);
 
 int aeCreateRemoteFileEvent(aeEventLoop *eventLoop, int fd, int mask,
-        aeFileProc *proc, void *clientData, int fSynchronous);
+        aeFileProc *proc, void *clientData);
 
 void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask);
 void aeDeleteFileEventAsync(aeEventLoop *eventLoop, int fd, int mask);
@@ -164,6 +164,7 @@ 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/cluster.cpp

@@ -5066,12 +5066,15 @@ void mvccrestoreCommand(client *c) {
     setMvccTstamp(obj, mvcc);
 
     /* Create the key and set the TTL if any */
-    dbMerge(c->db,key,obj,true);
-    if (expire >= 0) {
-        setExpire(c,c->db,key,nullptr,expire);
+    if (dbMerge(c->db,key,obj,true)) {
+        if (expire >= 0) {
+            setExpire(c,c->db,key,nullptr,expire);
+        }
+        signalModifiedKey(c,c->db,key);
+        notifyKeyspaceEvent(NOTIFY_GENERIC,"restore",key,c->db->id);
+    } else {
+        decrRefCount(obj);
     }
-    signalModifiedKey(c,c->db,key);
-    notifyKeyspaceEvent(NOTIFY_GENERIC,"restore",key,c->db->id);
     addReply(c,shared.ok);
     g_pserver->dirty++;
 }

src/dict.cpp

@@ -127,6 +127,7 @@ int _dictInit(dict *d, dictType *type,
     d->privdata = privDataPtr;
     d->rehashidx = -1;
     d->iterators = 0;
+    d->asyncdata = nullptr;
     return DICT_OK;
 }
 
@@ -188,13 +189,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;
@@ -230,6 +231,144 @@ 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;
 
@@ -385,9 +524,14 @@ static dictEntry *dictGenericDelete(dict *d, const void *key, int nofree) {
                 else
                     d->ht[table].table[idx] = he->next;
                 if (!nofree) {
-                    dictFreeKey(d, he);
-                    dictFreeVal(d, he);
-                    zfree(he);
+                    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);
+                    }
                 }
                 d->ht[table].used--;
                 return he;
@@ -435,9 +579,14 @@ 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;
-    dictFreeKey(d, he);
-    dictFreeVal(d, he);
-    zfree(he);
+    if (d->asyncdata) {
+        he->next = d->asyncdata->deGCList;
+        d->asyncdata->deGCList = he;
+    } else { 
+        dictFreeKey(d, he);
+        dictFreeVal(d, he);
+        zfree(he);
+    }
 }
 
 /* Destroy an entire dictionary */
@@ -453,9 +602,14 @@ int _dictClear(dict *d, dictht *ht, void(callback)(void *)) {
         if ((he = ht->table[i]) == NULL) continue;
         while(he) {
             nextHe = he->next;
-            dictFreeKey(d, he);
-            dictFreeVal(d, he);
-            zfree(he);
+            if (d->asyncdata && i < d->rehashidx) {
+                he->next = d->asyncdata->deGCList;
+                d->asyncdata->deGCList = he;
+            } else {
+                dictFreeKey(d, he);
+                dictFreeVal(d, he);
+                zfree(he);
+            }
             ht->used--;
             he = nextHe;
         }
@@ -470,6 +624,10 @@ 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,6 +36,8 @@
 #include <stdint.h>
 
 #ifdef __cplusplus
+#include <vector>
+#include <atomic>
 extern "C" {
 #endif
 
@@ -81,12 +83,40 @@ 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
@@ -190,6 +220,12 @@ 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/expire.cpp

@@ -136,7 +136,7 @@ void activeExpireCycleExpire(redisDb *db, expireEntry &e, long long now) {
                 executeCronJobExpireHook(keyCopy, val);
                 sdsfree(keyCopy);
                 decrRefCount(val);
-            }, false, true /*fLock*/, true /*fForceQueue*/);
+            }, true /*fLock*/, true /*fForceQueue*/);
         }
             return;
 

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)
+extern "C" void fastlock_lock(struct fastlock *lock, spin_worker worker)
 {
     int pidOwner;
     __atomic_load(&lock->m_pidOwner, &pidOwner, __ATOMIC_ACQUIRE);
@@ -356,20 +356,32 @@ extern "C" void fastlock_lock(struct fastlock *lock)
     __atomic_load(&g_fHighCpuPressure, &fHighPressure, __ATOMIC_RELAXED);
     unsigned loopLimit = fHighPressure ? 0x10000 : 0x100000;
 
-    for (;;)
-    {
-        __atomic_load(&lock->m_ticket.u, &ticketT.u, __ATOMIC_ACQUIRE);
-        if ((ticketT.u & 0xffff) == myticket)
-            break;
+    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;
 
 #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,10 +6,16 @@
 extern "C" {
 #endif
 
+typedef int (*spin_worker)();
+
 /* Begin C API */
 struct fastlock;
 void fastlock_init(struct fastlock *lock, const char *name);
-void fastlock_lock(struct fastlock *lock);
+#ifdef __cplusplus
+void fastlock_lock(struct fastlock *lock, spin_worker worker = nullptr);
+#else
+void fastlock_lock(struct fastlock *lock, spin_worker worker);
+#endif
 int fastlock_trylock(struct fastlock *lock, int fWeak);
 void fastlock_unlock(struct fastlock *lock);
 void fastlock_free(struct fastlock *lock);
@@ -56,9 +62,9 @@ struct fastlock
         fastlock_init(this, name);
     }
 
-    inline void lock()
+    inline void lock(spin_worker worker = nullptr)
     {
-        fastlock_lock(this);
+        fastlock_lock(this, worker);
     }
 
     inline bool try_lock(bool fWeak = false)

src/fastlock_x64.asm

@@ -22,14 +22,22 @@ 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
-	push rdi                # we need our struct pointer (also balance the stack for the call)
-	.cfi_adjust_cfa_offset 8
+	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
+	
 	call gettid             # get our thread ID (TLS is nasty in ASM so don't bother inlining)
 	mov esi, eax            # back it up in esi
-	pop rdi                 # get our pointer back
-	.cfi_adjust_cfa_offset -8
+
+	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
 
 	cmp [rdi], esi          # Is the TID we got back the owner of the lock?
 	je .LLocked             # Don't spin in that case
@@ -47,6 +55,8 @@ 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]
@@ -83,6 +93,40 @@ 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/module.cpp

@@ -5045,7 +5045,7 @@ void RM_FreeThreadSafeContext(RedisModuleCtx *ctx) {
     zfree(ctx);
 }
 
-static bool g_fModuleThread = false;
+thread_local bool g_fModuleThread = false;
 /* Acquire the server lock before executing a thread safe API call.
  * This is not needed for `RedisModule_Reply*` calls when there is
  * a blocked client connected to the thread safe context. */
@@ -5104,7 +5104,14 @@ void moduleAcquireGIL(int fServerThread) {
     }
     else
     {
-        s_mutexModule.lock();
+        // It is possible that another module thread holds the GIL (and s_mutexModule as a result). 
+        // When said thread goes to release the GIL, it will wait for s_mutex, which this thread owns. 
+        // This thread is however waiting for the GIL (and s_mutexModule) that the other thread owns.
+        // As a result, a deadlock has occured. 
+        // We release the lock on s_mutex and wait until we are able to safely acquire the GIL 
+        // in order to prevent this deadlock from occuring. 
+        while (!s_mutexModule.try_lock())
+            s_cv.wait(lock);         
         ++s_cAcquisitionsModule;
         fModuleGILWlocked++;
     }
@@ -5643,6 +5650,9 @@ int moduleTimerHandler(struct aeEventLoop *eventLoop, long long id, void *client
  * (If the time it takes to execute 'callback' is negligible the two
  * statements above mean the same) */
 RedisModuleTimerID RM_CreateTimer(RedisModuleCtx *ctx, mstime_t period, RedisModuleTimerProc callback, void *data) {
+    static uint64_t pending_key;
+    static mstime_t pending_period = -1; 
+
     RedisModuleTimer *timer = (RedisModuleTimer*)zmalloc(sizeof(*timer), MALLOC_LOCAL);
     timer->module = ctx->module;
     timer->callback = callback;
@@ -5661,32 +5671,40 @@ RedisModuleTimerID RM_CreateTimer(RedisModuleCtx *ctx, mstime_t period, RedisMod
         }
     }
 
+    bool fNeedPost = (pending_period < 0);    // If pending_period is already set, then a PostFunction is in flight and we don't need to set a new one
+    if (pending_period < 0 || period < pending_period) {
+        pending_period = period;
+        pending_key = key;
+    }
+
     /* We need to install the main event loop timer if it's not already
      * installed, or we may need to refresh its period if we just installed
      * a timer that will expire sooner than any other else (i.e. the timer
      * we just installed is the first timer in the Timers rax). */
-    if (aeTimer != -1) {
-        raxIterator ri;
-        raxStart(&ri,Timers);
-        raxSeek(&ri,"^",NULL,0);
-        raxNext(&ri);
-        if (memcmp(ri.key,&key,sizeof(key)) == 0) {
-            /* This is the first key, we need to re-install the timer according
-             * to the just added event. */
-            aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [&]{
-                aeDeleteTimeEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el,aeTimer);
-            }, true /* synchronous */, false /* fLock */);
-            aeTimer = -1;
-        }
-        raxStop(&ri);
-    }
+    if (fNeedPost) {
+        aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, []{
+            if (aeTimer != -1) {
+                raxIterator ri;
+                raxStart(&ri,Timers);
+                raxSeek(&ri,"^",NULL,0);
+                raxNext(&ri);
+                if (memcmp(ri.key,&pending_key,sizeof(key)) == 0) {
+                    /* This is the first key, we need to re-install the timer according
+                    * to the just added event. */
+                    aeDeleteTimeEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el,aeTimer);
+                    aeTimer = -1;
+                }
+                raxStop(&ri);
+            }
 
-    /* If we have no main timer (the old one was invalidated, or this is the
-     * first module timer we have), install one. */
-    if (aeTimer == -1) {
-        aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [&]{
-            aeTimer = aeCreateTimeEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el,period,moduleTimerHandler,NULL,NULL);
-        }, true /* synchronous */, false /* fLock */);
+            /* If we have no main timer (the old one was invalidated, or this is the
+            * first module timer we have), install one. */
+            if (aeTimer == -1) {
+                aeTimer = aeCreateTimeEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el,pending_period,moduleTimerHandler,NULL,NULL);
+            }
+
+            pending_period = -1;
+        });
     }
 
     return key;

src/networking.cpp

@@ -3486,6 +3486,16 @@ 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,13 +1508,12 @@ 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 1 if some rehashing was performed, otherwise 0
+ * The function returns the number of rehashes if some rehashing was performed, otherwise 0
  * is returned. */
 int incrementallyRehash(int dbid) {
     /* Keys dictionary */
     if (dictIsRehashing(g_pserver->db[dbid].dict)) {
-        dictRehashMilliseconds(g_pserver->db[dbid].dict,1);
-        return 1; /* already used our millisecond for this loop... */
+        return dictRehashMilliseconds(g_pserver->db[dbid].dict,1);
     }
     return 0;
 }
@@ -1772,22 +1771,32 @@ 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(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();
+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();
+            }
         }
-    }
 
-    /* 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
@@ -1804,28 +1813,69 @@ void databasesCron(void) {
         /* Don't test more DBs than we have. */
         if (dbs_per_call > cserver.dbnum) dbs_per_call = cserver.dbnum;
 
-        /* Resize */
-        for (j = 0; j < dbs_per_call; j++) {
-            tryResizeHashTables(resize_db % cserver.dbnum);
-            resize_db++;
+        if (fMainThread) {
+            /* 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++) {
-                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 (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;
-                } else {
-                    /* If this db didn't need rehash, we'll try the next one. */
+                
+                // 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) {
+                    /* 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);
+                    }
+                    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. */
                     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
@@ -2065,7 +2115,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     clientsCron(IDX_EVENT_LOOP_MAIN);
 
     /* Handle background operations on Redis databases. */
-    databasesCron();
+    databasesCron(true /* fMainThread */);
 
     /* Start a scheduled AOF rewrite if this was requested by the user while
      * a BGSAVE was in progress. */
@@ -2215,6 +2265,9 @@ 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();
     
@@ -4087,7 +4140,7 @@ bool client::postFunction(std::function<void(client *)> fn, bool fLock) {
         std::lock_guard<decltype(this->lock)> lock(this->lock);
         fn(this);
         --casyncOpsPending;
-    }, false, fLock) == AE_OK;
+    }, fLock) == AE_OK;
 }
 
 /*================================== Shutdown =============================== */

src/server.h

@@ -1393,6 +1393,7 @@ struct redisServerThreadVars {
     long unsigned commandsExecuted = 0;
     bool fRetrySetAofEvent = false;
     std::vector<client*> vecclientsProcess;
+    dictAsyncRehashCtl *rehashCtl = nullptr;
 };
 
 struct redisMaster {

tests/integration/logging.tcl

@@ -1,7 +1,8 @@
 set server_path [tmpdir server.log]
 set system_name [string tolower [exec uname -s]]
+set is_musl [catch {exec ldd --version}]
 
-if {$system_name eq {linux} || $system_name eq {darwin}} {
+if {$system_name eq {linux} && $is_musl eq 0 || $system_name eq {darwin}} {
     start_server [list overrides [list dir $server_path]] {
         test "Server is able to generate a stack trace on selected systems" {
             r config set watchdog-period 200

tests/modules/Makefile

@@ -25,7 +25,7 @@ TEST_MODULES = \
     auth.so \
     keyspace_events.so \
     blockedclient.so \
-    getkeys.so
+    timers.so
 
 
 .PHONY: all

tests/modules/timers.c

@@ -0,0 +1,76 @@
+#define REDISMODULE_EXPERIMENTAL_API
+#include "redismodule.h"
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <sched.h>
+#include <unistd.h>
+#include <string.h>
+
+static long current_timer_count = 0;
+static long total_timer_count;
+
+void timerHandler(RedisModuleCtx *ctx, void *data) {
+    REDISMODULE_NOT_USED(data);
+    current_timer_count++;
+    RedisModule_Log(ctx, "notice", "Timer %ld went off", current_timer_count);
+    if (current_timer_count == total_timer_count){
+        RedisModule_Log(ctx, "notice", "All timers fired successfully");
+    }
+}
+
+int elapsed(RedisModuleCtx *ctx, RedisModuleString **argv, int argc){
+    REDISMODULE_NOT_USED(argc);
+    REDISMODULE_NOT_USED(argv);
+    RedisModule_ReplyWithLongLong(ctx, (long long)current_timer_count);
+    return REDISMODULE_OK;
+} 
+
+enum TimerOrder {
+    TIMERS_ASCENDING,
+    TIMERS_DESCENDING,
+    TIMERS_SAME
+};
+
+int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    if (RedisModule_Init(ctx,"timer-spawn",1,REDISMODULE_APIVER_1)
+            == REDISMODULE_ERR) return REDISMODULE_ERR;
+
+    enum TimerOrder mode;
+    if (argc >= 1){
+        const char * modeString = RedisModule_StringPtrLen(argv[0], NULL);
+        if (strcmp(modeString, "ascending") == 0){
+            mode = TIMERS_ASCENDING;
+        } else if (strcmp(modeString, "descending") == 0){
+            mode = TIMERS_DESCENDING;
+        } else if (strcmp(modeString, "same") == 0){
+            mode = TIMERS_SAME;
+        } else {
+            RedisModule_Log(ctx, "warning", "Invalid mode specified as first argument. Valid modes are: ascending, descending, and same");
+            return REDISMODULE_ERR;
+        }
+    } else {
+        mode = TIMERS_DESCENDING;
+    }
+
+    long long user_timer_count;
+    total_timer_count = (argc >= 2 && RedisModule_StringToLongLong(argv[1], &user_timer_count) == REDISMODULE_OK) ? 
+        (long)user_timer_count : 2500;
+
+    if (RedisModule_CreateCommand(ctx,"timer.elapsed", elapsed,"",0,0,0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    for (long i = 0; i < total_timer_count; i++){
+        mstime_t period;
+        if (mode == TIMERS_ASCENDING){
+            period = i;
+        } else if (mode == TIMERS_DESCENDING){
+            period = total_timer_count - i;
+        } else if (mode == TIMERS_SAME){
+            period = total_timer_count;
+        }
+        RedisModule_CreateTimer(ctx, period, timerHandler, NULL);
+    }
+
+    return REDISMODULE_OK;
+}

tests/unit/moduleapi/hooks.tcl

@@ -132,6 +132,7 @@ tags "modules" {
             }
 
             $replica replicaof no one
+            after 300
 
             test {Test role-master hook} {
                 assert_equal [r hooks.event_count role-replica] 1

tests/unit/moduleapi/timers.tcl

@@ -0,0 +1,35 @@
+set testmodule [file normalize tests/modules/timers.so]
+set timercount 4000
+
+
+tags "modules" {
+    test {Ascending module timers can load in correctly} {
+        start_server [list overrides [list loadmodule "$testmodule ascending $timercount"]] {
+            wait_for_condition [expr round($timercount/10)] 20 {
+                [r timer.elapsed] == $timercount
+            } else {
+                fail "Server failed to load in timers with ascending periods"
+            }
+        }
+    }
+    
+    test {Descending module timers can load in correctly} {
+        start_server [list overrides [list loadmodule "$testmodule descending $timercount"]] {
+            wait_for_condition [expr round($timercount/10)] 20 {
+                [r timer.elapsed] == $timercount
+            } else {
+                fail "Server failed to load in timers with descending periods"
+            }
+        }
+    }
+
+    test {Module timers with the same period can load in correctly} {
+        start_server [list overrides [list loadmodule "$testmodule same $timercount"]] {
+            wait_for_condition [expr round($timercount/10)] 20 {
+                [r timer.elapsed] == $timercount
+            } else {
+                fail "Server failed to load in timers with the same period"
+            }
+        }
+    }
+}