Merge branch 'unstable' into RELEASE_5

Allow active replicas to connect to each other when replica-serve-stale-data is set
Don't send build log to stderr
2020-03-30 00:51:55 -04:00 · 2020-03-30 00:51:40 -04:00 · 2020-03-28 23:36:57 -04:00 · 2020-03-27 14:36:35 -04:00 · 2020-03-27 12:46:25 -04:00 · 2020-03-26 15:21:35 -04:00
36 changed files with 1075 additions and 302 deletions
@@ -4,6 +4,8 @@

 ##### Need Help? Check out our extensive [documentation](https://docs.keydb.dev).

+##### Have feedback?  Take our quick survey: https://www.surveymonkey.com/r/Y9XNS93
+
 What is KeyDB?
 --------------

@@ -13,7 +15,9 @@ KeyDB maintains full compatibility with the Redis protocol, modules, and scripts

 On the same hardware KeyDB can perform twice as many queries per second as Redis, with 60% lower latency. Active-Replication simplifies hot-spare failover allowing you to easily distribute writes over replicas and use simple TCP based load balancing/failover. KeyDB's higher performance allows you to do more on less hardware which reduces operation costs and complexity.

-<img src=https://cdn-images-1.medium.com/max/1400/1*s7mTb7Qb0kxc951mz8bdgA.png width=420 height=300/><img src=https://cdn-images-1.medium.com/max/1400/1*R00A5U4AFGohGOYHMfT6fA.png height=300/>
+<img src="https://keydb.dev/assets/img/blog/5x_opspersecVSdatasize.PNG"/>
+
+See the full benchmark results and setup information here: https://docs.keydb.dev/blog/2019/10/07/blog-post/

 Why fork Redis?
 ---------------
@@ -82,6 +82,10 @@ ifneq (,$(findstring armv,$(uname_M)))
 endif
 endif

+ifneq (,$(findstring armv,$(uname_M)))
+	FINAL_LIBS+=-latomic
+endif	
+
 # Backwards compatibility for selecting an allocator
 ifeq ($(USE_TCMALLOC),yes)
 	MALLOC=tcmalloc
@@ -105,7 +109,7 @@ endif
 FINAL_CFLAGS=$(STD) $(WARN) $(OPT) $(DEBUG) $(CFLAGS) $(REDIS_CFLAGS)
 FINAL_CXXFLAGS=$(CXX_STD) $(WARN) $(OPT) $(DEBUG) $(CXXFLAGS) $(REDIS_CFLAGS)
 FINAL_LDFLAGS=$(LDFLAGS) $(REDIS_LDFLAGS) $(DEBUG)
-FINAL_LIBS=-lm
+FINAL_LIBS+=-lm -lcurl
 DEBUG=-g -ggdb

 ifeq ($(uname_S),SunOS)
@@ -206,16 +210,16 @@ MAKECOLOR="\033[32;1m"
 ENDCOLOR="\033[0m"

 ifndef V
-QUIET_CC = @printf '    %b %b\n' $(CCCOLOR)CC$(ENDCOLOR) $(SRCCOLOR)$@$(ENDCOLOR) 1>&2;
-QUIET_LINK = @printf '    %b %b\n' $(LINKCOLOR)LINK$(ENDCOLOR) $(BINCOLOR)$@$(ENDCOLOR) 1>&2;
-QUIET_INSTALL = @printf '    %b %b\n' $(LINKCOLOR)INSTALL$(ENDCOLOR) $(BINCOLOR)$@$(ENDCOLOR) 1>&2;
+QUIET_CC = @printf '    %b %b\n' $(CCCOLOR)CC$(ENDCOLOR) $(SRCCOLOR)$@$(ENDCOLOR);
+QUIET_LINK = @printf '    %b %b\n' $(LINKCOLOR)LINK$(ENDCOLOR) $(BINCOLOR)$@$(ENDCOLOR);
+QUIET_INSTALL = @printf '    %b %b\n' $(LINKCOLOR)INSTALL$(ENDCOLOR) $(BINCOLOR)$@$(ENDCOLOR);
 endif

 REDIS_SERVER_NAME=keydb-server
 REDIS_SENTINEL_NAME=keydb-sentinel
-REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o acl.o storage.o rdb-s3.o fastlock.o new.o tracking.o $(ASM_OBJ)
+REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o acl.o storage.o rdb-s3.o fastlock.o new.o tracking.o cron.o motd.o $(ASM_OBJ)
 REDIS_CLI_NAME=keydb-cli
-REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o redis-cli-cpphelper.o zmalloc.o release.o anet.o ae.o crc64.o siphash.o crc16.o storage-lite.o fastlock.o new.o $(ASM_OBJ)
+REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o redis-cli-cpphelper.o zmalloc.o release.o anet.o ae.o crc64.o siphash.o crc16.o storage-lite.o fastlock.o new.o motd.o $(ASM_OBJ)
 REDIS_BENCHMARK_NAME=keydb-benchmark
 REDIS_BENCHMARK_OBJ=ae.o anet.o redis-benchmark.o adlist.o dict.o zmalloc.o siphash.o redis-benchmark.o storage-lite.o fastlock.o new.o $(ASM_OBJ)
 REDIS_CHECK_RDB_NAME=keydb-check-rdb
@@ -285,7 +289,7 @@ $(REDIS_CHECK_AOF_NAME): $(REDIS_SERVER_NAME)

 # keydb-cli
 $(REDIS_CLI_NAME): $(REDIS_CLI_OBJ)
-	$(REDIS_LD) -o $@ $^ ../deps/hiredis/libhiredis.a ../deps/linenoise/linenoise.o $(FINAL_LIBS) -lcurl
+	$(REDIS_LD) -o $@ $^ ../deps/hiredis/libhiredis.a ../deps/linenoise/linenoise.o $(FINAL_LIBS)

 # keydb-benchmark
 $(REDIS_BENCHMARK_NAME): $(REDIS_BENCHMARK_OBJ)
@@ -273,7 +273,8 @@ int aePostFunction(aeEventLoop *eventLoop, aePostFunctionProc *proc, void *arg)
    cmd.proc = proc;
    cmd.clientData = arg;
    auto size = write(eventLoop->fdCmdWrite, &cmd, sizeof(cmd));
-    AE_ASSERT(size == sizeof(cmd));
+    if (size != sizeof(cmd))
+        return AE_ERR;
    return AE_OK;
 }

@@ -296,6 +297,8 @@ int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fSynch
    }

    auto size = write(eventLoop->fdCmdWrite, &cmd, sizeof(cmd));
+    if (size != sizeof(cmd))
+        return AE_ERR;
    AE_ASSERT(size == sizeof(cmd));
    int ret = AE_OK;
    if (fSynchronous)
@@ -387,10 +390,18 @@ extern "C" void aeDeleteEventLoop(aeEventLoop *eventLoop) {
    aeApiFree(eventLoop);
    zfree(eventLoop->events);
    zfree(eventLoop->fired);
-    zfree(eventLoop);
    fastlock_free(&eventLoop->flock);
    close(eventLoop->fdCmdRead);
    close(eventLoop->fdCmdWrite);
+
+    auto *te = eventLoop->timeEventHead;
+    while (te)
+    {
+        auto *teNext = te->next;
+        zfree(te);
+        te = teNext;
+    }
+    zfree(eventLoop);
 }

 extern "C" void aeStop(aeEventLoop *eventLoop) {
@@ -124,6 +124,21 @@ void aofChildWriteDiffData(aeEventLoop *el, int fd, void *privdata, int mask) {
    }
 }

+void installAofRewriteEvent()
+{
+    serverTL->fRetrySetAofEvent = false;
+    if (!g_pserver->aof_rewrite_pending) {
+        g_pserver->aof_rewrite_pending = true;
+        int res = aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [] {
+            g_pserver->aof_rewrite_pending = false;
+            if (g_pserver->aof_pipe_write_data_to_child >= 0)
+                aeCreateFileEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, g_pserver->aof_pipe_write_data_to_child, AE_WRITABLE, aofChildWriteDiffData, NULL);
+        });
+        if (res != AE_OK)
+            serverTL->fRetrySetAofEvent = true;
+    }
+}
+
 /* Append data to the AOF rewrite buffer, allocating new blocks if needed. */
 void aofRewriteBufferAppend(unsigned char *s, unsigned long len) {
    listNode *ln = listLast(g_pserver->aof_rewrite_buf_blocks);
@@ -165,10 +180,7 @@ void aofRewriteBufferAppend(unsigned char *s, unsigned long len) {

    /* Install a file event to send data to the rewrite child if there is
     * not one already. */
-    aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, []{
-        if (g_pserver->aof_pipe_write_data_to_child >= 0)
-            aeCreateFileEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, g_pserver->aof_pipe_write_data_to_child, AE_WRITABLE, aofChildWriteDiffData, NULL);
-    });
+    installAofRewriteEvent();
 }

 /* Write the buffer (possibly composed of multiple blocks) into the specified
@@ -344,6 +356,9 @@ void flushAppendOnlyFile(int force) {
    int sync_in_progress = 0;
    mstime_t latency;

+    if (serverTL->fRetrySetAofEvent)
+        installAofRewriteEvent();
+
    if (sdslen(g_pserver->aof_buf) == 0) {
        /* Check if we need to do fsync even the aof buffer is empty,
         * because previously in AOF_FSYNC_EVERYSEC mode, fsync is
@@ -1559,16 +1574,18 @@ error:

 void aofClosePipes(void) {
    int fdAofAckPipe = g_pserver->aof_pipe_read_ack_from_child;
-    aePostFunction(g_pserver->el_alf_pip_read_ack_from_child, [fdAofAckPipe]{
+    int res = aePostFunction(g_pserver->el_alf_pip_read_ack_from_child, [fdAofAckPipe]{
        aeDeleteFileEventAsync(serverTL->el,fdAofAckPipe,AE_READABLE);
        close (fdAofAckPipe);
    });
+    serverAssert(res == AE_OK);

    int fdAofWritePipe = g_pserver->aof_pipe_write_data_to_child;
-    aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [fdAofWritePipe]{
+    res = aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [fdAofWritePipe]{
        aeDeleteFileEventAsync(serverTL->el,fdAofWritePipe,AE_WRITABLE);
        close(fdAofWritePipe);
    });
+    serverAssert(res == AE_OK);
    g_pserver->aof_pipe_write_data_to_child = -1;
    
    close(g_pserver->aof_pipe_read_data_from_parent);
@@ -36,7 +36,6 @@ const char *ascii_logo =
 "                                        Port: %d\n"
 "                                        PID: %ld\n"
 "                                                                     \n"
-"                     Like KeyDB? Star us on GitHub!                  \n"
+"                     %s\n"
 "                                                                     \n"
-"                     https://github.com/JohnSully/KeyDB              \n"
-"                                                                     \n\n";
+"                                                                     \n";
@@ -295,6 +295,15 @@ int clusterLoadConfig(char *filename) {
    if (clusterGetMaxEpoch() > g_pserver->cluster->currentEpoch) {
        g_pserver->cluster->currentEpoch = clusterGetMaxEpoch();
    }
+
+    if (dictSize(g_pserver->cluster->nodes) > 1 && cserver.thread_min_client_threshold < 100)
+    {
+        // Because we expect the individual load of a client to be much less in a cluster (it will spread over multiple server)
+        //  we can increase the grouping of clients on a single thread within reason
+        cserver.thread_min_client_threshold *= dictSize(g_pserver->cluster->nodes);
+        cserver.thread_min_client_threshold = std::min(cserver.thread_min_client_threshold, 200);
+        serverLog(LL_NOTICE, "Expanding min-clients-per-thread to %d due to cluster", cserver.thread_min_client_threshold);
+    }
    return C_OK;

 fmterr:
@@ -623,9 +632,10 @@ void freeClusterLink(clusterLink *link) {
        if (link->node)
            link->node->link = NULL;
        link->node = nullptr;
-        aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [link]{
+        int res = aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [link]{
            freeClusterLink(link);
        });
+        serverAssert(res == AE_OK);
        return;
    }
    if (link->fd != -1) {
@@ -405,7 +405,7 @@ void loadServerConfigFromString(char *config) {
        } else if ((!strcasecmp(argv[0],"slaveof") ||
                    !strcasecmp(argv[0],"replicaof")) && argc == 3) {
            slaveof_linenum = linenum;
-            replicationAddMaster(sdsnew(argv[1]), atoi(argv[2]));
+            replicationAddMaster(argv[1], atoi(argv[2]));
        } else if ((!strcasecmp(argv[0],"repl-ping-slave-period") ||
                    !strcasecmp(argv[0],"repl-ping-replica-period")) &&
                    argc == 2)
@@ -804,6 +804,12 @@ void loadServerConfigFromString(char *config) {
            // NOP, handled in main
        } else if (!strcasecmp(argv[0],"enable-pro")) {
            cserver.fUsePro = true;
+            break;
+        } else if (!strcasecmp(argv[0],"min-clients-per-thread") && argc == 2) {
+            cserver.thread_min_client_threshold = atoi(argv[1]);
+            if (cserver.thread_min_client_threshold < 0 || cserver.thread_min_client_threshold > 400) {
+                err = "min-thread-client must be between 0 and 400"; goto loaderr;
+            }
        } else {
            err = "Bad directive or wrong number of arguments"; goto loaderr;
        }
@@ -1490,8 +1496,15 @@ void configGetCommand(client *c) {
            listIter li;
            listNode *ln;
            listRewind(g_pserver->masters, &li);
+            bool fFirst = true;
            while ((ln = listNext(&li)))
            {
+                if (!fFirst)
+                {
+                    addReplyBulkCString(c,optname);
+                    matches++;
+                }
+                fFirst = false;
                struct redisMaster *mi = (struct redisMaster*)listNodeValue(ln);
                snprintf(buf,sizeof(buf),"%s %d",
                    mi->masterhost, mi->masterport);
@@ -0,0 +1,135 @@
+#include "server.h"
+#include "cron.h"
+
+void freeCronObject(robj_roptr o)
+{
+    delete reinterpret_cast<const cronjob*>(ptrFromObj(o));
+}
+
+// CRON [name] [single shot] [optional: start] [delay] [script] [numkeys] [key N] [arg N]
+void cronCommand(client *c)
+{
+    int arg_offset = 0;
+    static const int ARG_NAME = 1;
+    static const int ARG_SINGLESHOT = 2;
+    static const int ARG_EXPIRE = 3;
+    #define ARG_SCRIPT (4+arg_offset)
+    #define ARG_NUMKEYS (5+arg_offset)
+    #define ARG_KEYSTART (6+arg_offset)
+
+    bool fSingleShot = false;
+    if (strcasecmp("single", szFromObj(c->argv[ARG_SINGLESHOT])) == 0) {
+        fSingleShot = true;
+    } else {
+        if (strcasecmp("repeat", szFromObj(c->argv[ARG_SINGLESHOT])) != 0) {
+            addReply(c, shared.syntaxerr);
+            return;
+        }
+    }
+
+    long long interval;
+    if (getLongLongFromObjectOrReply(c, c->argv[ARG_EXPIRE], &interval, "missing expire time") != C_OK)
+        return;
+
+    long long base = g_pserver->mstime;
+    if (getLongLongFromObject(c->argv[ARG_EXPIRE+1], &base) == C_OK) {
+        arg_offset++;
+        std::swap(base, interval);
+    }
+
+    if (interval <= 0)
+    {
+        addReplyError(c, "interval must be positive");
+        return;
+    }
+
+    long numkeys = 0;
+    if (c->argc > ARG_NUMKEYS)
+    {
+        if (getLongFromObjectOrReply(c, c->argv[ARG_NUMKEYS], &numkeys, NULL) != C_OK)
+            return;
+        
+        if (c->argc < (6 + numkeys)) {
+            addReplyError(c, "Missing arguments or numkeys is too big");
+        }
+    }
+
+    std::unique_ptr<cronjob> spjob = std::make_unique<cronjob>();
+    spjob->script = sdsstring(sdsdup(szFromObj(c->argv[ARG_SCRIPT])));
+    spjob->interval = (uint64_t)interval;
+    spjob->startTime = (uint64_t)base;
+    spjob->fSingleShot = fSingleShot;
+    spjob->dbNum = c->db - g_pserver->db;
+    for (long i = 0; i < numkeys; ++i)
+        spjob->veckeys.emplace_back(sdsdup(szFromObj(c->argv[ARG_KEYSTART+i])));
+    for (long i = ARG_KEYSTART + numkeys; i < c->argc; ++i)
+        spjob->vecargs.emplace_back(sdsdup(szFromObj(c->argv[i])));
+
+    robj *o = createObject(OBJ_CRON, spjob.release());
+    setKey(c->db, c->argv[ARG_NAME], o);
+    decrRefCount(o);
+    // use an expire to trigger execution.  Note: We use a subkey expire here so legacy clients don't delete it.
+    setExpire(c, c->db, c->argv[ARG_NAME], c->argv[ARG_NAME], base + interval);
+    addReply(c, shared.ok);
+}
+
+void executeCronJobExpireHook(const char *key, robj *o)
+{
+    serverAssert(o->type == OBJ_CRON);
+    cronjob *job = (cronjob*)ptrFromObj(o);
+    
+    client *cFake = createClient(-1, IDX_EVENT_LOOP_MAIN);
+    cFake->lock.lock();
+    cFake->authenticated = 1;
+    cFake->puser = nullptr;
+    selectDb(cFake, job->dbNum);
+    serverAssert(cFake->argc == 0);
+
+    // Setup the args for the EVAL command
+    cFake->argc = 3 + job->veckeys.size() + job->vecargs.size();
+    cFake->argv = (robj**)zmalloc(sizeof(robj*) * cFake->argc, MALLOC_LOCAL);
+    cFake->argv[0] = createStringObject("EVAL", 4);
+    cFake->argv[1] = createStringObject(job->script.get(), job->script.size());
+    cFake->argv[2] = createStringObjectFromLongLong(job->veckeys.size());
+    for (size_t i = 0; i < job->veckeys.size(); ++i)
+        cFake->argv[3+i] = createStringObject(job->veckeys[i].get(), job->veckeys[i].size());
+    for (size_t i = 0; i < job->vecargs.size(); ++i)
+        cFake->argv[3+job->veckeys.size()+i] = createStringObject(job->vecargs[i].get(), job->vecargs[i].size());
+
+    evalCommand(cFake);
+    resetClient(cFake);
+
+    robj *keyobj = createStringObject(key,sdslen(key));
+    int dbId = job->dbNum;
+    if (job->fSingleShot)
+    {
+        dbSyncDelete(cFake->db, keyobj);
+    }
+    else
+    {
+        job->startTime += job->interval;
+        if (job->startTime < (uint64_t)g_pserver->mstime)
+        {
+            // If we are more than one interval in the past then fast forward to
+            //  the first interval still in the future.  If startTime wasn't zero align
+            //  this to the original startTime, if it was zero align to now
+            if (job->startTime == job->interval)
+            {   // startTime was 0
+                job->startTime = g_pserver->mstime + job->interval;
+            }
+            else
+            {
+                auto delta = g_pserver->mstime - job->startTime;
+                auto multiple = (delta / job->interval)+1;
+                job->startTime += job->interval * multiple;
+            }
+        }
+        setExpire(cFake, cFake->db, keyobj, keyobj, job->startTime);
+    }
+
+    notifyKeyspaceEvent(NOTIFY_KEYEVENT, "CRON Executed", keyobj, dbId);
+    decrRefCount(keyobj);
+
+    // o is invalid at this point
+    freeClient(cFake);
+}
@@ -0,0 +1,16 @@
+#pragma once
+
+struct cronjob
+{
+    sdsstring script;
+    uint64_t interval;
+    uint64_t startTime;
+    std::vector<sdsstring> veckeys;
+    std::vector<sdsstring> vecargs;
+    int dbNum = 0;
+    bool fSingleShot = false;
+};
+
+void freeCronObject(robj_roptr o);
+void executeCronJobExpireHook(const char *key, robj *o);
+void cronCommand(client *c);
@@ -177,9 +177,8 @@ robj_roptr lookupKeyRead(redisDb *db, robj *key) {
 * Returns the linked value object if the key exists or NULL if the key
 * does not exist in the specified DB. */
 robj *lookupKeyWrite(redisDb *db, robj *key) {
+    expireIfNeeded(db,key);
    robj *o = lookupKey(db,key,LOOKUP_UPDATEMVCC);
-    if (expireIfNeeded(db,key))
-        o = NULL;
    return o;
 }

@@ -978,7 +977,7 @@ void shutdownCommand(client *c) {
     * Also when in Sentinel mode clear the SAVE flag and force NOSAVE. */
    if (g_pserver->loading || g_pserver->sentinel_mode)
        flags = (flags & ~SHUTDOWN_SAVE) | SHUTDOWN_NOSAVE;
-    if (prepareForShutdown(flags) == C_OK) exit(0);
+    if (prepareForShutdown(flags) == C_OK) throw ShutdownException();
    addReplyError(c,"Errors trying to SHUTDOWN. Check logs.");
 }

@@ -1306,6 +1305,14 @@ void setExpire(client *c, redisDb *db, robj *key, robj *subkey, long long when)
        rememberSlaveKeyWithExpire(db,key);
 }

+redisDb::~redisDb()
+{
+    dictRelease(watched_keys);
+    dictRelease(ready_keys);
+    dictRelease(blocking_keys);
+    listRelease(defrag_later);
+}
+
 void setExpire(client *c, redisDb *db, robj *key, expireEntry &&e)
 {
    dictEntry *kde;
@@ -685,8 +685,26 @@ NULL
    } else if (!strcasecmp(szFromObj(c->argv[1]),"stringmatch-test") && c->argc == 2) {
        stringmatchlen_fuzz_test();
        addReplyStatus(c,"Apparently Redis did not crash: test passed");
-    } else if (!strcasecmp(szFromObj(c->argv[1]), "force-master") && c->argc == 2) {
+    } else if (!strcasecmp(szFromObj(c->argv[1]), "force-master") && c->argc == 3) {
        c->flags |= CLIENT_MASTER | CLIENT_MASTER_FORCE_REPLY;
+        if (!strcasecmp(szFromObj(c->argv[2]), "yes"))
+        {
+            redisMaster *mi = (redisMaster*)zcalloc(sizeof(redisMaster), MALLOC_LOCAL);
+            mi->master = c;
+            listAddNodeHead(g_pserver->masters, mi);
+        }
+        else if (strcasecmp(szFromObj(c->argv[2]), "flagonly")) // if we didn't set flagonly assume its an unset
+        {
+            serverAssert(c->flags & CLIENT_MASTER);
+            if (listLength(g_pserver->masters))
+            {
+                redisMaster *mi = (redisMaster*)listNodeValue(listFirst(g_pserver->masters));
+                serverAssert(mi->master == c);
+                listDelNode(g_pserver->masters, listFirst(g_pserver->masters));
+                zfree(mi);
+            }
+            c->flags &= ~(CLIENT_MASTER | CLIENT_MASTER_FORCE_REPLY);
+        }
        addReply(c, shared.ok);
    } else {
        addReplySubcommandSyntaxError(c);
@@ -31,6 +31,7 @@
 */

 #include "server.h"
+#include "cron.h"

 void activeExpireCycleExpireFullKey(redisDb *db, const char *key) {
    robj *keyobj = createStringObject(key,sdslen(key));
@@ -121,6 +122,10 @@ void activeExpireCycleExpire(redisDb *db, expireEntry &e, long long now) {
            }
            break;

+        case OBJ_CRON:
+            executeCronJobExpireHook(e.key(), val);
+            return;
+
        case OBJ_LIST:
        default:
            serverAssert(false);
@@ -130,6 +135,15 @@ void activeExpireCycleExpire(redisDb *db, expireEntry &e, long long now) {

    if (deleted)
    {
+        if (!pfat->FEmpty())
+        {
+            // We need to resort the expire entry since it may no longer be in the correct position
+            auto itr = db->setexpire->find(e.key());
+            expireEntry eT = std::move(e);
+            db->setexpire->erase(itr);
+            db->setexpire->insert(eT);
+        }
+
        robj objT;
        switch (val->type)
        {
@@ -27,6 +27,7 @@
 * POSSIBILITY OF SUCH DAMAGE.
 */

+#include "fmacros.h"
 #include "fastlock.h"
 #include <unistd.h>
 #include <sys/syscall.h>
@@ -43,6 +44,7 @@
 #include <string.h>
 #include <stdarg.h>
 #include <stdio.h>
+#include "config.h"

 #ifdef __APPLE__
 #include <TargetConditionals.h>
@@ -60,6 +62,11 @@
 #define UNUSED(x) ((void)x)
 #endif

+#ifdef HAVE_BACKTRACE
+#include <ucontext.h>
+__attribute__((weak)) void logStackTrace(ucontext_t *) {}
+#endif
+
 extern int g_fInCrash;

 /****************************************************
@@ -74,6 +81,10 @@ extern int g_fInCrash;
    #define __has_feature(x) 0
 #endif

+#ifdef __linux__
+extern "C" void unlock_futex(struct fastlock *lock, uint16_t ifutex);
+#endif
+
 #if __has_feature(thread_sanitizer)

    /* Report that a lock has been created at address "lock". */
@@ -145,6 +156,43 @@ __attribute__((weak)) void serverLog(int , const char *fmt, ...)
    printf("\n");
 }

+extern "C" pid_t gettid()
+{
+    static thread_local int pidCache = -1;
+#ifdef __linux__
+    if (pidCache == -1)
+        pidCache = syscall(SYS_gettid);
+#else
+	if (pidCache == -1) {
+		uint64_t tidT;
+		pthread_threadid_np(nullptr, &tidT);
+		assert(tidT < UINT_MAX);
+		pidCache = (int)tidT;
+	}
+#endif
+    return pidCache;
+}
+
+void printTrace()
+{
+#ifdef HAVE_BACKTRACE
+    serverLog(3 /*LL_WARNING*/, "printing backtrace for thread %d", gettid());
+    ucontext_t ctxt;
+    getcontext(&ctxt);
+    logStackTrace(&ctxt);
+#endif
+}
+
+
+#ifdef __linux__
+static int futex(volatile unsigned *uaddr, int futex_op, int val,
+    const struct timespec *timeout, int val3)
+{
+    return syscall(SYS_futex, uaddr, futex_op, val,
+                    timeout, uaddr, val3);
+}
+#endif
+
 class DeadlockDetector
 {
    std::map<pid_t, fastlock *> m_mapwait;
@@ -152,9 +200,19 @@ class DeadlockDetector
 public:
    void registerwait(fastlock *lock, pid_t thispid)
    {
+        static volatile bool fInDeadlock = false;
+
        if (lock == &m_lock || g_fInCrash)
            return;
        fastlock_lock(&m_lock);
+        
+        if (fInDeadlock)
+        {
+            printTrace();
+            fastlock_unlock(&m_lock);
+            return;
+        }
+
        m_mapwait.insert(std::make_pair(thispid, lock));

        // Detect cycles
@@ -180,6 +238,19 @@ public:
                    if (pidCheck == thispid)
                        break;
                }
+                // Wake All sleeping threads so they can print their callstacks
+#ifdef HAVE_BACKTRACE
+#ifdef __linux__
+                int mask = -1;
+                fInDeadlock = true;
+                fastlock_unlock(&m_lock);
+                futex(&lock->m_ticket.u, FUTEX_WAKE_BITSET_PRIVATE, INT_MAX, nullptr, mask);
+                futex(&itr->second->m_ticket.u, FUTEX_WAKE_BITSET_PRIVATE, INT_MAX, nullptr, mask);
+                sleep(2);
+                fastlock_lock(&m_lock);
+                printTrace();
+#endif
+#endif
                serverLog(3 /*LL_WARNING*/, "!!! KeyDB Will Now Crash !!!");
                _serverPanic(__FILE__, __LINE__, "Deadlock detected");
            }
@@ -206,6 +277,11 @@ DeadlockDetector g_dlock;
 static_assert(sizeof(pid_t) <= sizeof(fastlock::m_pidOwner), "fastlock::m_pidOwner not large enough");
 uint64_t g_longwaits = 0;

+extern "C" void fastlock_panic(struct fastlock *lock)
+{
+    _serverPanic(__FILE__, __LINE__, "fastlock lock/unlock mismatch for: %s", lock->szName);
+}
+
 uint64_t fastlock_getlongwaitcount()
 {
    uint64_t rval;
@@ -213,32 +289,6 @@ uint64_t fastlock_getlongwaitcount()
    return rval;
 }

-#ifdef __linux__
-static int futex(volatile unsigned *uaddr, int futex_op, int val,
-    const struct timespec *timeout, int val3)
-{
-    return syscall(SYS_futex, uaddr, futex_op, val,
-                    timeout, uaddr, val3);
-}
-#endif
-
-extern "C" pid_t gettid()
-{
-    static thread_local int pidCache = -1;
-#ifdef __linux__
-    if (pidCache == -1)
-        pidCache = syscall(SYS_gettid);
-#else
-	if (pidCache == -1) {
-		uint64_t tidT;
-		pthread_threadid_np(nullptr, &tidT);
-		assert(tidT < UINT_MAX);
-		pidCache = (int)tidT;
-	}
-#endif
-    return pidCache;
-}
-
 extern "C" void fastlock_sleep(fastlock *lock, pid_t pid, unsigned wake, unsigned mask)
 {
 #ifdef __linux__
@@ -290,7 +340,7 @@ extern "C" void fastlock_lock(struct fastlock *lock)

 #if defined(__i386__) || defined(__amd64__)
        __asm__ __volatile__ ("pause");
-#elif defined(__arm__)
+#elif defined(__aarch64__)
        __asm__ __volatile__ ("yield");
 #endif
        if ((++cloops % 0x100000) == 0)
@@ -326,7 +376,7 @@ extern "C" int fastlock_trylock(struct fastlock *lock, int fWeak)

    struct ticket ticket_expect { { { active, active } } };
    struct ticket ticket_setiflocked { { { active, next } } };
-    if (__atomic_compare_exchange(&lock->m_ticket, &ticket_expect, &ticket_setiflocked, fWeak /*weak*/, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED))
+    if (__atomic_compare_exchange(&lock->m_ticket.u, &ticket_expect.u, &ticket_setiflocked.u, fWeak /*weak*/, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED))
    {
        lock->m_depth = 1;
        tid = gettid();
@@ -337,31 +387,6 @@ extern "C" int fastlock_trylock(struct fastlock *lock, int fWeak)
    return false;
 }

-#ifdef __linux__
-#define ROL32(v, shift) ((v << shift) | (v >> (32-shift)))
-void unlock_futex(struct fastlock *lock, uint16_t ifutex)
-{
-    unsigned mask = (1U << (ifutex % 32));
-    unsigned futexT;
-    __atomic_load(&lock->futex, &futexT, __ATOMIC_RELAXED);
-    futexT &= mask;
-    
-    if (futexT == 0)
-        return;
-    
-    for (;;)
-    {
-        __atomic_load(&lock->futex, &futexT, __ATOMIC_ACQUIRE);
-        futexT &= mask;
-        if (!futexT)
-            break;
-
-        if (futex(&lock->m_ticket.u, FUTEX_WAKE_BITSET_PRIVATE, INT_MAX, nullptr, mask) == 1)
-            break;
-    }
-}
-#endif
-
 extern "C" void fastlock_unlock(struct fastlock *lock)
 {
    --lock->m_depth;
@@ -384,6 +409,26 @@ extern "C" void fastlock_unlock(struct fastlock *lock)
 }
 #endif

+#ifdef __linux__
+#define ROL32(v, shift) ((v << shift) | (v >> (32-shift)))
+extern "C" void unlock_futex(struct fastlock *lock, uint16_t ifutex)
+{
+    unsigned mask = (1U << (ifutex % 32));
+    unsigned futexT;
+    
+    for (;;)
+    {
+        __atomic_load(&lock->futex, &futexT, __ATOMIC_ACQUIRE);
+        futexT &= mask;
+        if (!futexT)
+            break;
+
+        if (futex(&lock->m_ticket.u, FUTEX_WAKE_BITSET_PRIVATE, INT_MAX, nullptr, mask) == 1)
+            break;
+    }
+}
+#endif
+
 extern "C" void fastlock_free(struct fastlock *lock)
 {
    // NOP
@@ -413,4 +458,4 @@ void fastlock_lock_recursive(struct fastlock *lock, int nesting)
 {
    fastlock_lock(lock);
    lock->m_depth = nesting;
-}
+}
@@ -126,6 +126,7 @@ fastlock_trylock:

 .ALIGN 16
 .global fastlock_unlock
+.type   fastlock_unlock,@function
 fastlock_unlock:
 	# RDI points to the struct:
 	#	int32_t m_pidOwner
@@ -133,34 +134,19 @@ fastlock_unlock:
 	# [rdi+64] ...
 	#	uint16_t active
 	#	uint16_t avail
-	push r11
 	sub dword ptr [rdi+4], 1     # decrement m_depth, don't use dec because it partially writes the flag register and we don't know its state
 	jnz .LDone                   # if depth is non-zero this is a recursive unlock, and we still hold it
 	mov dword ptr [rdi], -1      # pidOwner = -1 (we don't own it anymore)
-	mov ecx, [rdi+64]            # get current active (this one)
-	inc ecx                      # bump it to the next thread
-	mov [rdi+64], cx             # give up our ticket (note: lock is not required here because the spinlock itself guards this variable)
+	mov esi, [rdi+64]            # get current active (this one)
+	inc esi                      # bump it to the next thread
+	mov word ptr [rdi+64], si    # give up our ticket (note: lock is not required here because the spinlock itself guards this variable)
 	mfence                       # sync other threads
 	# At this point the lock is removed, however we must wake up any pending futexs
-	mov r9d, 1                   # eax is the bitmask for 2 threads
-	rol r9d, cl                  # place the mask in the right spot for the next 2 threads
-	add rdi, 64                  # rdi now points to the token
+	mov edx, [rdi+64+4]          # load the futex mask
+	bt edx, esi                  # is the next thread waiting on a futex?
+	jc unlock_futex              # unlock the futex if necessary
+	ret                          # if not we're done.
 .ALIGN 16
-.LRetryWake:
-	mov r11d, [rdi+4]            # load the futex mask
-	and r11d, r9d                # are any threads waiting on a futex?
-	jz .LDone                    # if not we're done.
-	# we have to wake the futexs
-                                 # rdi ARG1 futex (already in rdi)
-	mov esi, (10 | 128)          # rsi ARG2 FUTEX_WAKE_BITSET_PRIVATE
-	mov edx, 0x7fffffff          # rdx ARG3 INT_MAX (number of threads to wake)
-	xor r10d, r10d               # r10 ARG4 NULL
-	mov r8, rdi                  # r8 ARG5 dup rdi
-                                 # r9 ARG6 mask (already set above)
-	mov eax, 202                 # sys_futex
-	syscall
-	cmp eax, 1                   # did we wake as many as we expected?
-	jnz .LRetryWake
 .LDone:
-	pop r11
+	js fastlock_panic            # panic if we made m_depth negative
 	ret
@@ -1168,7 +1168,13 @@ struct commandHelp {
    "destination numkeys key [key ...] [WEIGHTS weight] [AGGREGATE SUM|MIN|MAX]",
    "Add multiple sorted sets and store the resulting sorted set in a new key",
    4,
-    "2.0.0" }
+    "2.0.0" },
+    { "KEYDB.CRON",
+    "name [single/repeat] [optional: start] delay script numkeys [key N] [arg N]",
+    "Run a specified script after start + delay, optionally repeating every delay interval.  The job may be cancelled by deleting the key associated with the job (name parameter)",
+    10,
+    "6.5.2"
+    }
 };

 #endif
@@ -0,0 +1,132 @@
+#include "sds.h"
+#include <cstring>
+#include <unistd.h>
+#include <sys/types.h>
+#include <pwd.h>
+#include <sys/stat.h>
+#include "motd.h"
+
+/*------------------------------------------------------------------------------
+ * Message of the day
+ *--------------------------------------------------------------------------- */
+#ifdef MOTD
+#include <curl/curl.h> 
+
+static const char *szMotdCachePath()
+{
+    static sds sdsMotdCachePath = NULL;
+    if (sdsMotdCachePath != NULL)
+        return sdsMotdCachePath;
+
+    struct passwd *pw = getpwuid(getuid());
+    if (pw == NULL)
+        return "";
+    const char *homedir = pw->pw_dir;
+    sdsMotdCachePath = sdsnew(homedir);
+    sdsMotdCachePath = sdscat(sdsMotdCachePath, motd_cache_file);
+    return sdsMotdCachePath;
+}
+static size_t motd_write_callback(void *ptr, size_t size, size_t nmemb, sds *str)
+{
+    *str = sdscatlen(*str, ptr, size*nmemb);
+    return (size*nmemb);
+}
+
+static char *fetchMOTDFromCache()
+{
+    struct stat attrib;
+    if (stat(szMotdCachePath(), &attrib) != 0)
+        return NULL;
+    time_t t = attrib.st_mtim.tv_sec;
+    time_t now = time(NULL);
+    if ((now - t) < 14400)
+    {
+        // If our cache was updated no more than 4 hours ago use it instead of fetching the MOTD
+        FILE *pf = fopen(szMotdCachePath(), "rb");
+        if (pf == NULL)
+            return NULL;
+        fseek(pf, 0L, SEEK_END);
+        long cb = ftell(pf);
+        fseek(pf, 0L, SEEK_SET);    // rewind
+        sds str = sdsnewlen(NULL, cb);
+        size_t cbRead = fread(str, 1, cb, pf);
+        fclose(pf);
+        if ((long)cbRead != cb)
+        {
+            sdsfree(str);
+            return NULL;
+        }
+        return str;
+    }
+    return NULL;
+}
+
+static void setMOTDCache(const char *sz)
+{
+    FILE *pf = fopen(szMotdCachePath(), "wb");
+    if (pf == NULL)
+        return;
+    size_t celem = fwrite(sz, strlen(sz), 1, pf);
+    (void)celem;    // best effort
+    fclose(pf);
+}
+
+extern "C" char *fetchMOTD(int cache)
+{
+    sds str;
+    CURL *curl;
+    CURLcode res;
+
+    /* First try and get the string from the cache */
+    if (cache) {
+        str = fetchMOTDFromCache();
+        if (str != NULL)
+            return str;
+    }
+
+    str = sdsnew("");
+    curl = curl_easy_init();
+    if(curl) {
+        curl_easy_setopt(curl, CURLOPT_URL, motd_url);
+        curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L); // follow redirects
+        curl_easy_setopt(curl, CURLOPT_TIMEOUT, 2); // take no more than two seconds
+        curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, motd_write_callback);
+        curl_easy_setopt(curl, CURLOPT_WRITEDATA, &str);
+
+        /* Perform the request, res will get the return code */ 
+        res = curl_easy_perform(curl);
+        /* Check for errors */ 
+        if(res != CURLE_OK)
+        {
+            sdsfree(str);
+            str = NULL;
+        }
+        else
+        {
+            long response_code;
+            curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &response_code);
+            if ((response_code / 100) != 2)
+            {
+                // An error code not in the 200s implies an error
+                sdsfree(str);
+                str = NULL;
+            }
+        }
+
+        /* always cleanup */ 
+        curl_easy_cleanup(curl);
+
+        if (str != NULL && cache)
+            setMOTDCache(str);
+    }
+    return str;
+}
+
+#else
+
+extern "C" char *fetchMOTD(int /* cache */)
+{
+    return NULL;
+}
+
+#endif
@@ -0,0 +1,14 @@
+#pragma once
+
+extern const char *motd_url;
+extern const char *motd_cache_file;
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+    
+char *fetchMOTD(int fCache);
+
+#ifdef __cplusplus
+}
+#endif
@@ -28,6 +28,7 @@
 */

 #include "server.h"
+bool FInReplicaReplay();

 /* ================================ MULTI/EXEC ============================== */

@@ -172,12 +173,15 @@ void execCommand(client *c) {
         * This way we'll deliver the MULTI/..../EXEC block as a whole and
         * both the AOF and the replication link will have the same consistency
         * and atomicity guarantees. */
-        if (!must_propagate && !(c->cmd->flags & (CMD_READONLY|CMD_ADMIN))) {
+        if (!must_propagate && !(c->cmd->flags & (CMD_READONLY|CMD_ADMIN)) && !(FInReplicaReplay())) {
            execCommandPropagateMulti(c);
            must_propagate = 1;
        }

-        call(c,g_pserver->loading ? CMD_CALL_NONE : CMD_CALL_FULL);
+        int flags = g_pserver->loading ? CMD_CALL_NONE : CMD_CALL_FULL;
+        if (FInReplicaReplay())
+            flags &= ~CMD_CALL_PROPAGATE;
+        call(c,flags);

        /* Commands may alter argc/argv, restore mstate. */
        c->mstate.commands[j].argc = c->argc;
@@ -177,6 +177,7 @@ client *createClient(int fd, int iel) {
    c->bufposAsync = 0;
    c->client_tracking_redirection = 0;
    c->casyncOpsPending = 0;
+    c->master_error = 0;
    memset(c->uuid, 0, UUID_BINARY_LEN);

    listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid);
@@ -432,6 +433,34 @@ void addReplyProtoAsync(client *c, const char *s, size_t len) {
    addReplyProtoCore(c, s, len, true);
 }

+std::string escapeString(sds str)
+{
+    std::string newstr;
+    size_t len = sdslen(str);
+    for (size_t ich = 0; ich < len; ++ich)
+    {
+        char ch = str[ich];
+        switch (ch)
+        {
+        case '\n':
+            newstr += "\\n";
+            break;
+
+        case '\t':
+            newstr += "\\t";
+            break;
+
+        case '\r':
+            newstr += "\\r";
+            break;
+        
+        default:
+            newstr += ch;
+        }
+    }
+    return newstr;
+}
+
 /* Low level function called by the addReplyError...() functions.
 * It emits the protocol for a Redis error, in the form:
 *
@@ -464,6 +493,12 @@ void addReplyErrorLengthCore(client *c, const char *s, size_t len, bool fAsync)
        serverLog(LL_WARNING,"== CRITICAL == This %s is sending an error "
                             "to its %s: '%s' after processing the command "
                             "'%s'", from, to, s, cmdname);
+
+        if (c->querybuf && sdslen(c->querybuf)) {
+            std::string str = escapeString(c->querybuf);
+            printf("\tquerybuf: %s\n", str.c_str());
+        }
+        c->master_error = 1;
    }
 }

@@ -1003,6 +1038,27 @@ int clientHasPendingReplies(client *c) {
    return (c->bufpos || listLength(c->reply)) && !(c->flags & CLIENT_CLOSE_ASAP);
 }

+static std::atomic<int> rgacceptsInFlight[MAX_EVENT_LOOPS];
+int chooseBestThreadForAccept()
+{
+    int ielMinLoad = 0;
+    int cclientsMin = INT_MAX;
+    for (int iel = 0; iel < cserver.cthreads; ++iel)
+    {
+        int cclientsThread;
+        atomicGet(g_pserver->rgthreadvar[iel].cclients, cclientsThread);
+        cclientsThread += rgacceptsInFlight[iel].load(std::memory_order_relaxed);
+        if (cclientsThread < cserver.thread_min_client_threshold)
+            return iel;
+        if (cclientsThread < cclientsMin)
+        {
+            cclientsMin = cclientsThread;
+            ielMinLoad = iel;
+        }
+    }
+    return ielMinLoad;
+}
+
 #define MAX_ACCEPTS_PER_CALL 1000
 static void acceptCommonHandler(int fd, int flags, char *ip, int iel) {
    client *c;
@@ -1105,7 +1161,25 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {

        if (!g_fTestMode)
        {
-            // We always accept on the same thread
+            {
+            int ielTarget = chooseBestThreadForAccept();
+            rgacceptsInFlight[ielTarget].fetch_add(1, std::memory_order_relaxed);
+            if (ielTarget != ielCur)
+            {
+                char *szT = (char*)zmalloc(NET_IP_STR_LEN, MALLOC_LOCAL);
+                memcpy(szT, cip, NET_IP_STR_LEN);
+                int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [cfd, ielTarget, szT]{
+                    acceptCommonHandler(cfd,0,szT, ielTarget);
+                    rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
+                    zfree(szT);
+                });
+
+                if (res == AE_OK)
+                    continue;
+            }
+            rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
+            }
+
        LLocalThread:
            aeAcquireLock();
            acceptCommonHandler(cfd,0,cip, ielCur);
@@ -1122,10 +1196,15 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
                goto LLocalThread;
            char *szT = (char*)zmalloc(NET_IP_STR_LEN, MALLOC_LOCAL);
            memcpy(szT, cip, NET_IP_STR_LEN);
-            aePostFunction(g_pserver->rgthreadvar[iel].el, [cfd, iel, szT]{
+            int res = aePostFunction(g_pserver->rgthreadvar[iel].el, [cfd, iel, szT]{
                acceptCommonHandler(cfd,0,szT, iel);
                zfree(szT);
            });
+            if (res != AE_OK)
+            {
+                zfree(szT);
+                goto LLocalThread;
+            }
        }
    }
 }
@@ -1151,13 +1230,16 @@ void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
        int ielTarget = rand() % cserver.cthreads;
        if (ielTarget == ielCur)
        {
+        LLocalThread:
            acceptCommonHandler(cfd,CLIENT_UNIX_SOCKET,NULL, ielCur);
        }
        else
        {
-            aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [cfd, ielTarget]{
+            int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [cfd, ielTarget]{
                acceptCommonHandler(cfd,CLIENT_UNIX_SOCKET,NULL, ielTarget);
            });
+            if (res != AE_OK)
+                goto LLocalThread;
        }
        aeReleaseLock();
        
@@ -1551,7 +1633,10 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
        c->lock.lock();
        ae.arm(c);
        if (c->flags & CLIENT_CLOSE_ASAP)
-            freeClient(c);
+        {
+            if (!freeClient(c))
+                c->lock.unlock();
+        }
    }
 }

@@ -2529,7 +2614,7 @@ NULL
                {
                    int iel = client->iel;
                    freeClientAsync(client);
-                    aePostFunction(g_pserver->rgthreadvar[client->iel].el, [iel] {
+                    aePostFunction(g_pserver->rgthreadvar[client->iel].el, [iel] {  // note: failure is OK
                        freeClientsInAsyncFreeQueue(iel);
                    });
                }
@@ -3018,28 +3103,55 @@ void unpauseClientsIfNecessary()
 *
 * The function returns the total number of events processed. */
 int processEventsWhileBlocked(int iel) {
+    serverAssert(GlobalLocksAcquired());
    int iterations = 4; /* See the function top-comment. */
    int count = 0;

-    client *c = serverTL->current_client;
-    if (c != nullptr)
+    std::vector<client*> vecclients;
+    listIter li;
+    listNode *ln;
+    listRewind(g_pserver->clients, &li);
+
+    // All client locks must be acquired *after* the global lock is reacquired to prevent deadlocks
+    //  so unlock here, and save them for reacquisition later
+    while ((ln = listNext(&li)) != nullptr)
    {
-        serverAssert(c->flags & CLIENT_PROTECTED);
-        c->lock.unlock();
+        client *c = (client*)listNodeValue(ln);
+        if (c->lock.fOwnLock()) {
+            serverAssert(c->flags & CLIENT_PROTECTED);  // If the client is not protected we have no gurantee they won't be free'd in the event loop
+            c->lock.unlock();
+            vecclients.push_back(c);
+        }
    }
+    
+
    aeReleaseLock();
-    while (iterations--) {
-        int events = 0;
-        events += aeProcessEvents(g_pserver->rgthreadvar[iel].el, AE_FILE_EVENTS|AE_DONT_WAIT);
-        events += handleClientsWithPendingWrites(iel);
-        if (!events) break;
-        count += events;
+    try
+    {
+        while (iterations--) {
+            int events = 0;
+            events += aeProcessEvents(g_pserver->rgthreadvar[iel].el, AE_FILE_EVENTS|AE_DONT_WAIT);
+            events += handleClientsWithPendingWrites(iel);
+            if (!events) break;
+            count += events;
+        }
    }
+    catch (...)
+    {
+        // Caller expects us to be locked so fix and rethrow
+        AeLocker locker;
+        locker.arm(nullptr);
+        locker.release();
+        for (client *c : vecclients)
+            c->lock.lock();
+        throw;
+    }
+    
    AeLocker locker;
-    if (c != nullptr)
-        c->lock.lock();
-    locker.arm(c);
+    locker.arm(nullptr);
    locker.release();
+    for (client *c : vecclients)
+        c->lock.lock();
    return count;
 }

@@ -29,6 +29,7 @@
 */

 #include "server.h"
+#include "cron.h"
 #include <math.h>
 #include <ctype.h>

@@ -369,6 +370,7 @@ void decrRefCount(robj_roptr o) {
        case OBJ_HASH: freeHashObject(o); break;
        case OBJ_MODULE: freeModuleObject(o); break;
        case OBJ_STREAM: freeStreamObject(o); break;
+        case OBJ_CRON: freeCronObject(o); break;
        default: serverPanic("Unknown object type"); break;
        }
        zfree(o.unsafe_robjcast());
@@ -33,6 +33,7 @@
 #include "endianconv.h"
 #include "stream.h"
 #include "storage.h"
+#include "cron.h"

 #include <math.h>
 #include <sys/types.h>
@@ -523,6 +524,11 @@ void *rdbGenericLoadStringObject(rio *rdb, int flags, size_t *lenptr) {
    }
 }

+sdsstring rdbLoadString(rio *rdb){
+    sds str = (sds)rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL);
+    return sdsstring(str);
+}
+
 robj *rdbLoadStringObject(rio *rdb) {
    return (robj*)rdbGenericLoadStringObject(rdb,RDB_LOAD_NONE,NULL);
 }
@@ -657,6 +663,8 @@ int rdbSaveObjectType(rio *rdb, robj_roptr o) {
        return rdbSaveType(rdb,RDB_TYPE_STREAM_LISTPACKS);
    case OBJ_MODULE:
        return rdbSaveType(rdb,RDB_TYPE_MODULE_2);
+    case OBJ_CRON:
+        return rdbSaveType(rdb,RDB_TYPE_CRON);
    default:
        serverPanic("Unknown object type");
    }
@@ -986,6 +994,17 @@ ssize_t rdbSaveObject(rio *rdb, robj_roptr o, robj *key) {
            zfree(io.ctx);
        }
        return io.error ? -1 : (ssize_t)io.bytes;
+    } else if (o->type == OBJ_CRON) {
+        cronjob *job = (cronjob*)ptrFromObj(o);
+        rdbSaveRawString(rdb, (const unsigned char*)job->script.get(), job->script.size());
+        rdbSaveMillisecondTime(rdb, job->startTime);
+        rdbSaveMillisecondTime(rdb, job->interval);
+        rdbSaveLen(rdb, job->veckeys.size());
+        for (auto &key : job->veckeys)
+            rdbSaveRawString(rdb, (const unsigned char*)key.get(), key.size());
+        rdbSaveLen(rdb, job->vecargs.size());
+        for (auto &arg : job->vecargs)
+            rdbSaveRawString(rdb, (const unsigned char*)arg.get(), arg.size());
    } else {
        serverPanic("Unknown object type");
    }
@@ -1526,9 +1545,15 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key, uint64_t mvcc_tstamp) {
                == NULL) return NULL;

            if (rdbtype == RDB_TYPE_ZSET_2) {
-                if (rdbLoadBinaryDoubleValue(rdb,&score) == -1) return NULL;
+                if (rdbLoadBinaryDoubleValue(rdb,&score) == -1) {
+                    sdsfree(sdsele);
+                    return NULL;
+                }
            } else {
-                if (rdbLoadDoubleValue(rdb,&score) == -1) return NULL;
+                if (rdbLoadDoubleValue(rdb,&score) == -1) {
+                    sdsfree(sdsele);
+                    return NULL;
+                }
            }

            /* Don't care about integer-encoded strings. */
@@ -1848,6 +1873,18 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key, uint64_t mvcc_tstamp) {
            exit(1);
        }
        o = createModuleObject(mt,ptr);
+    } else if (rdbtype == RDB_TYPE_CRON) {
+        std::unique_ptr<cronjob> spjob = std::make_unique<cronjob>();
+        spjob->script = rdbLoadString(rdb);
+        spjob->startTime = rdbLoadMillisecondTime(rdb,RDB_VERSION);
+        spjob->interval = rdbLoadMillisecondTime(rdb,RDB_VERSION);
+        auto ckeys = rdbLoadLen(rdb,NULL);
+        for (uint64_t i = 0; i < ckeys; ++i)
+            spjob->veckeys.push_back(rdbLoadString(rdb));
+        auto cargs = rdbLoadLen(rdb,NULL);
+        for (uint64_t i = 0; i < cargs; ++i)
+            spjob->vecargs.push_back(rdbLoadString(rdb));
+        o = createObject(OBJ_CRON, spjob.release());
    } else {
        rdbExitReportCorruptRDB("Unknown RDB encoding type %d",rdbtype);
    }
@@ -2058,7 +2095,7 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
                incrRefCount(subexpireKey);
            } else if (!strcasecmp(szFromObj(auxkey), "keydb-subexpire-when")) {
                if (key == nullptr || subexpireKey == nullptr) {
-                    serverLog(LL_WARNING, "Corrupt subexpire entry in RDB skipping.");
+                    serverLog(LL_WARNING, "Corrupt subexpire entry in RDB skipping. key: %s subkey: %s", key != nullptr ? szFromObj(key) : "(null)", subexpireKey != nullptr ? szFromObj(subexpireKey) : "(null)");
                }
                else {
                    setExpire(NULL, db, key, subexpireKey, strtoll(szFromObj(auxval), nullptr, 10));
@@ -2160,7 +2197,10 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
    }

    if (key != nullptr)
+    {
        decrRefCount(key);
+        key = nullptr;
+    }

    if (subexpireKey != nullptr)
    {
@@ -2187,6 +2227,17 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
    return C_OK;

 eoferr: /* unexpected end of file is handled here with a fatal exit */
+    if (key != nullptr)
+    {
+        decrRefCount(key);
+        key = nullptr;
+    }
+    if (subexpireKey != nullptr)
+    {
+        decrRefCount(subexpireKey);
+        subexpireKey = nullptr;
+    }
+
    serverLog(LL_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now.");
    rdbExitReportCorruptRDB("Unexpected EOF reading RDB file");
    return C_ERR; /* Just to avoid warning */
@@ -91,10 +91,13 @@
 #define RDB_TYPE_HASH_ZIPLIST  13
 #define RDB_TYPE_LIST_QUICKLIST 14
 #define RDB_TYPE_STREAM_LISTPACKS 15
+
+/* KeyDB Specific Object Types */
+#define RDB_TYPE_CRON 64
 /* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */

 /* Test if a type is an object type. */
-#define rdbIsObjectType(t) ((t >= 0 && t <= 7) || (t >= 9 && t <= 15))
+#define rdbIsObjectType(t) ((t >= 0 && t <= 7) || (t >= 9 && t <= 15) || (t == RDB_TYPE_CRON))

 /* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
 #define RDB_OPCODE_MODULE_AUX 247   /* Module auxiliary data. */
@@ -57,6 +57,7 @@
 #include "anet.h"
 #include "ae.h"
 #include "storage.h"
+#include "motd.h"

 #include "redis-cli.h"

@@ -92,6 +93,9 @@ int spectrum_palette_size;

 int g_fInCrash = 0;

+const char *motd_url = "http://api.keydb.dev/motd/motd_cli.txt";
+const char *motd_cache_file = "/.keydb-cli-motd";
+
 /*------------------------------------------------------------------------------
 * Utility functions
 *--------------------------------------------------------------------------- */
@@ -6536,118 +6540,6 @@ static void intrinsicLatencyMode(void) {
    }
 }

-/*------------------------------------------------------------------------------
- * Message of the day
- *--------------------------------------------------------------------------- */
-#ifdef MOTD
-#include <curl/curl.h> 
-
-static const char *szMotdCachePath()
-{
-    static sds sdsMotdCachePath = NULL;
-    if (sdsMotdCachePath != NULL)
-        return sdsMotdCachePath;
-
-    struct passwd *pw = getpwuid(getuid());
-    if (pw == NULL)
-        return "";
-    const char *homedir = pw->pw_dir;
-    sdsMotdCachePath = sdsnew(homedir);
-    sdsMotdCachePath = sdscat(sdsMotdCachePath, "/.keydb-cli-motd");
-    return sdsMotdCachePath;
-}
-static size_t motd_write_callback(void *ptr, size_t size, size_t nmemb, sds *str)
-{
-    *str = sdscatlen(*str, ptr, size*nmemb);
-    return (size*nmemb);
-}
-
-static char *fetchMOTDFromCache()
-{
-    struct stat attrib;
-    if (stat(szMotdCachePath(), &attrib) != 0)
-        return NULL;
-    time_t t = attrib.st_mtim.tv_sec;
-    time_t now = time(NULL);
-    if ((now - t) < 14400)
-    {
-        // If our cache was updated no more than 4 hours ago use it instead of fetching the MOTD
-        FILE *pf = fopen(szMotdCachePath(), "rb");
-        if (pf == NULL)
-            return NULL;
-        fseek(pf, 0L, SEEK_END);
-        long cb = ftell(pf);
-        fseek(pf, 0L, SEEK_SET);    // rewind
-        sds str = sdsnewlen(NULL, cb);
-        size_t cbRead = fread(str, 1, cb, pf);
-        fclose(pf);
-        if ((long)cbRead != cb)
-        {
-            sdsfree(str);
-            return NULL;
-        }
-        return str;
-    }
-    return NULL;
-}
-
-static void setMOTDCache(const char *sz)
-{
-    FILE *pf = fopen(szMotdCachePath(), "wb");
-    if (pf == NULL)
-        return;
-    size_t celem = fwrite(sz, strlen(sz), 1, pf);
-    (void)celem;    // best effort
-    fclose(pf);
-}
-
-static char *fetchMOTD()
-{
-    sds str;
-    CURL *curl;
-    CURLcode res;
-
-    /* First try and get the string from the cache */
-    str = fetchMOTDFromCache();
-    if (str != NULL)
-        return str;
-
-    str = sdsnew("");
-    curl = curl_easy_init();
-    if(curl) {
-        curl_easy_setopt(curl, CURLOPT_URL, "http://api.keydb.dev/motd/motd.txt");
-        curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L); // follow redirects
-        curl_easy_setopt(curl, CURLOPT_TIMEOUT, 2); // take no more than two seconds
-        curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, motd_write_callback);
-        curl_easy_setopt(curl, CURLOPT_WRITEDATA, &str);
-
-        /* Perform the request, res will get the return code */ 
-        res = curl_easy_perform(curl);
-        /* Check for errors */ 
-        if(res != CURLE_OK)
-        {
-            sdsfree(str);
-            str = NULL;
-        }
-
-        /* always cleanup */ 
-        curl_easy_cleanup(curl);
-
-        if (str != NULL)
-            setMOTDCache(str);
-    }
-    return str;
-}
-
-#else
-
-static char *fetchMOTD()
-{
-    return NULL;
-}
-
-#endif
-
 /*------------------------------------------------------------------------------
 * Program main()
 *--------------------------------------------------------------------------- */
@@ -6818,7 +6710,7 @@ int main(int argc, char **argv) {
    if (argc == 0 && !config.eval) {
        /* Show the message of the day if we are interactive */
        if (config.output == OUTPUT_STANDARD) {
-            char *szMotd = fetchMOTD();
+            char *szMotd = fetchMOTD(1 /* cache */);
            if (szMotd != NULL) {
                printf("Message of the day:\n  %s\n", szMotd);
                sdsfree(szMotd);
@@ -42,6 +42,8 @@
 #include <algorithm>
 #include <uuid/uuid.h>
 #include <chrono>
+#include <unordered_map>
+#include <string>

 void replicationDiscardCachedMaster(redisMaster *mi);
 void replicationResurrectCachedMaster(redisMaster *mi, int newfd);
@@ -276,7 +278,7 @@ void replicationFeedSlave(client *replica, int dictid, robj **argv, int argc, bo
        if (g_pserver->repl_backlog && fSendRaw) feedReplicationBacklogWithObject(selectcmd);

        /* Send it to slaves */
-        addReply(replica,selectcmd);
+        addReplyAsync(replica,selectcmd);

        if (dictid < 0 || dictid >= PROTO_SHARED_SELECT_CMDS)
            decrRefCount(selectcmd);
@@ -288,12 +290,12 @@ void replicationFeedSlave(client *replica, int dictid, robj **argv, int argc, bo
     * or are already in sync with the master. */

    /* Add the multi bulk length. */
-    addReplyArrayLen(replica,argc);
+    addReplyArrayLenAsync(replica,argc);

    /* Finally any additional argument that was not stored inside the
        * static buffer if any (from j to argc). */
    for (int j = 0; j < argc; j++)
-        addReplyBulk(replica,argv[j]);
+        addReplyBulkAsync(replica,argv[j]);
 }

 /* Propagate write commands to slaves, and populate the replication backlog
@@ -353,9 +355,10 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
    cchDbNum = std::min<int>(cchDbNum, sizeof(szDbNum)); // snprintf is tricky like that

    char szMvcc[128];
+    incrementMvccTstamp();
    uint64_t mvccTstamp = getMvccTstamp();
-    int cchMvccNum = snprintf(szMvcc, sizeof(szMvcc), "%lu", mvccTstamp);
-    int cchMvcc = snprintf(szMvcc, sizeof(szMvcc), "$%d\r\n%lu\r\n", cchMvccNum, mvccTstamp);
+    int cchMvccNum = snprintf(szMvcc, sizeof(szMvcc), "%" PRIu64, mvccTstamp);
+    int cchMvcc = snprintf(szMvcc, sizeof(szMvcc), "$%d\r\n%" PRIu64 "\r\n", cchMvccNum, mvccTstamp);
    cchMvcc = std::min<int>(cchMvcc, sizeof(szMvcc));    // tricky snprintf

    /* Write the command to the replication backlog if any. */
@@ -432,6 +435,7 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
            clientReplyBlock* reply = (clientReplyBlock*)listNodeValue(lnReply);
            addReplyProtoAsync(replica, reply->buf(), reply->used);
        }
+
        if (!fSendRaw)
        {
            addReplyAsync(replica,shared.crlf);
@@ -1255,19 +1259,19 @@ void updateSlavesWaitingBgsave(int bgsaveerr, int type)
                {
                    aePostFunction(g_pserver->rgthreadvar[replica->iel].el, [replica] {
                        // Because the client could have been closed while the lambda waited to run we need to
-			// verify the replica is still connected
+			            // verify the replica is still connected
                        listIter li;
-			listNode *ln;
-			listRewind(g_pserver->slaves,&li);
-			bool fFound = false;
-			while ((ln = listNext(&li))) {
-			    if (listNodeValue(ln) == replica) {
-			        fFound = true;
-				break;
-			     }
-			}
-			if (!fFound)
-			    return;
+                        listNode *ln;
+                        listRewind(g_pserver->slaves,&li);
+                        bool fFound = false;
+                        while ((ln = listNext(&li))) {
+                            if (listNodeValue(ln) == replica) {
+                                fFound = true;
+                                break;
+                            }
+                        }
+                        if (!fFound)
+                            return;
                        aeDeleteFileEvent(g_pserver->rgthreadvar[replica->iel].el,replica->fd,AE_WRITABLE);
                        if (aeCreateFileEvent(g_pserver->rgthreadvar[replica->iel].el, replica->fd, AE_WRITABLE, sendBulkToSlave, replica) == AE_ERR) {
                            freeClient(replica);
@@ -2325,10 +2329,11 @@ int connectWithMaster(redisMaster *mi) {
 void undoConnectWithMaster(redisMaster *mi) {
    int fd = mi->repl_transfer_s;

-    aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [fd]{
+    int res = aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [fd]{
        aeDeleteFileEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el,fd,AE_READABLE|AE_WRITABLE);
        close(fd);
    });
+    serverAssert(res == AE_OK);
    mi->repl_transfer_s = -1;
 }

@@ -3344,16 +3349,40 @@ public:
        return m_cnesting == 1;
    }

+    redisMaster *getMi(client *c)
+    {
+        if (m_mi == nullptr)
+            m_mi = MasterInfoFromClient(c);
+        return m_mi;
+    }
+
+    int nesting() const { return m_cnesting; }
+
 private:
    int m_cnesting = 0;
    bool m_fCancelled = false;
+    redisMaster *m_mi = nullptr;
 };

+static thread_local std::unique_ptr<ReplicaNestState> s_pstate;
+
+bool FInReplicaReplay()
+{
+    return s_pstate != nullptr && s_pstate->nesting() > 0;
+}
+
+struct RemoteMasterState
+{
+    uint64_t mvcc = 0;
+    client *cFake = nullptr;
+};
+
+static std::unordered_map<std::string, RemoteMasterState> g_mapremote;
+
 void replicaReplayCommand(client *c)
 {
-    static thread_local ReplicaNestState *s_pstate = nullptr;
    if (s_pstate == nullptr)
-        s_pstate = new (MALLOC_LOCAL) ReplicaNestState;
+        s_pstate = std::make_unique<ReplicaNestState>();

    // the replay command contains two arguments: 
    //  1: The UUID of the source
@@ -3375,9 +3404,10 @@ void replicaReplayCommand(client *c)
        return;
    }

-    unsigned char uuid[UUID_BINARY_LEN];
+    std::string uuid;
+    uuid.resize(UUID_BINARY_LEN);
    if (c->argv[1]->type != OBJ_STRING || sdslen((sds)ptrFromObj(c->argv[1])) != 36 
-        || uuid_parse((sds)ptrFromObj(c->argv[1]), uuid) != 0)
+        || uuid_parse((sds)ptrFromObj(c->argv[1]), (unsigned char*)uuid.data()) != 0)
    {
        addReplyError(c, "Expected UUID arg1");
        s_pstate->Cancel();
@@ -3413,7 +3443,7 @@ void replicaReplayCommand(client *c)
        }
    }

-    if (FSameUuidNoNil(uuid, cserver.uuid))
+    if (FSameUuidNoNil((unsigned char*)uuid.data(), cserver.uuid))
    {
        addReply(c, shared.ok);
        s_pstate->Cancel();
@@ -3423,33 +3453,51 @@ void replicaReplayCommand(client *c)
    if (!s_pstate->FPush())
        return;

+    RemoteMasterState &remoteState = g_mapremote[uuid];
+    if (remoteState.cFake == nullptr)
+        remoteState.cFake = createClient(-1, c->iel);
+    else
+        remoteState.cFake->iel = c->iel;
+
+    client *cFake = remoteState.cFake;
+
+    if (mvcc != 0 && remoteState.mvcc >= mvcc)
+    {
+        s_pstate->Cancel();
+        s_pstate->Pop();
+        return;
+    }
+
    // OK We've recieved a command lets execute
    client *current_clientSave = serverTL->current_client;
-    client *cFake = createClient(-1, c->iel);
    cFake->lock.lock();
    cFake->authenticated = c->authenticated;
    cFake->puser = c->puser;
    cFake->querybuf = sdscatsds(cFake->querybuf,(sds)ptrFromObj(c->argv[2]));
    selectDb(cFake, c->db->id);
    auto ccmdPrev = serverTL->commandsExecuted;
+    cFake->flags |= CLIENT_MASTER | CLIENT_PREVENT_REPL_PROP;
    processInputBuffer(cFake, (CMD_CALL_FULL & (~CMD_CALL_PROPAGATE)));
+    cFake->flags &= ~(CLIENT_MASTER | CLIENT_PREVENT_REPL_PROP);
    bool fExec = ccmdPrev != serverTL->commandsExecuted;
    cFake->lock.unlock();
-    if (fExec)
+    if (cFake->master_error)
+    {
+        addReplyError(c, "Error in rreplay command, please check logs.");
+    }
+    if (fExec || cFake->flags & CLIENT_MULTI)
    {
        addReply(c, shared.ok);
        selectDb(c, cFake->db->id);
-        redisMaster *mi = MasterInfoFromClient(c);
-        if (mi != nullptr)  // this should never be null but I'd prefer not to crash
-        {
-            mi->mvccLastSync = mvcc;
-        }
+        if (mvcc > remoteState.mvcc)
+            remoteState.mvcc = mvcc;
    }
    else
    {
+        serverLog(LL_WARNING, "Command didn't execute: %s", cFake->buf);
        addReplyError(c, "command did not execute");
    }
-    freeClient(cFake);
+    serverAssert(sdslen(cFake->querybuf) == 0);
    serverTL->current_client = current_clientSave;

    // call() will not propogate this for us, so we do so here
@@ -396,6 +396,20 @@ public:
        other.m_str = nullptr;
    }

+    sdsstring &operator=(const sdsstring &other)
+    {
+        sdsfree(m_str);
+        m_str = sdsdup(other.m_str);
+        return *this;
+    }
+
+    sdsstring &operator=(sds other)
+    {
+        sdsfree(m_str);
+        m_str = sdsdup(other);
+        return *this;
+    }
+
    ~sdsstring()
    {
        sdsfree(m_str);
@@ -35,6 +35,7 @@
 #include "latency.h"
 #include "atomicvar.h"
 #include "storage.h"
+#include "cron.h"
 #include <thread>
 #include <time.h>
 #include <signal.h>
@@ -61,8 +62,11 @@
 #include <uuid/uuid.h>
 #include <mutex>
 #include "aelocker.h"
+#include "motd.h"

 int g_fTestMode = false;
+const char *motd_url = "http://api.keydb.dev/motd/motd_server.txt";
+const char *motd_cache_file = "/.keydb-server-motd";

 /* Our shared "common" objects */

@@ -1023,7 +1027,11 @@ struct redisCommand redisCommandTable[] = {

    {"rreplay",replicaReplayCommand,-3,
     "read-only fast noprop",
-     0,NULL,0,0,0,0,0,0}
+     0,NULL,0,0,0,0,0,0},
+
+    {"keydb.cron",cronCommand,-5,
+     "write use-memory",
+     0,NULL,1,1,1,0,0,0},
 };

 /*============================ Utility functions ============================ */
@@ -1031,6 +1039,7 @@ struct redisCommand redisCommandTable[] = {
 /* We use a private localtime implementation which is fork-safe. The logging
 * function of Redis may be called from other threads. */
 extern "C" void nolocks_localtime(struct tm *tmp, time_t t, time_t tz, int dst);
+extern "C" pid_t gettid();

 /* Low level logging. To use only for very big messages, otherwise
 * serverLog() is to prefer. */
@@ -1068,8 +1077,8 @@ void serverLogRaw(int level, const char *msg) {
        } else {
            role_char = (listLength(g_pserver->masters) ? 'S':'M'); /* Slave or Master. */
        }
-        fprintf(fp,"%d:%c %s %c %s\n",
-            (int)getpid(),role_char, buf,c[level],msg);
+        fprintf(fp,"%d:%d:%c %s %c %s\n",
+            (int)getpid(),(int)gettid(),role_char, buf,c[level],msg);
    }
    fflush(fp);

@@ -1685,7 +1694,7 @@ void clientsCron(int iel) {
            /* The following functions do different service checks on the client.
            * The protocol is that they return non-zero if the client was
            * terminated. */
-            if (clientsCronHandleTimeout(c,now)) goto LContinue;
+            if (clientsCronHandleTimeout(c,now)) continue;  // Client free'd so don't release the lock
            if (clientsCronResizeQueryBuffer(c)) goto LContinue;
            if (clientsCronTrackExpansiveClients(c)) goto LContinue;
        LContinue:
@@ -1889,7 +1898,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    /* We received a SIGTERM, shutting down here in a safe way, as it is
     * not ok doing so inside the signal handler. */
    if (g_pserver->shutdown_asap) {
-        if (prepareForShutdown(SHUTDOWN_NOFLAGS) == C_OK) exit(0);
+        if (prepareForShutdown(SHUTDOWN_NOFLAGS) == C_OK) throw ShutdownException();
        serverLog(LL_WARNING,"SIGTERM received but errors trying to shut down the server, check the logs for more information");
        g_pserver->shutdown_asap = 0;
    }
@@ -2879,6 +2888,7 @@ static void initServerThread(struct redisServerThreadVars *pvar, int fMain)
    pvar->el = aeCreateEventLoop(g_pserver->maxclients+CONFIG_FDSET_INCR);
    pvar->current_client = nullptr;
    pvar->clients_paused = 0;
+    pvar->fRetrySetAofEvent = false;
    if (pvar->el == NULL) {
        serverLog(LL_WARNING,
            "Failed creating the event loop. Error message: '%s'",
@@ -3583,12 +3593,12 @@ int processCommand(client *c, int callFlags) {
            return C_OK;
        }
    }
-
-    incrementMvccTstamp();
    
    if (!locker.isArmed())
        locker.arm(c);

+    incrementMvccTstamp();
+
    /* Handle the maxmemory directive.
     *
     * Note that we do not want to reclaim memory if we are here re-entering
@@ -3672,7 +3682,8 @@ int processCommand(client *c, int callFlags) {
     * link with master. */
    if (FBrokenLinkToMaster() &&
        g_pserver->repl_serve_stale_data == 0 &&
-        !(c->cmd->flags & CMD_STALE))
+        !(c->cmd->flags & CMD_STALE)
+        && !(g_pserver->fActiveReplica && c->cmd->proc == syncCommand))
    {
        flagTransaction(c);
        addReply(c, shared.masterdownerr);
@@ -3806,8 +3817,17 @@ int prepareForShutdown(int flags) {

    /* Close the listening sockets. Apparently this allows faster restarts. */
    closeListeningSockets(1);
+
+    for (int iel = 0; iel < cserver.cthreads; ++iel)
+    {
+        aePostFunction(g_pserver->rgthreadvar[iel].el, [iel]{
+            g_pserver->rgthreadvar[iel].el->stop = 1;
+        });
+    }
+
    serverLog(LL_WARNING,"%s is now ready to exit, bye bye...",
        g_pserver->sentinel_mode ? "Sentinel" : "KeyDB");
+
    return C_OK;
 }

@@ -4684,7 +4704,7 @@ void daemonize(void) {
 }

 void version(void) {
-    printf("Redis server v=%s sha=%s:%d malloc=%s bits=%d build=%llx\n",
+    printf("KeyDB server v=%s sha=%s:%d malloc=%s bits=%d build=%llx\n",
        KEYDB_REAL_VERSION,
        redisGitSHA1(),
        atoi(redisGitDirty()) > 0,
@@ -4734,14 +4754,18 @@ void redisAsciiArt(void) {
            mode, g_pserver->port
        );
    } else {
+        sds motd = fetchMOTD(true);
        snprintf(buf,1024*16,ascii_logo,
            KEYDB_REAL_VERSION,
            redisGitSHA1(),
            strtol(redisGitDirty(),NULL,10) > 0,
            (sizeof(long) == 8) ? "64" : "32",
            mode, g_pserver->port,
-            (long) getpid()
+            (long) getpid(),
+            motd ? motd : ""
        );
+        if (motd)
+            sdsfree(motd);
        serverLogRaw(LL_NOTICE|LL_RAW,buf);
    }
    zfree(buf);
@@ -4995,14 +5019,15 @@ void incrementMvccTstamp()
    msPrev >>= MVCC_MS_SHIFT;  // convert to milliseconds

    long long mst;
-    __atomic_load(&g_pserver->mstime, &mst, __ATOMIC_RELAXED);
+    __atomic_load(&g_pserver->mstime, &mst, __ATOMIC_ACQUIRE);
    if (msPrev >= (uint64_t)mst)  // we can be greater if the count overflows
    {
-        atomicIncr(g_pserver->mvcc_tstamp, 1);
+        __atomic_fetch_add(&g_pserver->mvcc_tstamp, 1, __ATOMIC_RELEASE);
    }
    else
    {
-        atomicSet(g_pserver->mvcc_tstamp, ((uint64_t)g_pserver->mstime) << MVCC_MS_SHIFT);
+        uint64_t val = ((uint64_t)g_pserver->mstime) << MVCC_MS_SHIFT;
+        __atomic_store(&g_pserver->mvcc_tstamp, &val, __ATOMIC_RELEASE);
    }
 }

@@ -5013,7 +5038,7 @@ void OnTerminate()
        The easiest way to achieve that is to acutally segfault, so we assert
        here.
    */
-    serverAssert(false);
+    serverPanic("std::teminate() called");
 }

 void *workerThreadMain(void *parg)
@@ -5027,8 +5052,16 @@ void *workerThreadMain(void *parg)
    aeEventLoop *el = g_pserver->rgthreadvar[iel].el;
    aeSetBeforeSleepProc(el, isMainThread ? beforeSleep : beforeSleepLite, isMainThread ? 0 : AE_SLEEP_THREADSAFE);
    aeSetAfterSleepProc(el, afterSleep, AE_SLEEP_THREADSAFE);
-    aeMain(el);
+    try
+    {
+        aeMain(el);
+    }
+    catch (ShutdownException)
+    {
+    }
+    serverAssert(!GlobalLocksAcquired());
    aeDeleteEventLoop(el);
+
    return NULL;
 }

@@ -5098,10 +5131,6 @@ int main(int argc, char **argv) {
    dictSetHashFunctionSeed((uint8_t*)hashseed);
    g_pserver->sentinel_mode = checkForSentinelMode(argc,argv);
    initServerConfig();
-    for (int iel = 0; iel < MAX_EVENT_LOOPS; ++iel)
-    {
-        initServerThread(g_pserver->rgthreadvar+iel, iel == IDX_EVENT_LOOP_MAIN);
-    }
    serverTL = &g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN];
    aeAcquireLock();    // We own the lock on boot

@@ -5199,7 +5228,10 @@ int main(int argc, char **argv) {
    }

    if (cserver.fUsePro) {
-        execv("keydb-pro-server", argv);
+        const char *keydb_pro_dir = getenv("KEYDB_PRO_DIRECTORY");
+        sds path = sdsnew(keydb_pro_dir);
+        path = sdscat(path, "keydb-pro-server");
+        execv(path, argv);
        perror("Failed launch the pro binary");
        exit(EXIT_FAILURE);
    }
@@ -5225,6 +5257,10 @@ int main(int argc, char **argv) {
    int background = cserver.daemonize && !cserver.supervised;
    if (background) daemonize();

+    for (int iel = 0; iel < MAX_EVENT_LOOPS; ++iel)
+    {
+        initServerThread(g_pserver->rgthreadvar+iel, iel == IDX_EVENT_LOOP_MAIN);
+    }
    initServer();
    initNetworking(cserver.cthreads > 1 /* fReusePort */);

@@ -5325,7 +5361,9 @@ int main(int argc, char **argv) {
    /* The main thread sleeps until all the workers are done.
        this is so that all worker threads are orthogonal in their startup/shutdown */
    void *pvRet;
-    pthread_join(rgthread[IDX_EVENT_LOOP_MAIN], &pvRet);
+    for (int iel = 0; iel < cserver.cthreads; ++iel)
+        pthread_join(rgthread[iel], &pvRet);
+
    return 0;
 }

@@ -55,6 +55,7 @@
 #include <algorithm>
 #include <memory>
 #include <map>
+#include <string>
 #ifdef __cplusplus
 extern "C" {
 #include <lua.h>
@@ -683,6 +684,8 @@ public:
 * encoding version. */
 #define OBJ_MODULE 5    /* Module object. */
 #define OBJ_STREAM 6    /* Stream object. */
+#define OBJ_CRON 7      /* CRON job */
+

 /* Extract encver / signature from a module type ID. */
 #define REDISMODULE_TYPE_ENCVER_BITS 10
@@ -1125,10 +1128,13 @@ typedef struct clientReplyBlock {
 /* Redis database representation. There are multiple databases identified
 * by integers from 0 (the default database) up to the max configured
 * database. The database number is the 'id' field in the structure. */
-typedef struct redisDb {
+struct redisDb {
    redisDb() 
        : expireitr(nullptr)
    {};
+    
+    ~redisDb();
+
    dict *pdict;                 /* The keyspace for this DB */
    expireset *setexpire;
    expireset::setiter expireitr;
@@ -1140,7 +1146,7 @@ typedef struct redisDb {
    long long last_expire_set;  /* when the last expire was set */
    double avg_ttl;             /* Average TTL, just for stats */
    list *defrag_later;         /* List of key names to attempt to defrag one by one, gradually. */
-} redisDb;
+};

 /* Client MULTI/EXEC state */
 typedef struct multiCmd {
@@ -1332,6 +1338,7 @@ typedef struct client {

    int iel; /* the event loop index we're registered with */
    struct fastlock lock;
+    int master_error;
 } client;

 struct saveparam {
@@ -1524,6 +1531,7 @@ struct redisServerThreadVars {
    struct fastlock lockPendingWrite { "thread pending write" };
    char neterr[ANET_ERR_LEN];   /* Error buffer for anet.c */
    long unsigned commandsExecuted = 0;
+    bool fRetrySetAofEvent = false;
 };

 struct redisMaster {
@@ -1600,6 +1608,7 @@ struct redisServerConst {

    unsigned char uuid[UUID_BINARY_LEN];         /* This server's UUID - populated on boot */
    bool fUsePro = false;
+    int thread_min_client_threshold = 50;
 };

 struct redisServer {
@@ -1733,6 +1742,7 @@ struct redisServer {
    int aof_stop_sending_diff;     /* If true stop sending accumulated diffs
                                      to child process. */
    sds aof_child_diff;             /* AOF diff accumulator child side. */
+    int aof_rewrite_pending = 0;    /* is a call to aofChildWriteDiffData already queued? */
    /* RDB persistence */
    long long dirty;                /* Changes to DB from the last save */
    long long dirty_before_bgsave;  /* Used to restore dirty on failed BGSAVE */
@@ -2912,6 +2922,9 @@ inline int FCorrectThread(client *c)
 }
 #define AssertCorrectThread(c) serverAssert(FCorrectThread(c))

+class ShutdownException
+{};
+
 #define redisDebug(fmt, ...) \
    printf("DEBUG %s:%d > " fmt "\n", __FILE__, __LINE__, __VA_ARGS__)
 #define redisDebugMark() \
@@ -1,3 +1,3 @@
-#define KEYDB_REAL_VERSION "5.3.0"
+#define KEYDB_REAL_VERSION "5.3.3"
 extern const char *KEYDB_SET_VERSION;   // Unlike real version, this can be overriden by the config

@@ -59,6 +59,24 @@ start_server {tags {"active-repl"} overrides {active-replica yes}} {
            }
        }

+        test {Active replicas propogate transaction} {
+            $master set testkey 0
+            $master multi
+            $master incr testkey
+            $master incr testkey
+            after 5000
+            $master get testkey
+            $master exec
+            assert_equal 2 [$master get testkey]
+            after 500
+            wait_for_condition 50 500 {
+                [string match "2" [$slave get testkey]]
+            } else {
+                fail "Transaction failed to replicate"
+            }
+            $master flushall
+        }
+
        test {Active replicas WAIT} {
            # Test that wait succeeds since replicas should be syncronized
            $master set testkey foo
@@ -0,0 +1,74 @@
+foreach topology {mesh ring} {
+start_server {tags {"multi-master"} overrides {hz 500 active-replica yes multi-master yes}} {
+start_server {overrides {hz 500 active-replica yes multi-master yes}} {
+start_server {overrides {hz 500 active-replica yes multi-master yes}} {
+start_server {overrides {hz 500 active-replica yes multi-master yes}} {
+
+    for {set j 0} {$j < 4} {incr j} {
+        set R($j) [srv [expr 0-$j] client]
+        set R_host($j) [srv [expr 0-$j] host]
+        set R_port($j) [srv [expr 0-$j] port]
+    }
+
+    # Initialize as mesh
+    if [string equal $topology "mesh"] {
+    for {set j 0} {$j < 4} {incr j} {
+        for {set k 0} {$k < 4} {incr k} {
+            if $j!=$k {
+                $R($j) replicaof $R_host($k) $R_port($k)
+                after 100
+            }
+        }
+    }}
+    #Else Ring
+    if [string equal $topology "ring"] {
+        $R(0) replicaof $R_host(3) $R_port(3)
+        after 100
+        $R(1) replicaof $R_host(0) $R_port(0)
+        after 100
+        $R(2) replicaof $R_host(1) $R_port(1)
+        after 100
+        $R(3) replicaof $R_host(2) $R_port(2)
+    }
+
+    after 2000
+
+    test "$topology replicates to all nodes" {
+        $R(0) set testkey foo
+        after 500
+        assert_equal foo [$R(1) get testkey] "replicates to 1"
+        assert_equal foo [$R(2) get testkey] "replicates to 2"
+    }
+
+    test "$topology replicates only once" {
+        $R(0) set testkey 1
+        after 500
+        $R(1) incr testkey
+        after 500
+        $R(2) incr testkey
+        after 500
+        assert_equal 3 [$R(0) get testkey]
+        assert_equal 3 [$R(1) get testkey]
+        assert_equal 3 [$R(2) get testkey]
+        assert_equal 3 [$R(3) get testkey]
+    }
+
+    test "$topology transaction replicates only once" {
+        for {set j 0} {$j < 1000} {incr j} {
+            $R(0) set testkey 1
+            $R(0) multi
+            $R(0) incr testkey
+            $R(0) incr testkey
+            $R(0) exec
+            after 1
+            assert_equal 3 [$R(0) get testkey] "node 0"
+            assert_equal 3 [$R(1) get testkey] "node 1"
+            assert_equal 3 [$R(2) get testkey] "node 2"
+            assert_equal 3 [$R(3) get testkey] "node 3"
+        }
+    }
+}
+}
+}
+}
+}
@@ -309,3 +309,5 @@ start_server {tags {"repl"}} {
        }
    }
 }
+
+
@@ -36,6 +36,8 @@ set ::all_tests {
    unit/aofrw
    unit/acl
    unit/rreplay
+    unit/cron
+    unit/replication
    integration/block-repl
    integration/replication
    integration/replication-2
@@ -43,6 +45,7 @@ set ::all_tests {
    integration/replication-4
    integration/replication-psync
    integration/replication-active
+    integration/replication-multimaster
    integration/aof
    integration/rdb
    integration/convert-zipmap-hash-on-load
@@ -0,0 +1,47 @@
+start_server {tags {"CRON"}} {
+    test {keydb.cron singleshot past tense} {
+        r flushall
+        r keydb.cron testjob single 0 1 {redis.call("incr", "testkey")} 1 testkey
+        after 300
+        assert_equal 1 [r get testkey]
+        assert_equal 0 [r exists testjob]
+    }
+
+    test {keydb.cron repeat past tense next exec is in the future} {
+        r flushall
+        r keydb.cron testjob repeat 0 1000000 {redis.call("incr", "testkey")} 1 testkey
+        after 300
+        assert_equal 1 [r get testkey]
+        assert_equal 1 [r exists testjob]
+        r del testjob
+    }
+
+    test {keydb.cron repeat works} {
+        r flushall
+        r keydb.cron testjob repeat 0 600 {redis.call("incr","testkey")}
+        after 1000
+        assert_equal 2 [r get testkey]
+    }
+
+    test {keydb.cron overwrite works} {
+        r flushall
+        r keydb.cron testjob single 500 {redis.call("set","testkey","a")} 1 testkey
+        r keydb.cron testjob single 500 {redis.call("set","anotherkey","b")} 1 anotherkey
+        after 1000
+        assert_equal 0 [r exists testkey]
+        assert_equal b [r get anotherkey]
+    }
+
+    test {keydb.cron delete key stops job} {
+        r flushall
+        r keydb.cron testjob single 500 {redis.call("set","testkey","a")}
+        r del testjob
+        after 1000
+        assert_equal 0 [r exists testkey]
+    }
+
+    test {keydb.cron zero interval rejected} {
+        catch {r keydb.cron testjob single 0 0 {redis.call("incr","testkey")} 1 testkey} e
+        assert_match {ERR*} $e
+    }
+}
@@ -220,7 +220,7 @@ start_server {tags {"expire"}} {
        assert {$ttl <= 98 && $ttl > 90}
    }

-    test { EXPIREMEMBER works (set) } {
+    test {EXPIREMEMBER works (set)} {
        r flushall
        r sadd testkey foo bar baz
        r expiremember testkey foo 1
@@ -228,7 +228,7 @@ start_server {tags {"expire"}} {
        assert_equal {2} [r scard testkey]
    }

-    test { EXPIREMEMBER works (hash) } {
+    test {EXPIREMEMBER works (hash)} {
        r flushall
        r hset testkey foo bar
        r expiremember testkey foo 1
@@ -236,7 +236,7 @@ start_server {tags {"expire"}} {
        r exists testkey
    } {0}

-    test { EXPIREMEMBER works (zset) } {
+    test {EXPIREMEMBER works (zset)} {
        r flushall
        r zadd testkey 1 foo
        r zadd testkey 2 bar
@@ -246,10 +246,19 @@ start_server {tags {"expire"}} {
        assert_equal {1} [r zcard testkey]
    }

-    test { TTL for subkey expires works } {
+    test {TTL for subkey expires works} {
        r flushall
        r sadd testkey foo bar baz
        r expiremember testkey foo 10000
        assert [expr [r ttl testkey foo] > 0]
    }
+
+    test {Roundtrip for subkey expires works} {
+        r flushall
+        r sadd testkey foo bar baz
+        r expiremember testkey foo 10000
+        r save
+        r debug reload
+        assert [expr [r ttl testkey foo] > 0]
+    }
 }
@@ -0,0 +1,12 @@
+
+start_server {tags {"repl"}} {
+        test "incr of expired key on replica doesn't cause a crash" {
+            r debug force-master yes
+            r set testkey 1
+            r pexpire testkey 1
+            after 500
+            r incr testkey
+            r incr testkey
+            r debug force-master no
+        }
+}
@@ -1,7 +1,7 @@
-start_server {tags {"rreplay"}} {
+start_server {tags {"rreplay"} overrides {active-replica yes}} {

    test {RREPLAY use current db} {
-        r debug force-master
+        r debug force-master yes
        r select 4
        r set dbnum invalid
        r rreplay "f4d5b2b5-4f07-4ee5-a4f2-5dc98507dfce" "*3\r\n\$3\r\nSET\r\n\$5\r\ndbnum\r\n\$4\r\nfour\r\n"
@@ -10,7 +10,7 @@ start_server {tags {"rreplay"}} {
    reconnect

    test {RREPLAY db different} {
-        r debug force-master
+        r debug force-master yes
        r select 4
        r set testkey four
        r rreplay "f4d5b2b5-4f07-4ee5-a4f2-5dc98507dfce" "*3\r\n\$3\r\nSET\r\n\$7\r\ntestkey\r\n\$4\r\nbebe\r\n" 2
Author	SHA1	Message	Date
John Sully	07cb1f4581	Merge branch 'unstable' into RELEASE_5	2020-03-30 00:51:55 -04:00
John Sully	f2ca2aa1f7	Allow active replicas to connect to each other when replica-serve-stale-data is set	2020-03-30 00:51:40 -04:00
John Sully	b20bf10fda	Don't send build log to stderr	2020-03-28 23:36:57 -04:00
John Sully	fa6870a268	Merge branch 'unstable' into RELEASE_5	2020-03-27 14:36:35 -04:00
John Sully	c3d3c82f3a	Raspberry Pi fixes (compile and replication)	2020-03-27 12:46:25 -04:00
John Sully	f534b5b3a2	OS X Build break fix	2020-03-26 15:21:35 -04:00
John Sully	27da034819	bump version	2020-03-26 01:19:21 -04:00
John Sully	6ec1641294	Fix issue #143	2020-03-25 21:55:31 -04:00
John Sully	7b79ec360b	Give a better error when handling std::terminate	2020-03-25 16:27:24 -04:00
John Sully	56fc6b7deb	Add roundtrip test for subkey expires	2020-03-25 15:34:30 -04:00
John Sully	528a43bd6c	Fix failure to load subkey expires	2020-03-25 01:38:58 -04:00
John Sully	b357803362	Expire entry needs to be resorted after a subkey expires	2020-03-25 01:06:40 -04:00
John Sully	448808529a	Unstable branch should always be version 0.0.0	2020-03-19 15:52:24 -04:00
John Sully	a0bc208412	Merge branch 'RELEASE_5' into unstable	2020-03-19 15:38:16 -04:00
John Sully	6b0172b9bf	Print stack traces of other threads in the deadlock detector	2020-03-19 15:37:49 -04:00
John Sully	bc1eccb66d	Log which thread a message came from	2020-03-19 15:37:24 -04:00
John Sully	a9249d4a82	Fix lock inversion in processEventsWhileBlocked	2020-03-19 15:28:39 -04:00
John Sully	3c373494d6	Prevent dangling lock when we can't free the client	2020-03-19 14:34:23 -04:00
John Sully	5aef606d3b	Fix issue #153 : Config get replicaof returns a corrupt response	2020-03-16 20:43:46 -04:00
John Sully	2ec0b2cd20	Handle HTTP error codes with MOTD	2020-03-15 18:47:01 -04:00
John Sully	cd76a71de3	Update MOTD URL	2020-03-15 18:32:05 -04:00
John Sully	e523afa741	Fix CLANG build break	2020-03-04 17:24:00 -05:00
John Sully	c2a98a60f9	Fix CLANG build break	2020-03-04 17:23:40 -05:00
John Sully	13cf8bbeb7	Merge branch 'unstable' into RELEASE_5	2020-03-04 17:13:32 -05:00
John Sully	60c0d4104d	Add missing files from last checkin	2020-03-04 17:13:18 -05:00
John Sully	693ba4ed4a	Merge branch 'unstable' into RELEASE_5	2020-03-04 17:09:29 -05:00
John Sully	5397f0b033	Add extra logging when reporting errors from masters - especially in rreplay	2020-03-04 17:09:12 -05:00
John Sully	6c78ec7c41	MOTD refactor	2020-03-04 16:22:55 -05:00
John Sully	c0882864f2	Update README.md	2020-03-03 13:58:48 -08:00
John Sully	faa87ed839	Merge branch 'unstable' into RELEASE_5	2020-03-01 21:43:21 -05:00
John Sully	83f585e30a	Update tests to reflect new CRON name (keydb.cron)	2020-03-01 21:42:01 -05:00
benschermel	7a244bb847	Merge branch 'unstable' into RELEASE_5	2020-03-01 19:53:32 -05:00
John Sully	3eb847e28f	Rename cron command to KEYDB.CRON and add help	2020-03-01 19:17:58 -05:00
John Sully	3ce6441a0c	Bump version	2020-02-28 23:53:54 -05:00
John Sully	6d00a1a5cf	Merge branch 'unstable' into RELEASE_5	2020-02-28 23:53:00 -05:00
John Sully	aaca32a5cd	Fix compile failure on raspberry pi machines, issue #141	2020-02-28 23:49:17 -05:00
John Sully	54c96d1995	Fix not respecting max clients config, issue #147	2020-02-28 22:04:42 -05:00
John Sully	e5a2d59446	Fix issue #146	2020-02-28 21:21:05 -05:00
John Sully	d95bead383	Fix crash propogating stale keys	2020-02-26 21:49:40 -05:00
John Sully	6ad6c1d780	Change Redis to KeyDB	2020-02-17 23:45:40 -05:00
John Sully	98aefac09b	Add double unlock detection and improve fastlock_unlock assmebly	2020-02-17 19:54:05 -05:00
John Sully	1ef187533c	processEventsWhileBlocked not exception safe	2020-02-17 19:03:49 -05:00
John Sully	b9db899f6c	Graceful shutdown of server threads when quit is requested	2020-02-16 18:53:46 -05:00
John Sully	06ad1c15d7	Fix memory leak in RDB load	2020-02-16 18:41:50 -05:00
John Sully	63ae679fe1	Merge branch 'unstable' of https://github.com/JohnSully/KeyDB into unstable	2020-02-16 18:12:11 -05:00
John Sully	b92bbf4de8	Memory leak fix on config, and redisDb dtor	2020-02-16 17:08:00 -05:00
John Sully	4781eda722	Fix memory leak of ReplicaNestState on shutdown	2020-02-16 03:43:29 -05:00
John Sully	2fd93c5789	aeDeleteEventLoop use after free and leak fixes	2020-02-16 03:34:25 -05:00
John Sully	52434a583a	Fix race condition in allocating connections to threads	2020-02-11 03:44:28 -05:00
John Sully	16a019dba0	Implement an error handler so bug #125 can't happen	2020-02-11 01:41:00 -05:00
John Sully	1a4c3224c9	Fix higher latency at low load by grouping clients to threads. This fixes slow perf in cluster benchmarks mentioned in issue #102	2020-02-11 01:00:21 -05:00
John Sully	c3317686f8	Fix cases where duplicate RREPLAY is applied	2020-02-11 00:59:07 -05:00
John Sully	4622472123	Fix issue #119	2020-02-10 19:52:57 -05:00
John Sully	f1748f8c76	Fix memory leak in cron	2020-02-10 18:26:04 -05:00
John Sully	0c101dccc8	Add missing test file	2020-02-10 18:15:29 -05:00
John Sully	dc78bf1ccb	CLANG build fix	2020-02-10 00:29:26 -05:00
John Sully	a7cc3aac28	Ensure multi-master works for ring topologies	2020-02-10 00:25:03 -05:00
John Sully	fb2bdf7d05	Addmissing test file	2020-02-08 16:49:41 -05:00
John Sully	050d58007f	Fix issue #137 and #132	2020-02-06 23:31:31 -05:00
John Sully	49d86746ed	Add test to detect issue #137 and #132	2020-02-06 23:31:12 -05:00
John Sully	1069d0b4bf	CRON jobs were running half as often as required	2020-01-28 21:40:40 -05:00
John Sully	c983d57f73	Missing cron files	2020-01-21 19:51:16 -05:00
John Sully	3204a39ada	Initial implementation of the CRON command	2020-01-21 19:50:28 -05:00
John Sully	f4319c1280	Bump version	2020-01-20 19:20:50 -05:00
John Sully	762396d997	Merge branch 'unstable' of https://github.com/JohnSully/KeyDB into unstable	2020-01-11 16:35:06 -05:00
John Sully	c575e7df94	Avoid crash due to excessive posted functions for AOF rewrite	2020-01-11 16:34:30 -05:00
John Sully	aa6409f2e8	Avoid crash due to excessive posted functions for AOF rewrite	2020-01-11 16:34:09 -05:00
John Sully	e603e57588	Merge branch 'unstable' into RELEASE_5	2020-01-10 17:46:23 -05:00
John Sully	049b48ef79	Ignore other flags after pro is enabled since we won't understand them	2020-01-10 17:46:02 -05:00
John Sully	0ba0a6ab0a	Update README.md	2020-01-10 14:21:11 -05:00
John Sully	7475ea1b0b	Update graphs	2020-01-10 14:19:44 -05:00
John Sully	d0654af74f	Merge branch 'unstable' into RELEASE_5	2020-01-06 14:24:49 -05:00
John Sully	d5f8df5997	pro launch should respect KEYDB_PRO_DIRECTORY	2020-01-06 14:24:37 -05:00