Postpone handling incoming messages ?

src/messenger.h

@@ -134,6 +134,7 @@ protected:
     msgr_rdma_context_t *rdma_context = NULL;
     uint64_t rdma_max_sge = 0, rdma_max_send = 0, rdma_max_recv = 0;
     uint64_t rdma_max_msg = 0;
+    std::vector<rdma_hb_t> rdma_handle_buffers;
 #endif
 
     std::vector<int> read_ready_clients;

src/msgr_rdma.cpp

@@ -492,13 +492,9 @@ void osd_messenger_t::handle_rdma_events()
             if (!is_send)
             {
                 rc->cur_recv--;
-                if (!handle_read_buffer(cl, rc->recv_buffers[rc->next_recv_buf], wc[i].byte_len))
+                rdma_handle_buffers.push_back((rdma_hb_t){ .peer_fd = client_id, .buf = rc->recv_buffers[0], .len = wc[i].byte_len });
-                {
+                rc->recv_buffers.erase(rc->recv_buffers.begin(), rc->recv_buffers.begin()+1);
-                    // handle_read_buffer may stop the client
+                try_recv_rdma(cl);
-                    continue;
-                }
-                try_recv_rdma_wr(cl, rc->recv_buffers[rc->next_recv_buf]);
-                rc->next_recv_buf = (rc->next_recv_buf+1) % rc->recv_buffers.size();
             }
             else
             {
@@ -547,6 +543,16 @@ void osd_messenger_t::handle_rdma_events()
             }
         }
     } while (event_count > 0);
+    for (auto & hb: rdma_handle_buffers)
+    {
+        auto cl_it = clients.find(hb.peer_fd);
+        if (cl_it != clients.end())
+        {
+            handle_read_buffer(cl_it->second, hb.buf, hb.len);
+        }
+        free(hb.buf);
+    }
+    rdma_handle_buffers.clear();
     for (auto cb: set_immediate)
     {
         cb();

src/msgr_rdma.h

@@ -57,3 +57,10 @@ struct msgr_rdma_connection_t
     static msgr_rdma_connection_t *create(msgr_rdma_context_t *ctx, uint32_t max_send, uint32_t max_recv, uint32_t max_sge, uint32_t max_msg);
     int connect(msgr_rdma_address_t *dest);
 };
+
+struct rdma_hb_t
+{
+    int peer_fd;
+    void *buf;
+    uint64_t len;
+};