Bump version because we added new features

Add a Gitter chat badge to README.md

.travis.yml

@@ -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

Dockerfile

@@ -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

README.md

@@ -1,3 +1,6 @@
+![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?
 --------------
 
@@ -9,6 +12,13 @@ KeyDB has full compatibility with the Redis protocol, modules, and scripts.  Thi
 
 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?
 ---------------
 
@@ -201,6 +211,26 @@ 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
 -----------------
 

deps/hiredis/sds.h

@@ -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)) {

redis.conf

@@ -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

src/Makefile

@@ -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)

src/ae_kqueue.c

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

src/aof.c

@@ -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);

src/cluster.c

@@ -4776,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;
 
@@ -4784,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:
      * ----------------+---------------------+---------------+
@@ -4842,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);
@@ -4915,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;
@@ -5203,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)));

src/config.c

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

src/fastlock.cpp

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

src/fastlock.h

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

src/fastlock_x64.asm

@@ -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

src/fmacros.h

@@ -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

src/module.c

@@ -3439,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
  * -------------------------------------------------------------------------- */
@@ -5195,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);

src/networking.cpp

@@ -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();
 }
 

src/rdb.c

@@ -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

src/rdb.h

@@ -144,9 +144,9 @@ int rdbSaveFile(char *filename, rdbSaveInfo *rsi);
 int rdbSaveFd(int fd, rdbSaveInfo *rsi);
 int rdbSaveS3(char *path, rdbSaveInfo *rsi);
 int rdbLoadS3(char *path, rdbSaveInfo *rsi);
-ssize_t rdbSaveObject(rio *rdb, robj *o);
+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);

src/redis-check-rdb.c

@@ -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++;

src/redismodule.h

@@ -278,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);
@@ -442,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);

src/replication.cpp

@@ -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);
     }
 }
@@ -1339,7 +1411,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
             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

src/sds.h

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

src/server.cpp

@@ -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 */
 
@@ -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);
     }
 
@@ -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;

src/server.h

@@ -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;
 };
 
@@ -1420,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;
 };
 

src/sort.c

@@ -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);

src/storage-lite.c

@@ -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

src/t_set.c

@@ -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

src/version.h

@@ -1 +1 @@
-#define REDIS_VERSION "0.9.2-r2"
+#define REDIS_VERSION "0.9.3"

src/zmalloc.c

@@ -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);