diff --git a/src/brpc/rdma/rdma_endpoint.cpp b/src/brpc/rdma/rdma_endpoint.cpp index ca6f3929af..d0fad7536f 100644 --- a/src/brpc/rdma/rdma_endpoint.cpp +++ b/src/brpc/rdma/rdma_endpoint.cpp @@ -1434,6 +1434,11 @@ void RdmaEndpoint::PollCq(Socket* m) { if (Socket::Address(ep->_socket->id(), &s) < 0) { return; } + // A queued callback may outlive Reset() and see the main Socket after + // it has been revived with another CQ. + if (m->id() != ep->_cq_sid) { + return; + } auto* rdma_transport = static_cast(s->_transport.get()); CHECK(ep == rdma_transport->_rdma_ep); diff --git a/test/brpc_rdma_unittest.cpp b/test/brpc_rdma_unittest.cpp index e30ae09f35..0435641e74 100644 --- a/test/brpc_rdma_unittest.cpp +++ b/test/brpc_rdma_unittest.cpp @@ -186,6 +186,40 @@ class RdmaRpcTest : public RdmaTest, int _saved_handshake_version = 2; }; +TEST_F(RdmaTest, stale_cq_callback_does_not_poll_new_generation) { + SocketOptions main_options; + main_options.socket_mode = SOCKET_MODE_RDMA; + SocketId main_sid; + ASSERT_EQ(0, Socket::Create(main_options, &main_sid)); + + SocketUniquePtr main_socket; + ASSERT_EQ(0, Socket::Address(main_sid, &main_socket)); + RdmaTransport* transport = + static_cast(main_socket->_transport.get()); + rdma::RdmaEndpoint* ep = transport->_rdma_ep; + + SocketOptions cq_options; + cq_options.user = ep; + SocketId stale_cq_sid; + SocketId current_cq_sid; + ASSERT_EQ(0, Socket::Create(cq_options, &stale_cq_sid)); + ASSERT_EQ(0, Socket::Create(cq_options, ¤t_cq_sid)); + + SocketUniquePtr stale_cq_socket; + SocketUniquePtr current_cq_socket; + ASSERT_EQ(0, Socket::Address(stale_cq_sid, &stale_cq_socket)); + ASSERT_EQ(0, Socket::Address(current_cq_sid, ¤t_cq_socket)); + ep->_cq_sid = current_cq_sid; + + rdma::RdmaEndpoint::PollCq(stale_cq_socket.get()); + + stale_cq_socket->_user = NULL; + current_cq_socket->_user = NULL; + stale_cq_socket->SetFailed(); + current_cq_socket->SetFailed(); + main_socket->SetFailed(); +} + TEST_F(RdmaTest, client_close_before_hello_send) { StartServer();