Fix bvar worker starvation with recursive bthread mutex

BUILD.bazel

@@ -370,6 +370,23 @@ cc_library(
     }),
 )
 
+cc_library(
+    # Header-only interface so :bvar can use RecursiveMutex without depending on
+    # :bthread (which itself depends on :bvar). Implementation lives in :bthread.
+    name = "bthread_recursive_mutex",
+    hdrs = [
+        "src/bthread/recursive_mutex.h",
+        "src/bthread/types.h",
+    ],
+    includes = [
+        "src/",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":butil",
+    ],
+)
+
 cc_library(
     name = "bvar",
     srcs = glob(
@@ -405,6 +422,7 @@ cc_library(
     linkopts = LINKOPTS,
     visibility = ["//visibility:public"],
     deps = [
+        ":bthread_recursive_mutex",
         ":butil",
     ]  + select({
         "//bazel/config:with_babylon_counter": ["@babylon//:concurrent_counter"],
@@ -417,10 +435,16 @@ cc_library(
     srcs = glob([
         "src/bthread/*.cpp",
     ]),
-    hdrs = glob([
-        "src/bthread/*.h",
-        "src/bthread/*.list",
-    ]),
+    hdrs = glob(
+        [
+            "src/bthread/*.h",
+            "src/bthread/*.list",
+        ],
+        exclude = [
+            "src/bthread/recursive_mutex.h",
+            "src/bthread/types.h",
+        ],
+    ),
     defines = [] + select({
         "//bazel/config:brpc_with_bthread_tracer": [
             "BRPC_BTHREAD_TRACER",
@@ -434,6 +458,7 @@ cc_library(
     linkopts = LINKOPTS,
     visibility = ["//visibility:public"],
     deps = [
+        ":bthread_recursive_mutex",
         ":butil",
         ":bvar",
     ] + select({

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

@@ -22,7 +22,7 @@
 #ifndef  BTHREAD_MUTEX_H
 #define  BTHREAD_MUTEX_H
 
-#include "bthread/types.h"
+#include "bthread/recursive_mutex.h"
 #include "butil/scoped_lock.h"
 #include "bvar/utils/lock_timer.h"
 

src/bthread/recursive_mutex.h

@@ -0,0 +1,78 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// bthread - An M:N threading library to make applications more concurrent.
+
+// Split from mutex.h so bvar can use RecursiveMutex without depending on the
+// full bthread mutex header (which includes bvar utilities and would form a
+// bazel dependency cycle: bvar -> bthread -> bvar).
+
+#ifndef BTHREAD_RECURSIVE_MUTEX_H
+#define BTHREAD_RECURSIVE_MUTEX_H
+
+#include <system_error>
+
+#include "bthread/types.h"
+#include "butil/macros.h"
+
+__BEGIN_DECLS
+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);
+__END_DECLS
+
+namespace bthread {
+
+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 bthread
+
+#endif  // BTHREAD_RECURSIVE_MUTEX_H

src/bthread/types.h

@@ -197,6 +197,19 @@ typedef struct bthread_mutex_t {
     mutex_owner_t owner;
 } bthread_mutex_t;
 
+typedef struct bthread_recursive_mutex_t {
+#if defined(__cplusplus)
+    bthread_recursive_mutex_t()
+        : mutex(), owner(0), owner_kind(0), recursion(0) {}
+
+    DISALLOW_COPY_AND_ASSIGN(bthread_recursive_mutex_t);
+#endif
+    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/recursive_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/BUILD.bazel

@@ -185,7 +185,9 @@ cc_test(
 
 cc_test(
     name = "bvar_unittests",
-    srcs = glob(["bvar_*_unittest.cpp"]),
+    srcs = glob(["bvar_*_unittest.cpp"]) + [
+        "bvar_bthread_stubs.cpp",
+    ],
     deps = [
         ":gperftools_helper",
         "//:bvar",
@@ -223,11 +225,29 @@ root_runfiles(
     ],
 )
 
+# brpc_bvar_mutex_unittest provides its own main() for fork/exec child modes.
+cc_test(
+    name = "brpc_bvar_mutex_unittest",
+    srcs = ["brpc_bvar_mutex_unittest.cpp"],
+    copts = COPTS,
+    deps = [
+        ":gperftools_helper",
+        "//:brpc",
+        ":cc_test_proto",
+        "@com_google_googletest//:gtest",
+    ] + TCMALLOC_DEP_UNLESS_ASAN,
+)
+
 generate_unittests(
     name = "brpc_unittests",
-    srcs = glob([
-        "brpc_*_unittest.cpp",
-    ]),
+    srcs = glob(
+        [
+            "brpc_*_unittest.cpp",
+        ],
+        exclude = [
+            "brpc_bvar_mutex_unittest.cpp",
+        ],
+    ),
     deps = [
         ":gperftools_helper",
         "//:brpc",

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