Copilot commented on code in PR #3647:
URL: https://github.com/apache/brpc/pull/3647#discussion_r4357790294
##########
src/brpc/span.cpp:
##########
@@ -506,10 +506,15 @@ 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;
+// Serializes persistence with shutdown.
+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;
+}
Review Comment:
`has_span_db()` now locks `g_span_db_mutex`. This makes it unsafe to call
from any code path that might already hold `g_span_db_mutex` (non-recursive
mutex => self-deadlock). If the intent is to provide a safe external query,
please document that callers must not hold `g_span_db_mutex`, or provide a
separate unlocked/internal helper for use under the mutex.
##########
src/brpc/span.cpp:
##########
@@ -847,22 +874,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);
Review Comment:
`g_span_dump_mutex` is held across the entire `dump_to_db()` body, including
`SpanDB::Open()` and `db->Index(...)` (potentially slow I/O). This serializes
all span persistence and could become a bottleneck under load. If the goal is
only to make shutdown wait for active dumps, consider replacing the global
mutex with an active-dump counter + condition variable (or narrowing the locked
region to DB publication/teardown) so concurrent dumps can proceed while still
allowing `StopSpanIndexing()` to block until in-flight dumps complete.
##########
src/brpc/span.cpp:
##########
@@ -847,22 +874,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();
Review Comment:
`g_span_dump_mutex` is held across the entire `dump_to_db()` body, including
`SpanDB::Open()` and `db->Index(...)` (potentially slow I/O). This serializes
all span persistence and could become a bottleneck under load. If the goal is
only to make shutdown wait for active dumps, consider replacing the global
mutex with an active-dump counter + condition variable (or narrowing the locked
region to DB publication/teardown) so concurrent dumps can proceed while still
allowing `StopSpanIndexing()` to block until in-flight dumps complete.
##########
test/brpc_span_shutdown_unittest.cpp:
##########
@@ -0,0 +1,232 @@
+// 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 <thread>
Review Comment:
`pthread_spin_lock`/`pthread_spin_unlock` are used in this test but
`<pthread.h>` is not included. Relying on transitive includes can break builds
on different platforms/toolchains; add an explicit `#include <pthread.h>` here.
##########
test/brpc_span_shutdown_unittest.cpp:
##########
@@ -0,0 +1,232 @@
+// 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 <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.
+ ::testing::FLAGS_gtest_death_test_style = "threadsafe";
+ 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");
Review Comment:
This test directly accesses `Span` internals (`_info_spinlock`), which makes
the test fragile to refactors and can break if visibility changes. Consider
adding a small test-only hook (e.g., a friend helper in `Span` guarded by
`#ifdef BRPC_UNIT_TEST`, or a dedicated internal API) so the test doesn't
depend on a private member layout.
##########
src/brpc/span.h:
##########
@@ -50,6 +50,11 @@ void* CreateBthreadSpanAsVoid();
void DestroyRpczParentSpan(void* ptr);
void EndBthreadSpan();
+// Permanently stop span persistence and wait for any active dump to finish.
+// Further dump attempts are ignored, including already queued spans.
Review Comment:
`StopSpanIndexing()` is added to a public header, which effectively makes it
part of the public API surface. If this is intended as an internal shutdown
primitive (primarily for exit handling/tests), consider moving it to an
internal header/namespace or clearly documenting its thread-safety, lifetime
expectations, and whether it is supported for external callers.
##########
test/brpc_span_shutdown_unittest.cpp:
##########
@@ -0,0 +1,232 @@
+// 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 <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");
Review Comment:
This test directly accesses `Span` internals (`_info_spinlock`), which makes
the test fragile to refactors and can break if visibility changes. Consider
adding a small test-only hook (e.g., a friend helper in `Span` guarded by
`#ifdef BRPC_UNIT_TEST`, or a dedicated internal API) so the test doesn't
depend on a private member layout.
##########
test/brpc_controller_unittest.cpp:
##########
@@ -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));
Review Comment:
The test reaches into `Controller` internals (`cntl._oncancel_id`). This
makes the test sensitive to implementation changes and visibility. Prefer
asserting via a public/testing API (e.g., a helper to wait for cancel callback
completion), or add a narrow test-only accessor so the test doesn't depend on
private fields.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
