wwbmmm commented on code in PR #3634:
URL: https://github.com/apache/brpc/pull/3634#discussion_r4348989733
##########
src/brpc/urma/urma_endpoint.cpp:
##########
@@ -1705,11 +1712,14 @@ int UrmaEndpoint::PollingModeInitialize(
}
void UrmaEndpoint::PollingModeRelease(bthread_tag_t tag) {
- if (!FLAGS_urma_use_polling || tag >= _poller_groups.size()) {
+ auto& poller_groups = PollerGroups();
+ if (!FLAGS_urma_use_polling || tag >= poller_groups.size()) {
Review Comment:
[replied by brpc-oncall robot] The exit path is still gated on
`FLAGS_urma_use_polling`: `ReleasePollingModeAtExit()` is registered only when
the flag was true at init time, but `PollingModeRelease()` re-reads the flag's
current value. If anything clears the flag before exit (application
re-configuration, a test using `FlagSaver`), the atexit handler silently skips
the join and the ordering hazard this PR fixes comes back. The flag check is
not needed for safety here because `tag >= poller_groups.size()` already
returns for the empty vector case; consider dropping it (at least for the
global/exit path) so shutdown depends only on `group.running`.
##########
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];
Review Comment:
[replied by brpc-oncall robot] The new tests access members that are private
in `urma_endpoint.h`: `PollerGroups()` (and the nested `PollerGroup` / `Poller`
/ `CqSidOp` types), `PollerAddCqSid()` / `PollerRemoveCqSid()`, `_cq_sid` and
`_resource` (all below the `private:` at line 190 of the header).
`UrmaEndpoint` has no friend declaration for these test classes, so with
`BRPC_WITH_URMA=1` this file does not compile (`'PollerGroups' is private`
etc.). The current CI has no URMA build, so the green CI does not catch this.
Please either declare the concrete test classes as friends in
`urma_endpoint.h`, or add a small test-only accessor (e.g. under `#if
defined(UNIT_TEST)`) that exposes the per-group `running`/`tid` state and lets
a test endpoint set `_cq_sid`/`_resource`.
##########
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;
Review Comment:
[replied by brpc-oncall robot] False positive: `UrmaResource` gives every
member a default member initializer (`next{nullptr}`, `jfc{nullptr}`,
`jfce{nullptr}`, `jfr`, `jetty`, `remote_jetty`, `remote_seg`), so `new
urma::UrmaResource` runs the defaulted default constructor and initializes all
handles to nullptr. There are no uninitialized members here; no change is
needed.
--
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]
