UCT/IB/RC/VERBS: Release the descriptor of a failed receive completion

src/uct/ib/rc/verbs/rc_verbs_impl.h

@@ -94,11 +94,12 @@ uct_rc_verbs_iface_poll_rx_common(uct_rc_verbs_iface_t *iface)
         desc = (uct_ib_iface_recv_desc_t *)(uintptr_t)wc[i].wr_id;
         hdr  = (uct_rc_hdr_t *)uct_ib_iface_recv_desc_hdr(&iface->super.super, desc);
         if (ucs_unlikely(wc[i].status != IBV_WC_SUCCESS)) {
-            if (wc[i].status == IBV_WC_REM_ABORT_ERR) {
-                continue;
-            }
-            /* we can get flushed messages during ep destroy */
-            if (wc[i].status == IBV_WC_WR_FLUSH_ERR) {
+            /* A failed receive does not deliver an active message, so return
+             * its descriptor to the pool. A flushed one arrives during
+             * endpoint destruction. */
+            if ((wc[i].status == IBV_WC_REM_ABORT_ERR) ||
+                (wc[i].status == IBV_WC_WR_FLUSH_ERR)) {
+                ucs_mpool_put_inline(desc);
                 continue;
             }
             UCT_IB_IFACE_VERBS_COMPLETION_FATAL("receive", &iface->super.super, i, wc);

test/gtest/uct/ib/test_rc.cc

@@ -181,6 +181,94 @@ UCS_TEST_SKIP_COND_P(test_rc, relaxed_order_required_rejects_verbs,
     EXPECT_EQ(UCS_ERR_UNSUPPORTED, status);
 }
 
+static struct {
+    struct ibv_cq      *cq;
+    void               *desc;
+    enum ibv_wc_status status;
+    int                inject;
+} failed_rx;
+
+static int failed_rx_poll_cq(struct ibv_cq *cq, int num_entries,
+                             struct ibv_wc *wc)
+{
+    if ((cq != failed_rx.cq) || (num_entries < 1) || !failed_rx.inject) {
+        return 0;
+    }
+
+    failed_rx.inject = 0;
+    wc[0].status     = failed_rx.status;
+    wc[0].wr_id      = (uintptr_t)failed_rx.desc;
+    wc[0].qp_num     = 0;
+    wc[0].byte_len   = 0;
+    wc[0].slid       = 0;
+    wc[0].imm_data   = 0;
+    return 1;
+}
+
+static bool is_desc_in_pool(ucs_mpool_t *mp, void *desc)
+{
+    ucs_mpool_elem_t *elem;
+
+    for (elem = mp->freelist; elem != NULL; elem = elem->next) {
+        VALGRIND_MAKE_MEM_DEFINED(elem, sizeof(*elem));
+        if (UCS_PTR_BYTE_OFFSET(elem, sizeof(*elem)) == desc) {
+            return true;
+        }
+    }
+
+    return false;
+}
+
+/* Synthesize an error receive completion by replacing the poll_cq() operation
+ * of the device context: receive work requests go to the shared receive queue,
+ * which verbs can not move to an error state. */
+class test_rc_failed_rx : public test_rc {
+public:
+    void check_failed_rx(enum ibv_wc_status status)
+    {
+        uct_rc_iface_t *rc = rc_iface(m_e2);
+        void *desc;
+
+        if (ucs_derived_of(m_e2->iface(), uct_rc_verbs_iface_t)->srq == NULL) {
+            UCS_TEST_SKIP_R("SRQ is disabled");
+        }
+
+        desc = ucs_mpool_get(&rc->rx.mp);
+        if (desc == NULL) {
+            UCS_TEST_SKIP_R("no receive descriptor is available");
+        }
+
+        failed_rx.cq     = rc->super.cq[UCT_IB_DIR_RX];
+        failed_rx.desc   = desc;
+        failed_rx.status = status;
+        failed_rx.inject = 1;
+
+        ucs::mock mock;
+        mock.setup(&failed_rx.cq->context->ops.poll_cq, failed_rx_poll_cq);
+
+        progress();
+
+        ASSERT_FALSE(failed_rx.inject);
+        EXPECT_TRUE(is_desc_in_pool(&rc->rx.mp, desc))
+                << "receive descriptor leaked";
+    }
+};
+
+/* The repost batch must stay above the one completion reported here, so that
+ * the released descriptor is not posted again, and at or below 25, since
+ * uct_rc_verbs_iface_post_recv_always() allocates a batch on the stack. */
+UCS_TEST_P(test_rc_failed_rx, flushed, "IB_RX_MAX_BATCH=16")
+{
+    check_failed_rx(IBV_WC_WR_FLUSH_ERR);
+}
+
+UCS_TEST_P(test_rc_failed_rx, remote_abort, "IB_RX_MAX_BATCH=16")
+{
+    check_failed_rx(IBV_WC_REM_ABORT_ERR);
+}
+
+_UCT_INSTANTIATE_TEST_CASE(test_rc_failed_rx, rc_verbs)
+
 UCT_INSTANTIATE_RC_TEST_CASE(test_rc)
 
 #ifdef HAVE_MLX5_DV