msg/async/rdma: handle buffers after close msg

fixing an issue of OSD’s going down during recovery process while traffic
is running in the background.
fix is making sure that messages previously pulled from the cqe and are
now stored in the socket queue “buffers” will always be handled and fd
will be notified even if cqe is empty.

issue: 1025030

Change-Id: I6ced6b8ee0152e233632700ccc9ec19970e66b6a
Signed-off-by: Alexander Mikheev <alexm@mellanox.com>
Signed-off-by: Adir Lev <adirl@mellanox.com>

diff --git a/src/msg/async/rdma/RDMAConnectedSocketImpl.cc b/src/msg/async/rdma/RDMAConnectedSocketImpl.cc
index 7fa4c5ba944417e5e5ce47ba8369696c5daeec86..7bd8a80fd0899adc1e57ac7adac036b35b43fc4f 100644 (file)
--- a/src/msg/async/rdma/RDMAConnectedSocketImpl.cc
+++ b/src/msg/async/rdma/RDMAConnectedSocketImpl.cc
@@ -201,7 +201,7 @@ int RDMAConnectedSocketImpl::try_connect(const entity_addr_t& peer_addr, const S
 }
 
 void RDMAConnectedSocketImpl::handle_connection() {
-  ldout(cct, 20) << __func__ << " QP: " << my_msg.qpn << " tcp_fd: " << tcp_fd << " fd: " << notify_fd << dendl;
+  ldout(cct, 20) << __func__ << " QP: " << my_msg.qpn << " tcp_fd: " << tcp_fd << " notify_fd: " << notify_fd << dendl;
   int r = infiniband->recv_msg(cct, tcp_fd, peer_msg);
   if (r < 0) {
     if (r != -EAGAIN) {
@@ -256,16 +256,25 @@ ssize_t RDMAConnectedSocketImpl::read(char* buf, size_t len)
   uint64_t i = 0;
   int r = ::read(notify_fd, &i, sizeof(i));
   ldout(cct, 20) << __func__ << " notify_fd : " << i << " in " << my_msg.qpn << " r = " << r << dendl;
-  if (error)
-    return -error;
   ssize_t read = 0;
   if (!buffers.empty())
     read = read_buffers(buf,len);
 
   std::vector<ibv_wc> cqe;
   get_wc(cqe);
-  if (cqe.empty())
-    return read == 0 ? -EAGAIN : read;
+  if (cqe.empty()) {
+    if (!buffers.empty()) {
+      notify();
+    }
+    if (read > 0) {
+      return read;
+    }
+    if (error) {
+      return -error;
+    } else {
+      return -EAGAIN;
+    }
+  }
 
   ldout(cct, 20) << __func__ << " poll queue got " << cqe.size() << " responses. QP: " << my_msg.qpn << dendl;
   for (size_t i = 0; i < cqe.size(); ++i) {
@@ -307,6 +316,10 @@ ssize_t RDMAConnectedSocketImpl::read(char* buf, size_t len)
     submit(false);
   }
 
+  if (!buffers.empty()) {
+    notify();
+  }
+
   if (read == 0 && error)
     return -error;
   return read == 0 ? -EAGAIN : read;