This is an automated email from the ASF dual-hosted git repository.
wwbmmm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new f760a4f78 Fix rpcz span persistence races during process shutdown
(#3647)
f760a4f78 is described below
commit f760a4f7893ace039b4b6a8989f572b7131b6095
Author: Bright Chen <[email protected]>
AuthorDate: Sun Oct 25 17:20:30 2026 +0800
Fix rpcz span persistence races during process shutdown (#3647)
* Fix rpcz span persistence races during process shutdown
* Fix comments and ut
---
src/brpc/span.cpp | 60 +++++++--
src/brpc/span.h | 7 ++
test/brpc_controller_unittest.cpp | 13 +-
test/brpc_span_shutdown_unittest.cpp | 238 +++++++++++++++++++++++++++++++++++
4 files changed, 302 insertions(+), 16 deletions(-)
diff --git a/src/brpc/span.cpp b/src/brpc/span.cpp
index 304ba5d2b..b94caa381 100644
--- a/src/brpc/span.cpp
+++ b/src/brpc/span.cpp
@@ -506,10 +506,17 @@ static int64_t g_last_delete_tm = 0;
// Following variables are monitored by builtin services, thus non-static.
static pthread_mutex_t g_span_db_mutex = PTHREAD_MUTEX_INITIALIZER;
+// The collector already dumps samples on a single thread. This mutex also
+// serializes persistence with shutdown and is intentionally never destroyed,
+// so collector callbacks can still acquire it after the exit guard has run.
+static pthread_mutex_t g_span_dump_mutex = PTHREAD_MUTEX_INITIALIZER;
static bool g_span_ending = false; // don't open span again if this var is
true.
// Can't use intrusive_ptr which has ctor/dtor issues.
static SpanDB* g_span_db = nullptr;
-bool has_span_db() { return !!g_span_db; }
+bool has_span_db() {
+ BAIDU_SCOPED_LOCK(g_span_db_mutex);
+ return g_span_db != nullptr;
+}
bvar::CollectorSpeedLimit g_span_sl = BVAR_COLLECTOR_SPEED_LIMIT_INITIALIZER;
static bvar::DisplaySamplingRatio s_display_sampling_ratio(
"rpcz_sampling_ratio", &g_span_sl);
@@ -543,10 +550,15 @@ bvar::CollectorPreprocessor* Span::preprocessor() {
return g_span_prep;
}
-static void ResetSpanDB(SpanDB* db) {
+static bool ResetSpanDB(SpanDB* db) {
SpanDB* old_db = nullptr;
{
BAIDU_SCOPED_LOCK(g_span_db_mutex);
+ // Opening a DB can overlap with StopSpanIndexing(). Never publish it
+ // after shutdown has started.
+ if (db != nullptr && g_span_ending) {
+ return false;
+ }
old_db = g_span_db;
g_span_db = db;
if (g_span_db) {
@@ -556,15 +568,32 @@ static void ResetSpanDB(SpanDB* db) {
if (old_db) {
old_db->RemoveRefManually();
}
+ return true;
}
-static void RemoveSpanDB() {
- g_span_ending = true;
+void StopSpanIndexing() {
+ // Reject new dumps before waiting for the current one. Do not hold the DB
+ // mutex while acquiring the dump mutex: the dumper also needs the DB
mutex.
+ {
+ BAIDU_SCOPED_LOCK(g_span_db_mutex);
+ g_span_ending = true;
+ }
+ BAIDU_SCOPED_LOCK(g_span_dump_mutex);
ResetSpanDB(nullptr);
}
+// Defined after the flags in this translation unit so this destructor runs
+// before their destructors. Registering cleanup lazily in StartSpanIndexing is
+// too late when the first dump starts during process exit.
+struct SpanIndexingExitGuard {
+ ~SpanIndexingExitGuard() { StopSpanIndexing(); }
+};
+// No explicit reference is needed: the compiler registers its destructor for
+// normal process exit (e.g. via __cxa_atexit). When this translation unit is
+// linked in, optimization must preserve the StopSpanIndexing() side effects.
+static SpanIndexingExitGuard g_span_indexing_exit_guard;
+
static void StartSpanIndexing() {
- atexit(RemoveSpanDB);
g_span_prep = new SpanPreprocessor;
started_span_indexing = true;
}
@@ -847,22 +876,31 @@ leveldb::Status SpanDB::RemoveSpansBefore(int64_t tm) {
// Write span into leveldb.
void Span::dump_to_db() {
+ // This must outlive all locals, including the local DB reference: shutdown
+ // must wait for both persistence and any resulting DB destruction.
+ BAIDU_SCOPED_LOCK(g_span_dump_mutex);
+ {
+ BAIDU_SCOPED_LOCK(g_span_db_mutex);
+ if (g_span_ending) {
+ return;
+ }
+ }
StartIndexingIfNeeded();
std::string value_buf;
butil::intrusive_ptr<SpanDB> db;
if (GetSpanDB(&db) != 0) {
- if (g_span_ending) {
+ // Own the DB before publishing it. If shutdown started during Open(),
+ // the rejected DB is released before we leave the dump mutex.
+ db.reset(SpanDB::Open());
+ if (!db) {
+ LOG(WARNING) << "Fail to open SpanDB";
return;
}
- SpanDB* db2 = SpanDB::Open();
- if (db2 == nullptr) {
- LOG(WARNING) << "Fail to open SpanDB";
+ if (!ResetSpanDB(db.get())) {
return;
}
- ResetSpanDB(db2);
- db.reset(db2);
}
leveldb::Status st = db->Index(shared_from_this(), &value_buf);
diff --git a/src/brpc/span.h b/src/brpc/span.h
index e62357e7b..d9f1771c4 100644
--- a/src/brpc/span.h
+++ b/src/brpc/span.h
@@ -50,6 +50,13 @@ void* CreateBthreadSpanAsVoid();
void DestroyRpczParentSpan(void* ptr);
void EndBthreadSpan();
+// Internal shutdown/test hook; not a supported user API.
+// Permanently stop span persistence and wait for any active dump to finish.
+// Further dump attempts are ignored, including already queued spans.
+// Thread-safe and idempotent; must not be called from an active span dump.
+// Called automatically before this module's global resources are destroyed.
+void StopSpanIndexing();
+
DECLARE_bool(enable_rpcz);
class Span;
diff --git a/test/brpc_controller_unittest.cpp
b/test/brpc_controller_unittest.cpp
index d000d41ad..24ab694da 100644
--- a/test/brpc_controller_unittest.cpp
+++ b/test/brpc_controller_unittest.cpp
@@ -24,6 +24,7 @@
#include "butil/logging.h"
#include "butil/time.h"
#include "butil/macros.h"
+#include "bthread/id.h"
#include "brpc/socket.h"
#include "brpc/server.h"
#include "brpc/channel.h"
@@ -42,9 +43,11 @@ void MyCancelCallback(bool* cancel_flag) {
*cancel_flag = true;
}
-bool WaitForSpanToExpire(const std::weak_ptr<brpc::Span>& span) {
- const int64_t deadline_us = butil::gettimeofday_us() + 5000000L;
- while (!span.expired() && butil::gettimeofday_us() < deadline_us) {
+static bool WaitForSpanToExpire(const std::weak_ptr<brpc::Span>& span) {
+ // The collector keeps the span alive until dump_to_db() returns (or the
+ // sample is dropped). Wait for it to release this test's submission.
+ const int64_t deadline_ms = butil::monotonic_time_ms() + 10000;
+ while (!span.expired() && butil::monotonic_time_ms() < deadline_ms) {
usleep(1000);
}
return span.expired();
@@ -62,8 +65,8 @@ TEST_F(ControllerTest, notify_on_failed) {
cntl.NotifyOnCancel(brpc::NewCallback(&MyCancelCallback, &cancel));
// Trigger callback
brpc::Socket::SetFailed(id);
- usleep(20000); // sleep a while to wait for the canceling which will be
- // happening in another thread.
+ // The callback destroys this id after writing cancel.
+ ASSERT_EQ(0, bthread_id_join(cntl._oncancel_id));
ASSERT_TRUE(cancel);
ASSERT_TRUE(cntl.IsCanceled());
}
diff --git a/test/brpc_span_shutdown_unittest.cpp
b/test/brpc_span_shutdown_unittest.cpp
new file mode 100644
index 000000000..c71c0960e
--- /dev/null
+++ b/test/brpc_span_shutdown_unittest.cpp
@@ -0,0 +1,238 @@
+// 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 <atomic>
+#include <cstdlib>
+#include <cstring>
+#include <pthread.h>
+#include <thread>
+#include <unistd.h>
+#include <gflags/gflags.h>
+#include <gtest/gtest.h>
+#include "butil/files/scoped_temp_dir.h"
+#include "butil/time.h"
+#include "brpc/span.h"
+
+namespace brpc {
+DECLARE_string(rpcz_database_dir);
+bool has_span_db();
+}
+
+namespace {
+
+const char kDirectoryEnv[] = "BRPC_SPAN_SHUTDOWN_TEST_DIR";
+
+void Require(bool condition, const char* message) {
+ if (!condition) {
+ (void)write(STDERR_FILENO, message, strlen(message));
+ (void)write(STDERR_FILENO, "\n", 1);
+ std::_Exit(1);
+ }
+}
+
+template <typename Predicate>
+bool WaitUntil(Predicate predicate, int64_t timeout_ms = 10000) {
+ const int64_t deadline_ms = butil::monotonic_time_ms() + timeout_ms;
+ while (!predicate() && butil::monotonic_time_ms() < deadline_ms) {
+ usleep(1000);
+ }
+ return predicate();
+}
+
+void PrepareChild() {
+ // Also bound direct shutdown calls and thread joins if a regression hangs.
+ alarm(15);
+ const char* directory = getenv(kDirectoryEnv);
+ Require(directory != nullptr, "missing test database directory");
+ brpc::FLAGS_rpcz_database_dir = directory;
+ brpc::ClearTlsParentSpan();
+ Require(!brpc::has_span_db(), "child already has a span database");
+}
+
+std::shared_ptr<brpc::Span> NewSpan() {
+ const int64_t now = butil::cpuwide_time_us();
+ std::shared_ptr<brpc::Span> span = brpc::Span::CreateClientSpan(
+ "test.Shutdown/Call", butil::gettimeofday_us() - now);
+ Require(static_cast<bool>(span), "cannot create span");
+ span->set_start_send_us(now);
+ span->set_start_callback_us(now + 1);
+ return span;
+}
+
+std::shared_ptr<brpc::Span> StartBlockedDump() {
+ std::shared_ptr<brpc::Span> span = NewSpan();
+ // The dumper publishes the DB before Index() reads info(). This lock keeps
+ // the dump in flight without adding a hook to the production collector.
+ Require(pthread_spin_lock(&span->_info_spinlock) == 0,
+ "cannot lock span annotations");
+ brpc::Span::Submit(span, butil::cpuwide_time_us());
+ Require(WaitUntil([] { return brpc::has_span_db(); }),
+ "collector did not open the span database");
+ return span;
+}
+
+struct BlockerState {
+ std::atomic<bool> entered{false};
+ std::atomic<bool> release{false};
+};
+
+class CollectorBlocker : public bvar::Collected {
+public:
+ explicit CollectorBlocker(BlockerState* state) : _state(state) {}
+
+ void dump_and_destroy(size_t) override {
+ _state->entered.store(true);
+ Require(WaitUntil([this] { return _state->release.load(); }),
+ "collector blocker was not released");
+ destroy();
+ }
+
+ void destroy() override { delete this; }
+ bvar::CollectorSpeedLimit* speed_limit() override { return nullptr; }
+
+private:
+ BlockerState* _state;
+};
+
+class SpanShutdownDeathTest : public ::testing::Test {
+protected:
+ void SetUp() override {
+ // Exec each child: forking an existing collector would leave its
+ // mutexes and queues without their worker threads.
+#ifdef GTEST_FLAG_SET
+ GTEST_FLAG_SET(death_test_style, "threadsafe");
+#else
+ // GoogleTest 1.11 and earlier do not provide the setter macro.
+ ::testing::GTEST_FLAG(death_test_style) = "threadsafe";
+#endif
+ if (getenv(kDirectoryEnv) == nullptr) {
+ ASSERT_TRUE(_directory.CreateUniqueTempDir());
+ ASSERT_EQ(0, setenv(kDirectoryEnv,
_directory.path().value().c_str(), 1));
+ _owns_directory = true;
+ }
+ }
+
+ void TearDown() override {
+ if (_owns_directory) {
+ EXPECT_EQ(0, unsetenv(kDirectoryEnv));
+ }
+ }
+
+private:
+ // The parent removes the child's files after EXPECT_EXIT has reaped it.
+ butil::ScopedTempDir _directory;
+ bool _owns_directory = false;
+};
+
+TEST_F(SpanShutdownDeathTest, stop_waits_for_inflight_dump) {
+ EXPECT_EXIT({
+ PrepareChild();
+ std::shared_ptr<brpc::Span> span = StartBlockedDump();
+ std::atomic<bool> stopping{false};
+ std::atomic<bool> stopped{false};
+ std::thread stopper([&] {
+ stopping.store(true);
+ brpc::StopSpanIndexing();
+ stopped.store(true);
+ });
+ Require(WaitUntil([&] { return stopping.load(); }),
+ "shutdown thread did not start");
+ Require(!WaitUntil([&] { return stopped.load(); }, 100),
+ "shutdown returned while a span dump was blocked");
+ Require(pthread_spin_unlock(&span->_info_spinlock) == 0,
+ "cannot release span annotations");
+ Require(WaitUntil([&] { return stopped.load(); }),
+ "shutdown did not finish after the dump was released");
+ stopper.join();
+ Require(!brpc::has_span_db(), "shutdown retained the span database");
+ std::_Exit(0);
+ }, ::testing::ExitedWithCode(0), "");
+}
+
+TEST_F(SpanShutdownDeathTest, stop_skips_queued_first_span) {
+ EXPECT_EXIT({
+ PrepareChild();
+ BlockerState blocker;
+ (new CollectorBlocker(&blocker))->submit();
+ Require(WaitUntil([&] { return blocker.entered.load(); }),
+ "collector blocker did not start");
+ std::shared_ptr<brpc::Span> span = NewSpan();
+ std::weak_ptr<brpc::Span> weak_span = span;
+ brpc::Span::Submit(span, butil::cpuwide_time_us());
+ span.reset();
+ Require(!weak_span.expired(), "queued span was not retained");
+
+ // No span has reached dump_to_db(), so shutdown must already be
+ // available without relying on lazy database initialization.
+ brpc::StopSpanIndexing();
+ Require(!brpc::has_span_db(), "shutdown opened the span database");
+ blocker.release.store(true);
+ Require(WaitUntil([&] { return weak_span.expired(); }),
+ "collector did not release the queued span");
+ Require(!brpc::has_span_db(), "late collector dump reopened the
database");
+ std::_Exit(0);
+ }, ::testing::ExitedWithCode(0), "");
+}
+
+TEST_F(SpanShutdownDeathTest, stop_is_idempotent_and_late_dump_is_skipped) {
+ EXPECT_EXIT({
+ PrepareChild();
+ std::shared_ptr<brpc::Span> span = NewSpan();
+ brpc::StopSpanIndexing();
+ brpc::StopSpanIndexing();
+ span->dump_to_db();
+ Require(!brpc::has_span_db(), "direct late dump reopened the
database");
+ std::_Exit(0);
+ }, ::testing::ExitedWithCode(0), "");
+}
+
+TEST_F(SpanShutdownDeathTest, process_exit_waits_for_inflight_dump) {
+ EXPECT_EXIT({
+ PrepareChild();
+ std::shared_ptr<brpc::Span> span = NewSpan();
+ std::atomic<bool> locked{false};
+ std::atomic<bool> exiting{false};
+ std::thread([span, &locked, &exiting] {
+ // Keep lock and unlock on the same thread, as required by POSIX.
+ Require(pthread_spin_lock(&span->_info_spinlock) == 0,
+ "cannot lock span annotations during exit");
+ locked.store(true);
+ Require(WaitUntil([&] { return exiting.load(); }),
+ "main thread did not begin process exit");
+ usleep(200000);
+ // Print while the dump is still blocked. Without an exit barrier
+ // the process finishes before this thread can emit the marker.
+ const char marker[] = "releasing span during process exit\n";
+ Require(write(STDERR_FILENO, marker, sizeof(marker) - 1) ==
+ static_cast<ssize_t>(sizeof(marker) - 1),
+ "cannot write the exit marker");
+ Require(pthread_spin_unlock(&span->_info_spinlock) == 0,
+ "cannot release span annotations during exit");
+ }).detach();
+ Require(WaitUntil([&] { return locked.load(); }),
+ "annotation holder did not start");
+ brpc::Span::Submit(span, butil::cpuwide_time_us());
+ Require(WaitUntil([] { return brpc::has_span_db(); }),
+ "collector did not open the span database");
+ // Exercise the registered process-exit hook, with no explicit stop
+ // and no wait for the collector to release this span.
+ exiting.store(true);
+ std::exit(0);
+ }, ::testing::ExitedWithCode(0), "releasing span during process exit");
+}
+
+} // namespace
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
