OS X build break

src/ae.cpp

@@ -30,6 +30,7 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
+#include "fmacros.h"
 #include <condition_variable>
 #include <atomic>
 #include <mutex>
@@ -48,6 +49,7 @@
 #include "fastlock.h"
 #include "zmalloc.h"
 #include "config.h"
+#include "mcs_lock.h"
 
 #ifdef USE_MUTEX
 thread_local int cOwnLock = 0;
@@ -80,7 +82,45 @@ public:
 mutex_wrapper g_lock;
 
 #else
-fastlock g_lock("AE (global)");
+//fastlock g_lock("AE (global)");
+McsLock g_lock;
+
+thread_local McsLock::node m_node;
+class McsLockGuard
+{
+    McsLock &m_lock;
+    bool fReleased = false;
+public:
+
+    McsLockGuard(McsLock &lock)
+        : m_lock(lock)
+    {
+        m_lock.lock(&m_node);
+    }
+
+    McsLockGuard(McsLock &lock, std::defer_lock_t)
+        : m_lock(lock)
+    {
+        fReleased = true;
+    }
+
+    void lock() {
+        m_lock.lock(&m_node);
+        fReleased = false;
+    }
+
+    void unlock() {
+        m_lock.unlock(&m_node);
+        fReleased = true;
+    }
+
+    ~McsLockGuard()
+    {
+        if (!fReleased)
+            m_lock.unlock(&m_node);
+    }
+};
+
 #endif
 thread_local aeEventLoop *g_eventLoopThisThread = NULL;
 
@@ -167,7 +207,7 @@ void aeProcessCmd(aeEventLoop *eventLoop, int fd, void *, int )
 
         case AE_ASYNC_OP::PostFunction:
             {
-            std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
+            McsLockGuard ulock(g_lock, std::defer_lock);
             if (cmd.fLock)
                 ulock.lock();
             ((aePostFunctionProc*)cmd.proc)(cmd.clientData);
@@ -179,7 +219,7 @@ void aeProcessCmd(aeEventLoop *eventLoop, int fd, void *, int )
             if (cmd.pctl != nullptr)
                 cmd.pctl->mutexcv.lock();
             
-            std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
+            McsLockGuard ulock(g_lock, std::defer_lock);
             if (cmd.fLock)
                 ulock.lock();
             (*cmd.pfn)();
@@ -585,7 +625,7 @@ static aeTimeEvent *aeSearchNearestTimer(aeEventLoop *eventLoop)
 
 /* Process time events */
 static int processTimeEvents(aeEventLoop *eventLoop) {
-    std::unique_lock<decltype(g_lock)> ulock(g_lock);
+    McsLockGuard ulock(g_lock);
     int processed = 0;
     aeTimeEvent *te;
     long long maxId;
@@ -671,7 +711,7 @@ static int processTimeEvents(aeEventLoop *eventLoop) {
 extern "C" void ProcessEventCore(aeEventLoop *eventLoop, aeFileEvent *fe, int mask, int fd)
 {
 #define LOCK_IF_NECESSARY(fe, tsmask) \
-    std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock); \
+    McsLockGuard ulock(g_lock, std::defer_lock); \
     if (!(fe->mask & tsmask)) \
         ulock.lock()
 
@@ -798,7 +838,7 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
         }
 
         if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP) {
-            std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
+            McsLockGuard ulock(g_lock, std::defer_lock);
             if (!(eventLoop->beforesleepFlags & AE_SLEEP_THREADSAFE))
                 ulock.lock();
             eventLoop->beforesleep(eventLoop);
@@ -810,7 +850,7 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
 
         /* After sleep callback. */
         if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP) {
-            std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
+            McsLockGuard ulock(g_lock, std::defer_lock);
             if (!(eventLoop->aftersleepFlags & AE_SLEEP_THREADSAFE))
                 ulock.lock();
             eventLoop->aftersleep(eventLoop);
@@ -861,7 +901,7 @@ void aeMain(aeEventLoop *eventLoop) {
     g_eventLoopThisThread = eventLoop;
     while (!eventLoop->stop) {
         if (eventLoop->beforesleep != NULL) {
-            std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
+            McsLockGuard ulock(g_lock, std::defer_lock);
             if (!(eventLoop->beforesleepFlags & AE_SLEEP_THREADSAFE))
                 ulock.lock();
             eventLoop->beforesleep(eventLoop);
@@ -888,20 +928,20 @@ void aeSetAfterSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *aftersleep,
 
 void aeAcquireLock()
 {
-    g_lock.lock();
+    g_lock.lock(&m_node);
 }
 
 int aeTryAcquireLock(int fWeak)
 {
-    return g_lock.try_lock(!!fWeak);
+    return g_lock.try_lock(&m_node, fWeak);
 }
 
 void aeReleaseLock()
 {
-    g_lock.unlock();
+    g_lock.unlock(&m_node);
 }
 
 int aeThreadOwnsLock()
 {
-    return g_lock.fOwnLock();
+    return m_node.depth > 0;
 }

src/fastlock.cpp

@@ -47,6 +47,7 @@
 #include <stdio.h>
 #include "config.h"
 #include "serverassert.h"
+#include "mcs_lock.h"
 
 #ifdef __APPLE__
 #include <TargetConditionals.h>
@@ -73,6 +74,12 @@ extern int g_fInCrash;
 extern int g_fTestMode;
 int g_fHighCpuPressure = false;
 
+#if defined(__i386__) || defined(__amd64__)
+#define asm_yield() __asm__ __volatile__ ("pause");
+#elif defined(__aarch64__)
+#define asm_yield() __asm__ __volatile__ ("yield");
+#endif
+
 /****************************************************
  *
  *      Implementation of a fair spinlock.  To promote fairness we
@@ -310,6 +317,11 @@ extern "C" void fastlock_sleep(fastlock *lock, pid_t pid, unsigned wake, unsigne
     
     __atomic_fetch_and(&lock->futex, ~mask, __ATOMIC_RELEASE);
     g_dlock.clearwait(lock, pid);
+#else
+    UNUSED(lock);
+    UNUSED(pid);
+    UNUSED(wake);
+    UNUSED(myticket);
 #endif
     __atomic_fetch_add(&g_longwaits, 1, __ATOMIC_RELAXED);
 }
@@ -351,11 +363,7 @@ extern "C" void fastlock_lock(struct fastlock *lock)
         if ((ticketT.u & 0xffff) == myticket)
             break;
 
-#if defined(__i386__) || defined(__amd64__)
-        __asm__ __volatile__ ("pause");
-#elif defined(__aarch64__)
-        __asm__ __volatile__ ("yield");
-#endif
+        asm_yield();
         if ((++cloops % loopLimit) == 0)
         {
             fastlock_sleep(lock, tid, ticketT.u, myticket);
@@ -493,4 +501,114 @@ void fastlock_auto_adjust_waits()
 #else
     g_fHighCpuPressure = g_fTestMode;
 #endif
+}
+
+#ifdef ASM_SPINLOCK
+extern "C" int mcs_spin_core(unsigned *locked, unsigned limit);
+#endif
+
+void McsLock::lock(node *pnode)
+{
+    serverAssert(pnode->depth >= 0);
+    if (pnode->depth > 0) {
+        ++pnode->depth;
+        return;
+    }
+
+    pnode->pnext = nullptr;
+    pnode->depth = 0;
+    
+    McsLock::node *predecessor = nullptr;
+    __atomic_exchange(&m_root, &pnode, &predecessor, __ATOMIC_ACQ_REL);
+    
+    //if its null, we now own the lock and can leave, else....
+    if (predecessor != nullptr){
+        //when the predecessor unlocks, it will give us the lock
+        pnode->locked = MCS_LOCKED;
+        __atomic_store(&predecessor->pnext, &pnode, __ATOMIC_RELEASE);
+
+#ifdef __linux__
+        unsigned loopLimit = g_fHighCpuPressure ? 0x10000 : 0x100000;
+        unsigned loopIter = 0;
+#endif
+
+        for (;;) {
+            unsigned locked;
+            __atomic_load(&pnode->locked, &locked, __ATOMIC_ACQUIRE);
+            if (!locked)
+                break;
+#ifdef __linux__
+            ++loopIter;
+            if (loopIter == loopLimit) {
+                unsigned lockedSet = MCS_FUTEX_LOCKED;
+                unsigned lockedExpect = MCS_LOCKED;
+                if (__atomic_compare_exchange(&pnode->locked, &lockedExpect, &lockedSet, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED) || lockedExpect == MCS_FUTEX_LOCKED) {
+                    futex(&pnode->locked, FUTEX_WAIT, MCS_FUTEX_LOCKED, nullptr, 0);
+                }
+                loopIter = 0;
+            } 
+            else 
+#endif
+            {
+                asm_yield();
+            }
+        }
+    }
+    ANNOTATE_RWLOCK_ACQUIRED(this, true);
+    pnode->depth = 1;
+}
+
+bool McsLock::try_lock(node *pnode, bool fWeak)
+{
+    if (pnode->depth > 0) {
+        ++pnode->depth;
+        return true;
+    }
+
+    pnode->pnext = nullptr;
+    pnode->depth = 0;
+    unsigned lockedSet = MCS_LOCKED;
+    __atomic_store(&pnode->locked, &lockedSet, __ATOMIC_RELEASE);
+
+    McsLock::node *expected = nullptr;
+    if (!__atomic_compare_exchange(&m_root, &expected, &pnode, fWeak, __ATOMIC_ACQ_REL, __ATOMIC_RELAXED))
+    {
+        return false;
+    }
+    pnode->depth = 1;
+    ANNOTATE_RWLOCK_ACQUIRED(this, true);
+    return true;
+}
+
+void McsLock::unlock(node *pnode)
+{
+    serverAssert(pnode->depth > 0);
+    pnode->depth--;
+    if (pnode->depth == 0)
+    {
+        McsLock::node *pnext;
+        __atomic_load(&pnode->pnext, &pnext, __ATOMIC_ACQUIRE);
+        if (pnext == nullptr) {
+            node *desired = nullptr;
+            auto *nodeT = pnode;
+            if (__atomic_compare_exchange(&m_root, &nodeT, &desired, false, __ATOMIC_RELEASE, __ATOMIC_ACQUIRE)) {
+                return;
+            }
+
+            do {
+                __atomic_load(&pnode->pnext, &pnext, __ATOMIC_ACQUIRE);
+            } while(pnext == nullptr);
+        }
+
+        unsigned lockedSet = MCS_UNLOCKED;
+        unsigned lockedActual;
+        __atomic_exchange(&pnext->locked, &lockedSet, &lockedActual, __ATOMIC_RELEASE);
+        ANNOTATE_RWLOCK_RELEASED(this, true);
+        
+#ifdef __linux__
+        if (lockedActual == MCS_FUTEX_LOCKED) {
+            futex(&pnext->locked, FUTEX_WAKE, INT_MAX, nullptr, 0);
+        }
+#endif
+    }
 }

src/mcs_lock.h

@@ -0,0 +1,27 @@
+#pragma once
+#include <atomic>
+#include <stdio.h>
+#include "serverassert.h"
+
+#define MCS_UNLOCKED 0
+#define MCS_LOCKED 1
+#define MCS_FUTEX_LOCKED 2
+class McsLock
+{
+public:
+    struct node 
+    {
+        char front_padding[64];
+        struct node* pnext = nullptr;
+        int depth = 0;
+        unsigned locked = false;
+        char back_padding[64];
+    };
+
+    void lock(McsLock::node *pnode);
+    bool try_lock(McsLock::node *pnode, bool fWeak);
+    void unlock(McsLock::node *pnode);
+
+private:
+    McsLock::node *m_root = nullptr;
+};