Merge branch 'unstable' into RELEASE_0_9

.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 [email protected]: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,22 +58,22 @@ 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);
     zfree(state);
 }
 
 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,15 +4776,15 @@ 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;
 
     /* Serialize the object in a RDB-like format. It consist of an object type
      * 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 @@ void migrateCommand(client *c) {
 
         /* 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
 }