Copilot commented on code in PR #3634:
URL: https://github.com/apache/brpc/pull/3634#discussion_r4332922359
##########
src/brpc/urma/urma_helper.cpp:
##########
@@ -620,12 +627,16 @@ static bool GlobalUrmaInitializeImpl() {
return false;
}
+ if (FLAGS_urma_use_polling) {
+ atexit(ReleasePollingModeAtExit);
+ }
Review Comment:
The return value of `atexit(...)` is ignored. If registration fails
(non-zero), the process will exit without stopping/joining pollers,
reintroducing the shutdown-order hazard this PR aims to fix. Consider checking
the return value and logging (or failing initialization) to ensure the shutdown
handler is reliably installed.
##########
test/brpc_urma_unittest.cpp:
##########
@@ -21,26 +21,180 @@
#include <gflags/gflags.h>
#if BRPC_WITH_URMA
+#include <atomic>
+#include <cstdlib>
#include "butil/atomicops.h"
#include "butil/sys_byteorder.h"
#include "urma_api.h"
+#include "brpc/urma/urma_endpoint.h"
#include "brpc/urma/urma_handshake.h"
#include "brpc/urma/urma_handshake.pb.h"
#include "brpc/urma/urma_helper.h"
#include "urma_types.h"
using namespace brpc;
+DECLARE_int32(task_group_ntags);
+
namespace brpc {
namespace urma {
DECLARE_int32(urma_client_handshake_version);
+DECLARE_int32(urma_buffer_size);
+DECLARE_int32(urma_buffer_count);
+DECLARE_int32(urma_sq_size);
+DECLARE_int32(urma_rq_size);
+DECLARE_int32(urma_prepared_jetty_cnt);
extern bool g_skip_urma_init;
extern butil::atomic<bool> g_urma_available;
} // namespace urma
} // namespace brpc
+class UrmaPollingTest : public ::testing::TestWithParam<bool> {
+protected:
+ void SetUp() override {
+ urma::FLAGS_urma_use_polling = true;
+ urma::FLAGS_urma_poller_num = 2;
+ urma::FLAGS_urma_disable_bthread = GetParam();
+ ASSERT_EQ(0, urma::UrmaEndpoint::GlobalInitialize());
+ }
+
+ void TearDown() override {
+ urma::UrmaEndpoint::PollingModeRelease(0);
+ }
+
+ GFLAGS_NAMESPACE::FlagSaver _flags;
+};
+
+TEST_P(UrmaPollingTest, release_joins_and_can_be_repeated) {
+ std::atomic<int> initialized{0};
+ std::atomic<int> released{0};
+ ASSERT_EQ(0, urma::UrmaEndpoint::PollingModeInitialize(
+ 0, nullptr, [&] { ++initialized; }, [&] { ++released; }));
+ urma::UrmaEndpoint::PollingModeRelease(0);
+ EXPECT_EQ(2, initialized.load());
+ EXPECT_EQ(2, released.load());
+ auto& group = urma::UrmaEndpoint::PollerGroups()[0];
+ EXPECT_FALSE(group.running.load());
+ for (const auto& poller : group.pollers) {
+ EXPECT_EQ(INVALID_BTHREAD, poller.tid);
+ }
+ urma::UrmaEndpoint::PollingModeRelease(0);
+ EXPECT_EQ(2, released.load());
+
+ ASSERT_EQ(0, urma::UrmaEndpoint::PollingModeInitialize(
+ 0, nullptr, nullptr, [&] { ++released; }));
+ urma::UrmaEndpoint::PollingModeRelease(0);
+ EXPECT_EQ(4, released.load());
+}
+
+TEST_P(UrmaPollingTest, late_cq_operations_skip_stopped_pollers) {
+ ASSERT_EQ(0, urma::UrmaEndpoint::PollingModeInitialize(
+ 0, nullptr, nullptr, nullptr));
+ urma::UrmaEndpoint::PollingModeRelease(0);
+ urma::UrmaEndpoint endpoint(nullptr);
+ endpoint._cq_sid = 7;
+ endpoint.PollerAddCqSid();
+ endpoint.PollerRemoveCqSid();
+ for (auto& poller : urma::UrmaEndpoint::PollerGroups()[0].pollers) {
+ urma::UrmaEndpoint::CqSidOp op;
+ EXPECT_FALSE(poller.op_queue.Dequeue(op));
+ }
+}
+
+INSTANTIATE_TEST_SUITE_P(BthreadAndPthread, UrmaPollingTest,
+ ::testing::Bool());
+
+namespace {
+std::atomic<int> g_exit_pollers_released{0};
+urma::UrmaEndpoint* g_exit_endpoint = nullptr;
+
+void CheckPollingExit() {
+ if (g_exit_pollers_released.load() != 2) {
+ std::_Exit(91);
+ }
+ // This handler was registered before URMA initialization, so it runs
+ // after the polling exit handler, like a late Socket/endpoint destructor.
+ delete g_exit_endpoint;
+ g_exit_endpoint = nullptr;
+}
+} // namespace
+
+TEST(UrmaPollingDeathTest, global_shutdown_joins_all_tags) {
+ ::testing::FLAGS_gtest_death_test_style = "threadsafe";
+ EXPECT_EXIT({
+ urma::FLAGS_urma_use_polling = true;
+ urma::FLAGS_urma_poller_num = 2;
+ FLAGS_task_group_ntags = 2;
+ if (urma::UrmaEndpoint::GlobalInitialize() != 0 ||
+ urma::UrmaEndpoint::PollingModeInitialize(
+ 0, nullptr, nullptr,
+ [] { ++g_exit_pollers_released; }) != 0) {
+ std::_Exit(92);
+ }
+ if (urma::UrmaEndpoint::PollingModeInitialize(
+ 1, nullptr, nullptr,
+ [] { ++g_exit_pollers_released; }) != 0) {
+ std::_Exit(92);
+ }
+ urma::UrmaEndpoint::GlobalPollingModeRelease();
+ urma::UrmaEndpoint::GlobalPollingModeRelease();
+ if (g_exit_pollers_released.load() != 4) {
+ std::_Exit(93);
+ }
+ std::_Exit(0);
+ }, ::testing::ExitedWithCode(0), "");
+}
+
+TEST(UrmaPollingDeathTest, exit_stops_pollers_before_late_endpoint_cleanup) {
+ // Only the link-time mock can provide device resources independently of
+ // the host. Release the probe initialization before starting the child.
+ urma_init_attr_t init_attr{};
+ const urma_status_t status = urma_init(&init_attr);
+ if (status != URMA_SUCCESS && status != URMA_EEXIST) {
+ GTEST_SKIP() << "Polling exit test requires the URMA link-time mock";
+ }
+ int num_devices = 0;
+ urma_device_t** devices = urma_get_device_list(&num_devices);
+ const bool using_mock = devices && num_devices > 0 &&
+ strcmp(devices[0]->name, "mock_urma_device") == 0;
+ if (devices) {
+ urma_free_device_list(devices);
+ }
+ if (status == URMA_SUCCESS) {
+ ASSERT_EQ(URMA_SUCCESS, urma_uninit());
+ }
+ if (!using_mock) {
+ GTEST_SKIP() << "Polling exit test requires the URMA link-time mock";
+ }
+ ::testing::FLAGS_gtest_death_test_style = "threadsafe";
+ EXPECT_EXIT({
+ urma::FLAGS_urma_use_polling = true;
+ urma::FLAGS_urma_poller_num = 2;
+ urma::FLAGS_urma_buffer_size = 4096;
+ urma::FLAGS_urma_buffer_count = 512;
+ urma::FLAGS_urma_sq_size = 16;
+ urma::FLAGS_urma_rq_size = 16;
+ urma::FLAGS_urma_prepared_jetty_cnt = 0;
+ urma::g_skip_urma_init = false;
+ if (std::atexit(CheckPollingExit) != 0) {
+ std::_Exit(94);
+ }
+ urma::GlobalUrmaInitializeOrDie();
+ if (!urma::IsUrmaAvailable() ||
+ urma::UrmaEndpoint::PollingModeInitialize(
+ 0, nullptr, nullptr,
+ [] { ++g_exit_pollers_released; }) != 0) {
+ std::_Exit(95);
+ }
+ g_exit_endpoint = new urma::UrmaEndpoint(nullptr);
+ g_exit_endpoint->_resource = new urma::UrmaResource;
+ g_exit_endpoint->_cq_sid = 7;
Review Comment:
`new urma::UrmaResource;` performs default-initialization, which leaves
POD/aggregate members uninitialized. If `UrmaResource` doesn't have a
user-provided default constructor that zeroes members, this can lead to
undefined behavior during endpoint cleanup/destruction (e.g., freeing garbage
handles). Prefer value-initialization (e.g., `new urma::UrmaResource()` / `{}`)
or construct a properly initialized resource via existing factory/init code
used in production.
##########
test/brpc_urma_unittest.cpp:
##########
@@ -21,26 +21,180 @@
#include <gflags/gflags.h>
#if BRPC_WITH_URMA
+#include <atomic>
+#include <cstdlib>
#include "butil/atomicops.h"
#include "butil/sys_byteorder.h"
#include "urma_api.h"
+#include "brpc/urma/urma_endpoint.h"
#include "brpc/urma/urma_handshake.h"
#include "brpc/urma/urma_handshake.pb.h"
#include "brpc/urma/urma_helper.h"
#include "urma_types.h"
using namespace brpc;
+DECLARE_int32(task_group_ntags);
+
namespace brpc {
namespace urma {
DECLARE_int32(urma_client_handshake_version);
+DECLARE_int32(urma_buffer_size);
+DECLARE_int32(urma_buffer_count);
+DECLARE_int32(urma_sq_size);
+DECLARE_int32(urma_rq_size);
+DECLARE_int32(urma_prepared_jetty_cnt);
extern bool g_skip_urma_init;
extern butil::atomic<bool> g_urma_available;
} // namespace urma
} // namespace brpc
+class UrmaPollingTest : public ::testing::TestWithParam<bool> {
+protected:
+ void SetUp() override {
+ urma::FLAGS_urma_use_polling = true;
+ urma::FLAGS_urma_poller_num = 2;
+ urma::FLAGS_urma_disable_bthread = GetParam();
+ ASSERT_EQ(0, urma::UrmaEndpoint::GlobalInitialize());
+ }
+
+ void TearDown() override {
+ urma::UrmaEndpoint::PollingModeRelease(0);
+ }
+
+ GFLAGS_NAMESPACE::FlagSaver _flags;
+};
+
+TEST_P(UrmaPollingTest, release_joins_and_can_be_repeated) {
+ std::atomic<int> initialized{0};
+ std::atomic<int> released{0};
+ ASSERT_EQ(0, urma::UrmaEndpoint::PollingModeInitialize(
+ 0, nullptr, [&] { ++initialized; }, [&] { ++released; }));
+ urma::UrmaEndpoint::PollingModeRelease(0);
+ EXPECT_EQ(2, initialized.load());
+ EXPECT_EQ(2, released.load());
+ auto& group = urma::UrmaEndpoint::PollerGroups()[0];
+ EXPECT_FALSE(group.running.load());
+ for (const auto& poller : group.pollers) {
+ EXPECT_EQ(INVALID_BTHREAD, poller.tid);
+ }
+ urma::UrmaEndpoint::PollingModeRelease(0);
+ EXPECT_EQ(2, released.load());
+
+ ASSERT_EQ(0, urma::UrmaEndpoint::PollingModeInitialize(
+ 0, nullptr, nullptr, [&] { ++released; }));
+ urma::UrmaEndpoint::PollingModeRelease(0);
+ EXPECT_EQ(4, released.load());
+}
+
+TEST_P(UrmaPollingTest, late_cq_operations_skip_stopped_pollers) {
+ ASSERT_EQ(0, urma::UrmaEndpoint::PollingModeInitialize(
+ 0, nullptr, nullptr, nullptr));
+ urma::UrmaEndpoint::PollingModeRelease(0);
+ urma::UrmaEndpoint endpoint(nullptr);
+ endpoint._cq_sid = 7;
Review Comment:
These tests directly depend on `UrmaEndpoint` internal state
(`PollerGroups()`, `PollerGroup::running/pollers`, and `_cq_sid`). This tightly
couples the tests to private implementation details and makes refactors risky.
Consider exposing a minimal, stable test/diagnostic API (e.g., a public query
that returns running/tid state, or a `#if defined(UNIT_TEST)` accessor / friend
test fixture) so the behavior can be validated without reaching into internals.
##########
test/brpc_urma_unittest.cpp:
##########
@@ -21,26 +21,180 @@
#include <gflags/gflags.h>
#if BRPC_WITH_URMA
+#include <atomic>
+#include <cstdlib>
#include "butil/atomicops.h"
#include "butil/sys_byteorder.h"
#include "urma_api.h"
+#include "brpc/urma/urma_endpoint.h"
#include "brpc/urma/urma_handshake.h"
#include "brpc/urma/urma_handshake.pb.h"
#include "brpc/urma/urma_helper.h"
#include "urma_types.h"
using namespace brpc;
+DECLARE_int32(task_group_ntags);
+
namespace brpc {
namespace urma {
DECLARE_int32(urma_client_handshake_version);
+DECLARE_int32(urma_buffer_size);
+DECLARE_int32(urma_buffer_count);
+DECLARE_int32(urma_sq_size);
+DECLARE_int32(urma_rq_size);
+DECLARE_int32(urma_prepared_jetty_cnt);
extern bool g_skip_urma_init;
extern butil::atomic<bool> g_urma_available;
} // namespace urma
} // namespace brpc
+class UrmaPollingTest : public ::testing::TestWithParam<bool> {
+protected:
+ void SetUp() override {
+ urma::FLAGS_urma_use_polling = true;
+ urma::FLAGS_urma_poller_num = 2;
+ urma::FLAGS_urma_disable_bthread = GetParam();
+ ASSERT_EQ(0, urma::UrmaEndpoint::GlobalInitialize());
+ }
+
+ void TearDown() override {
+ urma::UrmaEndpoint::PollingModeRelease(0);
+ }
+
+ GFLAGS_NAMESPACE::FlagSaver _flags;
+};
+
+TEST_P(UrmaPollingTest, release_joins_and_can_be_repeated) {
+ std::atomic<int> initialized{0};
+ std::atomic<int> released{0};
+ ASSERT_EQ(0, urma::UrmaEndpoint::PollingModeInitialize(
+ 0, nullptr, [&] { ++initialized; }, [&] { ++released; }));
+ urma::UrmaEndpoint::PollingModeRelease(0);
+ EXPECT_EQ(2, initialized.load());
+ EXPECT_EQ(2, released.load());
+ auto& group = urma::UrmaEndpoint::PollerGroups()[0];
+ EXPECT_FALSE(group.running.load());
+ for (const auto& poller : group.pollers) {
+ EXPECT_EQ(INVALID_BTHREAD, poller.tid);
+ }
Review Comment:
These tests directly depend on `UrmaEndpoint` internal state
(`PollerGroups()`, `PollerGroup::running/pollers`, and `_cq_sid`). This tightly
couples the tests to private implementation details and makes refactors risky.
Consider exposing a minimal, stable test/diagnostic API (e.g., a public query
that returns running/tid state, or a `#if defined(UNIT_TEST)` accessor / friend
test fixture) so the behavior can be validated without reaching into internals.
--
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]
