From 9c1b7671f0e95e4ec2f24baacd4df042827f5dda Mon Sep 17 00:00:00 2001 From: walterzhaoJR <519362600@qq.com> Date: Mon, 27 Jul 2026 18:57:16 +0800 Subject: [PATCH] Fix bvar worker starvation with recursive bthread mutex Replace the recursive pthread mutex protecting bvar's VarMap with a bthread-aware recursive mutex so yielding metric callbacks do not block worker pthreads. Add regression coverage for worker starvation, recursive ownership across bthread migration, pthread use, and high contention. Related issue: https://github.com/apache/brpc/issues/2888 --- src/bthread/mutex.cpp | 103 +++++++++ src/bthread/mutex.h | 39 ++++ src/bthread/types.h | 7 + src/bvar/variable.cpp | 11 +- test/CMakeLists.txt | 9 +- test/brpc_bvar_mutex_unittest.cpp | 361 ++++++++++++++++++++++++++++++ test/bthread_mutex_unittest.cpp | 124 ++++++++++ test/bvar_bthread_stubs.cpp | 60 +++++ 8 files changed, 705 insertions(+), 9 deletions(-) create mode 100644 test/brpc_bvar_mutex_unittest.cpp create mode 100644 test/bvar_bthread_stubs.cpp diff --git a/src/bthread/mutex.cpp b/src/bthread/mutex.cpp index 1e6d244168..0eeac1859a 100644 --- a/src/bthread/mutex.cpp +++ b/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( + reinterpret_cast(&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; diff --git a/src/bthread/mutex.h b/src/bthread/mutex.h index 11c7eae821..3cac09104c 100644 --- a/src/bthread/mutex.h +++ b/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 { diff --git a/src/bthread/types.h b/src/bthread/types.h index d46de1e835..139a3c0243 100644 --- a/src/bthread/types.h +++ b/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; diff --git a/src/bvar/variable.cpp b/src/bvar/variable.cpp index 17ffed302b..1cb386d61f 100644 --- a/src/bvar/variable.cpp +++ b/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 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* 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 mu(m.mutex); + std::unique_lock 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*/) { diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 6aebe271f4..cc7237e7c8 100644 --- a/test/CMakeLists.txt +++ b/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_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} $) + 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}) diff --git a/test/brpc_bvar_mutex_unittest.cpp b/test/brpc_bvar_mutex_unittest.cpp new file mode 100644 index 0000000000..a1d9250fd3 --- /dev/null +++ b/test/brpc_bvar_mutex_unittest.cpp @@ -0,0 +1,361 @@ +// 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. + +// brpc - A framework to host and access services throughout Baidu. + +// Date: Mon Jul 27 15:58:00 CST 2026 + +#include +#include +#include +#include +#include + +#include +#include + +#include "bthread/bthread.h" +#include "butil/compat.h" +#include "butil/time.h" +#include "bvar/bvar.h" + +namespace { + +const char kVariableName[] = "bvar_mutex_worker_starvation"; +const char kRecursiveOuterName[] = "bvar_recursive_outer"; +const char kRecursiveInnerName[] = "bvar_recursive_inner25"; +const int64_t kWatchdogTimeoutUs = 1000000; +const int64_t kWaiterSettleUs = 100000; +const int64_t kProbeTimeoutUs = 1000000; +const int64_t kChildTimeoutUs = 3000000; +const int kWaiterCount = BTHREAD_MIN_CONCURRENCY * 4; + +struct TestState; +int BlockingPassiveGetter(void* raw); +struct RecursiveTestState; +int RecursiveOuterGetter(void* raw); + +struct TestState { + std::atomic holder_in_describe{false}; + std::atomic holder_worker{0}; + std::atomic release_holder{false}; + std::atomic holder_finished{false}; + std::atomic waiter_started{0}; + std::atomic waiters_may_contend{false}; + std::atomic waiter_finished{0}; + std::atomic probe_started{false}; + std::atomic probe_finished{false}; + std::atomic failures{0}; + bvar::BasicPassiveStatus variable; + + TestState() : variable(kVariableName, BlockingPassiveGetter, this) {} +}; + +struct RecursiveTestState { + std::atomic inner_described{false}; + std::atomic outer_described{false}; + bvar::BasicPassiveStatus inner; + bvar::BasicPassiveStatus outer; + + RecursiveTestState() + : inner(kRecursiveInnerName, [](void*) { return 7; }, nullptr), + outer(kRecursiveOuterName, RecursiveOuterGetter, this) {} +}; + +int RecursiveOuterGetter(void* raw) { + auto* state = static_cast(raw); + bthread_usleep(1000); + if (bvar::Variable::describe_exposed(kRecursiveInnerName) == "7") { + state->inner_described.store(true, std::memory_order_release); + } + return 9; +} + +int BlockingPassiveGetter(void* raw) { + auto* state = static_cast(raw); + state->holder_worker.store(pthread_numeric_id(), std::memory_order_release); + state->holder_in_describe.store(true, std::memory_order_release); + while (!state->release_holder.load(std::memory_order_acquire)) { + bthread_usleep(1000); + } + return 1; +} + +void* DescribeAsBthreadHolder(void* raw) { + auto* state = static_cast(raw); + if (bvar::Variable::describe_exposed(kVariableName) != "1") { + state->failures.fetch_add(1, std::memory_order_relaxed); + } + state->holder_finished.store(true, std::memory_order_release); + return nullptr; +} + +void* DescribeAsBthreadWaiter(void* raw) { + auto* state = static_cast(raw); + state->waiter_started.fetch_add(1, std::memory_order_release); + while (!state->waiters_may_contend.load(std::memory_order_acquire)) { + bthread_usleep(1000); + } + while (!state->release_holder.load(std::memory_order_acquire) && + pthread_numeric_id() == + state->holder_worker.load(std::memory_order_acquire)) { + bthread_yield(); + } + if (bvar::Variable::describe_exposed(kVariableName) != "1") { + state->failures.fetch_add(1, std::memory_order_relaxed); + } + state->waiter_finished.fetch_add(1, std::memory_order_release); + return nullptr; +} + +void* RunProbe(void* raw) { + auto* state = static_cast(raw); + while (!state->release_holder.load(std::memory_order_acquire) && + pthread_numeric_id() == + state->holder_worker.load(std::memory_order_acquire)) { + bthread_yield(); + } + state->probe_started.store(true, std::memory_order_release); + state->probe_finished.store(true, std::memory_order_release); + return nullptr; +} + +void* WarmupBthreadRuntime(void*) { + return nullptr; +} + +template +bool WaitForAtLeast(const std::atomic& value, T expected, + int64_t timeout_us) { + const int64_t deadline_us = butil::gettimeofday_us() + timeout_us; + while (value.load(std::memory_order_acquire) < expected && + butil::gettimeofday_us() < deadline_us) { + usleep(1000); + } + return value.load(std::memory_order_acquire) >= expected; +} + +bool WaitForTrue(const std::atomic& value, int64_t timeout_us) { + const int64_t deadline_us = butil::gettimeofday_us() + timeout_us; + while (!value.load(std::memory_order_acquire) && + butil::gettimeofday_us() < deadline_us) { + usleep(1000); + } + return value.load(std::memory_order_acquire); +} + +void WriteStage(const char* stage) { + dprintf(STDERR_FILENO, "[bvar-mutex-test] %s\n", stage); +} + +bool RunWorkerStarvationScenario() { + // The holder yields as a bthread while owning the VarMap shard lock. + // Waiters that contend on the pthread mutex block their worker pthreads. + // Once all workers are blocked, neither the holder nor an unrelated probe + // can resume. + WriteStage("set concurrency"); + if (bthread_setconcurrency(BTHREAD_MIN_CONCURRENCY) != 0) { + WriteStage("set concurrency failed"); + return false; + } + + // TaskControl initialization exposes bthread's own bvars. It must finish + // before the holder owns a VarMap lock, otherwise bthread_start_background + // blocks during runtime setup rather than exercising mutex contention. + bthread_t warmup; + if (bthread_start_background(&warmup, nullptr, WarmupBthreadRuntime, + nullptr) != 0 || + bthread_join(warmup, nullptr) != 0) { + WriteStage("bthread runtime warmup failed"); + return false; + } + WriteStage("bthread runtime warmed up"); + + auto* state = new TestState; + WriteStage("state created"); + bthread_t holder; + bthread_t waiters[kWaiterCount]; + bthread_t probe; + bool holder_created = false; + int waiters_created = 0; + bool probe_created = false; + + if (bthread_start_background(&holder, nullptr, DescribeAsBthreadHolder, + state) != 0) { + WriteStage("holder create failed"); + return false; + } + holder_created = true; + if (!WaitForTrue(state->holder_in_describe, kWatchdogTimeoutUs)) { + WriteStage("holder did not enter describe"); + state->release_holder.store(true, std::memory_order_release); + bthread_join(holder, nullptr); + return false; + } + WriteStage("holder entered describe"); + + for (int i = 0; i < kWaiterCount; ++i) { + if (bthread_start_background(&waiters[i], nullptr, + DescribeAsBthreadWaiter, state) != 0) { + WriteStage("waiter create failed"); + state->release_holder.store(true, std::memory_order_release); + bthread_join(holder, nullptr); + return false; + } + ++waiters_created; + } + if (!WaitForAtLeast(state->waiter_started, kWaiterCount, + kWatchdogTimeoutUs)) { + WriteStage("waiters did not start"); + state->release_holder.store(true, std::memory_order_release); + bthread_join(holder, nullptr); + return false; + } + WriteStage("waiters started"); + state->waiters_may_contend.store(true, std::memory_order_release); + + // Release all waiters together and give them time to occupy every worker + // in the contended pthread mutex before enqueueing the unrelated probe. + usleep(kWaiterSettleUs); + if (bthread_start_background(&probe, nullptr, RunProbe, state) != 0) { + WriteStage("probe create failed"); + state->release_holder.store(true, std::memory_order_release); + bthread_join(holder, nullptr); + return false; + } + probe_created = true; + + // On the unfixed implementation, all workers are blocked in the VarMap + // pthread mutex while the holder is suspended, so the unrelated probe + // cannot run. With bthread_mutex_t, the waiters park and the probe runs. + const bool probe_ran_before_release = + WaitForTrue(state->probe_started, kProbeTimeoutUs); + WriteStage(probe_ran_before_release ? "probe ran" : "probe did not run"); + + state->release_holder.store(true, std::memory_order_release); + WriteStage("holder released"); + const bool completed = + WaitForTrue(state->holder_finished, kWatchdogTimeoutUs) && + WaitForAtLeast(state->waiter_finished, kWaiterCount, + kWatchdogTimeoutUs) && + WaitForTrue(state->probe_finished, kWatchdogTimeoutUs); + WriteStage(completed ? "workers completed" : "workers did not complete"); + + if (holder_created) { + bthread_join(holder, nullptr); + } + for (int i = 0; i < waiters_created; ++i) { + bthread_join(waiters[i], nullptr); + } + if (probe_created) { + bthread_join(probe, nullptr); + } + WriteStage("workers joined"); + + const bool ok = probe_ran_before_release && completed && + state->failures.load(std::memory_order_relaxed) == 0; + delete state; + return ok; +} + +void* DescribeRecursiveVariable(void* raw) { + auto* state = static_cast(raw); + const std::string description = + bvar::Variable::describe_exposed(kRecursiveOuterName); + const bool inner_described = + state->inner_described.load(std::memory_order_acquire); + state->outer_described.store(description == "9" && inner_described, + std::memory_order_release); + return nullptr; +} + +bool RunRecursiveDescribeScenario() { + RecursiveTestState state; + bthread_t thread; + if (bthread_start_background(&thread, nullptr, DescribeRecursiveVariable, + &state) != 0) { + return false; + } + return bthread_join(thread, nullptr) == 0 && + state.outer_described.load(std::memory_order_acquire); +} + +bool RunChildAndWait(const char* child_argument, const char* failure_message) { + const pid_t pid = fork(); + if (pid == -1) { + return false; + } + if (pid == 0) { + execl("/proc/self/exe", "brpc_bvar_mutex_unittest", child_argument, + nullptr); + _exit(127); + } + + int status = 0; + pid_t child = 0; + const int64_t deadline_us = butil::gettimeofday_us() + kChildTimeoutUs; + while ((child = waitpid(pid, &status, WNOHANG)) == 0 && + butil::gettimeofday_us() < deadline_us) { + usleep(1000); + } + if (child == 0) { + kill(pid, SIGKILL); + EXPECT_EQ(pid, waitpid(pid, &status, 0)); + ADD_FAILURE() << failure_message; + return false; + } + EXPECT_EQ(pid, child); + EXPECT_TRUE(WIFEXITED(status)); + return WIFEXITED(status) && WEXITSTATUS(status) == 0; +} + +TEST(BvarMutexTest, DoesNotStarveWorkersWhenBthreadDescribeYields) { + EXPECT_TRUE(RunChildAndWait( + "--bvar-mutex-worker-starvation-child", + "bvar worker-starvation child did not finish before watchdog; " + "bthread workers could not make progress while the VarMap lock was " + "held")); +} + +TEST(BvarMutexTest, SupportsRecursiveDescribeAfterBthreadYield) { + EXPECT_TRUE(RunChildAndWait( + "--bvar-mutex-recursive-describe-child", + "bvar recursive describe child did not finish before watchdog; " + "VarMap lock did not preserve recursive bthread ownership")); +} + +TEST(BvarMutexTest, SupportsRecursiveDescribeFromPthread) { + RecursiveTestState state; + EXPECT_EQ("9", bvar::Variable::describe_exposed(kRecursiveOuterName)); + EXPECT_TRUE(state.inner_described.load(std::memory_order_acquire)); +} + +} // namespace + +int main(int argc, char** argv) { + if (argc == 2 && + strcmp(argv[1], "--bvar-mutex-worker-starvation-child") == 0) { + _exit(RunWorkerStarvationScenario() ? 0 : 1); + } + if (argc == 2 && + strcmp(argv[1], "--bvar-mutex-recursive-describe-child") == 0) { + _exit(RunRecursiveDescribeScenario() ? 0 : 1); + } + testing::InitGoogleTest(&argc, argv); + GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true); + const int rc = RUN_ALL_TESTS(); + _exit(rc); +} diff --git a/test/bthread_mutex_unittest.cpp b/test/bthread_mutex_unittest.cpp index 121f1ebb91..c76851c00f 100644 --- a/test/bthread_mutex_unittest.cpp +++ b/test/bthread_mutex_unittest.cpp @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +#include #include #include "butil/compat.h" #include "butil/time.h" @@ -140,6 +141,129 @@ TEST(MutexTest, cpp_wrapper) { mutex.unlock(); } +struct RecursiveMutexArgs { + bthread_recursive_mutex_t* mutex; + std::atomic* completed; +}; + +void* lock_recursive_mutex_after_yield(void* arg) { + RecursiveMutexArgs* args = static_cast(arg); + EXPECT_EQ(0, bthread_recursive_mutex_lock(args->mutex)); + bthread_usleep(1000); + EXPECT_EQ(0, bthread_recursive_mutex_trylock(args->mutex)); + EXPECT_EQ(0, bthread_recursive_mutex_unlock(args->mutex)); + EXPECT_EQ(0, bthread_recursive_mutex_unlock(args->mutex)); + args->completed->store(true, std::memory_order_release); + return NULL; +} + +TEST(MutexTest, recursive_mutex_tracks_bthread_across_yield) { + bthread_recursive_mutex_t mutex; + ASSERT_EQ(0, bthread_recursive_mutex_init(&mutex)); + std::atomic completed(false); + RecursiveMutexArgs args = {&mutex, &completed}; + bthread_t thread; + ASSERT_EQ(0, bthread_start_background( + &thread, NULL, lock_recursive_mutex_after_yield, &args)); + ASSERT_EQ(0, bthread_join(thread, NULL)); + EXPECT_TRUE(completed.load(std::memory_order_acquire)); + EXPECT_EQ(0, bthread_recursive_mutex_destroy(&mutex)); +} + +TEST(MutexTest, recursive_mutex_supports_pthread_and_cpp_wrapper) { + bthread_recursive_mutex_t mutex; + ASSERT_EQ(0, bthread_recursive_mutex_init(&mutex)); + ASSERT_EQ(0, bthread_recursive_mutex_lock(&mutex)); + ASSERT_EQ(0, bthread_recursive_mutex_trylock(&mutex)); + ASSERT_EQ(0, bthread_recursive_mutex_unlock(&mutex)); + ASSERT_EQ(0, bthread_recursive_mutex_unlock(&mutex)); + ASSERT_EQ(0, bthread_recursive_mutex_destroy(&mutex)); + + bthread::RecursiveMutex cpp_mutex; + cpp_mutex.lock(); + EXPECT_TRUE(cpp_mutex.try_lock()); + cpp_mutex.unlock(); + cpp_mutex.unlock(); +} + +struct RecursiveMutexStressArgs { + bthread_recursive_mutex_t* mutex; + std::atomic* ready; + std::atomic* start; + std::atomic* failures; + int64_t* counter; + std::atomic* migrations; +}; + +void* stress_recursive_mutex(void* arg) { + RecursiveMutexStressArgs* args = + static_cast(arg); + args->ready->fetch_add(1, std::memory_order_release); + while (!args->start->load(std::memory_order_acquire)) { + bthread_usleep(1000); + } + + const int kIterations = 10000; + const int kRecursionDepth = 4; + for (int i = 0; i < kIterations; ++i) { + for (int depth = 0; depth < kRecursionDepth; ++depth) { + if (bthread_recursive_mutex_lock(args->mutex) != 0) { + args->failures->fetch_add(1, std::memory_order_relaxed); + return NULL; + } + } + + const uint64_t worker_before_yield = pthread_numeric_id(); + bthread_yield(); + if (pthread_numeric_id() != worker_before_yield) { + args->migrations->fetch_add(1, std::memory_order_relaxed); + } + ++*args->counter; + + for (int depth = 0; depth < kRecursionDepth; ++depth) { + if (bthread_recursive_mutex_unlock(args->mutex) != 0) { + args->failures->fetch_add(1, std::memory_order_relaxed); + return NULL; + } + } + } + return NULL; +} + +TEST(MutexTest, recursive_mutex_high_contention) { + const int kThreadCount = 64; + const int kIterations = 10000; + bthread_recursive_mutex_t mutex; + ASSERT_EQ(0, bthread_recursive_mutex_init(&mutex)); + + std::atomic ready(0); + std::atomic start(false); + std::atomic failures(0); + std::atomic migrations(0); + int64_t counter = 0; + RecursiveMutexStressArgs args = { + &mutex, &ready, &start, &failures, &counter, &migrations}; + bthread_t threads[kThreadCount]; + for (int i = 0; i < kThreadCount; ++i) { + ASSERT_EQ(0, bthread_start_background( + &threads[i], NULL, stress_recursive_mutex, &args)); + } + while (ready.load(std::memory_order_acquire) != kThreadCount) { + usleep(1000); + } + start.store(true, std::memory_order_release); + for (int i = 0; i < kThreadCount; ++i) { + ASSERT_EQ(0, bthread_join(threads[i], NULL)); + } + + EXPECT_EQ(0, failures.load(std::memory_order_relaxed)); + EXPECT_EQ(static_cast(kThreadCount) * kIterations, counter); + LOG(INFO) << "Recursive mutex stress observed " + << migrations.load(std::memory_order_relaxed) + << " cross-worker migrations"; + EXPECT_EQ(0, bthread_recursive_mutex_destroy(&mutex)); +} + bool g_started = false; bool g_stopped = false; diff --git a/test/bvar_bthread_stubs.cpp b/test/bvar_bthread_stubs.cpp new file mode 100644 index 0000000000..a4b48bc2ff --- /dev/null +++ b/test/bvar_bthread_stubs.cpp @@ -0,0 +1,60 @@ +// 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. + +#include +#include + +#include "bthread/mutex.h" + +namespace { + +// test_bvar verifies bvar behavior without linking the bthread runtime. +std::recursive_mutex* native_mutex(bthread_recursive_mutex_t* mutex) { + return reinterpret_cast(mutex->mutex.butex); +} + +} // namespace + +extern "C" { + +int bthread_recursive_mutex_init(bthread_recursive_mutex_t* mutex) { + mutex->mutex.butex = + reinterpret_cast(new std::recursive_mutex); + return 0; +} + +int bthread_recursive_mutex_destroy(bthread_recursive_mutex_t* mutex) { + delete native_mutex(mutex); + mutex->mutex.butex = nullptr; + return 0; +} + +int bthread_recursive_mutex_trylock(bthread_recursive_mutex_t* mutex) { + return native_mutex(mutex)->try_lock() ? 0 : EBUSY; +} + +int bthread_recursive_mutex_lock(bthread_recursive_mutex_t* mutex) { + native_mutex(mutex)->lock(); + return 0; +} + +int bthread_recursive_mutex_unlock(bthread_recursive_mutex_t* mutex) { + native_mutex(mutex)->unlock(); + return 0; +} + +} // extern "C"