[#2888] Fix bvar worker starvation with recursive bthread mutex

src/bthread/mutex.cpp

@@ -1190,6 +1190,46 @@ bool FastPthreadMutex::timed_lock(const struct timespec* abstime) {
 }
 #endif // BTHREAD_USE_FAST_PTHREAD_MUTEX HAS_PTHREAD_MUTEX_TIMEDLOCK
 
+enum RecursiveMutexOwnerKind {
+    RECURSIVE_MUTEX_UNOWNED = 0,
+    RECURSIVE_MUTEX_BTHREAD = 1,
+    RECURSIVE_MUTEX_PTHREAD = 2,
+};
+
+struct RecursiveMutexOwner {
+    uint64_t id;
+    uint32_t kind;
+};
+
+static __thread char tls_recursive_mutex_owner;
+
+static RecursiveMutexOwner current_recursive_mutex_owner() {
+    const bthread_t tid = bthread_self();
+    if (tid != INVALID_BTHREAD) {
+        return {tid, RECURSIVE_MUTEX_BTHREAD};
+    }
+    return {
+        static_cast<uint64_t>(
+            reinterpret_cast<uintptr_t>(&tls_recursive_mutex_owner)),
+        RECURSIVE_MUTEX_PTHREAD};
+}
+
+static bool recursive_mutex_owned_by(
+    const bthread_recursive_mutex_t* mutex,
+    const RecursiveMutexOwner& owner) {
+    return __atomic_load_n(&mutex->owner_kind, __ATOMIC_ACQUIRE) ==
+               owner.kind &&
+           __atomic_load_n(&mutex->owner, __ATOMIC_RELAXED) == owner.id;
+}
+
+static void set_recursive_mutex_owner(
+    bthread_recursive_mutex_t* mutex,
+    const RecursiveMutexOwner& owner) {
+    __atomic_store_n(&mutex->owner, owner.id, __ATOMIC_RELAXED);
+    mutex->recursion = 1;
+    __atomic_store_n(&mutex->owner_kind, owner.kind, __ATOMIC_RELEASE);
+}
+
 } // namespace bthread
 
 __BEGIN_DECLS
@@ -1288,6 +1328,69 @@ int bthread_mutex_unlock(bthread_mutex_t* m) {
     return 0;
 }
 
+int bthread_recursive_mutex_init(bthread_recursive_mutex_t* mutex) {
+    mutex->owner = 0;
+    mutex->owner_kind = bthread::RECURSIVE_MUTEX_UNOWNED;
+    mutex->recursion = 0;
+    return bthread_mutex_init(&mutex->mutex, NULL);
+}
+
+int bthread_recursive_mutex_destroy(bthread_recursive_mutex_t* mutex) {
+    if (__atomic_load_n(&mutex->owner_kind, __ATOMIC_ACQUIRE) !=
+        bthread::RECURSIVE_MUTEX_UNOWNED) {
+        return EBUSY;
+    }
+    return bthread_mutex_destroy(&mutex->mutex);
+}
+
+int bthread_recursive_mutex_trylock(bthread_recursive_mutex_t* mutex) {
+    const bthread::RecursiveMutexOwner owner =
+        bthread::current_recursive_mutex_owner();
+    if (bthread::recursive_mutex_owned_by(mutex, owner)) {
+        if (mutex->recursion == UINT32_MAX) {
+            return EAGAIN;
+        }
+        ++mutex->recursion;
+        return 0;
+    }
+    const int rc = bthread_mutex_trylock(&mutex->mutex);
+    if (rc == 0) {
+        bthread::set_recursive_mutex_owner(mutex, owner);
+    }
+    return rc;
+}
+
+int bthread_recursive_mutex_lock(bthread_recursive_mutex_t* mutex) {
+    const bthread::RecursiveMutexOwner owner =
+        bthread::current_recursive_mutex_owner();
+    if (bthread::recursive_mutex_owned_by(mutex, owner)) {
+        if (mutex->recursion == UINT32_MAX) {
+            return EAGAIN;
+        }
+        ++mutex->recursion;
+        return 0;
+    }
+    const int rc = bthread_mutex_lock(&mutex->mutex);
+    if (rc == 0) {
+        bthread::set_recursive_mutex_owner(mutex, owner);
+    }
+    return rc;
+}
+
+int bthread_recursive_mutex_unlock(bthread_recursive_mutex_t* mutex) {
+    const bthread::RecursiveMutexOwner owner =
+        bthread::current_recursive_mutex_owner();
+    if (!bthread::recursive_mutex_owned_by(mutex, owner)) {
+        return EPERM;
+    }
+    if (--mutex->recursion != 0) {
+        return 0;
+    }
+    __atomic_store_n(&mutex->owner_kind, bthread::RECURSIVE_MUTEX_UNOWNED,
+                     __ATOMIC_RELEASE);
+    return bthread_mutex_unlock(&mutex->mutex);
+}
+
 int bthread_mutexattr_init(bthread_mutexattr_t* attr) {
     attr->enable_csite = true;
     return 0;

src/bthread/mutex.h

@@ -35,6 +35,11 @@ extern int bthread_mutex_lock(bthread_mutex_t* mutex);
 extern int bthread_mutex_timedlock(bthread_mutex_t* __restrict mutex,
                                    const struct timespec* __restrict abstime);
 extern int bthread_mutex_unlock(bthread_mutex_t* mutex);
+extern int bthread_recursive_mutex_init(bthread_recursive_mutex_t* mutex);
+extern int bthread_recursive_mutex_destroy(bthread_recursive_mutex_t* mutex);
+extern int bthread_recursive_mutex_trylock(bthread_recursive_mutex_t* mutex);
+extern int bthread_recursive_mutex_lock(bthread_recursive_mutex_t* mutex);
+extern int bthread_recursive_mutex_unlock(bthread_recursive_mutex_t* mutex);
 extern bthread_t bthread_self(void);
 __END_DECLS
 
@@ -73,6 +78,40 @@ class Mutex {
     bthread_mutex_t _mutex;   
 };
 
+class RecursiveMutex {
+public:
+    typedef bthread_recursive_mutex_t* native_handler_type;
+
+    RecursiveMutex() {
+        const int ec = bthread_recursive_mutex_init(&_mutex);
+        if (ec != 0) {
+            throw std::system_error(
+                std::error_code(ec, std::system_category()),
+                "RecursiveMutex constructor failed");
+        }
+    }
+    ~RecursiveMutex() {
+        CHECK_EQ(0, bthread_recursive_mutex_destroy(&_mutex));
+    }
+    native_handler_type native_handler() { return &_mutex; }
+    void lock() {
+        const int ec = bthread_recursive_mutex_lock(&_mutex);
+        if (ec != 0) {
+            throw std::system_error(
+                std::error_code(ec, std::system_category()),
+                "RecursiveMutex lock failed");
+        }
+    }
+    void unlock() { CHECK_EQ(0, bthread_recursive_mutex_unlock(&_mutex)); }
+    bool try_lock() {
+        return bthread_recursive_mutex_trylock(&_mutex) == 0;
+    }
+
+private:
+    DISALLOW_COPY_AND_ASSIGN(RecursiveMutex);
+    bthread_recursive_mutex_t _mutex;
+};
+
 namespace internal {
 #ifdef BTHREAD_USE_FAST_PTHREAD_MUTEX
 class FastPthreadMutex {

src/bthread/types.h

@@ -197,6 +197,13 @@ typedef struct bthread_mutex_t {
     mutex_owner_t owner;
 } bthread_mutex_t;
 
+typedef struct {
+    bthread_mutex_t mutex;
+    uint64_t owner;
+    uint32_t owner_kind;
+    uint32_t recursion;
+} bthread_recursive_mutex_t;
+
 typedef struct {
     bool enable_csite;
 } bthread_mutexattr_t;

src/bvar/variable.cpp

@@ -31,6 +31,7 @@
 #include "butil/file_util.h"                     // butil::FilePath
 #include "butil/threading/platform_thread.h"
 #include "butil/reloadable_flags.h"
+#include "bthread/mutex.h"
 #include "bvar/gflag.h"
 #include "bvar/variable.h"
 #include "bvar/mvariable.h"
@@ -73,18 +74,12 @@ class VarEntry {
 typedef butil::FlatMap<std::string, VarEntry> VarMap;
 
 struct VarMapWithLock : public VarMap {
-    pthread_mutex_t mutex;
+    bthread::RecursiveMutex mutex;
 
     VarMapWithLock() {
         if (init(1024) != 0) {
             LOG(WARNING) << "Fail to init VarMap";
         }
-
-        pthread_mutexattr_t attr;
-        pthread_mutexattr_init(&attr);
-        pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);
-        pthread_mutex_init(&mutex, &attr);
-        pthread_mutexattr_destroy(&attr);
     }
 };
 
@@ -212,7 +207,7 @@ void Variable::list_exposed(std::vector<std::string>* names,
     VarMapWithLock* var_maps = get_var_maps();
     for (size_t i = 0; i < SUB_MAP_COUNT; ++i) {
         VarMapWithLock& m = var_maps[i];
-        std::unique_lock<pthread_mutex_t> mu(m.mutex);
+        std::unique_lock<bthread::RecursiveMutex> mu(m.mutex);
         size_t n = 0;
         for (VarMap::const_iterator it = m.begin(); it != m.end(); ++it) {
             if (++n >= 256/*max iterated one pass*/) {

test/CMakeLists.txt

@@ -245,7 +245,9 @@ list(REMOVE_ITEM BVAR_SOURCES ${PROJECT_SOURCE_DIR}/src/bvar/default_variables.c
 add_library(BVAR_DEBUG_LIB OBJECT ${BVAR_SOURCES})
 target_link_libraries(BVAR_DEBUG_LIB PRIVATE brpc_test_config)
 file(GLOB TEST_BVAR_SRCS "bvar_*_unittest.cpp")
+# Keep test_bvar independent of the bthread runtime and pthread interceptors.
 add_executable(test_bvar ${TEST_BVAR_SRCS}
+                         bvar_bthread_stubs.cpp
                          $<TARGET_OBJECTS:BVAR_DEBUG_LIB>
                          $<TARGET_OBJECTS:BUTIL_DEBUG_LIB>)
 target_link_libraries(test_bvar PRIVATE
@@ -272,9 +274,14 @@ file(GLOB BRPC_UNITTESTS "brpc_*_unittest.cpp")
 foreach(BRPC_UT ${BRPC_UNITTESTS})
     get_filename_component(BRPC_UT_WE ${BRPC_UT} NAME_WE)
     add_executable(${BRPC_UT_WE} ${BRPC_UT} $<TARGET_OBJECTS:TEST_PROTO_LIB>)
+    if(BRPC_UT_WE STREQUAL "brpc_bvar_mutex_unittest")
+        set(BRPC_GTEST_LIBRARY gtest)
+    else()
+        set(BRPC_GTEST_LIBRARY gtest_main)
+    endif()
     target_link_libraries(${BRPC_UT_WE} PRIVATE
         brpc-shared-debug
-        gtest_main
+        ${BRPC_GTEST_LIBRARY}
         brpc_test_config
         ${GPERFTOOLS_LIBRARIES})
     add_test(NAME ${BRPC_UT_WE} COMMAND ${BRPC_UT_WE})