Deallocate polling cq

Author: chenBright

src/brpc/rdma/rdma_endpoint.cpp

@@ -1298,14 +1298,11 @@ int RdmaEndpoint::BringUpQp(uint16_t lid, ibv_gid gid, uint32_t qp_num) {
     return 0;
 }
 
-void RdmaEndpoint::DeallocateCq(ibv_cq* cq, unsigned int cq_events) {
+void DeallocateCq(ibv_cq* cq) {
     if (NULL == cq) {
         return;
     }
 
-    if (cq_events > 0) {
-        IbvAckCqEvents(cq, cq_events);
-    }
     int err = IbvDestroyCq(cq);
     LOG_IF(WARNING, 0 != err) << "Fail to destroy CQ: " << berror(err);
 }
@@ -1328,6 +1325,13 @@ void RdmaEndpoint::DeallocateResources() {
         }
     }
 
+    if (NULL != _resource->send_cq) {
+        IbvAckCqEvents(_resource->send_cq, _send_cq_events);
+    }
+    if (NULL != _resource->recv_cq) {
+        IbvAckCqEvents(_resource->recv_cq, _recv_cq_events);
+    }
+
     bool remove_consumer = true;
     if (!move_to_rdma_resource_list) {
         if (NULL != _resource->qp) {
@@ -1336,8 +1340,9 @@ void RdmaEndpoint::DeallocateResources() {
             _resource->qp = NULL;
         }
 
-        DeallocateCq(_resource->send_cq, _send_cq_events);
-        DeallocateCq(_resource->recv_cq, _recv_cq_events);
+        DeallocateCq(_resource->polling_cq);
+        DeallocateCq(_resource->send_cq);
+        DeallocateCq(_resource->recv_cq);
 
         if (NULL != _resource->comp_channel) {
             // Destroy send_comp_channel will destroy this fd,
@@ -1369,12 +1374,6 @@ void RdmaEndpoint::DeallocateResources() {
     }
 
     if (move_to_rdma_resource_list) {
-        if (NULL != _resource->send_cq) {
-            IbvAckCqEvents(_resource->send_cq, _send_cq_events);
-        }
-        if (NULL != _resource->recv_cq) {
-            IbvAckCqEvents(_resource->recv_cq, _recv_cq_events);
-        }
         BAIDU_SCOPED_LOCK(*g_rdma_resource_mutex);
         _resource->next = g_rdma_resource_list;
         g_rdma_resource_list = _resource;

src/brpc/rdma/rdma_endpoint.h

@@ -134,9 +134,6 @@ friend class Socket;
     // Process handshake at the server
     static void* ProcessHandshakeAtServer(void* arg);
 
-    // Deallocate CQ resource.
-    static void DeallocateCq(ibv_cq* cq, unsigned int cq_events);
-
     // Allocate resources
     // Return 0 if success, -1 if failed and errno set
     int AllocateResources();
@@ -277,8 +274,6 @@ friend class Socket;
     butil::atomic<uint16_t> _sq_window_size;
     // The number of new WRs posted in the local Recv Queue
     butil::atomic<uint16_t> _new_rq_wrs;
-    // The number of inflight send IMM.
-    butil::atomic<uint16_t> _imm_inflight;
 
     // butex for inform read events on TCP fd during handshake
     butil::atomic<int> *_read_butex;

test/brpc_rdma_unittest.cpp

@@ -209,8 +209,7 @@ TEST_F(RdmaTest, client_hello_msg_invalid_magic_str) {
 
     uint8_t data[RDMA_HELLO_MSG_LEN];
     memcpy(data, "PRPC", 4);  // send as normal baidu_std protocol
-    memset(data + 4, 0, 32);
-    ASSERT_EQ(38, write(sockfd, data, 38));
+    ASSERT_EQ(4, write(sockfd, data, 4));
     usleep(100000);  // wait for server to handle the msg
     ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, s->_rdma_ep->_state);
 
@@ -660,9 +659,6 @@ TEST_F(RdmaTest, client_send_data_on_tcp_after_ack_send) {
     ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, s->_rdma_ep->_state);
     ASSERT_EQ(sizeof(flags), write(sockfd1, &flags, sizeof(flags)));
     usleep(100000);
-    ASSERT_EQ(rdma::RdmaEndpoint::FALLBACK_TCP, s->_rdma_ep->_state);
-    close(sockfd1);
-    usleep(100000);  // wait for server to handle the msg
     ASSERT_EQ(NULL, GetSocketFromServer(0));
 
     butil::fd_guard sockfd2(socket(AF_INET, SOCK_STREAM, 0));