check ziplist len to avoid crash on empty ziplist convert

Change min tested version to 18.04

.github/workflows/ci.yml

@@ -30,7 +30,7 @@ jobs:
           ./runtest-moduleapi
         
   build-ubuntu-old:
-    runs-on: ubuntu-16.04
+    runs-on: ubuntu-18.04
     steps:
     - uses: actions/checkout@v1
     - name: make -j2

README.md

@@ -33,19 +33,13 @@ Because of this difference of opinion features which are right for KeyDB may not
 Additional Resources
 --------------------
 
-Try our docker container: https://hub.docker.com/r/eqalpha/keydb
+Check out KeyDB's [Docker Image](https://hub.docker.com/r/eqalpha/keydb)
 
-Talk on Gitter: https://gitter.im/KeyDB
+Join us on [Slack](https://docs.keydb.dev/slack/)
 
-Visit our Website: https://keydb.dev
+Post to the [Community Forum](https://community.keydb.dev)
 
-See options for channel partners and support contracts: https://keydb.dev/support.html
-
-Learn with KeyDB’s official documentation site: https://docs.keydb.dev
-
-[Subscribe to the KeyDB mailing list](https://eqalpha.us20.list-manage.com/subscribe/post?u=978f486c2f95589b24591a9cc&id=4ab9220500)
-
-Management GUI: We recommend [FastoNoSQL](https://fastonosql.com/) which has official KeyDB support.
+Learn more through KeyDB's [Documentation & Learning Center](https://docs.keydb.dev)
 
 
 Benchmarking KeyDB

src/ae.cpp

@@ -53,6 +53,7 @@
 #include "zmalloc.h"
 #include "config.h"
 #include "serverassert.h"
+#include "readwritelock.h"
 
 #ifdef USE_MUTEX
 thread_local int cOwnLock = 0;
@@ -87,6 +88,7 @@ mutex_wrapper g_lock;
 #else
 fastlock g_lock("AE (global)");
 #endif
+readWriteLock g_forkLock("Fork (global)");
 thread_local aeEventLoop *g_eventLoopThisThread = NULL;
 
 /* Include the best multiplexing layer supported by this system.
@@ -151,8 +153,11 @@ void aeProcessCmd(aeEventLoop *eventLoop, int fd, void *, int )
         case AE_ASYNC_OP::PostFunction:
             {
             std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
-            if (cmd.fLock)
+            if (cmd.fLock) {
+                g_forkLock.releaseRead();
                 ulock.lock();
+                g_forkLock.acquireRead();
+            }
             ((aePostFunctionProc*)cmd.proc)(cmd.clientData);
             break;
             }
@@ -160,8 +165,11 @@ void aeProcessCmd(aeEventLoop *eventLoop, int fd, void *, int )
         case AE_ASYNC_OP::PostCppFunction:
         {
             std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
-            if (cmd.fLock)
+            if (cmd.fLock) {
+                g_forkLock.releaseRead();
                 ulock.lock();
+                g_forkLock.acquireRead();
+            }
             (*cmd.pfn)();
 
             delete cmd.pfn;
@@ -545,7 +553,11 @@ static int processTimeEvents(aeEventLoop *eventLoop) {
             if (te->next)
                 te->next->prev = te->prev;
             if (te->finalizerProc) {
-                if (!ulock.owns_lock()) ulock.lock();
+                if (!ulock.owns_lock()) {
+                    g_forkLock.releaseRead();
+                    ulock.lock();
+                    g_forkLock.acquireRead();
+                }
                 te->finalizerProc(eventLoop, te->clientData);
                 now = getMonotonicUs();
             }
@@ -565,7 +577,11 @@ static int processTimeEvents(aeEventLoop *eventLoop) {
         }
 
         if (te->when <= now) {
-            if (!ulock.owns_lock()) ulock.lock();
+            if (!ulock.owns_lock()) {
+                g_forkLock.releaseRead();
+                ulock.lock();
+                g_forkLock.acquireRead();
+            }
             int retval;
 
             id = te->id;
@@ -589,8 +605,11 @@ extern "C" void ProcessEventCore(aeEventLoop *eventLoop, aeFileEvent *fe, int ma
 {
 #define LOCK_IF_NECESSARY(fe, tsmask) \
     std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock); \
-    if (!(fe->mask & tsmask)) \
-        ulock.lock()
+    if (!(fe->mask & tsmask)) { \
+        g_forkLock.releaseRead(); \
+        ulock.lock(); \
+        g_forkLock.acquireRead(); \
+    }
 
     int fired = 0; /* Number of events fired for current fd. */
 
@@ -702,8 +721,11 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
 
         if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP) {
             std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
-            if (!(eventLoop->beforesleepFlags & AE_SLEEP_THREADSAFE))
+            if (!(eventLoop->beforesleepFlags & AE_SLEEP_THREADSAFE)) {
+                g_forkLock.releaseRead();
                 ulock.lock();
+                g_forkLock.acquireRead();
+            }
             eventLoop->beforesleep(eventLoop);
         }
 
@@ -714,8 +736,11 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
         /* After sleep callback. */
         if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP) {
             std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
-            if (!(eventLoop->aftersleepFlags & AE_SLEEP_THREADSAFE))
+            if (!(eventLoop->aftersleepFlags & AE_SLEEP_THREADSAFE)) {
+                g_forkLock.releaseRead();
                 ulock.lock();
+                g_forkLock.acquireRead();
+            }
             eventLoop->aftersleep(eventLoop);
         }
 
@@ -789,9 +814,21 @@ void setAeLockSetThreadSpinWorker(spin_worker worker)
     tl_worker = worker;
 }
 
+void aeThreadOnline()
+{
+    g_forkLock.acquireRead();
+}
+
 void aeAcquireLock()
 {
+    g_forkLock.releaseRead();
     g_lock.lock(tl_worker);
+    g_forkLock.acquireRead();
+}
+
+void aeAcquireForkLock()
+{
+    g_forkLock.upgradeWrite();
 }
 
 int aeTryAcquireLock(int fWeak)
@@ -799,11 +836,21 @@ int aeTryAcquireLock(int fWeak)
     return g_lock.try_lock(!!fWeak);
 }
 
+void aeThreadOffline()
+{
+    g_forkLock.releaseRead();
+}
+
 void aeReleaseLock()
 {
     g_lock.unlock();
 }
 
+void aeReleaseForkLock()
+{
+    g_forkLock.downgradeWrite();
+}
+
 int aeThreadOwnsLock()
 {
     return g_lock.fOwnLock();

src/ae.h

@@ -164,9 +164,13 @@ void aeSetDontWait(aeEventLoop *eventLoop, int noWait);
 void aeClosePipesForForkChild(aeEventLoop *eventLoop);
 
 void setAeLockSetThreadSpinWorker(spin_worker worker);
+void aeThreadOnline();
 void aeAcquireLock();
+void aeAcquireForkLock();
 int aeTryAcquireLock(int fWeak);
+void aeThreadOffline();
 void aeReleaseLock();
+void aeReleaseForkLock();
 int aeThreadOwnsLock();
 
 #ifdef __cplusplus

src/module.cpp

@@ -365,7 +365,7 @@ typedef struct RedisModuleCommandFilter {
 static list *moduleCommandFilters;
 
 /* Module GIL Variables */
-static readWriteLock s_moduleGIL;
+static readWriteLock s_moduleGIL("Module GIL");
 thread_local bool g_fModuleThread = false;
 
 typedef void (*RedisModuleForkDoneHandler) (int exitcode, int bysignal, void *user_data);

src/networking.cpp

@@ -1321,7 +1321,7 @@ void acceptOnThread(connection *conn, int flags, char *cip)
             szT = (char*)zmalloc(NET_IP_STR_LEN, MALLOC_LOCAL);
             memcpy(szT, cip, NET_IP_STR_LEN);
         }
-        int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [conn, flags, ielTarget, szT, fBootLoad] {
+        int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [conn, flags, ielTarget, szT] {
             connMarshalThread(conn);
             acceptCommonHandler(conn,flags,szT,ielTarget);
             rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
@@ -1907,19 +1907,6 @@ void ProcessPendingAsyncWrites()
         zfree(c->replyAsync);
         c->replyAsync = nullptr;
         c->fPendingAsyncWrite = FALSE;
-
-        // Now install the write event handler
-        int ae_flags = AE_WRITABLE|AE_WRITE_THREADSAFE;
-        /* For the fsync=always policy, we want that a given FD is never
-            * served for reading and writing in the same event loop iteration,
-            * so that in the middle of receiving the query, and serving it
-            * to the client, we'll call beforeSleep() that will do the
-            * actual fsync of AOF to disk. AE_BARRIER ensures that. */
-        if (g_pserver->aof_state == AOF_ON &&
-            g_pserver->aof_fsync == AOF_FSYNC_ALWAYS)
-        {
-            ae_flags |= AE_BARRIER;
-        }
         
         if (!((c->replstate == REPL_STATE_NONE ||
          (c->replstate == SLAVE_STATE_ONLINE && !c->repl_put_online_on_ack))))

src/object.cpp

@@ -136,7 +136,7 @@ robj *createEmbeddedStringObject(const char *ptr, size_t len) {
  *
  * The current limit of 52 is chosen so that the biggest string object
  * we allocate as EMBSTR will still fit into the 64 byte arena of jemalloc. */
-#define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 52
+size_t OBJ_ENCODING_EMBSTR_SIZE_LIMIT = 52;
 
 //static_assert((sizeof(redisObject)+OBJ_ENCODING_EMBSTR_SIZE_LIMIT-8) == 64, "Max EMBSTR obj should be 64 bytes total");
 robj *createStringObject(const char *ptr, size_t len) {

src/readwritelock.h

@@ -2,22 +2,24 @@
 #include <condition_variable>
 
 class readWriteLock {
-    std::mutex m_readLock;
-    std::recursive_mutex m_writeLock;
-    std::condition_variable m_cv;
+    fastlock m_readLock;
+    fastlock m_writeLock;
+    std::condition_variable_any m_cv;
     int m_readCount = 0;
     int m_writeCount = 0;
     bool m_writeWaiting = false;
 public:
+    readWriteLock(const char *name) : m_readLock(name), m_writeLock(name) {}
+
     void acquireRead() {
-        std::unique_lock<std::mutex> rm(m_readLock);
+        std::unique_lock<fastlock> rm(m_readLock);
         while (m_writeCount > 0 || m_writeWaiting)
             m_cv.wait(rm);
         m_readCount++;
     }
     
     bool tryAcquireRead() {
-        std::unique_lock<std::mutex> rm(m_readLock, std::defer_lock);
+        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
         if (!rm.try_lock())
             return false;
         if (m_writeCount > 0 || m_writeWaiting)
@@ -27,7 +29,7 @@ public:
     }
 
     void acquireWrite(bool exclusive = true) {
-        std::unique_lock<std::mutex> rm(m_readLock);
+        std::unique_lock<fastlock> rm(m_readLock);
         m_writeWaiting = true;
         while (m_readCount > 0)
             m_cv.wait(rm);
@@ -43,24 +45,12 @@ public:
     }
 
     void upgradeWrite(bool exclusive = true) {
-        std::unique_lock<std::mutex> rm(m_readLock);
-        m_writeWaiting = true;
-        while (m_readCount > 1)
-            m_cv.wait(rm);
-        if (exclusive) {
-            /* Another thread might have the write lock while we have the read lock
-               but won't be able to release it until they can acquire the read lock
-               so release the read lock and try again instead of waiting to avoid deadlock */
-            while(!m_writeLock.try_lock())
-                m_cv.wait(rm);
-        }
-        m_writeCount++;
-        m_readCount--;
-        m_writeWaiting = false;
+        releaseRead();
+        acquireWrite(exclusive);
     }
 
     bool tryAcquireWrite(bool exclusive = true) {
-        std::unique_lock<std::mutex> rm(m_readLock, std::defer_lock);
+        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
         if (!rm.try_lock())
             return false;
         if (m_readCount > 0)
@@ -73,14 +63,13 @@ public:
     }
 
     void releaseRead() {
-        std::unique_lock<std::mutex> rm(m_readLock);
-        serverAssert(m_readCount > 0);
+        std::unique_lock<fastlock> rm(m_readLock);
         m_readCount--;
         m_cv.notify_all();
     }
 
     void releaseWrite(bool exclusive = true) {
-        std::unique_lock<std::mutex> rm(m_readLock);
+        std::unique_lock<fastlock> rm(m_readLock);
         serverAssert(m_writeCount > 0);
         if (exclusive)
             m_writeLock.unlock();
@@ -89,14 +78,8 @@ public:
     }
 
     void downgradeWrite(bool exclusive = true) {
-        std::unique_lock<std::mutex> rm(m_readLock);
-        serverAssert(m_writeCount > 0);
-        if (exclusive)
-            m_writeLock.unlock();
-        m_writeCount--;
-        while (m_writeCount > 0 || m_writeWaiting)
-            m_cv.wait(rm);
-        m_readCount++;
+        releaseWrite(exclusive);
+        acquireRead();
     }
 
     bool hasReader() {

src/server.cpp

@@ -62,7 +62,6 @@
 #include <sys/socket.h>
 #include <algorithm>
 #include <uuid/uuid.h>
-#include <mutex>
 #include <condition_variable>
 #include "aelocker.h"
 #include "motd.h"
@@ -92,14 +91,12 @@ double R_Zero, R_PosInf, R_NegInf, R_Nan;
 /* Global vars */
 namespace GlobalHidden {
 struct redisServer server; /* Server global state */
-readWriteLock forkLock;
 }
 redisServer *g_pserver = &GlobalHidden::server;
-readWriteLock *g_forkLock = &GlobalHidden::forkLock;
 struct redisServerConst cserver;
 __thread struct redisServerThreadVars *serverTL = NULL;   // thread local server vars
-std::mutex time_thread_mutex;
-std::condition_variable time_thread_cv;
+fastlock time_thread_lock("Time thread lock");
+std::condition_variable_any time_thread_cv;
 int sleeping_threads = 0;
 void wakeTimeThread();
 
@@ -2626,9 +2623,10 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
     if (!fSentReplies)
         handleClientsWithPendingWrites(iel, aof_state);
 
+    aeThreadOffline();
     // Scope lock_guard
     {
-        std::lock_guard<std::mutex> lock(time_thread_mutex);
+        std::unique_lock<fastlock> lock(time_thread_lock);
         sleeping_threads++;
         serverAssert(sleeping_threads <= cserver.cthreads);
     }
@@ -2653,6 +2651,7 @@ void afterSleep(struct aeEventLoop *eventLoop) {
        Otherwise you may double acquire the GIL and cause deadlocks in the module */
     if (!ProcessingEventsWhileBlocked) {
         if (serverTL->modulesEnabledThisAeLoop) moduleAcquireGIL(TRUE /*fServerThread*/);
+        aeThreadOnline();
         wakeTimeThread();
     }
 }
@@ -6290,7 +6289,7 @@ int redisFork(int purpose) {
         openChildInfoPipe();
     }
     long long startWriteLock = ustime();
-    g_forkLock->acquireWrite();
+    aeAcquireForkLock();
     latencyAddSampleIfNeeded("fork-lock",(ustime()-startWriteLock)/1000);
     if ((childpid = fork()) == 0) {
         /* Child */
@@ -6300,7 +6299,7 @@ int redisFork(int purpose) {
         closeChildUnusedResourceAfterFork();
     } else {
         /* Parent */
-        g_forkLock->releaseWrite();
+        aeReleaseForkLock();
         g_pserver->stat_total_forks++;
         g_pserver->stat_fork_time = ustime()-start;
         g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
@@ -6618,7 +6617,9 @@ void OnTerminate()
 
 void wakeTimeThread() {
     updateCachedTime();
-    std::lock_guard<std::mutex> lock(time_thread_mutex);
+    aeThreadOffline();
+    std::unique_lock<fastlock> lock(time_thread_lock);
+    aeThreadOnline();
     if (sleeping_threads >= cserver.cthreads)
         time_thread_cv.notify_one();
     sleeping_threads--;
@@ -6630,21 +6631,23 @@ void *timeThreadMain(void*) {
     delay.tv_sec = 0;
     delay.tv_nsec = 100;
     int cycle_count = 0;
-    g_forkLock->acquireRead();
+    aeThreadOnline();
     while (true) {
         {
-            std::unique_lock<std::mutex> lock(time_thread_mutex);
+            aeThreadOffline();
+            std::unique_lock<fastlock> lock(time_thread_lock);
+            aeThreadOnline();
             if (sleeping_threads >= cserver.cthreads) {
-                g_forkLock->releaseRead();
+                aeThreadOffline();
                 time_thread_cv.wait(lock);
-                g_forkLock->acquireRead();
+                aeThreadOnline();
                 cycle_count = 0;
             }
         }
         updateCachedTime();
         if (cycle_count == MAX_CYCLES_TO_HOLD_FORK_LOCK) {
-            g_forkLock->releaseRead();
-            g_forkLock->acquireRead();
+            aeThreadOffline();
+            aeThreadOnline();
             cycle_count = 0;
         }
 #if defined(__APPLE__)
@@ -6654,7 +6657,7 @@ void *timeThreadMain(void*) {
 #endif
         cycle_count++;
     }
-    g_forkLock->releaseRead();
+    aeThreadOffline();
 }
 
 void *workerThreadMain(void *parg)
@@ -6672,6 +6675,7 @@ void *workerThreadMain(void *parg)
     }
 
     moduleAcquireGIL(true); // Normally afterSleep acquires this, but that won't be called on the first run
+    aeThreadOnline();
     aeEventLoop *el = g_pserver->rgthreadvar[iel].el;
     try
     {
@@ -6680,6 +6684,7 @@ void *workerThreadMain(void *parg)
     catch (ShutdownException)
     {
     }
+    aeThreadOffline();
     moduleReleaseGIL(true);
     serverAssert(!GlobalLocksAcquired());
     aeDeleteEventLoop(el);
@@ -6817,8 +6822,8 @@ int main(int argc, char **argv) {
     g_pserver->sentinel_mode = checkForSentinelMode(argc,argv);
     initServerConfig();
     serverTL = &g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN];
+    aeThreadOnline();
     aeAcquireLock();    // We own the lock on boot
-
     ACLInit(); /* The ACL subsystem must be initialized ASAP because the
                   basic networking code and client creation depends on it. */
     moduleInitModulesSystem();
@@ -7048,6 +7053,7 @@ int main(int argc, char **argv) {
 
     redisSetCpuAffinity(g_pserver->server_cpulist);
     aeReleaseLock();    //Finally we can dump the lock
+    aeThreadOffline();
     moduleReleaseGIL(true);
     
     setOOMScoreAdj(-1);

src/server.h

@@ -2177,7 +2177,6 @@ typedef struct {
 
 //extern struct redisServer server;
 extern redisServer *g_pserver;
-extern readWriteLock *g_forkLock;
 extern struct redisServerConst cserver;
 extern __thread struct redisServerThreadVars *serverTL;   // thread local server vars
 extern struct sharedObjectsStruct shared;

src/storage.cpp

@@ -114,6 +114,7 @@ void pool_free(struct alloc_pool *ppool, void *pv)
     return;
 }
 
+extern size_t OBJ_ENCODING_EMBSTR_SIZE_LIMIT;
 #define EMBSTR_ROBJ_SIZE (sizeof(robj)+sizeof(struct sdshdr8)+OBJ_ENCODING_EMBSTR_SIZE_LIMIT+1)
 struct alloc_pool poolobj;
 struct alloc_pool poolembstrobj;

src/t_zset.cpp

@@ -1191,24 +1191,26 @@ void zsetConvert(robj *zobj, int encoding) {
         zs->dict = dictCreate(&zsetDictType,NULL);
         zs->zsl = zslCreate();
 
-        eptr = ziplistIndex(zl,0);
-        serverAssertWithInfo(NULL,zobj,eptr != NULL);
-        sptr = ziplistNext(zl,eptr);
-        serverAssertWithInfo(NULL,zobj,sptr != NULL);
+        if (ziplistLen(zl) > 0) {
+            eptr = ziplistIndex(zl,0);
+            serverAssertWithInfo(NULL,zobj,eptr != NULL);
+            sptr = ziplistNext(zl,eptr);
+            serverAssertWithInfo(NULL,zobj,sptr != NULL);
 
-        while (eptr != NULL) {
-            score = zzlGetScore(sptr);
-            serverAssertWithInfo(NULL,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
-            if (vstr == NULL)
-                ele = sdsfromlonglong(vlong);
-            else
-                ele = sdsnewlen((char*)vstr,vlen);
+            while (eptr != NULL) {
+                score = zzlGetScore(sptr);
+                serverAssertWithInfo(NULL,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));
+                if (vstr == NULL)
+                    ele = sdsfromlonglong(vlong);
+                else
+                    ele = sdsnewlen((char*)vstr,vlen);
 
-            node = zslInsert(zs->zsl,score,ele);
-            serverAssert(dictAdd(zs->dict,ele,&node->score) == DICT_OK);
-            zzlNext(zl,&eptr,&sptr);
+                node = zslInsert(zs->zsl,score,ele);
+                serverAssert(dictAdd(zs->dict,ele,&node->score) == DICT_OK);
+                zzlNext(zl,&eptr,&sptr);
+            }
         }
-
+        
         zfree(zobj->m_ptr);
         zobj->m_ptr = zs;
         zobj->encoding = OBJ_ENCODING_SKIPLIST;

src/zmalloc.cpp

@@ -243,7 +243,7 @@ void *ztryrealloc_usable(void *ptr, size_t size, size_t *usable) {
 #else
     realptr = (char*)ptr-PREFIX_SIZE;
     oldsize = *((size_t*)realptr);
-    newptr = realloc(realptr,size+PREFIX_SIZE);
+    newptr = realloc(realptr,size+PREFIX_SIZE, MALLOC_LOCAL);
     if (newptr == NULL) {
         if (usable) *usable = 0;
         return NULL;

tests/assets/minimal.conf

@@ -1,4 +1,5 @@
 # Minimal configuration for testing.
+bind 127.0.0.1
 always-show-logo yes
 daemonize no
 pidfile /var/run/keydb.pid

tests/cluster/run.tcl

@@ -13,8 +13,9 @@ set ::tlsdir "../../tls"
 proc main {} {
     parse_options
     spawn_instance redis $::redis_base_port $::instances_count {
-        "cluster-enabled yes"
-        "appendonly yes"
+	"bind 127.0.0.1"
+	"cluster-enabled yes"
+	"appendonly yes"
 	"testmode yes"
 	"server-threads 3"
     }

tests/modules/Makefile

@@ -32,6 +32,7 @@ TEST_MODULES = \
     auth.so \
     keyspace_events.so \
     blockedclient.so \
+    timers.so \
     getkeys.so \
     test_lazyfree.so \
     timer.so \