Bump version because we added new features

Merge branch 'unstable' into RELEASE_0_9
Talk about active replica
2019-03-25 01:12:31 -04:00 · 2019-03-25 01:11:50 -04:00 · 2019-03-25 00:12:04 -04:00 · 2019-03-24 18:57:25 -04:00 · 2019-03-24 18:06:57 -04:00 · 2019-03-24 15:39:10 -04:00
51 changed files with 992 additions and 279 deletions
@@ -0,0 +1,26 @@
+language: generic
+os: osx
+matrix:
+  include:
+    - os: linux
+      script: make
+      env: COMPILER_NAME=gcc CXX=g++-5 CC=gcc-5
+      addons:
+        apt:
+          packages:
+            - g++-5
+            - nasm
+          sources: &sources
+            - llvm-toolchain-precise-3.8
+            - ubuntu-toolchain-r-test
+    - os: linux
+      script: make MALLOC=libc
+      env: COMPILER_NAME=clang CXX=clang++-3.8 CC=clang-3.8 CXXFLAGS="-stdlib=libc++" LDFLAGS="-stdlib=libc++"
+      addons:
+        apt:
+          packages:
+            - clang-3.8
+            - libc++-dev
+            - libc++abi-dev
+            - nasm
+          sources: *sources
@@ -0,0 +1,10 @@
+FROM ubuntu:18.04
+
+
+
+RUN apt-get update \
+    && DEBIAN_FRONTEND=noninteractive apt-get install -qqy \
+        build-essential nasm autotools-dev autoconf libjemalloc-dev tcl tcl-dev \
+    && apt-get clean
+
+CMD make
@@ -1,67 +0,0 @@
-[Note: this is the Redis manifesto, for general information about
-       installing and running Redis read the README file instead.]
-
-Redis Manifesto
-===============
-
-1 - A DSL for Abstract Data Types. Redis is a DSL (Domain Specific Language)
-    that manipulates abstract data types and implemented as a TCP daemon.
-    Commands manipulate a key space where keys are binary-safe strings and
-    values are different kinds of abstract data types. Every data type
-    represents an abstract version of a fundamental data structure. For instance
-    Redis Lists are an abstract representation of linked lists. In Redis, the
-    essence of a data type isn't just the kind of operations that the data types
-    support, but also the space and time complexity of the data type and the
-    operations performed upon it.
-
-2 - Memory storage is #1. The Redis data set, composed of defined key-value
-    pairs, is primarily stored in the computer's memory. The amount of memory in
-    all kinds of computers, including entry-level servers, is increasing
-    significantly each year. Memory is fast, and allows Redis to have very
-    predictable performance. Datasets composed of 10k or 40 millions keys will
-    perform similarly. Complex data types like Redis Sorted Sets are easy to
-    implement and manipulate in memory with good performance, making Redis very
-    simple. Redis will continue to explore alternative options (where data can
-    be optionally stored on disk, say) but the main goal of the project remains
-    the development of an in-memory database.
-
-3 - Fundamental data structures for a fundamental API. The Redis API is a direct
-    consequence of fundamental data structures. APIs can often be arbitrary but
-    not an API that resembles the nature of fundamental data structures. If we
-    ever meet intelligent life forms from another part of the universe, they'll
-    likely know, understand and recognize the same basic data structures we have
-    in our computer science books. Redis will avoid intermediate layers in API,
-    so that the complexity is obvious and more complex operations can be
-    performed as the sum of the basic operations.
-
-4 - Code is like a poem; it's not just something we write to reach some
-    practical result. Sometimes people that are far from the Redis philosophy
-    suggest using other code written by other authors (frequently in other
-    languages) in order to implement something Redis currently lacks. But to us
-    this is like if Shakespeare decided to end Enrico IV using the Paradiso from
-    the Divina Commedia. Is using any external code a bad idea? Not at all. Like
-    in "One Thousand and One Nights" smaller self contained stories are embedded
-    in a bigger story, we'll be happy to use beautiful self contained libraries
-    when needed. At the same time, when writing the Redis story we're trying to
-    write smaller stories that will fit in to other code.
-
-5 - We're against complexity. We believe designing systems is a fight against
-    complexity. We'll accept to fight the complexity when it's worthwhile but
-    we'll try hard to recognize when a small feature is not worth 1000s of lines
-    of code. Most of the time the best way to fight complexity is by not
-    creating it at all.
-
-6 - Two levels of API. The Redis API has two levels: 1) a subset of the API fits
-    naturally into a distributed version of Redis and 2) a more complex API that
-    supports multi-key operations. Both are useful if used judiciously but
-    there's no way to make the more complex multi-keys API distributed in an
-    opaque way without violating our other principles. We don't want to provide
-    the illusion of something that will work magically when actually it can't in
-    all cases. Instead we'll provide commands to quickly migrate keys from one
-    instance to another to perform multi-key operations and expose the tradeoffs
-    to the user.
-
-7 - We optimize for joy. We believe writing code is a lot of hard work, and the
-    only way it can be worth is by enjoying it. When there is no longer joy in
-    writing code, the best thing to do is stop. To prevent this, we'll avoid
-    taking paths that will make Redis less of a joy to develop.
@@ -1,10 +1,24 @@
+![Current Release](https://img.shields.io/github/release/JohnSully/KeyDB.svg)
+[![Build Status](https://travis-ci.org/JohnSully/KeyDB.svg?branch=unstable)](https://travis-ci.org/JohnSully/KeyDB) [![Join the chat at https://gitter.im/KeyDB/community](https://badges.gitter.im/KeyDB/community.svg)](https://gitter.im/KeyDB/community?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
+
 What is KeyDB?
 --------------

-KeyDB is a high performance fork of Redis focussing on multithreading, memory efficiency, and high throughput.  In addition to multithreading KeyDB also has features only available in Redis Enterprise such as FLASH storage support, and some not available at all such as direct backup to AWS S3.
+KeyDB is a high performance fork of Redis focusing on multithreading, memory efficiency, and high throughput.  In addition to multithreading KeyDB also has features only available in Redis Enterprise such as FLASH storage support, and some not available at all such as direct backup to AWS S3.

 On the same hardware KeyDB can perform twice as many queries per second as Redis, with 60% lower latency.

+KeyDB has full compatibility with the Redis protocol, modules, and scripts.  This includes full support for transactions, and atomic execution of scripts.  For more information see our architecture section below.
+
+Try our docker container: https://hub.docker.com/r/eqalpha/keydb
+
+Talk on Gitter: https://gitter.im/KeyDB
+
+New: Active Replica Support
+---------------------------
+
+New! KeyDB now has support for Active Replicas.  This feature greatly simplifies hot-spare failover and allows you to distribute writes over replicas instead of just a single master.  For more information [see the wiki page](https://github.com/JohnSully/KeyDB/wiki/KeyDB-(Redis-Fork):-Active-Replica-Support).
+
 Why fork Redis?
 ---------------

@@ -15,6 +29,8 @@ We plan to track the Redis repo closely and hope our projects can learn from eac
 Benchmarking KeyDB
 ------------------

+<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/>
+
 Please note keydb-benchmark and redis-benchmark are currently single threaded and too slow to properly benchmark KeyDB.  We recommend using a redis cluster benchmark tool such as [memtier](https://github.com/RedisLabs/memtier_benchmark).  Please ensure your machine has enough cores for both KeyDB and memteir if testing locally.  KeyDB expects exclusive use of any cores assigned to it.

 For more details on how we benchmarked KeyDB along with performance numbers check out our blog post: [Redis Should Be Multithreaded](https://medium.com/@john_63123/redis-should-be-multi-threaded-e28319cab744?source=friends_link&sk=7ce8e9fe3ec8224a4d27ef075d085457)
@@ -35,7 +51,7 @@ If you would like to use the FLASH backed storage this option configures the dir
    
    db-s3-object /path/to/bucket

-If you would like KeyDB to dump directly to AWS S3 this option specifies the bucket.  Using this option with the traditional RDB options will result in KeyDB backing up twice to both locations.  This requires the AWS CLI tools to be installed and configured which are used under the hood to transfer the data.
+If you would like KeyDB to dump and load directly to AWS S3 this option specifies the bucket.  Using this option with the traditional RDB options will result in KeyDB backing up twice to both locations.  If both are specified KeyDB will first attempt to load from the local dump file and if that fails load from S3.  This requires the AWS CLI tools to be installed and configured which are used under the hood to transfer the data.

 All other configuration options behave as you'd expect.  Your existing configuration files should continue to work unchanged.

@@ -175,6 +191,8 @@ for Ubuntu and Debian systems:
    % cd utils
    % ./install_server.sh

+_Note_: `install_server.sh` will not work on Mac OSX; it is built for Linux only.
+
 The script will ask you a few questions and will setup everything you need
 to run KeyDB properly as a background daemon that will start again on
 system reboots.
@@ -182,6 +200,37 @@ system reboots.
 You'll be able to stop and start KeyDB using the script named
 `/etc/init.d/keydb_<portnumber>`, for instance `/etc/init.d/keydb_6379`.

+Multithreading Architecture
+---------------------------
+
+KeyDB works by running the normal Redis event loop on multiple threads.  Network IO, and query parsing are done concurrently.  Each connection is assigned a thread on accept().  Access to the core hash table is guarded by spinlock.  Because the hashtable access is extremely fast this lock has low contention.  Transactions hold the lock for the duration of the EXEC command.  Modules work in concert with the GIL which is only acquired when all server threads are paused.  This maintains the atomicity guarantees modules expect.
+
+Unlike most databases the core data structure is the fastest part of the system.  Most of the query time comes from parsing the REPL protocol and copying data to/from the network.
+
+Future work:
+ - Allow rebalancing of connections to different threads after the connection
+ - Allow multiple readers access to the hashtable concurrently
+
+Docker Build
+------------
+
+Run the following commands for a full source download and build:
+
+```
+git clone git@github.com:JohnSully/KeyDB.git
+docker run -it --rm `pwd`/KeyDB:/build -w /build devopsdood/keydb-builder make
+```
+
+Then you have fresh binaries built, you can also pass any other options to the make command above after the word make. E.g.
+
+```docker run -it --rm `pwd`/KeyDB:/build -w /build devopsdood/keydb-builder make MAllOC=memkind```
+
+The above commands will build you binaries in the src directory. Standard `make install` without Docker command will work after if you wish to install
+
+If you'd prefer you can build the Dockerfile in the repo instead of pulling the above container for use:
+
+`docker build -t KeyDB .`
+
 Code contributions
 -----------------

@@ -83,7 +83,7 @@ struct __attribute__ ((__packed__)) sdshdr64 {
 #define SDS_HDR(T,s) ((struct sdshdr##T *)((s)-(sizeof(struct sdshdr##T))))
 #define SDS_TYPE_5_LEN(f) ((f)>>SDS_TYPE_BITS)

-inline size_t sdslen(const sds s) {
+static inline size_t sdslen(const sds s) {
    unsigned char flags = s[-1];
    
    switch(__builtin_expect((flags&SDS_TYPE_MASK), SDS_TYPE_5)) {
@@ -1562,3 +1562,8 @@ server-threads 2
 # Should KeyDB pin threads to CPUs? By default this is disabled, and KeyDB will not bind threads.
 # When enabled threads are bount to cores sequentially starting at core 0.
 # server-thread-affinity true
+
+# Uncomment the option below to enable Active Active support.  Note that
+# replicas will still sync in the normal way and incorrect ordering when
+# bringing up replicas can result in data loss (the first master will win).
+# active-replica yes
@@ -27,7 +27,7 @@ ifneq (,$(findstring FreeBSD,$(uname_S)))
  STD+=-Wno-c11-extensions
 endif
 endif
-WARN=-Wall -Werror -W -Wno-missing-field-initializers
+WARN=-Wall -W -Wno-missing-field-initializers
 OPT=$(OPTIMIZATION)

 PREFIX?=/usr/local
@@ -45,8 +45,9 @@ endif
 endif

 USEASM?=true
-# Do we use our assembly spinlock?
+# Do we use our assembly spinlock? X64 only
 ifeq ($(uname_S),Linux)
+ifeq ($(uname_M),x86_64)
 ifneq ($(@@),32bit)
 ifeq ($(USEASM),true)
 	ASM_OBJ+= fastlock_x64.o
@@ -55,6 +56,11 @@ ifeq ($(USEASM),true)
 endif
 endif
 endif
+endif
+
+ifeq ($(COMPILER_NAME),clang)
+	CXXFLAGS+= -stdlib=libc++ 
+endif

 # To get ARM stack traces if Redis crashes we need a special C flag.
 ifneq (,$(filter aarch64 armv,$(uname_M)))
@@ -86,7 +92,7 @@ endif
 -include .make-settings

 FINAL_CFLAGS=$(STD) $(WARN) $(OPT) $(DEBUG) $(CFLAGS) $(REDIS_CFLAGS)
-FINAL_CXXFLAGS=$(CXX_STD) $(WARN) $(OPT) $(DEBUG) $(CFLAGS) $(REDIS_CFLAGS)
+FINAL_CXXFLAGS=$(CXX_STD) $(WARN) $(OPT) $(DEBUG) $(CFLAGS) $(CXXFLAGS) $(REDIS_CFLAGS)
 FINAL_LDFLAGS=$(LDFLAGS) $(REDIS_LDFLAGS) $(DEBUG)
 FINAL_LIBS=-lm
 DEBUG=-g -ggdb
@@ -133,7 +139,7 @@ ifeq ($(uname_S),DragonFly)
 else
 	# All the other OSes (notably Linux)
 	FINAL_LDFLAGS+= -rdynamic
-	FINAL_LIBS+=-ldl -pthread -lrt
+	FINAL_LIBS+=-ldl -pthread -lrt -luuid
 endif
 endif
 endif
@@ -229,17 +235,15 @@ persist-settings: distclean
 .PHONY: persist-settings

 # Prerequisites target
-.make-prerequisites:
-	@touch $@
-
 # Clean everything, persist settings and build dependencies if anything changed
 ifneq ($(strip $(PREV_FINAL_CFLAGS)), $(strip $(FINAL_CFLAGS)))
 .make-prerequisites: persist-settings
-endif
-
-ifneq ($(strip $(PREV_FINAL_LDFLAGS)), $(strip $(FINAL_LDFLAGS)))
+else ifneq ($(strip $(PREV_FINAL_LDFLAGS)), $(strip $(FINAL_LDFLAGS)))
 .make-prerequisites: persist-settings
+else
+.make-prerequisites:
 endif
+	@touch $@

 # keydb-server
 $(REDIS_SERVER_NAME): $(REDIS_SERVER_OBJ)
@@ -343,3 +347,6 @@ install: all
 	$(REDIS_INSTALL) $(REDIS_CHECK_RDB_NAME) $(INSTALL_BIN)
 	$(REDIS_INSTALL) $(REDIS_CHECK_AOF_NAME) $(INSTALL_BIN)
 	@ln -sf $(REDIS_SERVER_NAME) $(INSTALL_BIN)/$(REDIS_SENTINEL_NAME)
+
+uninstall:
+	rm -f $(INSTALL_BIN)/{$(REDIS_SERVER_NAME),$(REDIS_BENCHMARK_NAME),$(REDIS_CLI_NAME),$(REDIS_CHECK_RDB_NAME),$(REDIS_CHECK_AOF_NAME),$(REDIS_SENTINEL_NAME)}
@@ -1389,6 +1389,8 @@ void ACLLoadUsersAtStartup(void) {
 * ACL SETUSER <username> ... acl rules ...
 * ACL DELUSER <username> [...]
 * ACL GETUSER <username>
+ * ACL GENPASS
+ * ACL WHOAMI
 */
 void aclCommand(client *c) {
    char *sub = ptrFromObj(c->argv[1]);
@@ -1571,6 +1573,10 @@ void aclCommand(client *c) {
        }
        dictReleaseIterator(di);
        setDeferredArrayLen(c,dl,arraylen);
+    } else if (!strcasecmp(sub,"genpass") && c->argc == 2) {
+        char pass[32]; /* 128 bits of actual pseudo random data. */
+        getRandomHexChars(pass,sizeof(pass));
+        addReplyBulkCBuffer(c,pass,sizeof(pass));
    } else if (!strcasecmp(sub,"help")) {
        const char *help[] = {
 "LOAD                              -- Reload users from the ACL file.",
@@ -1581,6 +1587,7 @@ void aclCommand(client *c) {
 "DELUSER <username> [...]          -- Delete a list of users.",
 "CAT                               -- List available categories.",
 "CAT <category>                    -- List commands inside category.",
+"GENPASS                           -- Generate a secure user password.",
 "WHOAMI                            -- Return the current connection username.",
 NULL
        };
@@ -39,10 +39,10 @@ typedef struct aeApiState {
 } aeApiState;

 static int aeApiCreate(aeEventLoop *eventLoop) {
-    aeApiState *state = zmalloc(sizeof(aeApiState));
+    aeApiState *state = (aeApiState*)zmalloc(sizeof(aeApiState), MALLOC_LOCAL);

    if (!state) return -1;
-    state->events = zmalloc(sizeof(struct kevent)*eventLoop->setsize, MALLOC_LOCAL);
+    state->events = (struct kevent*)zmalloc(sizeof(struct kevent)*eventLoop->setsize, MALLOC_LOCAL);
    if (!state->events) {
        zfree(state);
        return -1;
@@ -58,14 +58,14 @@ static int aeApiCreate(aeEventLoop *eventLoop) {
 }

 static int aeApiResize(aeEventLoop *eventLoop, int setsize) {
-    aeApiState *state = eventLoop->apidata;
+    aeApiState *state = (aeApiState*)eventLoop->apidata;

-    state->events = zrealloc(state->events, sizeof(struct kevent)*setsize, MALLOC_LOCAL);
+    state->events = (struct kevent*)zrealloc(state->events, sizeof(struct kevent)*setsize, MALLOC_LOCAL);
    return 0;
 }

 static void aeApiFree(aeEventLoop *eventLoop) {
-    aeApiState *state = eventLoop->apidata;
+    aeApiState *state = (aeApiState*)eventLoop->apidata;

    close(state->kqfd);
    zfree(state->events);
@@ -73,7 +73,7 @@ static void aeApiFree(aeEventLoop *eventLoop) {
 }

 static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
-    aeApiState *state = eventLoop->apidata;
+    aeApiState *state = (aeApiState*)eventLoop->apidata;
    struct kevent ke;

    if (mask & AE_READABLE) {
@@ -88,7 +88,7 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
 }

 static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
-    aeApiState *state = eventLoop->apidata;
+    aeApiState *state = (aeApiState*)eventLoop->apidata;
    struct kevent ke;

    if (mask & AE_READABLE) {
@@ -102,7 +102,7 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
 }

 static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
-    aeApiState *state = eventLoop->apidata;
+    aeApiState *state = (aeApiState*)eventLoop->apidata;
    int retval, numevents = 0;

    if (tvp != NULL) {
@@ -133,6 +133,6 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
    return numevents;
 }

-static char *aeApiName(void) {
+static const char *aeApiName(void) {
    return "kqueue";
 }
@@ -1248,7 +1248,7 @@ int rewriteModuleObject(rio *r, robj *key, robj *o) {
    RedisModuleIO io;
    moduleValue *mv = ptrFromObj(o);
    moduleType *mt = mv->type;
-    moduleInitIOContext(io,mt,r);
+    moduleInitIOContext(io,mt,r,key);
    mt->aof_rewrite(&io,key,mv->value);
    if (io.ctx) {
        moduleFreeContext(io.ctx);
@@ -1621,6 +1621,9 @@ void aofRemoveTempFile(pid_t childpid) {

    snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid);
    unlink(tmpfile);
+
+    snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) childpid);
+    unlink(tmpfile);
 }

 /* Update the server.aof_current_size field explicitly using stat(2)
@@ -77,10 +77,18 @@ int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb
 * is zero. */
 int getTimeoutFromObjectOrReply(client *c, robj *object, mstime_t *timeout, int unit) {
    long long tval;
+    long double ftval;

-    if (getLongLongFromObjectOrReply(c,object,&tval,
-        "timeout is not an integer or out of range") != C_OK)
-        return C_ERR;
+    if (unit == UNIT_SECONDS) {
+        if (getLongDoubleFromObjectOrReply(c,object,&ftval,
+            "timeout is not an float or out of range") != C_OK)
+            return C_ERR;
+        tval = (long long) (ftval * 1000.0);
+    } else {
+        if (getLongLongFromObjectOrReply(c,object,&tval,
+            "timeout is not an integer or out of range") != C_OK)
+            return C_ERR;
+    }

    if (tval < 0) {
        addReplyError(c,"timeout is negative");
@@ -88,7 +96,6 @@ int getTimeoutFromObjectOrReply(client *c, robj *object, mstime_t *timeout, int
    }

    if (tval > 0) {
-        if (unit == UNIT_SECONDS) tval *= 1000;
        tval += mstime();
    }
    *timeout = tval;
@@ -3031,6 +3031,7 @@ void clusterHandleSlaveFailover(void) {
        if (server.cluster->mf_end) {
            server.cluster->failover_auth_time = mstime();
            server.cluster->failover_auth_rank = 0;
+	    clusterDoBeforeSleep(CLUSTER_TODO_HANDLE_FAILOVER);
        }
        serverLog(LL_WARNING,
            "Start of election delayed for %lld milliseconds "
@@ -4775,7 +4776,7 @@ NULL

 /* Generates a DUMP-format representation of the object 'o', adding it to the
 * io stream pointed by 'rio'. This function can't fail. */
-void createDumpPayload(rio *payload, robj *o) {
+void createDumpPayload(rio *payload, robj *o, robj *key) {
    unsigned char buf[2];
    uint64_t crc;

@@ -4783,7 +4784,7 @@ void createDumpPayload(rio *payload, robj *o) {
     * byte followed by the serialized object. This is understood by RESTORE. */
    rioInitWithBuffer(payload,sdsempty());
    serverAssert(rdbSaveObjectType(payload,o));
-    serverAssert(rdbSaveObject(payload,o));
+    serverAssert(rdbSaveObject(payload,o,key));

    /* Write the footer, this is how it looks like:
     * ----------------+---------------------+---------------+
@@ -4841,7 +4842,7 @@ void dumpCommand(client *c) {
    }

    /* Create the DUMP encoded representation. */
-    createDumpPayload(&payload,o);
+    createDumpPayload(&payload,o,c->argv[1]);

    /* Transfer to the client */
    dumpobj = createObject(OBJ_STRING,payload.io.buffer.ptr);
@@ -4914,7 +4915,7 @@ void restoreCommand(client *c) {

    rioInitWithBuffer(&payload,ptrFromObj(c->argv[3]));
    if (((type = rdbLoadObjectType(&payload)) == -1) ||
-        ((obj = rdbLoadObject(type,&payload)) == NULL))
+        ((obj = rdbLoadObject(type,&payload,c->argv[1])) == NULL))
    {
        addReplyError(c,"Bad data format");
        return;
@@ -5202,7 +5203,7 @@ try_again:

        /* Emit the payload argument, that is the serialized object using
         * the DUMP format. */
-        createDumpPayload(&payload,ov[j]);
+        createDumpPayload(&payload,ov[j],kv[j]);
        serverAssertWithInfo(c,NULL,
            rioWriteBulkString(&cmd,payload.io.buffer.ptr,
                               sdslen(payload.io.buffer.ptr)));
@@ -849,6 +849,16 @@ void loadServerConfigFromString(char *config) {
                err = "Unknown argument: server-thread-affinity expects either true or false";
                goto loaderr;
            }
+        } else if (!strcasecmp(argv[0], "active-replica") && argc == 2) {
+            server.fActiveReplica = yesnotoi(argv[1]);
+            if (server.repl_slave_ro) {
+                server.repl_slave_ro = FALSE;
+                serverLog(LL_NOTICE, "Notice: \"active-replica yes\" implies \"replica-read-only no\"");
+            }
+            if (server.fActiveReplica == -1) {
+                server.fActiveReplica = CONFIG_DEFAULT_ACTIVE_REPLICA;
+                err = "argument must be 'yes' or 'no'"; goto loaderr;
+            }
        } else {
            err = "Bad directive or wrong number of arguments"; goto loaderr;
        }
@@ -1103,8 +1113,8 @@ void configSetCommand(client *c) {
            int soft_seconds;

            class = getClientTypeByName(v[j]);
-            hard = strtoll(v[j+1],NULL,10);
-            soft = strtoll(v[j+2],NULL,10);
+            hard = memtoll(v[j+1],NULL);
+            soft = memtoll(v[j+2],NULL);
            soft_seconds = strtoll(v[j+3],NULL,10);

            server.client_obuf_limits[class].hard_limit_bytes = hard;
@@ -2350,6 +2360,7 @@ int rewriteConfig(char *path) {
    rewriteConfigYesNoOption(state,"lazyfree-lazy-server-del",server.lazyfree_lazy_server_del,CONFIG_DEFAULT_LAZYFREE_LAZY_SERVER_DEL);
    rewriteConfigYesNoOption(state,"replica-lazy-flush",server.repl_slave_lazy_flush,CONFIG_DEFAULT_SLAVE_LAZY_FLUSH);
    rewriteConfigYesNoOption(state,"dynamic-hz",server.dynamic_hz,CONFIG_DEFAULT_DYNAMIC_HZ);
+    rewriteConfigYesNoOption(state,"active-replica",server.fActiveReplica,CONFIG_DEFAULT_ACTIVE_REPLICA);

    /* Rewrite Sentinel config if in Sentinel mode. */
    if (server.sentinel_mode) rewriteConfigSentinelOption(state);
@@ -362,7 +362,7 @@ NULL
        }
        emptyDb(-1,EMPTYDB_NO_FLAGS,NULL);
        protectClient(c);
-        int ret = rdbLoad(server.rdb_filename,NULL);
+        int ret = rdbLoad(NULL);
        unprotectClient(c);
        if (ret != C_OK) {
            addReplyError(c,"Error trying to load the RDB dump");
@@ -34,6 +34,20 @@
 #include <sched.h>
 #include <atomic>
 #include <assert.h>
+#include <pthread.h>
+#include <limits.h>
+
+#ifdef __APPLE__
+#include <TargetConditionals.h>
+#ifdef TARGET_OS_MAC
+/* The CLANG that ships with Mac OS doesn't have these builtins.
+    but on x86 they are just normal reads/writes anyways */
+#define __atomic_load_4(ptr, csq) (*(reinterpret_cast<const volatile uint32_t*>(ptr)))
+#define __atomic_load_2(ptr, csq) (*(reinterpret_cast<const volatile uint16_t*>(ptr)))
+
+#define __atomic_store_4(ptr, val, csq) (*(reinterpret_cast<volatile uint32_t*>(ptr)) = val)
+#endif
+#endif

 /****************************************************
 *
@@ -43,12 +57,28 @@
 ****************************************************/

 static_assert(sizeof(pid_t) <= sizeof(fastlock::m_pidOwner), "fastlock::m_pidOwner not large enough");
+uint64_t g_longwaits = 0;
+
+uint64_t fastlock_getlongwaitcount()
+{
+    return g_longwaits;
+}
+

 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;
 }

@@ -75,7 +105,10 @@ extern "C" void fastlock_lock(struct fastlock *lock)
    while (__atomic_load_2(&lock->m_ticket.m_active, __ATOMIC_ACQUIRE) != myticket)
    {
        if ((++cloops % 1024*1024) == 0)
+        {
            sched_yield();
+            ++g_longwaits;
+        }
 #if defined(__i386__) || defined(__amd64__)
        __asm__ ("pause");
 #endif
@@ -103,7 +136,7 @@ extern "C" int fastlock_trylock(struct fastlock *lock)

    struct ticket ticket_expect { active, active };
    struct ticket ticket_setiflocked { active, next };
-    if (__atomic_compare_exchange(&lock->m_ticket, &ticket_expect, &ticket_setiflocked, true /*strong*/, __ATOMIC_ACQUIRE, __ATOMIC_ACQUIRE))
+    if (__atomic_compare_exchange(&lock->m_ticket, &ticket_expect, &ticket_setiflocked, false /*weak*/, __ATOMIC_ACQUIRE, __ATOMIC_ACQUIRE))
    {
        lock->m_depth = 1;
        __atomic_store_4(&lock->m_pidOwner, gettid(), __ATOMIC_RELEASE);
@@ -137,4 +170,4 @@ extern "C" void fastlock_free(struct fastlock *lock)
 bool fastlock::fOwnLock()
 {
    return gettid() == m_pidOwner;
-}
+}
@@ -13,6 +13,8 @@ int fastlock_trylock(struct fastlock *lock);
 void fastlock_unlock(struct fastlock *lock);
 void fastlock_free(struct fastlock *lock);

+uint64_t fastlock_getlongwaitcount();   // this is a global value
+
 /* End C API */
 #ifdef __cplusplus
 }
@@ -2,6 +2,7 @@ section .text

 extern gettid
 extern sched_yield
+extern g_longwaits

 ;	This is the first use of assembly in this codebase, a valid question is WHY?
 ;	The spinlock we implement here is performance critical, and simply put GCC
@@ -49,6 +50,8 @@ ALIGN 16
 	syscall                 ; give up our timeslice we'll be here a while
 	pop rax
 	pop rsi
+	mov rcx, g_longwaits
+	inc qword [rcx]         ; increment our long wait counter
 	mov rdi, [rsp]          ; our struct pointer is on the stack already
 	xor ecx, ecx            ; Reset our loop counter
 	jmp .LLoop              ; Get back in the game
@@ -35,7 +35,12 @@
 #if defined(__linux__)
 #define _GNU_SOURCE 1
 #define _DEFAULT_SOURCE 1
+
+#include <linux/version.h>
+#if LINUX_VERSION_CODE >= KERNEL_VERSION(3,19,0)
+#define HAVE_SO_INCOMING_CPU 1
 #endif
+#endif  // __linux__

 #if defined(_AIX)
 #define _ALL_SOURCE
@@ -659,7 +659,7 @@ void georadiusGeneric(client *c, int flags) {
            zsetConvertToZiplistIfNeeded(zobj,maxelelen);
            setKey(c->db,storekey,zobj);
            decrRefCount(zobj);
-            notifyKeyspaceEvent(NOTIFY_LIST,"georadiusstore",storekey,
+            notifyKeyspaceEvent(NOTIFY_ZSET,"georadiusstore",storekey,
                                c->db->id);
            server.dirty += returned_items;
        } else if (dbDelete(c->db,storekey)) {
@@ -614,6 +614,7 @@ int hllSparseToDense(robj *o) {
        } else {
            runlen = HLL_SPARSE_VAL_LEN(p);
            regval = HLL_SPARSE_VAL_VALUE(p);
+            if ((runlen + idx) > HLL_REGISTERS) break; /* Overflow. */
            while(runlen--) {
                HLL_DENSE_SET_REGISTER(hdr->registers,idx,regval);
                idx++;
@@ -1013,7 +1014,12 @@ uint64_t hllCount(struct hllhdr *hdr, int *invalid) {
    double m = HLL_REGISTERS;
    double E;
    int j;
-    int reghisto[HLL_Q+2] = {0};
+    /* Note that reghisto size could be just HLL_Q+2, becuase HLL_Q+1 is
+     * the maximum frequency of the "000...1" sequence the hash function is
+     * able to return. However it is slow to check for sanity of the
+     * input: instead we history array at a safe size: overflows will
+     * just write data to wrong, but correctly allocated, places. */
+    int reghisto[64] = {0};

    /* Compute register histogram */
    if (hdr->encoding == HLL_DENSE) {
@@ -1088,6 +1094,7 @@ int hllMerge(uint8_t *max, size_t cmax, robj *hll) {
            } else {
                runlen = HLL_SPARSE_VAL_LEN(p);
                regval = HLL_SPARSE_VAL_VALUE(p);
+                if ((runlen + i) > HLL_REGISTERS) break; /* Overflow. */
                while(runlen--) {
                    if (i < 0 || (size_t)i >= cmax)
                        return C_ERR;
@@ -1239,7 +1246,7 @@ void pfcountCommand(client *c) {

            /* Merge with this HLL with our 'max' HHL by setting max[i]
             * to MAX(max[i],hll[i]). */
-            if (hllMerge(registers,sizeof(max),o) == C_ERR) {
+            if (hllMerge(registers,HLL_REGISTERS,o) == C_ERR) {
                addReplySds(c,sdsnew(invalid_hll_err));
                return;
            }
@@ -47,9 +47,21 @@ struct RedisModule {
    int ver;        /* Module version. We use just progressive integers. */
    int apiver;     /* Module API version as requested during initialization.*/
    list *types;    /* Module data types. */
+    list *usedby;   /* List of modules using APIs from this one. */
+    list *using;    /* List of modules we use some APIs of. */
 };
 typedef struct RedisModule RedisModule;

+/* This represents a shared API. Shared APIs will be used to populate
+ * the server.sharedapi dictionary, mapping names of APIs exported by
+ * modules for other modules to use, to their structure specifying the
+ * function pointer that can be called. */
+struct RedisModuleSharedAPI {
+    void *func;
+    RedisModule *module;
+};
+typedef struct RedisModuleSharedAPI RedisModuleSharedAPI;
+
 static dict *modules; /* Hash table of modules. SDS -> RedisModule ptr.*/

 /* Entries in the context->amqueue array, representing objects to free
@@ -510,6 +522,22 @@ void RedisModuleCommandDispatcher(client *c) {
    cp->func(&ctx,(void**)c->argv,c->argc);
    moduleHandlePropagationAfterCommandCallback(&ctx);
    moduleFreeContext(&ctx);
+
+    /* In some cases processMultibulkBuffer uses sdsMakeRoomFor to
+     * expand the query buffer, and in order to avoid a big object copy
+     * the query buffer SDS may be used directly as the SDS string backing
+     * the client argument vectors: sometimes this will result in the SDS
+     * string having unused space at the end. Later if a module takes ownership
+     * of the RedisString, such space will be wasted forever. Inside the
+     * Redis core this is not a problem because tryObjectEncoding() is called
+     * before storing strings in the key space. Here we need to do it
+     * for the module. */
+    for (int i = 0; i < c->argc; i++) {
+        /* Only do the work if the module took ownership of the object:
+         * in that case the refcount is no longer 1. */
+        if (c->argv[i]->refcount > 1)
+            trimStringObjectIfNeeded(c->argv[i]);
+    }
 }

 /* This function returns the list of keys, with the same interface as the
@@ -702,6 +730,8 @@ void RM_SetModuleAttribs(RedisModuleCtx *ctx, const char *name, int ver, int api
    module->ver = ver;
    module->apiver = apiver;
    module->types = listCreate();
+    module->usedby = listCreate();
+    module->using = listCreate();
    ctx->module = module;
 }

@@ -1360,6 +1390,9 @@ int RM_GetSelectedDb(RedisModuleCtx *ctx) {
 *
 *  * REDISMODULE_CTX_FLAGS_MULTI: The command is running inside a transaction
 *
+ *  * REDISMODULE_CTX_FLAGS_REPLICATED: The command was sent over the replication
+ *    link by the MASTER
+ *
 *  * REDISMODULE_CTX_FLAGS_MASTER: The Redis instance is a master
 *
 *  * REDISMODULE_CTX_FLAGS_SLAVE: The Redis instance is a slave
@@ -1392,6 +1425,9 @@ int RM_GetContextFlags(RedisModuleCtx *ctx) {
         flags |= REDISMODULE_CTX_FLAGS_LUA;
        if (ctx->client->flags & CLIENT_MULTI)
         flags |= REDISMODULE_CTX_FLAGS_MULTI;
+        /* Module command recieved from MASTER, is replicated. */
+        if (ctx->client->flags & CLIENT_MASTER)
+         flags |= REDISMODULE_CTX_FLAGS_REPLICATED;
    }

    if (server.cluster_enabled)
@@ -3403,6 +3439,14 @@ RedisModuleCtx *RM_GetContextFromIO(RedisModuleIO *io) {
    return io->ctx;
 }

+/* Returns a RedisModuleString with the name of the key currently saving or
+ * loading, when an IO data type callback is called.  There is no guarantee
+ * that the key name is always available, so this may return NULL.
+ */
+const RedisModuleString *RM_GetKeyNameFromIO(RedisModuleIO *io) {
+    return io->key;
+}
+
 /* --------------------------------------------------------------------------
 * Logging
 * -------------------------------------------------------------------------- */
@@ -3424,6 +3468,8 @@ void RM_LogRaw(RedisModule *module, const char *levelstr, const char *fmt, va_li
    else if (!strcasecmp(levelstr,"warning")) level = LL_WARNING;
    else level = LL_VERBOSE; /* Default. */

+    if (level < server.verbosity) return;
+
    name_len = snprintf(msg, sizeof(msg),"<%s> ", module->name);
    vsnprintf(msg + name_len, sizeof(msg) - name_len, fmt, ap);
    serverLogRaw(level,msg);
@@ -4647,6 +4693,121 @@ void RM_GetRandomHexChars(char *dst, size_t len) {
    getRandomHexChars(dst,len);
 }

+/* --------------------------------------------------------------------------
+ * Modules API exporting / importing
+ * -------------------------------------------------------------------------- */
+
+/* This function is called by a module in order to export some API with a
+ * given name. Other modules will be able to use this API by calling the
+ * symmetrical function RM_GetSharedAPI() and casting the return value to
+ * the right function pointer.
+ *
+ * The function will return REDISMODULE_OK if the name is not already taken,
+ * otherwise REDISMODULE_ERR will be returned and no operation will be
+ * performed.
+ *
+ * IMPORTANT: the apiname argument should be a string literal with static
+ * lifetime. The API relies on the fact that it will always be valid in
+ * the future. */
+int RM_ExportSharedAPI(RedisModuleCtx *ctx, const char *apiname, void *func) {
+    RedisModuleSharedAPI *sapi = zmalloc(sizeof(*sapi), MALLOC_LOCAL);
+    sapi->module = ctx->module;
+    sapi->func = func;
+    if (dictAdd(server.sharedapi, (char*)apiname, sapi) != DICT_OK) {
+        zfree(sapi);
+        return REDISMODULE_ERR;
+    }
+    return REDISMODULE_OK;
+}
+
+/* Request an exported API pointer. The return value is just a void pointer
+ * that the caller of this function will be required to cast to the right
+ * function pointer, so this is a private contract between modules.
+ *
+ * If the requested API is not available then NULL is returned. Because
+ * modules can be loaded at different times with different order, this
+ * function calls should be put inside some module generic API registering
+ * step, that is called every time a module attempts to execute a
+ * command that requires external APIs: if some API cannot be resolved, the
+ * command should return an error.
+ *
+ * Here is an exmaple:
+ *
+ *     int ... myCommandImplementation() {
+ *        if (getExternalAPIs() == 0) {
+ *             reply with an error here if we cannot have the APIs
+ *        }
+ *        // Use the API:
+ *        myFunctionPointer(foo);
+ *     }
+ *
+ * And the function registerAPI() is:
+ *
+ *     int getExternalAPIs(void) {
+ *         static int api_loaded = 0;
+ *         if (api_loaded != 0) return 1; // APIs already resolved.
+ *
+ *         myFunctionPointer = RedisModule_GetOtherModuleAPI("...");
+ *         if (myFunctionPointer == NULL) return 0;
+ *
+ *         return 1;
+ *     }
+ */
+void *RM_GetSharedAPI(RedisModuleCtx *ctx, const char *apiname) {
+    dictEntry *de = dictFind(server.sharedapi, apiname);
+    if (de == NULL) return NULL;
+    RedisModuleSharedAPI *sapi = dictGetVal(de);
+    if (listSearchKey(sapi->module->usedby,ctx->module) == NULL) {
+        listAddNodeTail(sapi->module->usedby,ctx->module);
+        listAddNodeTail(ctx->module->using,sapi->module);
+    }
+    return sapi->func;
+}
+
+/* Remove all the APIs registered by the specified module. Usually you
+ * want this when the module is going to be unloaded. This function
+ * assumes that's caller responsibility to make sure the APIs are not
+ * used by other modules.
+ *
+ * The number of unregistered APIs is returned. */
+int moduleUnregisterSharedAPI(RedisModule *module) {
+    int count = 0;
+    dictIterator *di = dictGetSafeIterator(server.sharedapi);
+    dictEntry *de;
+    while ((de = dictNext(di)) != NULL) {
+        const char *apiname = dictGetKey(de);
+        RedisModuleSharedAPI *sapi = dictGetVal(de);
+        if (sapi->module == module) {
+            dictDelete(server.sharedapi,apiname);
+            zfree(sapi);
+            count++;
+        }
+    }
+    dictReleaseIterator(di);
+    return count;
+}
+
+/* Remove the specified module as an user of APIs of ever other module.
+ * This is usually called when a module is unloaded.
+ *
+ * Returns the number of modules this module was using APIs from. */
+int moduleUnregisterUsedAPI(RedisModule *module) {
+    listIter li;
+    listNode *ln;
+    int count = 0;
+
+    listRewind(module->using,&li);
+    while((ln = listNext(&li))) {
+        RedisModule *used = ln->value;
+        listNode *ln = listSearchKey(used->usedby,module);
+        if (ln) {
+            listDelNode(module->using,ln);
+            count++;
+        }
+    }
+    return count;
+}
+
 /* --------------------------------------------------------------------------
 * Modules API internals
 * -------------------------------------------------------------------------- */
@@ -4792,6 +4953,8 @@ int moduleLoad(const char *path, void **module_argv, int module_argc) {
    if (onload((void*)&ctx,module_argv,module_argc) == REDISMODULE_ERR) {
        if (ctx.module) {
            moduleUnregisterCommands(ctx.module);
+            moduleUnregisterSharedAPI(ctx.module);
+            moduleUnregisterUsedAPI(ctx.module);
            moduleFreeModuleStructure(ctx.module);
        }
        dlclose(handle);
@@ -4821,14 +4984,17 @@ int moduleUnload(sds name) {
    if (module == NULL) {
        errno = ENOENT;
        return REDISMODULE_ERR;
-    }
-
-    if (listLength(module->types)) {
+    } else if (listLength(module->types)) {
        errno = EBUSY;
        return REDISMODULE_ERR;
+    } else if (listLength(module->usedby)) {
+        errno = EPERM;
+        return REDISMODULE_ERR;
    }

    moduleUnregisterCommands(module);
+    moduleUnregisterSharedAPI(module);
+    moduleUnregisterUsedAPI(module);

    /* Remove any notification subscribers this module might have */
    moduleUnsubscribeNotifications(module);
@@ -4909,7 +5075,12 @@ NULL
                errmsg = "no such module with that name";
                break;
            case EBUSY:
-                errmsg = "the module exports one or more module-side data types, can't unload";
+                errmsg = "the module exports one or more module-side data "
+                         "types, can't unload";
+                break;
+            case EPERM:
+                errmsg = "the module exports APIs used by other modules. "
+                         "Please unload them first and try again";
                break;
            default:
                errmsg = "operation not possible.";
@@ -4934,6 +5105,7 @@ size_t moduleCount(void) {
 * file so that's easy to seek it to add new entries. */
 void moduleRegisterCoreAPI(void) {
    server.moduleapi = dictCreate(&moduleAPIDictType,NULL);
+    server.sharedapi = dictCreate(&moduleAPIDictType,NULL);
    REGISTER_API(Alloc);
    REGISTER_API(Calloc);
    REGISTER_API(Realloc);
@@ -5031,6 +5203,7 @@ void moduleRegisterCoreAPI(void) {
    REGISTER_API(RetainString);
    REGISTER_API(StringCompare);
    REGISTER_API(GetContextFromIO);
+    REGISTER_API(GetKeyNameFromIO);
    REGISTER_API(BlockClient);
    REGISTER_API(UnblockClient);
    REGISTER_API(IsBlockedReplyRequest);
@@ -5084,4 +5257,6 @@ void moduleRegisterCoreAPI(void) {
    REGISTER_API(DictPrev);
    REGISTER_API(DictCompareC);
    REGISTER_API(DictCompare);
+    REGISTER_API(ExportSharedAPI);
+    REGISTER_API(GetSharedAPI);
 }
@@ -138,6 +138,7 @@ void linkClient(client *c) {
     * this way removing the client in unlinkClient() will not require
     * a linear scan, but just a constant time operation. */
    c->client_list_node = listLast(server.clients);
+    if (c->fd != -1) atomicIncr(server.rgthreadvar[c->iel].cclients, 1);
    uint64_t id = htonu64(c->id);
    raxInsert(server.clients_index,(unsigned char*)&id,sizeof(id),c,NULL);
 }
@@ -225,6 +226,7 @@ client *createClient(int fd, int iel) {
    c->bufAsync = NULL;
    c->buflenAsync = 0;
    c->bufposAsync = 0;
+    memset(c->uuid, 0, UUID_BINARY_LEN);

    listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid);
    listSetMatchMethod(c->pubsub_patterns,listMatchObjects);
@@ -1035,6 +1037,7 @@ static void acceptCommonHandler(int fd, int flags, char *ip, int iel) {
        return;
    }

+#ifdef HAVE_SO_INCOMING_CPU
    // Set thread affinity
    if (server.fThreadAffinity)
    {
@@ -1044,6 +1047,7 @@ static void acceptCommonHandler(int fd, int flags, char *ip, int iel) {
            serverLog(LL_WARNING, "Failed to set socket affinity");
        }
    }
+#endif

    /* If maxclient directive is set and this is one client more... close the
     * connection. Note that we create the client instead to check before
@@ -1206,6 +1210,8 @@ void unlinkClient(client *c) {
        aeDeleteFileEvent(server.rgthreadvar[c->iel].el,c->fd,AE_WRITABLE);
        close(c->fd);
        c->fd = -1;
+
+        atomicDecr(server.rgthreadvar[c->iel].cclients, 1);
    }

    /* Remove from the list of pending writes if needed. */
@@ -2130,7 +2136,7 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     * corresponding part of the replication stream, will be propagated to
     * the sub-slaves and to the replication backlog. */
    processInputBufferAndReplicate(c);
-    aelock.arm(nullptr);
+    aelock.arm(c);
    ProcessPendingAsyncWrites();
 }

@@ -417,6 +417,18 @@ int isObjectRepresentableAsLongLong(robj *o, long long *llval) {
    }
 }

+/* Optimize the SDS string inside the string object to require little space,
+ * in case there is more than 10% of free space at the end of the SDS
+ * string. This happens because SDS strings tend to overallocate to avoid
+ * wasting too much time in allocations when appending to the string. */
+void trimStringObjectIfNeeded(robj *o) {
+    if (o->encoding == OBJ_ENCODING_RAW &&
+        sdsavail(ptrFromObj(o)) > sdslen(ptrFromObj(o))/10)
+    {
+        o->m_ptr = sdsRemoveFreeSpace(ptrFromObj(o));
+    }
+}
+
 /* Try to encode a string object in order to save space */
 robj *tryObjectEncoding(robj *o) {
    long value;
@@ -486,11 +498,7 @@ robj *tryObjectEncoding(robj *o) {
     * We do that only for relatively large strings as this branch
     * is only entered if the length of the string is greater than
     * OBJ_ENCODING_EMBSTR_SIZE_LIMIT. */
-    if (o->encoding == OBJ_ENCODING_RAW &&
-        sdsavail(s) > len/10)
-    {
-        o->m_ptr = sdsRemoveFreeSpace(ptrFromObj(o));
-    }
+    trimStringObjectIfNeeded(o);

    /* Return the original object. */
    return o;
@@ -1197,7 +1205,7 @@ sds getMemoryDoctorReport(void) {

 /* Set the object LRU/LFU depending on server.maxmemory_policy.
 * The lfu_freq arg is only relevant if policy is MAXMEMORY_FLAG_LFU.
- * The lru_idle and lru_clock args are only relevant if policy 
+ * The lru_idle and lru_clock args are only relevant if policy
 * is MAXMEMORY_FLAG_LRU.
 * Either or both of them may be <0, in that case, nothing is set. */
 void objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle,
@@ -1208,16 +1216,20 @@ void objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle,
            val->lru = (LFUGetTimeInMinutes()<<8) | lfu_freq;
        }
    } else if (lru_idle >= 0) {
-        /* Serialized LRU idle time is in seconds. Scale
+        /* Provided LRU idle time is in seconds. Scale
         * according to the LRU clock resolution this Redis
         * instance was compiled with (normally 1000 ms, so the
         * below statement will expand to lru_idle*1000/1000. */
        lru_idle = lru_idle*1000/LRU_CLOCK_RESOLUTION;
-        val->lru = lru_clock - lru_idle;
-        /* If the lru field overflows (since LRU it is a wrapping
-         * clock), the best we can do is to provide the maximum
-         * representable idle time. */
-        if (val->lru < 0) val->lru = lru_clock+1;
+        long lru_abs = lru_clock - lru_idle; /* Absolute access time. */
+        /* If the LRU field underflows (since LRU it is a wrapping
+         * clock), the best we can do is to provide a large enough LRU
+         * that is half-way in the circlular LRU clock we use: this way
+         * the computed idle time for this object will stay high for quite
+         * some time. */
+        if (lru_abs < 0)
+            lru_abs = (lru_clock+(LRU_CLOCK_MAX/2)) % LRU_CLOCK_MAX;
+        val->lru = lru_abs;
    }
 }

@@ -48,4 +48,64 @@ extern "C" int rdbSaveS3(char *s3bucket, rdbSaveInfo *rsi)
        serverLog(LL_NOTICE,"DB saved on AWS S3");
        
    return (status == EXIT_SUCCESS) ? C_OK : C_ERR;
-}
+}
+
+
+int rdbLoadS3Core(int fd, rdbSaveInfo *rsi) 
+{
+    FILE *fp;
+    rio rdb;
+    int retval;
+
+    if ((fp = fdopen(fd, "rb")) == NULL) return C_ERR;
+    startLoading(fp);
+    rioInitWithFile(&rdb,fileno(fp));
+    retval = rdbLoadRio(&rdb,rsi,0);
+    fclose(fp);
+    stopLoading();
+    return retval;
+}
+
+int rdbLoadS3(char *s3bucket, rdbSaveInfo *rsi)
+{
+    int status = EXIT_FAILURE;
+    int fd[2];
+    if (pipe(fd) != 0)
+        return C_ERR;
+
+    pid_t pid = fork();
+    if (pid < 0)
+    {
+        close(fd[0]);
+        close(fd[1]);
+        return C_ERR;
+    }
+
+    if (pid == 0)
+    {
+        // child process
+        dup2(fd[1], STDOUT_FILENO);
+        close(fd[1]);
+        close(fd[0]);
+        execlp("aws", "aws", "s3", "cp", s3bucket, "-", nullptr);
+        exit(EXIT_FAILURE);
+    }
+    else
+    {
+        close(fd[1]);
+        if (rdbLoadS3Core(fd[0], rsi) != C_OK)
+        {
+            close(fd[0]);
+            return C_ERR;
+        }
+        close(fd[0]);
+        waitpid(pid, &status, 0);
+    }
+    
+    if (status != EXIT_SUCCESS)
+        serverLog(LL_WARNING, "Failed to load DB from AWS S3");
+    else
+        serverLog(LL_NOTICE,"DB loaded from AWS S3");
+        
+    return (status == EXIT_SUCCESS) ? C_OK : C_ERR;
+}
@@ -752,7 +752,7 @@ size_t rdbSaveStreamConsumers(rio *rdb, streamCG *cg) {

 /* Save a Redis object.
 * Returns -1 on error, number of bytes written on success. */
-ssize_t rdbSaveObject(rio *rdb, robj *o) {
+ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key) {
    ssize_t n = 0, nwritten = 0;

    if (o->type == OBJ_STRING) {
@@ -967,7 +967,7 @@ ssize_t rdbSaveObject(rio *rdb, robj *o) {
        RedisModuleIO io;
        moduleValue *mv = ptrFromObj(o);
        moduleType *mt = mv->type;
-        moduleInitIOContext(io,mt,rdb);
+        moduleInitIOContext(io,mt,rdb,key);

        /* Write the "module" identifier as prefix, so that we'll be able
         * to call the right module during loading. */
@@ -997,7 +997,7 @@ ssize_t rdbSaveObject(rio *rdb, robj *o) {
 * this length with very little changes to the code. In the future
 * we could switch to a faster solution. */
 size_t rdbSavedObjectLen(robj *o) {
-    ssize_t len = rdbSaveObject(NULL,o);
+    ssize_t len = rdbSaveObject(NULL,o,NULL);
    serverAssertWithInfo(NULL,o,len != -1);
    return len;
 }
@@ -1039,7 +1039,7 @@ int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime) {
    /* Save type, key, value */
    if (rdbSaveObjectType(rdb,val) == -1) return -1;
    if (rdbSaveStringObject(rdb,key) == -1) return -1;
-    if (rdbSaveObject(rdb,val) == -1) return -1;
+    if (rdbSaveObject(rdb,val,key) == -1) return -1;
    return 1;
 }

@@ -1401,7 +1401,7 @@ robj *rdbLoadCheckModuleValue(rio *rdb, char *modulename) {

 /* Load a Redis object of the specified type from the specified file.
 * On success a newly allocated object is returned, otherwise NULL. */
-robj *rdbLoadObject(int rdbtype, rio *rdb) {
+robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
    robj *o = NULL, *ele, *dec;
    uint64_t len;
    unsigned int i;
@@ -1788,7 +1788,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
            exit(1);
        }
        RedisModuleIO io;
-        moduleInitIOContext(io,mt,rdb);
+        moduleInitIOContext(io,mt,rdb,key);
        io.ver = (rdbtype == RDB_TYPE_MODULE) ? 1 : 2;
        /* Call the rdb_load method of the module providing the 10 bit
         * encoding version in the lower 10 bits of the module ID. */
@@ -2044,7 +2044,7 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
        /* Read key */
        if ((key = rdbLoadStringObject(rdb)) == NULL) goto eoferr;
        /* Read value */
-        if ((val = rdbLoadObject(type,rdb)) == NULL) goto eoferr;
+        if ((val = rdbLoadObject(type,rdb,key)) == NULL) goto eoferr;
        /* Check if the key already expired. This function is used when loading
         * an RDB file from disk, either at startup, or when an RDB was
         * received from the master. In the latter case, the master is
@@ -2097,6 +2097,19 @@ eoferr: /* unexpected end of file is handled here with a fatal exit */
    return C_ERR; /* Just to avoid warning */
 }

+int rdbLoadFile(char *filename, rdbSaveInfo *rsi);
+int rdbLoad(rdbSaveInfo *rsi)
+{
+    int err = C_ERR;
+    if (server.rdb_filename != NULL)
+        err = rdbLoadFile(server.rdb_filename, rsi);
+
+    if ((err == C_ERR) && server.rdb_s3bucketpath != NULL)
+        err = rdbLoadS3(server.rdb_s3bucketpath, rsi);
+
+    return err;
+}
+
 /* Like rdbLoadRio() but takes a filename instead of a rio stream. The
 * filename is open for reading and a rio stream object created in order
 * to do the actual loading. Moreover the ETA displayed in the INFO
@@ -2104,7 +2117,7 @@ eoferr: /* unexpected end of file is handled here with a fatal exit */
 *
 * If you pass an 'rsi' structure initialied with RDB_SAVE_OPTION_INIT, the
 * loading code will fiil the information fields in the structure. */
-int rdbLoad(char *filename, rdbSaveInfo *rsi) {
+int rdbLoadFile(char *filename, rdbSaveInfo *rsi) {
    FILE *fp;
    rio rdb;
    int retval;
@@ -135,7 +135,7 @@ uint64_t rdbLoadLen(rio *rdb, int *isencoded);
 int rdbLoadLenByRef(rio *rdb, int *isencoded, uint64_t *lenptr);
 int rdbSaveObjectType(rio *rdb, robj *o);
 int rdbLoadObjectType(rio *rdb);
-int rdbLoad(char *filename, rdbSaveInfo *rsi);
+int rdbLoad(rdbSaveInfo *rsi);
 int rdbSaveBackground(rdbSaveInfo *rsi);
 int rdbSaveToSlavesSockets(rdbSaveInfo *rsi);
 void rdbRemoveTempFile(pid_t childpid);
@@ -143,9 +143,10 @@ int rdbSave(rdbSaveInfo *rsi);
 int rdbSaveFile(char *filename, rdbSaveInfo *rsi);
 int rdbSaveFd(int fd, rdbSaveInfo *rsi);
 int rdbSaveS3(char *path, rdbSaveInfo *rsi);
-ssize_t rdbSaveObject(rio *rdb, robj *o);
+int rdbLoadS3(char *path, rdbSaveInfo *rsi);
+ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key);
 size_t rdbSavedObjectLen(robj *o);
-robj *rdbLoadObject(int type, rio *rdb);
+robj *rdbLoadObject(int type, rio *rdb, robj *key);
 void backgroundSaveDoneHandler(int exitcode, int bysignal);
 int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime);
 robj *rdbLoadStringObject(rio *rdb);
@@ -248,11 +248,11 @@ static redisConfig *getRedisConfig(const char *ip, int port,
        c = redisConnect(ip, port);
    else
        c = redisConnectUnix(hostsocket);
-    if (c->err) {
+    if (c == NULL || c->err) {
        fprintf(stderr,"Could not connect to Redis at ");
-        if (hostsocket == NULL)
-            fprintf(stderr,"%s:%d: %s\n",ip,port,c->errstr);
-        else fprintf(stderr,"%s: %s\n",hostsocket,c->errstr);
+        char *err = (c != NULL ? c->errstr : "");
+        if (hostsocket == NULL) fprintf(stderr,"%s:%d: %s\n",ip,port,err);
+        else fprintf(stderr,"%s: %s\n",hostsocket,err);
        goto fail;
    }
    redisAppendCommand(c, "CONFIG GET %s", "save");
@@ -277,18 +277,16 @@ static redisConfig *getRedisConfig(const char *ip, int port,
        case 1: cfg->appendonly = sdsnew(value); break;
        }
    }
-    if (reply) freeReplyObject(reply);
-    if (c) redisFree(c);
+    freeReplyObject(reply);
+    redisFree(c);
    return cfg;
 fail:
-    if (reply) freeReplyObject(reply);
-    if (c) redisFree(c);
-    zfree(cfg);
    fprintf(stderr, "ERROR: failed to fetch CONFIG from ");
-    if (c->connection_type == REDIS_CONN_TCP)
-        fprintf(stderr, "%s:%d\n", c->tcp.host, c->tcp.port);
-    else if (c->connection_type == REDIS_CONN_UNIX)
-        fprintf(stderr, "%s\n", c->unix_sock.path);
+    if (hostsocket == NULL) fprintf(stderr, "%s:%d\n", ip, port);
+    else fprintf(stderr, "%s\n", hostsocket);
+    freeReplyObject(reply);
+    redisFree(c);
+    zfree(cfg);
    return NULL;
 }
 static void freeRedisConfig(redisConfig *cfg) {
@@ -346,7 +344,9 @@ static void randomizeClientKey(client c) {

    for (i = 0; i < c->randlen; i++) {
        char *p = c->randptr[i]+11;
-        size_t r = random() % config.randomkeys_keyspacelen;
+        size_t r = 0;
+        if (config.randomkeys_keyspacelen != 0)
+            r = random() % config.randomkeys_keyspacelen;
        size_t j;

        for (j = 0; j < 12; j++) {
@@ -448,27 +448,27 @@ static void readHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
                    }
                }

-                if (config.cluster_mode && is_err && c->cluster_node &&
-                    (!strncmp(r->str,"MOVED",5) ||
-                     !strcmp(r->str,"ASK")))
-                {
-                    /* Try to update slots configuration if the key of the
-                     * command is using the slot hash tag. */
-                    if (c->staglen && !fetchClusterSlotsConfiguration(c))
-                        exit(1);
-                    /*
-                    clusterNode *node = c->cluster_node;
-                    assert(node);
-                    if (++node->current_slot_index >= node->slots_count) {
-                        if (config.showerrors) {
-                            fprintf(stderr, "WARN: No more available slots in "
-                                    "node %s:%d\n", node->ip, node->port);
-                        }
-                        freeReplyObject(reply);
-                        freeClient(c);
-                        return;
+                /* Try to update slots configuration if reply error is
+                 * MOVED/ASK/CLUSTERDOWN and the key(s) used by the command
+                 * contain(s) the slot hash tag. */
+                if (is_err && c->cluster_node && c->staglen) {
+                    int fetch_slots = 0, do_wait = 0;
+                    if (!strncmp(r->str,"MOVED",5) || !strncmp(r->str,"ASK",3))
+                        fetch_slots = 1;
+                    else if (!strncmp(r->str,"CLUSTERDOWN",11)) {
+                        /* Usually the cluster is able to recover itself after
+                         * a CLUSTERDOWN error, so try to sleep one second
+                         * before requesting the new configuration. */
+                        fetch_slots = 1;
+                        do_wait = 1;
+                        printf("Error from server %s:%d: %s\n",
+                               c->cluster_node->ip,
+                               c->cluster_node->port,
+                               r->str);
                    }
-                    */
+                    if (do_wait) sleep(1);
+                    if (fetch_slots && !fetchClusterSlotsConfiguration(c))
+                        exit(1);
                }

                freeReplyObject(reply);
@@ -818,22 +818,37 @@ static void showLatencyReport(void) {
    }
 }

-static void benchmark(char *title, char *cmd, int len) {
+static void initBenchmarkThreads() {
    int i;
+    if (config.threads) freeBenchmarkThreads();
+    config.threads = zmalloc(config.num_threads * sizeof(benchmarkThread*), MALLOC_LOCAL);
+    for (i = 0; i < config.num_threads; i++) {
+        benchmarkThread *thread = createBenchmarkThread(i);
+        config.threads[i] = thread;
+    }
+}
+
+static void startBenchmarkThreads() {
+    int i;
+    for (i = 0; i < config.num_threads; i++) {
+        benchmarkThread *t = config.threads[i];
+        if (pthread_create(&(t->thread), NULL, execBenchmarkThread, t)){
+            fprintf(stderr, "FATAL: Failed to start thread %d.\n", i);
+            exit(1);
+        }
+    }
+    for (i = 0; i < config.num_threads; i++)
+        pthread_join(config.threads[i]->thread, NULL);
+}
+
+static void benchmark(char *title, char *cmd, int len) {
    client c;

    config.title = title;
    config.requests_issued = 0;
    config.requests_finished = 0;

-    if (config.num_threads) {
-        if (config.threads) freeBenchmarkThreads();
-        config.threads = zmalloc(config.num_threads * sizeof(benchmarkThread*), MALLOC_LOCAL);
-        for (i = 0; i < config.num_threads; i++) {
-            benchmarkThread *thread = createBenchmarkThread(i);
-            config.threads[i] = thread;
-        }
-    }
+    if (config.num_threads) initBenchmarkThreads();

    int thread_id = config.num_threads > 0 ? 0 : -1;
    c = createClient(cmd,len,NULL,thread_id);
@@ -841,17 +856,7 @@ static void benchmark(char *title, char *cmd, int len) {

    config.start = mstime();
    if (!config.num_threads) aeMain(config.el);
-    else {
-        for (i = 0; i < config.num_threads; i++) {
-            benchmarkThread *t = config.threads[i];
-            if (pthread_create(&(t->thread), NULL, execBenchmarkThread, t)){
-                fprintf(stderr, "FATAL: Failed to start thread %d.\n", i);
-                exit(1);
-            }
-        }
-        for (i = 0; i < config.num_threads; i++)
-            pthread_join(config.threads[i]->thread, NULL);
-    }
+    else startBenchmarkThreads();
    config.totlatency = mstime()-config.start;

    showLatencyReport();
@@ -1284,6 +1289,11 @@ int parseOptions(int argc, const char **argv) {
            if (config.pipeline <= 0) config.pipeline=1;
        } else if (!strcmp(argv[i],"-r")) {
            if (lastarg) goto invalid;
+            const char *next = argv[++i], *p = next;
+            if (*p == '-') {
+                p++;
+                if (*p < '0' || *p > '9') goto invalid;
+            }
            config.randomkeys = 1;
            config.randomkeys_keyspacelen = atoi(argv[++i]);
            if (config.randomkeys_keyspacelen < 0)
@@ -1555,9 +1565,15 @@ int main(int argc, const char **argv) {

    if (config.idlemode) {
        printf("Creating %d idle connections and waiting forever (Ctrl+C when done)\n", config.numclients);
-        c = createClient("",0,NULL,-1); /* will never receive a reply */
+        int thread_id = -1, use_threads = (config.num_threads > 0);
+        if (use_threads) {
+            thread_id = 0;
+            initBenchmarkThreads();
+        }
+        c = createClient("",0,NULL,thread_id); /* will never receive a reply */
        createMissingClients(c);
-        aeMain(config.el);
+        if (use_threads) startBenchmarkThreads();
+        else aeMain(config.el);
        /* and will wait for every */
    }

@@ -33,8 +33,8 @@

 #define ERROR(...) { \
    char __buf[1024]; \
-    sprintf(__buf, __VA_ARGS__); \
-    sprintf(error, "0x%16llx: %s", (long long)epos, __buf); \
+    snprintf(__buf, sizeof(__buf), __VA_ARGS__); \
+    snprintf(error, sizeof(error), "0x%16llx: %s", (long long)epos, __buf); \
 }

 static char error[1024];
@@ -285,7 +285,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
        rdbstate.keys++;
        /* Read value */
        rdbstate.doing = RDB_CHECK_DOING_READ_OBJECT_VALUE;
-        if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr;
+        if ((val = rdbLoadObject(type,&rdb,key)) == NULL) goto eoferr;
        /* Check if the key already expired. */
        if (expiretime != -1 && expiretime < now)
            rdbstate.already_expired++;
@@ -2269,7 +2269,7 @@ static clusterManagerNode *clusterManagerNewNode(char *ip, int port) {
 static sds clusterManagerGetNodeRDBFilename(clusterManagerNode *node) {
    assert(config.cluster_manager_command.backup_dir);
    sds filename = sdsnew(config.cluster_manager_command.backup_dir);
-    if (filename[sdslen(filename)] - 1 != '/')
+    if (filename[sdslen(filename) - 1] != '/')
        filename = sdscat(filename, "/");
    filename = sdscatprintf(filename, "redis-node-%s-%d-%s.rdb", node->ip,
                            node->port, node->name);
@@ -2733,6 +2733,36 @@ static sds clusterManagerNodeGetJSON(clusterManagerNode *node,
        json = sdscatprintf(json, ",\n    \"cluster_errors\": %lu",
                            error_count);
    }
+    if (node->migrating_count > 0 && node->migrating != NULL) {
+        int i = 0;
+        sds migrating = sdsempty();
+        for (; i < node->migrating_count; i += 2) {
+            sds slot = node->migrating[i];
+            sds dest = node->migrating[i + 1];
+            if (slot && dest) {
+                if (sdslen(migrating) > 0) migrating = sdscat(migrating, ",");
+                migrating = sdscatfmt(migrating, "\"%S\": \"%S\"", slot, dest);
+            }
+        }
+        if (sdslen(migrating) > 0)
+            json = sdscatfmt(json, ",\n    \"migrating\": {%S}", migrating);
+        sdsfree(migrating);
+    }
+    if (node->importing_count > 0 && node->importing != NULL) {
+        int i = 0;
+        sds importing = sdsempty();
+        for (; i < node->importing_count; i += 2) {
+            sds slot = node->importing[i];
+            sds from = node->importing[i + 1];
+            if (slot && from) {
+                if (sdslen(importing) > 0) importing = sdscat(importing, ",");
+                importing = sdscatfmt(importing, "\"%S\": \"%S\"", slot, from);
+            }
+        }
+        if (sdslen(importing) > 0)
+            json = sdscatfmt(json, ",\n    \"importing\": {%S}", importing);
+        sdsfree(importing);
+    }
    json = sdscat(json, "\n  }");
    sdsfree(replicate);
    sdsfree(slots);
@@ -2842,7 +2872,7 @@ static void clusterManagerShowClusterInfo(void) {
                    replicas++;
            }
            redisReply *reply = CLUSTER_MANAGER_COMMAND(node, "DBSIZE");
-            if (reply != NULL || reply->type == REDIS_REPLY_INTEGER)
+            if (reply != NULL && reply->type == REDIS_REPLY_INTEGER)
                dbsize = reply->integer;
            if (dbsize < 0) {
                char *err = "";
@@ -85,6 +85,9 @@
 #define REDISMODULE_CTX_FLAGS_OOM (1<<10)
 /* Less than 25% of memory available according to maxmemory. */
 #define REDISMODULE_CTX_FLAGS_OOM_WARNING (1<<11)
+/* The command was sent over the replication link. */
+#define REDISMODULE_CTX_FLAGS_REPLICATED (1<<12)
+

 #define REDISMODULE_NOTIFY_GENERIC (1<<2)     /* g */
 #define REDISMODULE_NOTIFY_STRING (1<<3)      /* $ */
@@ -275,6 +278,7 @@ int REDISMODULE_API_FUNC(RedisModule_StringAppendBuffer)(RedisModuleCtx *ctx, Re
 void REDISMODULE_API_FUNC(RedisModule_RetainString)(RedisModuleCtx *ctx, RedisModuleString *str);
 int REDISMODULE_API_FUNC(RedisModule_StringCompare)(RedisModuleString *a, RedisModuleString *b);
 RedisModuleCtx *REDISMODULE_API_FUNC(RedisModule_GetContextFromIO)(RedisModuleIO *io);
+const RedisModuleString *REDISMODULE_API_FUNC(RedisModule_GetKeyNameFromIO)(RedisModuleIO *io);
 long long REDISMODULE_API_FUNC(RedisModule_Milliseconds)(void);
 void REDISMODULE_API_FUNC(RedisModule_DigestAddStringBuffer)(RedisModuleDigest *md, unsigned char *ele, size_t len);
 void REDISMODULE_API_FUNC(RedisModule_DigestAddLongLong)(RedisModuleDigest *md, long long ele);
@@ -332,6 +336,8 @@ void REDISMODULE_API_FUNC(RedisModule_GetRandomBytes)(unsigned char *dst, size_t
 void REDISMODULE_API_FUNC(RedisModule_GetRandomHexChars)(char *dst, size_t len);
 void REDISMODULE_API_FUNC(RedisModule_SetDisconnectCallback)(RedisModuleBlockedClient *bc, RedisModuleDisconnectFunc callback);
 void REDISMODULE_API_FUNC(RedisModule_SetClusterFlags)(RedisModuleCtx *ctx, uint64_t flags);
+int REDISMODULE_API_FUNC(RedisModule_ExportSharedAPI)(RedisModuleCtx *ctx, const char *apiname, void *func);
+void *REDISMODULE_API_FUNC(RedisModule_GetSharedAPI)(RedisModuleCtx *ctx, const char *apiname);
 #endif

 /* This is included inline inside each Redis module. */
@@ -437,6 +443,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
    REDISMODULE_GET_API(RetainString);
    REDISMODULE_GET_API(StringCompare);
    REDISMODULE_GET_API(GetContextFromIO);
+    REDISMODULE_GET_API(GetKeyNameFromIO);
    REDISMODULE_GET_API(Milliseconds);
    REDISMODULE_GET_API(DigestAddStringBuffer);
    REDISMODULE_GET_API(DigestAddLongLong);
@@ -492,6 +499,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
    REDISMODULE_GET_API(GetRandomBytes);
    REDISMODULE_GET_API(GetRandomHexChars);
    REDISMODULE_GET_API(SetClusterFlags);
+    REDISMODULE_GET_API(ExportSharedAPI);
+    REDISMODULE_GET_API(GetSharedAPI);
 #endif

    if (RedisModule_IsModuleNameBusy && RedisModule_IsModuleNameBusy(name)) return REDISMODULE_ERR;
@@ -38,6 +38,7 @@
 #include <sys/socket.h>
 #include <sys/stat.h>
 #include <mutex>
+#include <uuid/uuid.h>

 void replicationDiscardCachedMaster(void);
 void replicationResurrectCachedMaster(int newfd);
@@ -74,6 +75,21 @@ char *replicationGetSlaveName(client *c) {
    return buf;
 }

+static bool FSameHost(client *clientA, client *clientB)
+{
+    const unsigned char *a = clientA->uuid;
+    const unsigned char *b = clientB->uuid;
+
+    unsigned char zeroCheck = 0;
+    for (int i = 0; i < UUID_BINARY_LEN; ++i)
+    {
+        if (a[i] != b[i])
+            return false;
+        zeroCheck |= a[i];
+    }
+    return (zeroCheck != 0);    // if the UUID is nil then it is never equal
+}
+
 /* ---------------------------------- MASTER -------------------------------- */

 void createReplicationBacklog(void) {
@@ -117,7 +133,13 @@ void resizeReplicationBacklog(long long newsize) {

 void freeReplicationBacklog(void) {
    serverAssert(GlobalLocksAcquired());
-    serverAssert(listLength(server.slaves) == 0);
+    listIter li;
+    listNode *ln;
+    listRewind(server.slaves, &li);
+    while ((ln = listNext(&li))) {
+        // server.slaves should be empty, or filled with clients pending close
+        serverAssert(((client*)listNodeValue(ln))->flags & CLIENT_CLOSE_ASAP);
+    }
    zfree(server.repl_backlog);
    server.repl_backlog = NULL;
 }
@@ -186,7 +208,7 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
     * propagate *identical* replication stream. In this way this slave can
     * advertise the same replication ID as the master (since it shares the
     * master replication history and has the same backlog and offsets). */
-    if (server.masterhost != NULL) return;
+    if (!server.fActiveReplica && server.masterhost != NULL) return;

    /* If there aren't slaves, and there is no backlog buffer to populate,
     * we can return ASAP. */
@@ -226,6 +248,7 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
        while((ln = listNext(&li))) {
            client *slave = (client*)ln->value;
            if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue;
+            if (server.current_client && FSameHost(server.current_client, slave)) continue;
            addReplyAsync(slave,selectcmd);
        }

@@ -268,6 +291,7 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {

        /* Don't feed slaves that are still waiting for BGSAVE to start */
        if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue;
+        if (server.current_client && FSameHost(server.current_client, slave)) continue;

        /* Feed slaves that are waiting for the initial SYNC (so these commands
         * are queued in the output buffer until the initial SYNC completes),
@@ -282,10 +306,10 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
            addReplyBulkAsync(slave,argv[j]);
    }

-    /* Release the lock on all slaves */
    listRewind(slaves,&li);
    while((ln = listNext(&li))) {
-        ((client*)ln->value)->lock.unlock();
+        client *slave = (client*)ln->value;
+        slave->lock.unlock();
    }
 }

@@ -311,6 +335,8 @@ void replicationFeedSlavesFromMasterStream(list *slaves, char *buf, size_t bufle
    while((ln = listNext(&li))) {
        client *slave = (client*)ln->value;
        std::lock_guard<decltype(slave->lock)> ulock(slave->lock);
+        if (FSameHost(slave, server.master))
+            continue;   // Active Active case, don't feed back

        /* Don't feed slaves that are still waiting for BGSAVE to start */
        if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue;
@@ -658,9 +684,11 @@ void syncCommand(client *c) {

    /* Refuse SYNC requests if we are a slave but the link with our master
     * is not ok... */
-    if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED) {
-        addReplySds(c,sdsnew("-NOMASTERLINK Can't SYNC while not connected with my master\r\n"));
-        return;
+    if (!server.fActiveReplica) {
+        if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED) {
+            addReplySds(c,sdsnew("-NOMASTERLINK Can't SYNC while not connected with my master\r\n"));
+            return;
+        }
    }

    /* SYNC can't be issued when the server has pending data to send to
@@ -790,6 +818,33 @@ void syncCommand(client *c) {
    return;
 }

+void processReplconfUuid(client *c, robj *arg)
+{
+    try
+    {
+        if (arg->type != OBJ_STRING)
+            throw "Invalid UUID";
+
+        const char *remoteUUID = (const char*)ptrFromObj(arg);
+        if (strlen(remoteUUID) != 36)
+            throw "Invalid UUID";
+
+        if (uuid_parse(remoteUUID, c->uuid) != 0)
+            throw "Invalid UUID";
+
+        char szServerUUID[36 + 2]; // 1 for the '+', another for '\0'
+        szServerUUID[0] = '+';
+        uuid_unparse(server.uuid, szServerUUID+1);
+        addReplyProto(c, szServerUUID, 37);
+        addReplyProto(c, "\r\n", 2);
+    } 
+    catch (const char *szErr)
+    {
+        addReplyError(c, szErr);
+        return;
+    }
+}
+
 /* REPLCONF <option> <value> <option> <value> ...
 * This command is used by a slave in order to configure the replication
 * process before starting it with the SYNC command.
@@ -860,6 +915,10 @@ void replconfCommand(client *c) {
             * to the slave. */
            if (server.masterhost && server.master) replicationSendAck();
            return;
+        } else if (!strcasecmp((const char*)ptrFromObj(c->argv[j]),"uuid")) {
+            /* REPLCONF uuid is used to set and send the UUID of each host */
+            processReplconfUuid(c, c->argv[j+1]);
+            return; // the process function replies to the client for both error and success
        } else {
            addReplyErrorFormat(c,"Unrecognized REPLCONF option: %s",
                (char*)ptrFromObj(c->argv[j]));
@@ -1134,6 +1193,10 @@ void replicationCreateMasterClient(int fd, int dbid) {
    server.master->reploff = server.master_initial_offset;
    server.master->read_reploff = server.master->reploff;
    server.master->puser = NULL; /* This client can do everything. */
+    
+    memcpy(server.master->uuid, server.master_uuid, UUID_BINARY_LEN);
+    memset(server.master_uuid, 0, UUID_BINARY_LEN); // make sure people don't use this temp storage buffer
+
    memcpy(server.master->replid, server.master_replid,
        sizeof(server.master_replid));
    /* If master offset is set to -1, this master is old and is not
@@ -1143,14 +1206,23 @@ void replicationCreateMasterClient(int fd, int dbid) {
    if (dbid != -1) selectDb(server.master,dbid);
 }

-void restartAOF() {
-    int retry = 10;
-    while (retry-- && startAppendOnly() == C_ERR) {
-        serverLog(LL_WARNING,"Failed enabling the AOF after successful master synchronization! Trying it again in one second.");
+/* This function will try to re-enable the AOF file after the
+ * master-replica synchronization: if it fails after multiple attempts
+ * the replica cannot be considered reliable and exists with an
+ * error. */
+void restartAOFAfterSYNC() {
+    unsigned int tries, max_tries = 10;
+    for (tries = 0; tries < max_tries; ++tries) {
+        if (startAppendOnly() == C_OK) break;
+        serverLog(LL_WARNING,
+            "Failed enabling the AOF after successful master synchronization! "
+            "Trying it again in one second.");
        sleep(1);
    }
-    if (!retry) {
-        serverLog(LL_WARNING,"FATAL: this replica instance finished the synchronization with its master, but the AOF can't be turned on. Exiting now.");
+    if (tries == max_tries) {
+        serverLog(LL_WARNING,
+            "FATAL: this replica instance finished the synchronization with "
+            "its master, but the AOF can't be turned on. Exiting now.");
        exit(1);
    }
 }
@@ -1334,12 +1406,12 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
        aeDeleteFileEvent(el,server.repl_transfer_s,AE_READABLE);
        serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Loading DB in memory");
        rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
-        if (rdbLoad(server.rdb_filename,&rsi) != C_OK) {
+        if (rdbLoad(&rsi) != C_OK) {
            serverLog(LL_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
            cancelReplicationHandshake();
            /* Re-enable the AOF if we disabled it earlier, in order to restore
             * the original configuration. */
-            if (aof_is_enabled) restartAOF();
+            if (aof_is_enabled) restartAOFAfterSYNC();
            return;
        }
        /* Final setup of the connected slave <- master link */
@@ -1364,7 +1436,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
        /* Restart the AOF subsystem now that we finished the sync. This
         * will trigger an AOF rewrite, and when done will start appending
         * to the new file. */
-        if (aof_is_enabled) restartAOF();
+        if (aof_is_enabled) restartAOFAfterSYNC();
    }
    return;

@@ -1729,7 +1801,7 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
            server.repl_state = REPL_STATE_RECEIVE_AUTH;
            return;
        } else {
-            server.repl_state = REPL_STATE_SEND_PORT;
+            server.repl_state = REPL_STATE_SEND_UUID;
        }
    }

@@ -1742,7 +1814,38 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
            goto error;
        }
        sdsfree(err);
+        server.repl_state = REPL_STATE_SEND_UUID;
+    }
+
+    /* Send UUID */
+    if (server.repl_state == REPL_STATE_SEND_UUID) {
+        char szUUID[37] = {0};
+        memset(server.master_uuid, 0, UUID_BINARY_LEN);
+        uuid_unparse((unsigned char*)server.uuid, szUUID);
+        err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF","uuid",szUUID,NULL);
+        if (err) goto write_error;
+        server.repl_state = REPL_STATE_RECEIVE_UUID;
+        return;
+    }
+
+    /* Receive UUID */
+    if (server.repl_state == REPL_STATE_RECEIVE_UUID) {
+        err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
+        if (err[0] == '-') {
+            serverLog(LL_WARNING, "non-fatal: Master doesn't understand REPLCONF uuid");
+        }
+        else {
+            if (strlen(err) != 37   // 36-byte UUID string and the leading '+'
+                || uuid_parse(err+1, server.master_uuid) != 0)   
+            {
+                serverLog(LL_WARNING, "Master replied with a UUID we don't understand");
+                sdsfree(err);
+                goto error;
+            }
+        }
+        sdsfree(err);
        server.repl_state = REPL_STATE_SEND_PORT;
+        // fallthrough
    }

    /* Set the slave port, so that Master's INFO command can list the
@@ -2102,14 +2205,26 @@ void replicaofCommand(client *c) {
    } else {
        long port;

+        if (c->flags & CLIENT_SLAVE)
+        {
+            /* If a client is already a replica they cannot run this command,
+             * because it involves flushing all replicas (including this
+             * client) */
+            addReplyError(c, "Command is not valid when client is a replica.");
+            return;
+        }
+
        if ((getLongFromObjectOrReply(c, c->argv[2], &port, NULL) != C_OK))
            return;

        /* Check if we are already attached to the specified slave */
        if (server.masterhost && !strcasecmp(server.masterhost,(const char*)ptrFromObj(c->argv[1]))
            && server.masterport == port) {
-            serverLog(LL_NOTICE,"REPLICAOF would result into synchronization with the master we are already connected with. No operation performed.");
-            addReplySdsAsync(c,sdsnew("+OK Already connected to specified master\r\n"));
+            serverLog(LL_NOTICE,"REPLICAOF would result into synchronization "
+                                "with the master we are already connected "
+                                "with. No operation performed.");
+            addReplySds(c,sdsnew("+OK Already connected to specified "
+                                 "master\r\n"));
            return;
        }
        /* There was no previous master or the user specified a different one,
@@ -259,8 +259,12 @@ sds sdsRemoveFreeSpace(sds s) {
    char type, oldtype = s[-1] & SDS_TYPE_MASK;
    int hdrlen, oldhdrlen = sdsHdrSize(oldtype);
    size_t len = sdslen(s);
+    size_t avail = sdsavail(s);
    sh = (char*)s-oldhdrlen;

+    /* Return ASAP if there is no space left. */
+    if (avail == 0) return s;
+
    /* Check what would be the minimum SDS header that is just good enough to
     * fit this string. */
    type = sdsReqType(len);
@@ -98,7 +98,7 @@ struct __attribute__ ((__packed__)) sdshdr64 {
 #define SDS_HDR(T,s) ((struct sdshdr##T *)((s)-(sizeof(struct sdshdr##T))))
 #define SDS_TYPE_5_LEN(f) ((f)>>SDS_TYPE_BITS)

-inline size_t sdslen(const sds s) {
+static inline size_t sdslen(const sds s) {
    unsigned char flags = s[-1];
    int type = flags & SDS_TYPE_MASK;

@@ -57,6 +57,8 @@
 #include <sys/utsname.h>
 #include <locale.h>
 #include <sys/socket.h>
+#include <algorithm>
+#include <uuid/uuid.h>

 /* Our shared "common" objects */

@@ -483,11 +485,11 @@ struct redisCommand redisCommandTable[] = {
     "write fast @sortedset",
     0,NULL,1,1,1,0,0,0},

-    {"bzpopmin",bzpopminCommand,-2,
+    {"bzpopmin",bzpopminCommand,-3,
     "write no-script fast @sortedset @blocking",
     0,NULL,1,-2,1,0,0,0},

-    {"bzpopmax",bzpopmaxCommand,-2,
+    {"bzpopmax",bzpopmaxCommand,-3,
     "write no-script fast @sortedset @blocking",
     0,NULL,1,-2,1,0,0,0},

@@ -718,7 +720,7 @@ struct redisCommand redisCommandTable[] = {

    {"touch",touchCommand,-2,
     "read-only fast @keyspace",
-     0,NULL,1,1,1,0,0,0},
+     0,NULL,1,-1,1,0,0,0},

    {"pttl",pttlCommand,2,
     "read-only fast random @keyspace",
@@ -866,7 +868,7 @@ struct redisCommand redisCommandTable[] = {
     "no-script @keyspace",
     0,NULL,0,0,0,0,0,0},

-    {"command",commandCommand,0,
+    {"command",commandCommand,-1,
     "ok-loading ok-stale random @connection",
     0,NULL,0,0,0,0,0,0},

@@ -2370,6 +2372,7 @@ void initServerConfig(void) {
    server.lazyfree_lazy_server_del = CONFIG_DEFAULT_LAZYFREE_LAZY_SERVER_DEL;
    server.always_show_logo = CONFIG_DEFAULT_ALWAYS_SHOW_LOGO;
    server.lua_time_limit = LUA_SCRIPT_TIME_LIMIT;
+    server.fActiveReplica = CONFIG_DEFAULT_ACTIVE_REPLICA;

    unsigned int lruclock = getLRUClock();
    atomicSet(server.lruclock,lruclock);
@@ -2814,6 +2817,7 @@ static void initServerThread(struct redisServerThreadVars *pvar, int fMain)
    pvar->unblocked_clients = listCreate();
    pvar->clients_pending_asyncwrite = listCreate();
    pvar->ipfd_count = 0;
+    pvar->cclients = 0;
    pvar->el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
    if (pvar->el == NULL) {
        serverLog(LL_WARNING,
@@ -2960,6 +2964,10 @@ void initServer(void) {
        server.maxmemory_policy = MAXMEMORY_NO_EVICTION;
    }

+    /* Generate UUID */
+    static_assert(sizeof(uuid_t) == sizeof(server.uuid), "UUIDs are standardized at 16-bytes");
+    uuid_generate((unsigned char*)server.uuid);
+
    if (server.cluster_enabled) clusterInit();
    replicationScriptCacheInit();
    scriptingInit(1);
@@ -3996,6 +4004,12 @@ extern "C" sds genRedisInfoString(const char *section) {
            listLength(server.clients)-listLength(server.slaves),
            maxin, maxout,
            server.blocked_clients);
+        for (int ithread = 0; ithread < server.cthreads; ++ithread)
+        {
+            info = sdscatprintf(info,
+                "thread_%d_clients:%d\r\n",
+                ithread, server.rgthreadvar[ithread].cclients);
+        }
    }

    /* Memory */
@@ -4400,11 +4414,15 @@ extern "C" sds genRedisInfoString(const char *section) {
        "used_cpu_sys:%ld.%06ld\r\n"
        "used_cpu_user:%ld.%06ld\r\n"
        "used_cpu_sys_children:%ld.%06ld\r\n"
-        "used_cpu_user_children:%ld.%06ld\r\n",
+        "used_cpu_user_children:%ld.%06ld\r\n"
+        "server_threads:%d\r\n"
+        "long_lock_waits:%" PRIu64 "\r\n",
        (long)self_ru.ru_stime.tv_sec, (long)self_ru.ru_stime.tv_usec,
        (long)self_ru.ru_utime.tv_sec, (long)self_ru.ru_utime.tv_usec,
        (long)c_ru.ru_stime.tv_sec, (long)c_ru.ru_stime.tv_usec,
-        (long)c_ru.ru_utime.tv_sec, (long)c_ru.ru_utime.tv_usec);
+        (long)c_ru.ru_utime.tv_sec, (long)c_ru.ru_utime.tv_usec,
+        server.cthreads,
+        fastlock_getlongwaitcount());
    }

    /* Command statistics */
@@ -4619,6 +4637,7 @@ static void sigShutdownHandler(int sig) {
        rdbRemoveTempFile(getpid());
        exit(1); /* Exit with an error since this was not a clean shutdown. */
    } else if (server.loading) {
+        serverLogFromHandler(LL_WARNING, "Received shutdown signal during loading, exiting now.");
        exit(0);
    }

@@ -4668,9 +4687,9 @@ void loadDataFromDisk(void) {
    if (server.aof_state == AOF_ON) {
        if (loadAppendOnlyFile(server.aof_filename) == C_OK)
            serverLog(LL_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000);
-    } else if (server.rdb_filename != NULL) {
+    } else if (server.rdb_filename != NULL || server.rdb_s3bucketpath != NULL) {
        rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
-        if (rdbLoad(server.rdb_filename,&rsi) == C_OK) {
+        if (rdbLoad(&rsi) == C_OK) {
            serverLog(LL_NOTICE,"DB loaded from disk: %.3f seconds",
                (float)(ustime()-start)/1000000);

@@ -5057,6 +5076,7 @@ int main(int argc, char **argv) {
        pthread_create(rgthread + iel, NULL, workerThreadMain, (void*)((int64_t)iel));
        if (server.fThreadAffinity)
        {
+#ifdef __linux__
            cpu_set_t cpuset;
            CPU_ZERO(&cpuset);
            CPU_SET(iel, &cpuset);
@@ -5064,6 +5084,9 @@ int main(int argc, char **argv) {
            {
                serverLog(LOG_INFO, "Binding thread %d to cpu %d", iel, iel);
            }
+#else
+			serverLog(LL_WARNING, "CPU pinning not available on this platform");
+#endif
        }
    }
    void *pvRet;
@@ -87,6 +87,8 @@ typedef long long mstime_t; /* millisecond time type. */
 extern "C" {
 #endif

+#define UUID_BINARY_LEN 16
+
 /* Error codes */
 #define C_OK                    0
 #define C_ERR                   -1
@@ -186,6 +188,8 @@ extern "C" {
 #define CONFIG_DEFAULT_THREADS 1
 #define CONFIG_DEFAULT_THREAD_AFFINITY 0

+#define CONFIG_DEFAULT_ACTIVE_REPLICA 0
+
 #define ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP 20 /* Loopkups per loop. */
 #define ACTIVE_EXPIRE_CYCLE_FAST_DURATION 1000 /* Microseconds */
 #define ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC 25 /* CPU max % for keys collection */
@@ -335,17 +339,19 @@ extern "C" {
 #define REPL_STATE_RECEIVE_PONG 3 /* Wait for PING reply */
 #define REPL_STATE_SEND_AUTH 4 /* Send AUTH to master */
 #define REPL_STATE_RECEIVE_AUTH 5 /* Wait for AUTH reply */
-#define REPL_STATE_SEND_PORT 6 /* Send REPLCONF listening-port */
-#define REPL_STATE_RECEIVE_PORT 7 /* Wait for REPLCONF reply */
-#define REPL_STATE_SEND_IP 8 /* Send REPLCONF ip-address */
-#define REPL_STATE_RECEIVE_IP 9 /* Wait for REPLCONF reply */
-#define REPL_STATE_SEND_CAPA 10 /* Send REPLCONF capa */
-#define REPL_STATE_RECEIVE_CAPA 11 /* Wait for REPLCONF reply */
-#define REPL_STATE_SEND_PSYNC 12 /* Send PSYNC */
-#define REPL_STATE_RECEIVE_PSYNC 13 /* Wait for PSYNC reply */
+#define REPL_STATE_SEND_UUID 6 /* send our UUID */
+#define REPL_STATE_RECEIVE_UUID 7 /* they should ack with their UUID */
+#define REPL_STATE_SEND_PORT 8 /* Send REPLCONF listening-port */
+#define REPL_STATE_RECEIVE_PORT 9 /* Wait for REPLCONF reply */
+#define REPL_STATE_SEND_IP 10 /* Send REPLCONF ip-address */
+#define REPL_STATE_RECEIVE_IP 11 /* Wait for REPLCONF reply */
+#define REPL_STATE_SEND_CAPA 12 /* Send REPLCONF capa */
+#define REPL_STATE_RECEIVE_CAPA 13 /* Wait for REPLCONF reply */
+#define REPL_STATE_SEND_PSYNC 14 /* Send PSYNC */
+#define REPL_STATE_RECEIVE_PSYNC 15 /* Wait for PSYNC reply */
 /* --- End of handshake states --- */
-#define REPL_STATE_TRANSFER 14 /* Receiving .rdb from master */
-#define REPL_STATE_CONNECTED 15 /* Connected to master */
+#define REPL_STATE_TRANSFER 16 /* Receiving .rdb from master */
+#define REPL_STATE_CONNECTED 17 /* Connected to master */

 /* State of slaves from the POV of the master. Used in client->replstate.
 * In SEND_BULK and ONLINE state the slave receives new updates
@@ -595,16 +601,18 @@ typedef struct RedisModuleIO {
    int ver;            /* Module serialization version: 1 (old),
                         * 2 (current version with opcodes annotation). */
    struct RedisModuleCtx *ctx; /* Optional context, see RM_GetContextFromIO()*/
+    struct redisObject *key;    /* Optional name of key processed */
 } RedisModuleIO;

 /* Macro to initialize an IO context. Note that the 'ver' field is populated
 * inside rdb.c according to the version of the value to load. */
-#define moduleInitIOContext(iovar,mtype,rioptr) do { \
+#define moduleInitIOContext(iovar,mtype,rioptr,keyptr) do { \
    iovar.prio = rioptr; \
    iovar.type = mtype; \
    iovar.bytes = 0; \
    iovar.error = 0; \
    iovar.ver = 0; \
+    iovar.key = keyptr; \
    iovar.ctx = NULL; \
 } while(0);

@@ -869,6 +877,11 @@ typedef struct client {
    sds peerid;             /* Cached peer ID. */
    listNode *client_list_node; /* list node in client list */

+    /* UUID announced by the client (default nil) - used to detect multiple connections to/from the same peer */
+    /* compliant servers will announce their UUIDs when a replica connection is started, and return when asked */
+    /* UUIDs are transient and lost when the server is shut down */
+    unsigned char uuid[UUID_BINARY_LEN];
+
    /* Response buffer */
    int bufpos;
    char buf[PROTO_REPLY_CHUNK_BYTES];
@@ -1053,6 +1066,7 @@ struct redisServerThreadVars {
    list *clients_pending_write; /* There is to write or install handler. */
    list *unblocked_clients;     /* list of clients to unblock before next loop NOT THREADSAFE */
    list *clients_pending_asyncwrite;
+    int cclients;
    struct fastlock lockPendingWrite;
 };

@@ -1087,7 +1101,9 @@ struct redisServer {
    size_t initial_memory_usage; /* Bytes used after initialization. */
    int always_show_logo;       /* Show logo even for non-stdout logging. */
    /* Modules */
-    dict *moduleapi;            /* Exported APIs dictionary for modules. */
+    dict *moduleapi;            /* Exported core APIs dictionary for modules. */
+    dict *sharedapi;            /* Like moduleapi but containing the APIs that
+                                   modules share with each other. */
    list *loadmodule_queue;     /* List of modules to load at startup. */
    int module_blocked_pipe[2]; /* Pipe used to awake the event loop if a
                                   client blocked on a module command needs
@@ -1418,6 +1434,11 @@ struct redisServer {
    pthread_mutex_t next_client_id_mutex;
    pthread_mutex_t unixtime_mutex;

+    int fActiveReplica;                          /* Can this replica also be a master? */
+    unsigned char uuid[UUID_BINARY_LEN];         /* This server's UUID - populated on boot */
+    unsigned char master_uuid[UUID_BINARY_LEN];  /* Used during sync with master, this is our master's UUID */
+                                                /* After we've connected with our master use the UUID in server.master */
+
    struct fastlock flock;
 };

@@ -1740,6 +1761,7 @@ int compareStringObjects(robj *a, robj *b);
 int collateStringObjects(robj *a, robj *b);
 int equalStringObjects(robj *a, robj *b);
 unsigned long long estimateObjectIdleTime(robj *o);
+void trimStringObjectIfNeeded(robj *o);
 #define sdsEncodedObject(objptr) (objptr->encoding == OBJ_ENCODING_RAW || objptr->encoding == OBJ_ENCODING_EMBSTR)

 /* Synchronous I/O with timeout */
@@ -1967,7 +1989,6 @@ void setTypeConvert(robj *subject, int enc);

 void hashTypeConvert(robj *o, int enc);
 void hashTypeTryConversion(robj *subject, robj **argv, int start, int end);
-void hashTypeTryObjectEncoding(robj *subject, robj **o1, robj **o2);
 int hashTypeExists(robj *o, sds key);
 int hashTypeDelete(robj *o, sds key);
 unsigned long hashTypeLength(const robj *o);
@@ -2328,11 +2349,13 @@ void lolwutCommand(client *c);
 void aclCommand(client *c);

 #if defined(__GNUC__)
+#ifndef __cplusplus
 void *calloc(size_t count, size_t size) __attribute__ ((deprecated));
 void free(void *ptr) __attribute__ ((deprecated));
 void *malloc(size_t size) __attribute__ ((deprecated));
 void *realloc(void *ptr, size_t size) __attribute__ ((deprecated));
 #endif
+#endif

 /* Debugging stuff */
 void _serverAssertWithInfo(const client *c, const robj *o, const char *estr, const char *file, int line);
@@ -58,7 +58,7 @@ redisSortOperation *createSortOperation(int type, robj *pattern) {
 *
 * The returned object will always have its refcount increased by 1
 * when it is non-NULL. */
-robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
+robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst, int writeflag) {
    char *p, *f, *k;
    sds spat, ssub;
    robj *keyobj, *fieldobj = NULL, *o;
@@ -106,7 +106,10 @@ robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
    decrRefCount(subst); /* Incremented by decodeObject() */

    /* Lookup substituted key */
-    o = lookupKeyRead(db,keyobj);
+    if (!writeflag)
+        o = lookupKeyRead(db,keyobj);
+    else
+        o = lookupKeyWrite(db,keyobj);
    if (o == NULL) goto noobj;

    if (fieldobj) {
@@ -198,30 +201,12 @@ void sortCommand(client *c) {
    robj *sortval, *sortby = NULL, *storekey = NULL;
    redisSortObject *vector; /* Resulting vector to sort */

-    /* Lookup the key to sort. It must be of the right types */
-    sortval = lookupKeyRead(c->db,c->argv[1]);
-    if (sortval && sortval->type != OBJ_SET &&
-                   sortval->type != OBJ_LIST &&
-                   sortval->type != OBJ_ZSET)
-    {
-        addReply(c,shared.wrongtypeerr);
-        return;
-    }
-
    /* Create a list of operations to perform for every sorted element.
     * Operations can be GET */
    operations = listCreate();
    listSetFreeMethod(operations,zfree);
    j = 2; /* options start at argv[2] */

-    /* Now we need to protect sortval incrementing its count, in the future
-     * SORT may have options able to overwrite/delete keys during the sorting
-     * and the sorted key itself may get destroyed */
-    if (sortval)
-        incrRefCount(sortval);
-    else
-        sortval = createQuicklistObject();
-
    /* The SORT command has an SQL-alike syntax, parse it */
    while(j < c->argc) {
        int leftargs = c->argc-j-1;
@@ -280,11 +265,33 @@ void sortCommand(client *c) {

    /* Handle syntax errors set during options parsing. */
    if (syntax_error) {
-        decrRefCount(sortval);
        listRelease(operations);
        return;
    }

+    /* Lookup the key to sort. It must be of the right types */
+    if (storekey)
+        sortval = lookupKeyRead(c->db,c->argv[1]);
+    else
+        sortval = lookupKeyWrite(c->db,c->argv[1]);
+    if (sortval && sortval->type != OBJ_SET &&
+                   sortval->type != OBJ_LIST &&
+                   sortval->type != OBJ_ZSET)
+    {
+        listRelease(operations);
+        addReply(c,shared.wrongtypeerr);
+        return;
+    }
+
+    /* Now we need to protect sortval incrementing its count, in the future
+     * SORT may have options able to overwrite/delete keys during the sorting
+     * and the sorted key itself may get destroyed */
+    if (sortval)
+        incrRefCount(sortval);
+    else
+        sortval = createQuicklistObject();
+
+
    /* When sorting a set with no sort specified, we must sort the output
     * so the result is consistent across scripting and replication.
     *
@@ -452,7 +459,7 @@ void sortCommand(client *c) {
            robj *byval;
            if (sortby) {
                /* lookup value to sort by */
-                byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
+                byval = lookupKeyByPattern(c->db,sortby,vector[j].obj,storekey!=NULL);
                if (!byval) continue;
            } else {
                /* use object itself to sort by */
@@ -515,7 +522,7 @@ void sortCommand(client *c) {
            while((ln = listNext(&li))) {
                redisSortOperation *sop = ln->value;
                robj *val = lookupKeyByPattern(c->db,sop->pattern,
-                    vector[j].obj);
+                    vector[j].obj,storekey!=NULL);

                if (sop->type == SORT_OP_GET) {
                    if (!val) {
@@ -545,7 +552,7 @@ void sortCommand(client *c) {
                while((ln = listNext(&li))) {
                    redisSortOperation *sop = ln->value;
                    robj *val = lookupKeyByPattern(c->db,sop->pattern,
-                        vector[j].obj);
+                        vector[j].obj,storekey!=NULL);

                    if (sop->type == SORT_OP_GET) {
                        if (!val) val = createStringObject("",0);
@@ -1,12 +1,13 @@
 #include <stdlib.h>
 #include <stdio.h>
 #include <sys/ioctl.h>
-#include <linux/fs.h>
 #include <unistd.h>
 #include <inttypes.h>
 #include "storage.h"
 #include <assert.h>
+#ifdef __linux__
 #include <malloc.h>
+#endif

 // initialize the memory subsystem. 
 //  NOTE: This may be called twice, first with NULL specifying we should use ram
@@ -42,7 +43,9 @@ void *srealloc(void *pv, size_t cb, enum MALLOC_CLASS class)
    return realloc(pv, cb);
 }

+#ifdef __linux__
 size_t salloc_usable_size(void *ptr)
 {
    return malloc_usable_size(ptr);
 }
+#endif
@@ -3,6 +3,10 @@

 #define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 44   // Note: also defined in object.c - should always match

+#ifdef __cplusplus
+extern "C" {
+#endif
+
 enum MALLOC_CLASS
 {
    MALLOC_LOCAL,
@@ -22,4 +26,8 @@ void sfree_objembstr(struct redisObject *obj);
 struct redisObject *salloc_obj();
 void sfree_obj(struct redisObject *obj);

+#ifdef __cplusplus
+}
+#endif
+
 #endif
@@ -615,6 +615,10 @@ void hincrbyfloatCommand(client *c) {
    }

    value += incr;
+    if (isnan(value) || isinf(value)) {
+        addReplyError(c,"increment would produce NaN or Infinity");
+        return;
+    }

    char buf[MAX_LONG_DOUBLE_CHARS];
    int len = ld2string(buf,sizeof(buf),value,1);
@@ -520,7 +520,7 @@ void lremCommand(client *c) {

    if (removed) {
        signalModifiedKey(c->db,c->argv[1]);
-        notifyKeyspaceEvent(NOTIFY_GENERIC,"lrem",c->argv[1],c->db->id);
+        notifyKeyspaceEvent(NOTIFY_LIST,"lrem",c->argv[1],c->db->id);
    }

    if (listTypeLength(subject) == 0) {
@@ -415,7 +415,7 @@ void spopWithCountCommand(client *c) {

    /* Make sure a key with the name inputted exists, and that it's type is
     * indeed a set. Otherwise, return nil */
-    if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp]))
+    if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.null[c->resp]))
        == NULL || checkType(c,set,OBJ_SET)) return;

    /* If count is zero, serve an empty multibulk ASAP to avoid special
@@ -1064,7 +1064,8 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
            sdsfree(ele);
        }
        setTypeReleaseIterator(si);
-        decrRefCount(dstset);
+        server.lazyfree_lazy_server_del ? freeObjAsync(dstset) :
+                                          decrRefCount(dstset);
    } else {
        /* If we have a target key where to store the resulting set
         * create this key with the result set inside */
@@ -1742,14 +1742,17 @@ NULL
    /* Everything but the "HELP" option requires a key and group name. */
    if (c->argc >= 4) {
        o = lookupKeyWrite(c->db,c->argv[2]);
-        if (o) s = ptrFromObj(o);
+        if (o) {
+            if (checkType(c,o,OBJ_STREAM)) return;
+            s = (stream*)ptrFromObj(o);
+        }
        grpname = ptrFromObj(c->argv[3]);
    }

    /* Check for missing key/group. */
    if (c->argc >= 4 && !mkstream) {
        /* At this point key must exist, or there is an error. */
-        if (o == NULL) {
+        if (s == NULL) {
            addReplyError(c,
                "The XGROUP subcommand requires the key to exist. "
                "Note that for CREATE you may want to use the MKSTREAM "
@@ -1757,8 +1760,6 @@ NULL
            return;
        }

-        if (checkType(c,o,OBJ_STREAM)) return;
-
        /* Certain subcommands require the group to exist. */
        if ((cg = streamLookupCG(s,grpname)) == NULL &&
            (!strcasecmp(opt,"SETID") ||
@@ -1786,7 +1787,8 @@ NULL
        }

        /* Handle the MKSTREAM option now that the command can no longer fail. */
-        if (s == NULL && mkstream) {
+        if (s == NULL) {
+            serverAssert(mkstream);
            o = createStreamObject();
            dbAdd(c->db,c->argv[2],o);
            s = ptrFromObj(o);
@@ -2284,8 +2286,13 @@ void xclaimCommand(client *c) {
            /* Update the consumer and idle time. */
            nack->consumer = consumer;
            nack->delivery_time = deliverytime;
-            /* Set the delivery attempts counter if given. */
-            if (retrycount >= 0) nack->delivery_count = retrycount;
+            /* Set the delivery attempts counter if given, otherwise 
+             * autoincrement unless JUSTID option provided */
+            if (retrycount >= 0) {
+                nack->delivery_count = retrycount;
+            } else if (!justid) {
+                nack->delivery_count++;
+            }
            /* Add the entry in the new consumer local PEL. */
            raxInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
            /* Send the reply for this entry. */
@@ -2906,7 +2906,10 @@ void genericZrangebylexCommand(client *c, int reverse) {
        while (remaining) {
            if (remaining >= 3 && !strcasecmp(ptrFromObj(c->argv[pos]),"limit")) {
                if ((getLongFromObjectOrReply(c, c->argv[pos+1], &offset, NULL) != C_OK) ||
-                    (getLongFromObjectOrReply(c, c->argv[pos+2], &limit, NULL) != C_OK)) return;
+                    (getLongFromObjectOrReply(c, c->argv[pos+2], &limit, NULL) != C_OK)) {
+                    zslFreeLexRange(&range);
+                    return;
+                }
                pos += 3; remaining -= 3;
            } else {
                zslFreeLexRange(&range);
@@ -3140,7 +3143,10 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
    if (countarg) {
        if (getLongFromObjectOrReply(c,countarg,&count,NULL) != C_OK)
            return;
-        if (count < 0) count = 1;
+        if (count <= 0) {
+            addReply(c,shared.emptyarray);
+            return;
+        }
    }

    /* Check type and break on the first error, otherwise identify candidate. */
@@ -447,7 +447,7 @@ int string2l(const char *s, size_t slen, long *lval) {
 * a double: no spaces or other characters before or after the string
 * representing the number are accepted. */
 int string2ld(const char *s, size_t slen, long double *dp) {
-    char buf[256];
+    char buf[MAX_LONG_DOUBLE_CHARS];
    long double value;
    char *eptr;

@@ -1 +1 @@
-#define REDIS_VERSION "0.9.2"
+#define REDIS_VERSION "0.9.3"
@@ -159,6 +159,10 @@ void *zrealloc(void *ptr, size_t size, enum MALLOC_CLASS class) {
    size_t oldsize;
    void *newptr;

+    if (size == 0 && ptr != NULL) {
+        zfree(ptr);
+        return NULL;
+    }
    if (ptr == NULL) return zmalloc(size, class);
 #ifdef HAVE_MALLOC_SIZE
    oldsize = zmalloc_size(ptr);
@@ -115,6 +115,34 @@ start_server {tags {"hll"}} {
        set e
    } {*WRONGTYPE*}

+    test {Fuzzing dense/sparse encoding: Redis should always detect errors} {
+        for {set j 0} {$j < 1000} {incr j} {
+            r del hll
+            set items {}
+            set numitems [randomInt 3000]
+            for {set i 0} {$i < $numitems} {incr i} {
+                lappend items [expr {rand()}]
+            }
+            r pfadd hll {*}$items
+
+            # Corrupt it in some random way.
+            for {set i 0} {$i < 5} {incr i} {
+                set len [r strlen hll]
+                set pos [randomInt $len]
+                set byte [randstring 1 1 binary]
+                r setrange hll $pos $byte
+                # Don't modify more bytes 50% of times
+                if {rand() < 0.5} break
+            }
+
+            # Use the hyperloglog to check if it crashes
+            # Redis in some way.
+            catch {
+                r pfcount hll
+            }
+        }
+    }
+
    test {PFADD, PFCOUNT, PFMERGE type checking works} {
        r set foo bar
        catch {r pfadd foo 1} e
@@ -436,8 +436,11 @@ start_server {

        test "$pop: with non-integer timeout" {
            set rd [redis_deferring_client]
-            $rd $pop blist1 1.1
-            assert_error "ERR*not an integer*" {$rd read}
+            r del blist1
+            $rd $pop blist1 0.1
+            r rpush blist1 foo
+            assert_equal {blist1 foo} [$rd read]
+            assert_equal 0 [r exists blist1]
        }

        test "$pop: with zero timeout should block indefinitely" {
@@ -195,6 +195,49 @@ start_server {
        assert_equal "" [lindex $reply 0]
    }

+    test {XCLAIM without JUSTID increments delivery count} {
+        # Add 3 items into the stream, and create a consumer group
+        r del mystream
+        set id1 [r XADD mystream * a 1]
+        set id2 [r XADD mystream * b 2]
+        set id3 [r XADD mystream * c 3]
+        r XGROUP CREATE mystream mygroup 0
+
+        # Client 1 reads item 1 from the stream without acknowledgements.
+        # Client 2 then claims pending item 1 from the PEL of client 1
+        set reply [
+            r XREADGROUP GROUP mygroup client1 count 1 STREAMS mystream >
+        ]
+        assert {[llength [lindex $reply 0 1 0 1]] == 2}
+        assert {[lindex $reply 0 1 0 1] eq {a 1}}
+        r debug sleep 0.2
+        set reply [
+            r XCLAIM mystream mygroup client2 10 $id1
+        ]
+        assert {[llength [lindex $reply 0 1]] == 2}
+        assert {[lindex $reply 0 1] eq {a 1}}
+
+        set reply [
+            r XPENDING mystream mygroup - + 10
+        ]
+        assert {[llength [lindex $reply 0]] == 4}
+        assert {[lindex $reply 0 3] == 2}
+
+        # Client 3 then claims pending item 1 from the PEL of client 2 using JUSTID
+        r debug sleep 0.2
+        set reply [
+            r XCLAIM mystream mygroup client3 10 $id1 JUSTID
+        ]
+        assert {[llength $reply] == 1}
+        assert {[lindex $reply 0] eq $id1}
+
+        set reply [
+            r XPENDING mystream mygroup - + 10
+        ]
+        assert {[llength [lindex $reply 0]] == 4}
+        assert {[lindex $reply 0 3] == 2}
+    }
+
    start_server {} {
        set master [srv -1 client]
        set master_host [srv -1 host]
@@ -43,6 +43,9 @@
 #
 # /!\ This script should be run as root
 #
+# NOTE: This script will not work on Mac OSX.
+#       It supports Debian and Ubuntu Linux.
+#
 ################################################################################

 die () {
Author	SHA1	Message	Date
John Sully	d7112df465	Bump version because we added new features	2019-03-25 01:12:31 -04:00
John Sully	ed021c88b9	Merge branch 'unstable' into RELEASE_0_9	2019-03-25 01:11:50 -04:00
John Sully	1840b893c2	Talk about active replica	2019-03-25 00:12:04 -04:00
John Sully	8bcde757ce	We don't corretly terminate our varargs	2019-03-24 18:57:25 -04:00
John Sully	1dff223c71	OS X build fixes	2019-03-24 18:06:57 -04:00
John Sully	a7aa2b0740	Active Replica Support	2019-03-24 15:39:10 -04:00
John Sully	fb45fefbca	Merge branch 'unstable' of https://github.com/JohnSully/redis2 into unstable	2019-03-21 23:11:55 -04:00
John Sully	7b914c2531	Fix kqueue build error	2019-03-21 23:11:19 -04:00
John Sully	75d517a27f	Merge pull request #16 from gitter-badger/gitter-badge Add a Gitter chat badge to README.md	2019-03-21 21:57:25 -04:00
John Sully	4c98bc690e	Update README.md	2019-03-21 21:26:00 -04:00
John Sully	550b3a3a99	Update README.md	2019-03-21 21:25:49 -04:00
The Gitter Badger	ef7e89352d	Add Gitter badge	2019-03-22 01:24:32 +00:00
John Sully	9322d604ee	Merge branch 'unstable' of https://github.com/antirez/redis into unstable	2019-03-21 20:15:59 -04:00
John Sully	5d92110630	Merge pull request #15 from DevopsDood/unstable Added docker builder for those that want it	2019-03-21 19:54:00 -04:00
Richard Silver	4056037cc2	Merge branch 'unstable' of github.com:JohnSully/KeyDB into unstable	2019-03-21 16:19:45 -07:00
Richard Silver	677ae5487f	updated readme and Dockerfile to allow for tests to be run in container	2019-03-21 16:18:37 -07:00
John Sully	19956b3807	comment issue	2019-03-21 22:18:48 +00:00
John Sully	ab9122f2cb	Polarity of the weak flag was wrong. We want to use the strong variant of atomic_compare_exchange always	2019-03-21 22:17:04 +00:00
Richard Silver	e62c6b0ed0	Added build Dockerfile	2019-03-21 15:15:11 -07:00
John Sully	27a420fbc2	Compile issues at O0 optimization	2019-03-21 21:57:18 +00:00
antirez	b3408e9a9b	More sensible name for function: restartAOFAfterSYNC(). Related to #3829.	2019-03-21 17:21:29 +01:00
antirez	9588fd52ac	Mostly aesthetic changes to restartAOF(). See #3829.	2019-03-21 17:18:24 +01:00
Salvatore Sanfilippo	544373ea66	Merge pull request #3829 from oranagra/restart_aof_pr bugfix to restartAOF, exit will never happen	2019-03-21 17:15:55 +01:00
Salvatore Sanfilippo	d75c36ee48	Merge pull request #5390 from oranagra/lookup_key_write change SORT and SPOP to use lookupKeyWrite rather than lookupKeyRead	2019-03-21 13:39:38 +01:00
antirez	9dabbd1ab0	Alter coding style in #4696 to conform to Redis code base.	2019-03-21 12:18:55 +01:00
Salvatore Sanfilippo	5c47e2e964	Merge pull request #4696 from oranagra/zrealloc_fix Fix zrealloc to behave similarly to je_realloc when size is 0	2019-03-21 12:18:04 +01:00
Salvatore Sanfilippo	5c21eca66e	Merge pull request #5243 from oranagra/sigterm_log Add log when server dies of SIGTERM during loading	2019-03-21 11:54:09 +01:00
Salvatore Sanfilippo	ce518cea79	Merge pull request #5391 from oranagra/touch_flags getKeysFromCommand for TOUCH only extracted the first key.	2019-03-21 11:53:27 +01:00
Salvatore Sanfilippo	684b140a13	Merge pull request #3650 from yossigo/key_name_from_module_io Add RedisModule_GetKeyNameFromIO()	2019-03-21 11:49:25 +01:00
oranagra	c9e2900efc	bugfix to restartAOF, exit will never happen since retry will get negative. also reduce an excess sleep	2019-03-20 17:20:07 +02:00
Oran Agra	747174388f	change SORT and SPOP to use lookupKeyWrite rather than lookupKeyRead like in SUNIONSTORE etc, commands that perform writes are expected to open all keys, even input keys, with lookupKeyWrite	2019-03-20 17:06:22 +02:00
Oran Agra	385f6190a3	getKeysFromCommand for TOUCH only extracted the first key. also, airty for COMMAND command was wrong.	2019-03-20 17:00:39 +02:00
John Sully	b775ebb382	ARM build fix: Don't use X64 asm here	2019-03-20 04:14:33 +00:00
John Sully	ac80a5c6a6	Add debugging stats to the INFO command	2019-03-19 22:04:33 -04:00
John Sully	c74100a552	Add release badge	2019-03-19 20:54:36 -04:00
John Sully	03379106d6	Add CI badge	2019-03-19 20:46:25 -04:00
John Sully	9ad229f663	Getting close! Maybe there was a better way to do this than checking in travis.yml files repeatedly but alas it works and doesn't affect people building locally	2019-03-19 20:40:17 -04:00
John Sully	0900edf776	Clang build with libc malloc	2019-03-19 20:34:09 -04:00
John Sully	ea520027b9	more clang travis fixes	2019-03-19 20:26:35 -04:00
John Sully	04679b3b3b	Fix race condition building dependencies with multithreaded builds	2019-03-19 20:25:30 -04:00
John Sully	2bf5c1324a	Will clang build now with travis?	2019-03-19 20:00:52 -04:00
John Sully	3d93ca1bf6	clang build failure	2019-03-19 19:55:03 -04:00
John Sully	c548431611	Travis clang dependencies	2019-03-19 19:50:28 -04:00
John Sully	0ab29f7394	Makefile ignores env CXXFLAGS	2019-03-19 19:29:14 -04:00
John Sully	30dab7ad14	No underscore	2019-03-19 19:22:27 -04:00
John Sully	0c5fa59221	Maybe this will make travis-ci stop using GCC headers when compiling with clang	2019-03-19 19:12:15 -04:00
John Sully	267e84f97d	Feature gate SO_INCOMING_CPU so we compile on older kernels	2019-03-19 18:55:42 -04:00
John Sully	344cca75c6	typo	2019-03-19 18:40:50 -04:00
John Sully	0dd8e682f4	It would be nice if our CI actually called make...	2019-03-19 18:37:53 -04:00
John Sully	da7e54819a	Fix issues with travis config	2019-03-19 18:31:45 -04:00
John Sully	4e70215307	Let's try travis-ci	2019-03-19 18:27:54 -04:00
John	6aa162f6ff	Fix linker errors on some compilers due to sdslen()	2019-03-19 22:08:20 +00:00
John Sully	5d360ed401	Bump the version to indicate this updated release has bug fixes.	2019-03-19 15:38:39 -04:00
John Sully	8164c93d5a	Merge branch 'unstable' into RELEASE_0_9	2019-03-19 15:24:42 -04:00
John Sully	d380074569	KeyDB has different aims than the Redis project. We believe software is for the user. If a feature requires complicating the codebase but simplifies our user's lives we will do it any time. This should not be taken as an outright rejection of the manifesto, as we agree on many points.	2019-03-19 01:36:03 -04:00
John Sully	c406e7ff51	Merge branch 'unstable' of https://github.com/JohnSully/redis2 into unstable	2019-03-19 01:26:22 -04:00
John Sully	31ff6cffee	Merge branch 'unstable' of https://github.com/antirez/redis into unstable	2019-03-19 01:26:10 -04:00
antirez	3eaa2cdc44	MANIFESTO: simplicity and lock-in.	2019-03-18 15:49:52 +01:00
antirez	a5af648fdd	MANIFESTO v2.	2019-03-18 15:38:43 +01:00
antirez	14b17c3615	replicaofCommand() refactoring: stay into 80 cols.	2019-03-18 11:34:40 +01:00
antirez	b78ac354f4	redis-check-aof: fix potential overflow. Bug signaled by @vattezhang in PR #5940 but fixed differently.	2019-03-18 11:16:23 +01:00
antirez	8ea906a3e8	HyperLogLog: fix comment in hllCount().	2019-03-16 09:15:57 +01:00
antirez	e216ceaf0e	HyperLogLog: handle wrong offset in the base case.	2019-03-15 17:16:06 +01:00
antirez	dca7358279	HyperLogLog: speedup fuzz test.	2019-03-15 17:13:19 +01:00
antirez	a4b90be9fc	HyperLogLog: enlarge reghisto variable for safety.	2019-03-15 17:10:16 +01:00
antirez	4208666797	HyperLogLog: dense/sparse repr parsing fuzz test.	2019-03-15 13:52:29 +01:00
John Sully	9f13b2bd49	Fix hyperloglog corruption	2019-03-15 13:13:01 +01:00
Yossi Gottlieb	a88264d934	Add RedisModule_GetKeyNameFromIO().	2019-03-15 10:23:27 +02:00
John Sully	a9e525c81e	Hack to prevent build errors on some machines due to missing throw()	2019-03-15 05:28:20 +00:00
John Sully	6157a5fc1b	Merge /home/john/repos/redis_clean into unstable	2019-03-14 14:22:41 -04:00
John Sully	be3cb1ad33	Merge branch 'unstable' of https://github.com/antirez/redis into unstable	2019-03-14 14:22:27 -04:00
John Sully	a89ab4d6b0	Fix hyperloglog corruption	2019-03-14 14:02:16 -04:00
antirez	74d6af8f80	Fix ZPOP return type when COUNT=0. Related to #5799 .	2019-03-14 17:51:20 +01:00
antirez	052e03495f	Fix objectSetLRUOrLFU() when LFU underflows.	2019-03-14 17:06:59 +01:00
antirez	d292a51618	Improve comments after merging #5834 .	2019-03-14 12:47:36 +01:00
Salvatore Sanfilippo	0cce98f2f9	Merge pull request #5834 from guybe7/trim_module_sds Trim SDS free space of retained module strings	2019-03-14 12:41:31 +01:00
Salvatore Sanfilippo	178114e766	Merge pull request #5931 from guybe7/fix_notify_classes Fix mismatching keyspace notification classes	2019-03-14 12:30:10 +01:00
Guy Benoish	b2eb48df89	Fix mismatching keyspace notification classes	2019-03-14 12:11:16 +01:00
Salvatore Sanfilippo	d3cd2e9a4a	Merge pull request #4742 from guybe7/zrangebylex_leak Fix zlexrangespec mem-leak in genericZrangebylexCommand	2019-03-14 11:44:56 +01:00
Salvatore Sanfilippo	ec03f2db6b	Merge pull request #4904 from guybe7/memtoll_fix Use memtoll() in 'CONFIG SET client-output-buffer-limit'	2019-03-14 11:43:55 +01:00
Salvatore Sanfilippo	190b63f993	Merge pull request #5812 from guybe7/fix_string2ld Increase string2ld's buffer size (and fix HINCRBYFLOAT)	2019-03-14 11:35:01 +01:00
Salvatore Sanfilippo	01ace0b240	Merge pull request #5694 from guybe7/rm_log_args Check server.verbosity in RM_LogRaw	2019-03-14 11:32:03 +01:00
Salvatore Sanfilippo	ebf7274192	Merge pull request #5799 from guybe7/zpop_zero_count ZPOP should return an empty array if COUNT=0	2019-03-14 11:29:24 +01:00
antirez	6fd4af1f12	Merge branch 'sharedapi' into unstable	2019-03-14 11:24:48 +01:00
John Sully	966f4a8144	Merge branch 'unstable' of https://github.com/JohnSully/redis2 into unstable	2019-03-13 18:08:59 -04:00
John Sully	261cf24efc	Merge branch 'unstable' of https://github.com/antirez/redis into unstable Lots of fixes and improvements from upstream.	2019-03-13 18:08:22 -04:00
JohnSully	a3da10f4eb	Update README.md	2019-03-13 16:57:40 -04:00
John Sully	6cb8063497	Merge branch 'unstable' of https://github.com/JohnSully/redis2 into unstable	2019-03-13 16:53:55 -04:00
John Sully	a45f212693	Implement load database dumps from S3. We already save.	2019-03-13 16:53:37 -04:00
JohnSully	7c72cd5471	Merge pull request #9 from 0xflotus/patch-1 fixed guarantees	2019-03-13 12:22:29 -04:00
0xflotus	66f2032708	fixed guarantees	2019-03-13 15:14:34 +01:00
Salvatore Sanfilippo	2537b2140f	Merge pull request #5787 from soloestoy/bugfix-xgroup-create-with-mkstream Streams: checkType before XGROUP CREATE	2019-03-13 12:34:29 +01:00
Salvatore Sanfilippo	71499ab1c6	Merge pull request #5893 from itamarhaber/bzpop-arity Fixes BZ[REV]POP's arity	2019-03-13 12:27:35 +01:00
Salvatore Sanfilippo	c49abcb794	Merge pull request #5898 from bsolomon1124/make-uninstall-tgt Provide an uninstall target in Makefile	2019-03-13 12:26:43 +01:00
Salvatore Sanfilippo	372cbc37e7	Merge pull request #5884 from chendq8/cluster_failover Reduce 100ms of cluster failover time on Manual mode	2019-03-13 12:22:23 +01:00
Salvatore Sanfilippo	46a8cb2d18	Merge pull request #5910 from bsolomon1124/install-server-osx-note Note that install_server.sh is not for Mac OSX	2019-03-13 12:19:03 +01:00
Salvatore Sanfilippo	84d6ce4f56	Merge pull request #5901 from artix75/bm_threads_cluster_dev Redis Benchmark: add multithread idle mode	2019-03-13 12:18:32 +01:00
Salvatore Sanfilippo	53d1dc170c	Merge pull request #5921 from vattezhang/unstable_temp fix: fix the if condition in clusterManagerShowClusterInfo	2019-03-13 12:18:08 +01:00
Salvatore Sanfilippo	3e648907ee	Merge pull request #5907 from spjwebster/xclaim-increment-delivery-count Increment delivery counter on XCLAIM unless RETRYCOUNT specified	2019-03-13 11:55:46 +01:00
Salvatore Sanfilippo	610aff3251	Merge pull request #5924 from swilly22/redismodule-replicated-cmd-ctx-flag document additional flag of RM_GetContextFlags	2019-03-13 11:52:07 +01:00
Salvatore Sanfilippo	02dab20026	Merge pull request #5894 from soloestoy/fix-rdb-warning Fix compile some warnings	2019-03-13 11:47:48 +01:00
zhaozhao.zz	5a3d3d8be4	Fix compile warning in redis-cli.c	2019-03-13 16:54:34 +08:00
swilly22	89bf4db4fa	document additional flag of RM_GetContextFlags	2019-03-13 08:22:40 +02:00
Steve Webster	dfcb227b50	Only increment delivery count if JUSTID option is omitted	2019-03-12 20:27:53 +00:00
JohnSully	7546c7d8b4	Update README.md	2019-03-12 14:38:03 -04:00
Salvatore Sanfilippo	e5acc5ef4f	Merge pull request #2774 from rouzier/blocking-list-commands-support-milliseconds-floating Added millisecond resolution for blpop command && friends	2019-03-12 18:10:28 +01:00
Salvatore Sanfilippo	fba6e26e87	Merge pull request #5913 from wurongxin1987/unstable fix a bufferoverflow bug	2019-03-12 17:25:21 +01:00
Salvatore Sanfilippo	b0bc602bb3	Merge pull request #5900 from artix75/cluster_backup Cluster Manager: add importing/migrating nodes to backup info	2019-03-12 17:21:17 +01:00
Salvatore Sanfilippo	c02b414e9a	Merge pull request #5903 from soloestoy/sunionstore-lazyfree try lazyfree temp set in SUNION & SDIFF	2019-03-12 17:20:52 +01:00
Salvatore Sanfilippo	45645aee7f	Merge pull request #5915 from swilly22/redismodule-replicated-cmd-ctx-flag Extend REDISMODULE_CTX_FLAGS to indicate if command was sent by master	2019-03-12 17:17:19 +01:00
artix	5284d67e37	Redis Benchmark: fix possible usage of freed pointer (getRedisConfig) Fixes issue #5912	2019-03-12 17:07:23 +01:00
Salvatore Sanfilippo	f63bbb4335	Merge pull request #5920 from chendq8/aofrewrite remove temp-rewriteaof-xxx.aof when interrupt aofrewrite	2019-03-12 17:03:02 +01:00
vattezhang	bdc783b472	fix: fix the if condition in clusterManagerShowClusterInfo	2019-03-12 22:01:02 +08:00
chendianqiang	3621223fbb	remove temp-rewriteaof-xxx.aof when interrupt aofrewrite	2019-03-12 20:46:40 +08:00
JohnSully	82621702cf	Merge pull request #6 from JohnSully/unstable Merge readme changes	2019-03-12 01:30:56 -04:00
JohnSully	703e4ad2fe	formatting fixes	2019-03-12 00:59:28 -04:00
JohnSully	0b98280a1f	Add charts to readme Hotlink from medium, hope they don't mind.	2019-03-12 00:55:03 -04:00
John Sully	4609bed6a4	Merge branch 'unstable' of https://github.com/JohnSully/redis2 into RELEASE_0_9	2019-03-11 20:54:14 -04:00
John Sully	99ccb11d89	Werror is good for me, but not for people compiling on other compilers	2019-03-11 20:53:57 -04:00
JohnSully	8a5a513662	Explain how multithreading works	2019-03-11 20:37:57 -04:00
JohnSully	bf0fd22309	Update README.md	2019-03-11 19:38:20 -04:00
swilly22	468860ae18	Extend REDISMODULE_CTX_FLAGS to indicate if command was sent by master	2019-03-11 10:02:19 +02:00
antirez	8a46d32be2	Make comment in #5911 stay inside 80 cols.	2019-03-10 09:48:06 +01:00
Salvatore Sanfilippo	55acac04b4	Merge pull request #5911 from JohnSully/unstable Replicas aren't allowed to run the replicaof command	2019-03-10 09:45:55 +01:00
wurongxin	0298d3ad18	fix a bufferoverflow bug	2019-03-10 15:30:32 +08:00
John Sully	cc98d10fc6	Make storage.h C++ safe	2019-03-09 13:42:08 -05:00
John Sully	a6a022faa2	sizeof is too big	2019-03-09 13:41:49 -05:00
John Sully	5b52bc738b	Replicas aren't allowed to run the replicaof command	2019-03-09 11:04:48 -05:00
Brad Solomon	79660e4ff4	Note that install_server.sh is not for Mac OSX It will fail pretty quickly since there is no -f readlink flag there.	2019-03-09 10:21:15 -05:00
Steve Webster	f1e7df4b7c	Increment delivery counter on XCLAIM unless RETRYCOUNT specified The XCLAIM docs state the XCLAIM increments the delivery counter for messages. This PR makes the code match the documentation - which seems like the desired behaviour - whilst still allowing RETRYCOUNT to be specified manually. My understanding of the way streamPropagateXCLAIM() works is that this change will safely propagate to replicas since retry count is pulled directly from the streamNACK struct. Fixes #5194	2019-03-08 17:09:11 +00:00
Salvatore Sanfilippo	0f03312c23	Merge pull request #5902 from zhouyuan/wip_cleanup server.h: remove dead code	2019-03-08 16:53:20 +01:00
artix	8fadebfcca	Redis Benchmark: handle CLUSTERDOWN error	2019-03-08 11:05:02 +01:00
zhaozhao.zz	0137f1a2e3	try lazyfree temp set in SUNION & SDIFF	2019-03-07 22:08:04 +08:00
Yuan Zhou	feb4ebff45	server.h: remove dead code hashTypeTryObjectEncoding() is not used now	2019-03-07 18:38:27 +08:00
artix	c389ad0d52	Redis Benchmark: fix key randomization with zero keyspacelen	2019-03-07 11:30:09 +01:00
artix	0e963e068d	Redis Benchmark: add multithread idle mode Fix issue #5891	2019-03-07 11:14:07 +01:00
Brad Solomon	d5b24d31d7	Provide an uninstall target in Makefile On `make uninstall`, removes: - /usr/local/bin/redis-benchmark - /usr/local/bin/redis-check-aof - /usr/local/bin/redis-check-rdb - /usr/local/bin/redis-cli - /usr/local/bin/redis-sentinel - /usr/local/bin/redis-server (Only the src/ versions are removed in `make clean`)	2019-03-06 21:24:45 -05:00
artix	93e51239ce	Cluster Manager: add importing/migrating nodes to backup info	2019-03-06 16:38:36 +01:00
antirez	c33cb4938a	ACL: GENPASS subcommand.	2019-03-05 15:51:37 +01:00
zhaozhao.zz	fb81d1b3f8	Fix compile warning when log aux field	2019-03-04 19:43:00 +08:00
Itamar Haber	cd2743c0e6	Fixes BZ[REV]POP's arity	2019-03-03 23:10:45 +02:00
chendianqiang	a16da79228	optimize cluster failover	2019-03-01 15:28:21 +08:00
Guy Benoish	bdd9a8002a	Trim SDS free space of retained module strings In some cases processMultibulkBuffer uses sdsMakeRoomFor to expand the querybuf, but later in some cases it uses that query buffer as is for an argv element (see "Optimization"), which means that the sds in argv may have a lot of wasted space, and then in case modules keep that argv RedisString inside their data structure, this space waste will remain for long (until restarted from rdb).	2019-02-12 14:21:21 +01:00
Guy Benoish	b0c8d6c227	Increase string2ld's buffer size (and fix HINCRBYFLOAT) The string representation of `long double` may take up to ~5000 chars (see PR #3745). Before this fix HINCRBYFLOAT would never overflow (since the string could not exceed 256 chars). Now it can.	2019-01-28 17:58:11 +02:00
Guy Benoish	2502956835	ZPOP should return an empty array if COUNT=0	2019-01-23 11:11:57 +02:00
zhaozhao.zz	645d44d545	Streams: checkType before XGROUP CREATE Fix issue #5785, in case create group on a key is not stream.	2019-01-16 19:19:14 +08:00
antirez	8a87de130f	Modules shared API: export new core APIs.	2018-12-20 17:57:49 +01:00
antirez	d3eb0028e9	Modules shared API: also unregister the module as user.	2018-12-20 17:57:46 +01:00
antirez	9403b3d7a3	Modules shared API: prevent unloading of used modules.	2018-12-20 17:57:43 +01:00
antirez	6bb8cdaebe	Modules shared API: unregister APIs function.	2018-12-20 17:57:40 +01:00
antirez	27f6e9bb9b	Modules shared API: initial core functions. Based on ideas and code in PR #5560 by @MeirShpilraien.	2018-12-20 17:57:35 +01:00
antirez	850b64c116	Revert shared APIs to modify the design.	2018-12-20 17:56:38 +01:00
MeirShpilraien	ab37289fa6	added module ability to register api to be used by other modules	2018-12-20 17:55:18 +01:00
Guy Benoish	ed88f77d6d	Check server.verbosity in RM_LogRaw	2018-12-13 13:57:38 +01:00
Oran Agra	17c5f17686	Add log when server dies of SIGTERM during loading this is very confusing to see the server disappears as if it got SIGKILL when it was not the case.	2018-08-13 17:43:29 +03:00
Guy Benoish	8c8e85df87	Use memtoll() in 'CONFIG SET client-output-buffer-limit'	2018-04-18 13:01:53 +03:00
Guy Benoish	b660fc2fbe	Fix zlexrangespec mem-leak in genericZrangebylexCommand	2018-03-07 10:40:37 +07:00
Oran Agra	5def65008f	Fix zrealloc to behave similarly to je_realloc when size is 0 According to C11, the behavior of realloc with size 0 is now deprecated. it can either behave as free(ptr) and return NULL, or return a valid pointer. but in zmalloc it can lead to zmalloc_oom_handler and panic. and that can affect modules that use it. It looks like both glibc allocator and jemalloc behave like so: realloc(malloc(32),0) returns NULL realloc(NULL,0) returns a valid pointer This commit changes zmalloc to behave the same	2018-02-21 11:04:13 +02:00
James Rouzier	eca0187370	If the unit of a timeout is seconds treat it a float	2016-02-23 12:05:29 -05:00