Merge branch 'unstable' into RELEASE_5

src/Makefile

@@ -47,8 +47,6 @@ endif
 USEASM?=true
 
 ifneq ($(SANITIZE),)
-	CC=clang
-	CXX=clang++
 	CFLAGS+= -fsanitize=$(SANITIZE) -DSANITIZE
 	CXXFLAGS+= -fsanitize=$(SANITIZE) -DSANITIZE
 	LDFLAGS+= -fsanitize=$(SANITIZE)
@@ -190,7 +188,7 @@ endif
 
 REDIS_CC=$(QUIET_CC)$(CC) $(FINAL_CFLAGS)
 REDIS_CXX=$(QUIET_CC)$(CC) $(FINAL_CXXFLAGS)
-REDIS_NASM=$(QUIET_CC)nasm -felf64 
+KEYDB_AS=$(QUIET_CC) as --64 -g
 REDIS_LD=$(QUIET_LINK)$(CXX) $(FINAL_LDFLAGS)
 REDIS_INSTALL=$(QUIET_INSTALL)$(INSTALL)
 
@@ -295,7 +293,7 @@ dict-benchmark: dict.cpp zmalloc.cpp sds.c siphash.c
 	$(REDIS_CXX) -c $<
 
 %.o: %.asm .make-prerequisites
-	$(REDIS_NASM) $<
+	$(KEYDB_AS) $< -o $@
 
 clean:
 	rm -rf $(REDIS_SERVER_NAME) $(REDIS_SENTINEL_NAME) $(REDIS_CLI_NAME) $(REDIS_BENCHMARK_NAME) $(REDIS_CHECK_RDB_NAME) $(REDIS_CHECK_AOF_NAME) *.o *.gcda *.gcno *.gcov redis.info lcov-html Makefile.dep dict-benchmark

src/ae.cpp

@@ -80,7 +80,7 @@ class mutex_wrapper
 mutex_wrapper g_lock;
 
 #else
-fastlock g_lock;
+fastlock g_lock("AE (global)");
 #endif
 thread_local aeEventLoop *g_eventLoopThisThread = NULL;
 
@@ -327,7 +327,7 @@ aeEventLoop *aeCreateEventLoop(int setsize) {
     for (i = 0; i < setsize; i++)
         eventLoop->events[i].mask = AE_NONE;
 
-    fastlock_init(&eventLoop->flock);
+    fastlock_init(&eventLoop->flock, "event loop");
     int rgfd[2];
     if (pipe(rgfd) < 0)
         goto err;

src/aof.cpp

@@ -678,7 +678,7 @@ client *createFakeClient(void) {
     c->puser = NULL;
     listSetFreeMethod(c->reply,freeClientReplyValue);
     listSetDupMethod(c->reply,dupClientReplyValue);
-    fastlock_init(&c->lock);
+    fastlock_init(&c->lock, "fake client");
     fastlock_lock(&c->lock);
     initClientMultiState(c);
     return c;

src/db.cpp

@@ -643,10 +643,6 @@ void keysCommand(client *c) {
     unsigned long numkeys = 0;
     void *replylen = addReplyDeferredLen(c);
 
-#ifdef MULTITHREADED_KEYS
-    aeReleaseLock();
-#endif
-
     di = dictGetSafeIterator(c->db->pdict);
     allkeys = (pattern[0] == '*' && pattern[1] == '\0');
     while((de = dictNext(di)) != NULL) {
@@ -664,14 +660,6 @@ void keysCommand(client *c) {
     }
     dictReleaseIterator(di);
     setDeferredArrayLen(c,replylen,numkeys);
-
-#ifdef MULTITHREADED_KEYS
-    fastlock_unlock(&c->db->lock);  // we must release the DB lock before acquiring the AE lock to prevent deadlocks
-    AeLocker lock;
-    lock.arm(c);
-    fastlock_lock(&c->db->lock);    // we still need the DB lock
-    lock.release();
-#endif
 }
 
 /* This callback is used by scanGenericCommand in order to collect elements

src/debug.cpp

@@ -55,7 +55,7 @@ typedef ucontext_t sigcontext_t;
 #endif
 #endif
 
-bool g_fInCrash = false;
+int g_fInCrash = false;
 
 /* ================================= Debugging ============================== */
 

src/fastlock.cpp

@@ -36,10 +36,13 @@
 #include <assert.h>
 #include <pthread.h>
 #include <limits.h>
+#include <map>
 #ifdef __linux__
 #include <linux/futex.h>
 #endif
 #include <string.h>
+#include <stdarg.h>
+#include <stdio.h>
 
 #ifdef __APPLE__
 #include <TargetConditionals.h>
@@ -57,6 +60,7 @@
 #define UNUSED(x) ((void)x)
 #endif
 
+extern int g_fInCrash;
 
 /****************************************************
  *
@@ -125,6 +129,80 @@
 
 #endif
 
+#pragma weak _serverPanic
+extern "C"  __attribute__((weak)) void _serverPanic(const char * /*file*/, int /*line*/, const char * /*msg*/, ...)
+{
+    *((char*)-1) = 'x';
+}
+
+#pragma weak serverLog
+__attribute__((weak)) void serverLog(int , const char *fmt, ...)
+{
+    va_list args;
+    va_start(args, fmt);
+    vprintf(fmt, args);
+    va_end(args);
+    printf("\n");
+}
+
+class DeadlockDetector
+{
+    std::map<pid_t, fastlock *> m_mapwait;
+    fastlock m_lock { "deadlock detector" };
+public:
+    void registerwait(fastlock *lock, pid_t thispid)
+    {
+        if (lock == &m_lock || g_fInCrash)
+            return;
+        fastlock_lock(&m_lock);
+        m_mapwait.insert(std::make_pair(thispid, lock));
+
+        // Detect cycles
+        pid_t pidCheck = thispid;
+        size_t cchecks = 0;
+        for (;;)
+        {
+            auto itr = m_mapwait.find(pidCheck);
+            if (itr == m_mapwait.end())
+                break;
+            pidCheck = itr->second->m_pidOwner;
+            if (pidCheck == thispid)
+            {
+                // Deadlock detected, printout some debugging info and crash
+                serverLog(3 /*LL_WARNING*/, "\n\n");
+                serverLog(3 /*LL_WARNING*/, "!!! ERROR: Deadlock detected !!!");
+                pidCheck = thispid;
+                for (;;)
+                {
+                    auto itr = m_mapwait.find(pidCheck);
+                    serverLog(3 /* LL_WARNING */, "\t%d: (%p) %s", pidCheck, itr->second, itr->second->szName);
+                    pidCheck = itr->second->m_pidOwner;
+                    if (pidCheck == thispid)
+                        break;
+                }
+                serverLog(3 /*LL_WARNING*/, "!!! KeyDB Will Now Crash !!!");
+                _serverPanic(__FILE__, __LINE__, "Deadlock detected");
+            }
+
+            if (cchecks > m_mapwait.size())
+                break;  // There is a cycle but we're not in it
+            ++cchecks;
+        }
+        fastlock_unlock(&m_lock);
+    }
+
+    void clearwait(fastlock *lock, pid_t thispid)
+    {
+        if (lock == &m_lock || g_fInCrash)
+            return;
+        fastlock_lock(&m_lock);
+        m_mapwait.erase(thispid);
+        fastlock_unlock(&m_lock);
+    }
+};
+
+DeadlockDetector g_dlock;
+
 static_assert(sizeof(pid_t) <= sizeof(fastlock::m_pidOwner), "fastlock::m_pidOwner not large enough");
 uint64_t g_longwaits = 0;
 
@@ -135,7 +213,6 @@ uint64_t fastlock_getlongwaitcount()
     return rval;
 }
 
-#ifndef ASM_SPINLOCK
 #ifdef __linux__
 static int futex(volatile unsigned *uaddr, int futex_op, int val,
     const struct timespec *timeout, int val3)
@@ -144,7 +221,6 @@ static int futex(volatile unsigned *uaddr, int futex_op, int val,
                     timeout, uaddr, val3);
 }
 #endif
-#endif
 
 extern "C" pid_t gettid()
 {
@@ -163,13 +239,26 @@ extern "C" pid_t gettid()
     return pidCache;
 }
 
-extern "C" void fastlock_init(struct fastlock *lock)
+extern "C" void fastlock_sleep(fastlock *lock, pid_t pid, unsigned wake, unsigned mask)
+{
+#ifdef __linux__
+    g_dlock.registerwait(lock, pid);
+    __atomic_fetch_or(&lock->futex, mask, __ATOMIC_ACQUIRE);
+    futex(&lock->m_ticket.u, FUTEX_WAIT_BITSET_PRIVATE, wake, nullptr, mask);
+    __atomic_fetch_and(&lock->futex, ~mask, __ATOMIC_RELEASE);
+    g_dlock.clearwait(lock, pid);
+#endif
+    __atomic_fetch_add(&g_longwaits, 1, __ATOMIC_RELAXED);
+}
+
+extern "C" void fastlock_init(struct fastlock *lock, const char *name)
 {
     lock->m_ticket.m_active = 0;
     lock->m_ticket.m_avail = 0;
     lock->m_depth = 0;
     lock->m_pidOwner = -1;
     lock->futex = 0;
+    lock->szName = name;
     ANNOTATE_RWLOCK_CREATE(lock);
 }
 
@@ -184,12 +273,12 @@ extern "C" void fastlock_lock(struct fastlock *lock)
         return;
     }
 
+    int tid = gettid();
     unsigned myticket = __atomic_fetch_add(&lock->m_ticket.m_avail, 1, __ATOMIC_RELEASE);
-#ifdef __linux__
     unsigned mask = (1U << (myticket % 32));
-#endif
     int cloops = 0;
     ticket ticketT;
+
     for (;;)
     {
         __atomic_load(&lock->m_ticket.u, &ticketT.u, __ATOMIC_ACQUIRE);
@@ -201,17 +290,11 @@ extern "C" void fastlock_lock(struct fastlock *lock)
 #endif
         if ((++cloops % 1024*1024) == 0)
         {
-#ifdef __linux__
-            __atomic_fetch_or(&lock->futex, mask, __ATOMIC_ACQUIRE);
-            futex(&lock->m_ticket.u, FUTEX_WAIT_BITSET_PRIVATE, ticketT.u, nullptr, mask);
-            __atomic_fetch_and(&lock->futex, ~mask, __ATOMIC_RELEASE);
-#endif
-            __atomic_fetch_add(&g_longwaits, 1, __ATOMIC_RELAXED);
+            fastlock_sleep(lock, tid, ticketT.u, mask);
         }
     }
 
     lock->m_depth = 1;
-    int tid = gettid();
     __atomic_store(&lock->m_pidOwner, &tid, __ATOMIC_RELEASE);
     ANNOTATE_RWLOCK_ACQUIRED(lock, true);
     std::atomic_thread_fence(std::memory_order_acquire);

src/fastlock.h

@@ -7,7 +7,7 @@ extern "C" {
 
 /* Begin C API */
 struct fastlock;
-void fastlock_init(struct fastlock *lock);
+void fastlock_init(struct fastlock *lock, const char *name);
 void fastlock_lock(struct fastlock *lock);
 int fastlock_trylock(struct fastlock *lock, int fWeak);
 void fastlock_unlock(struct fastlock *lock);
@@ -45,24 +45,25 @@ struct fastlock
     volatile int m_pidOwner;
     volatile int m_depth;
     unsigned futex;
+    const char *szName;
 
 #ifdef __cplusplus
-    fastlock()
+    fastlock(const char *name)
     {
-        fastlock_init(this);
+        fastlock_init(this, name);
     }
 
-    void lock()
+    inline void lock()
     {
         fastlock_lock(this);
     }
 
-    bool try_lock(bool fWeak = false)
+    inline bool try_lock(bool fWeak = false)
     {
         return !!fastlock_trylock(this, fWeak);
     }
 
-    void unlock()
+    inline void unlock()
     {
         fastlock_unlock(this);
     }