Copilot commented on code in PR #3634:
URL: https://github.com/apache/brpc/pull/3634#discussion_r4396798074
##########
test/brpc_urma_unittest.cpp:
##########
@@ -21,26 +21,182 @@
#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::FLAGS_urma_use_polling = false;
+ 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;
+ urma::FLAGS_urma_use_polling = false;
Review Comment:
Setting this flag to false makes `UrmaEndpoint::DeallocateResources()` skip
`PollerRemoveCqSid()` entirely (`urma_endpoint.cpp:518–519`). Consequently,
deleting `g_exit_endpoint` in `CheckPollingExit()` never exercises the late
CQ-removal path that this regression test is intended to protect. Leave polling
enabled here so the endpoint destructor reaches the stopped poller group;
`global_shutdown_joins_all_tags` already covers shutdown with the flag disabled.
##########
src/brpc/urma/urma_endpoint.cpp:
##########
@@ -1737,11 +1752,16 @@ void UrmaEndpoint::PollerAddCqSid() {
}
void UrmaEndpoint::PollerRemoveCqSid() {
- if (_cq_sid == INVALID_SOCKET_ID || _poller_groups.empty() ||
- _poller_tag >= _poller_groups.size()) {
+ auto& poller_groups = PollerGroups();
+ if (_cq_sid == INVALID_SOCKET_ID || poller_groups.empty() ||
+ _poller_tag >= poller_groups.size()) {
return;
}
- auto& pollers = _poller_groups[_poller_tag].pollers;
+ auto& group = poller_groups[_poller_tag];
+ if (!group.running.load(butil::memory_order_acquire)) {
+ return;
Review Comment:
If a worker stops before draining a queued ADD and the endpoint is then
deleted, this check drops the matching REMOVE but leaves the ADD queued for
restart. On restart, that stale ID remains in `cq_sids` because failed
`Socket::Address()` lookups do not erase it; with the default
`urma_poller_yield=false`, it also prevents yielding and makes the poller spin
on a dead socket. Synchronize queue cleanup with producers so
stopped-generation operations cannot survive into a restart, or preserve their
matching removals.
--
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]
