Deallocate polling cq

Author: chenBright

src/brpc/rdma/rdma_endpoint.cpp

@@ -230,6 +230,7 @@ void RdmaEndpoint::Reset() {
     _rq_received = 0;
     _local_window_capacity = 0;
     _remote_window_capacity = 0;
+    _sq_imm_window_size = 0;
     _remote_rq_window_size.store(0, butil::memory_order_relaxed);
     _sq_window_size.store(0, butil::memory_order_relaxed);
     _new_rq_wrs.store(0, butil::memory_order_relaxed);
@@ -841,8 +842,7 @@ ssize_t RdmaEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata) {
             }
 
             ssize_t len = data->cut_into_sglist_and_iobuf(
-                sglist, &sge_index, to, max_sge,
-                _remote_recv_block_size - this_len);
+                sglist, &sge_index, to, max_sge, _remote_recv_block_size - this_len);
             if (len < 0) {
                 return -1;
             }
@@ -957,6 +957,10 @@ int RdmaEndpoint::SendImm(uint32_t imm) {
         LOG(WARNING) << "Fail to ibv_post_send: " << berror(err) << " " << oss.str();
         return -1;
     }
+
+    // `_sq_imm_window_size' will never be negative.
+    // Because IMM can only be sent if
+    // `_sq_imm_window_size` is greater than 0.
     _sq_imm_window_size -= 1;
     return 0;
 }
@@ -967,6 +971,7 @@ ssize_t RdmaEndpoint::HandleCompletion(ibv_wc& wc) {
     case IBV_WC_SEND: {  // send completion
         if (0 == wc.wr_id) {
             _sq_imm_window_size += 1;
+            // If there are any unacknowledged recvs, send an ack.
             SendAck(0);
             return 0;
         }
@@ -1298,14 +1303,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) {
+static 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 +1330,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 +1345,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,
@@ -1350,6 +1360,7 @@ void RdmaEndpoint::DeallocateResources() {
 
         }
 
+        _resource->polling_cq = NULL;
         _resource->send_cq = NULL;
         _resource->recv_cq = NULL;
         _resource->comp_channel = NULL;
@@ -1369,12 +1380,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;
@@ -1399,7 +1404,7 @@ int RdmaEndpoint::GetAndAckEvents(SocketUniquePtr& s) {
         }
         if (cq == _resource->send_cq) {
             ++_send_cq_events;
-        } else {
+        } else if (cq == _resource->recv_cq) {
             ++_recv_cq_events;
         }
     }
@@ -1418,7 +1423,8 @@ int RdmaEndpoint::GetAndAckEvents(SocketUniquePtr& s) {
 
 int RdmaEndpoint::ReqNotifyCq(bool send_cq) {
     errno = ibv_req_notify_cq(
-        send_cq ? _resource->send_cq : _resource->recv_cq, send_cq ? 0 : 1);
+        send_cq ? _resource->send_cq : _resource->recv_cq,
+        send_cq ? 0 : 1);
     if (0 != errno) {
         const int saved_errno = errno;
         PLOG(WARNING) << "Fail to arm" << (send_cq ? "send" : "recv")

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));