diff --git a/BUILD.bazel b/BUILD.bazel index 5dc5fcf726..6d3ace33d3 100644 --- a/BUILD.bazel +++ b/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({ 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..dc170ed7eb 100644 --- a/src/bthread/mutex.h +++ b/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" diff --git a/src/bthread/recursive_mutex.h b/src/bthread/recursive_mutex.h new file mode 100644 index 0000000000..d9beb59aad --- /dev/null +++ b/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 + +#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 diff --git a/src/bthread/types.h b/src/bthread/types.h index d46de1e835..c8896e7e0e 100644 --- a/src/bthread/types.h +++ b/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; diff --git a/src/bvar/variable.cpp b/src/bvar/variable.cpp index 17ffed302b..355c1f74ce 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/recursive_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/BUILD.bazel b/test/BUILD.bazel index 27d63aed81..2c301c5e8f 100644 --- a/test/BUILD.bazel +++ b/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", 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/Makefile b/test/Makefile index 15169937ef..1d7a9b86f9 100644 --- a/test/Makefile +++ b/test/Makefile @@ -174,7 +174,7 @@ UT_DYNAMIC_LINKINGS = -lbrpc.dbg $(DYNAMIC_LINKINGS) TEST_BUTIL_OBJS = iobuf.pb.o $(addsuffix .o, $(basename $(TEST_BUTIL_SOURCES))) TEST_BVAR_SOURCES = $(wildcard bvar_*_unittest.cpp) -TEST_BVAR_OBJS = $(addsuffix .o, $(basename $(TEST_BVAR_SOURCES))) +TEST_BVAR_OBJS = $(addsuffix .o, $(basename $(TEST_BVAR_SOURCES))) bvar_bthread_stubs.o TEST_BTHREAD_SOURCES = $(wildcard bthread_*unittest.cpp) TEST_BTHREAD_OBJS = $(addsuffix .o, $(basename $(TEST_BTHREAD_SOURCES))) @@ -185,6 +185,9 @@ TEST_BRPC_OBJS = $(addsuffix .o, $(basename $(TEST_BRPC_SOURCES))) TEST_PROTO_SOURCES = $(wildcard *.proto) TEST_PROTO_OBJS = $(TEST_PROTO_SOURCES:.proto=.pb.o) +# brpc_bvar_mutex_unittest provides its own main() for fork/exec child modes. +BRPC_BVAR_MUTEX_UT = brpc_bvar_mutex_unittest + TEST_BINS = test_butil test_bvar $(TEST_BTHREAD_SOURCES:.cpp=) $(TEST_BRPC_SOURCES:.cpp=) .PHONY:all @@ -234,6 +237,15 @@ else ifeq ($(SYSTEM),Darwin) $(CXX) -o $@ $(LIBPATHS) $(SOPATHS) $^ $(GTEST_STATIC_LINKINGS) $(UT_DYNAMIC_LINKINGS) endif +# Override the generic brpc_%_unittest rule: this binary ships its own main(). +$(BRPC_BVAR_MUTEX_UT):$(TEST_PROTO_OBJS) $(BRPC_BVAR_MUTEX_UT).o | libbrpc.dbg.$(SOEXT) + @echo "> Linking $@" +ifeq ($(SYSTEM),Linux) + $(CXX) -o $@ $(LIBPATHS) $(SOPATHS) -Xlinker "-(" $^ -Xlinker "-)" $(filter-out %libgtest_main.a,$(STATIC_LINKINGS)) $(filter-out -lgtest_main,$(UT_DYNAMIC_LINKINGS)) +else ifeq ($(SYSTEM),Darwin) + $(CXX) -o $@ $(LIBPATHS) $(SOPATHS) $^ $(filter-out %libgtest_main.a,$(GTEST_STATIC_LINKINGS)) $(filter-out -lgtest_main,$(UT_DYNAMIC_LINKINGS)) +endif + brpc_%_unittest:$(TEST_PROTO_OBJS) brpc_%_unittest.o | libbrpc.dbg.$(SOEXT) @echo "> Linking $@" ifeq ($(SYSTEM),Linux) diff --git a/test/brpc_bvar_mutex_unittest.cpp b/test/brpc_bvar_mutex_unittest.cpp new file mode 100644 index 0000000000..041ff78545 --- /dev/null +++ b/test/brpc_bvar_mutex_unittest.cpp @@ -0,0 +1,373 @@ +// 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 + +#include "bthread/bthread.h" +#include "butil/compat.h" +#include "butil/process_util.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) { + // proc_pidpath() may require PROC_PIDPATHINFO_MAXSIZE, which is + // 4 * PATH_MAX on macOS. + char executable_path[PATH_MAX * 4]; + const ssize_t executable_path_length = butil::GetProcessAbsolutePath( + executable_path, sizeof(executable_path) - 1); + if (executable_path_length <= 0) { + return false; + } + executable_path[executable_path_length] = '\0'; + + const pid_t pid = fork(); + if (pid == -1) { + return false; + } + if (pid == 0) { + execl(executable_path, "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/brpc_rtmp_unittest.cpp b/test/brpc_rtmp_unittest.cpp index 61a63744fc..fb8507c596 100644 --- a/test/brpc_rtmp_unittest.cpp +++ b/test/brpc_rtmp_unittest.cpp @@ -267,9 +267,13 @@ class PlayingDummyStream : public brpc::RtmpServerStream { << " ms before responding play request"; bthread_usleep(_sleep_ms * 1000L); } + // Keep the stream alive until RunSendData exits. On a write failure, + // OnStop may be called synchronously from the sending bthread. + AddRefManually(); int rc = bthread_start_background(&_play_thread, NULL, RunSendData, this); if (rc) { + RemoveRefManually(); status->set_error(rc, "Fail to create thread"); return; } @@ -288,7 +292,9 @@ class PlayingDummyStream : public brpc::RtmpServerStream { LOG(INFO) << "OnStop of PlayingDummyStream=" << this; if (_state.exchange(STATE_STOPPED) == STATE_PLAYING) { bthread_stop(_play_thread); - bthread_join(_play_thread, NULL); + if (_play_thread != bthread_self()) { + bthread_join(_play_thread, NULL); + } } } @@ -296,7 +302,9 @@ class PlayingDummyStream : public brpc::RtmpServerStream { private: static void* RunSendData(void* arg) { - ((PlayingDummyStream*)arg)->SendData(); + butil::intrusive_ptr stream( + static_cast(arg), false); + stream->SendData(); return NULL; } @@ -322,8 +330,10 @@ void PlayingDummyStream::SendData() { vmsg.data.clear(); vmsg.data.append(butil::string_printf("video_%d(ms_id=%u)", i, stream_id())); - //failing to send is possible - SendVideoMessage(vmsg); + // Failing to send is possible and may synchronously stop the stream. + if (SendVideoMessage(vmsg) != 0) { + break; + } amsg.codec = brpc::FLV_AUDIO_AAC; amsg.rate = brpc::FLV_SOUND_RATE_44100HZ; @@ -332,7 +342,9 @@ void PlayingDummyStream::SendData() { amsg.data.clear(); amsg.data.append(butil::string_printf("audio_%d(ms_id=%u)", i, stream_id())); - SendAudioMessage(amsg); + if (SendAudioMessage(amsg) != 0) { + break; + } bthread_usleep(1000000); } 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..352f2b4448 --- /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/recursive_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"