UCT/UD: Remove stale private UD connections

src/uct/ib/ud/base/ud_ep.c

@@ -22,11 +22,6 @@
 #include <ucs/vfs/base/vfs_obj.h>
 
 
-/* Must be less then peer_timeout to avoid false positive errors taking into
- * account timer resolution and not too small to avoid performance degradation
- */
-#define UCT_UD_SLOW_TIMER_MAX_TICK(_iface)  ((_iface)->config.peer_timeout / 3)
-
 static void uct_ud_ep_do_pending_ctl(uct_ud_ep_t *ep, uct_ud_iface_t *iface);
 
 static void uct_ud_peer_name(uct_ud_peer_name_t *peer)
@@ -124,6 +119,7 @@ static void uct_ud_ep_reset(uct_ud_ep_t *ep)
                                         uct_ud_iface_t)->config.max_window;
     ep->tx.acked_psn   = UCT_UD_INITIAL_PSN - 1;
     ep->tx.pending.ops = UCT_UD_EP_OP_NONE;
+    ep->tx.send_time   = ucs_get_time();
     uct_ud_ep_reset_max_psn(ep);
     ucs_queue_head_init(&ep->tx.window);
 
@@ -136,6 +132,7 @@ static void uct_ud_ep_reset(uct_ud_ep_t *ep)
     ep->rx.acked_psn = UCT_UD_INITIAL_PSN - 1;
     ucs_frag_list_init(ep->tx.psn-1, &ep->rx.ooo_pkts, 0 /*TODO: ooo support */
                        UCS_STATS_ARG(ep->super.stats));
+    ep->rx.check_sn  = ucs_frag_list_sn(&ep->rx.ooo_pkts);
 }
 
 static ucs_status_t uct_ud_ep_free_by_timeout(uct_ud_ep_t *ep,
@@ -282,7 +279,10 @@ static unsigned uct_ud_ep_deferred_timeout_handler(void *arg)
     }
 
     if (ep->flags & UCT_UD_EP_FLAG_PRIVATE) {
-        ucs_assert(ucs_queue_is_empty(&ep->tx.window));
+        /* The peer is gone, so stop matching the endpoint by new connections
+         * already now, it is destroyed only after the linger timeout */
+        uct_ud_iface_cep_remove_ep(iface, ep);
+        uct_ud_ep_purge(ep, UCS_ERR_ENDPOINT_TIMEOUT);
         uct_ep_destroy(&ep->super.super);
         goto out;
     }
@@ -338,6 +338,31 @@ static void uct_ud_ep_handle_timeout(uct_ud_ep_t *ep)
     }
 }
 
+/* A private endpoint is not owned by the upper layer, so nobody checks whether
+ * its peer is still there. Probe the peer once it becomes silent, to release an
+ * endpoint whose peer is gone. */
+static void uct_ud_ep_check_peer(uct_ud_iface_t *iface, uct_ud_ep_t *ep)
+{
+    uct_ud_psn_t check_sn = ucs_frag_list_sn(&ep->rx.ooo_pkts);
+    ucs_time_t idle_time  = ucs_get_time() - ep->tx.send_time;
+
+    if (check_sn != ep->rx.check_sn) {
+        /* The peer is sending, so it does not have to be probed */
+        ep->rx.check_sn = check_sn;
+    } else if (idle_time >= iface->config.keepalive_interval) {
+        /* Ask for an ack, otherwise the probe is acked only after the resend
+         * flow requests it. An alive peer acks the probe, otherwise the
+         * endpoint is released by the peer timeout flow. */
+        uct_ud_ep_ctl_op_add(iface, ep, UCT_UD_EP_OP_ACK_REQ);
+        if (uct_ud_ep_check(&ep->super.super, 0, NULL) == UCS_OK) {
+            return;
+        }
+    }
+
+    ucs_wtimer_add(&iface->tx.timer, &ep->timer,
+                   iface->config.keepalive_interval);
+}
+
 static void uct_ud_ep_timer(ucs_wtimer_t *self)
 {
     uct_ud_ep_t    *ep    = ucs_container_of(self, uct_ud_ep_t, timer);
@@ -354,6 +379,8 @@ static void uct_ud_ep_timer(ucs_wtimer_t *self)
             if (status == UCS_INPROGRESS) {
                 goto timer_backoff;
             }
+        } else if (ep->flags & UCT_UD_EP_FLAG_PRIVATE) {
+            uct_ud_ep_check_peer(iface, ep);
         }
         return;
     }

src/uct/ib/ud/base/ud_ep.h

@@ -258,6 +258,7 @@ struct uct_ud_ep {
     } tx;
     struct {
         uct_ud_psn_t        acked_psn;    /* Last psn we acked */
+        uct_ud_psn_t        check_sn;     /* Last psn seen by liveness check */
         ucs_frag_list_t     ooo_pkts;     /* Out of order packets that can not be processed yet,
                                             also keeps last psn we successfully received and processed */
         UCS_STATS_NODE_DECLARE(stats)

src/uct/ib/ud/base/ud_iface.c

@@ -415,6 +415,7 @@ UCS_CLASS_INIT_FUNC(uct_ud_iface_t, uct_ud_iface_ops_t *ops,
                     const uct_ud_iface_config_t *config,
                     uct_ib_iface_init_attr_t *init_attr)
 {
+    ucs_time_t keepalive_interval;
     ucs_status_t status;
     size_t data_size;
     int mtu;
@@ -493,6 +494,15 @@ UCS_CLASS_INIT_FUNC(uct_ud_iface_t, uct_ud_iface_ops_t *ops,
         self->tx.tick = ucs_time_from_sec(config->timer_tick);
     }
 
+    /* A private endpoint is not owned by the upper layer, so it has to check
+     * its peer by itself, reusing the keepalive interval of the upper layer */
+    keepalive_interval = (params->field_mask &
+                          UCT_IFACE_PARAM_FIELD_KEEPALIVE_INTERVAL) ?
+                         params->keepalive_interval :
+                         UCT_UD_SLOW_TIMER_MAX_TICK(self);
+    self->config.keepalive_interval = ucs_max(keepalive_interval,
+                                              self->tx.tick);
+
     if (config->timer_backoff < UCT_UD_MIN_TIMER_TIMER_BACKOFF) {
         ucs_error("The timer back off must be >= %lf (%lf)",
                   UCT_UD_MIN_TIMER_TIMER_BACKOFF, config->timer_backoff);

src/uct/ib/ud/base/ud_iface.h

@@ -33,6 +33,12 @@ BEGIN_C_DECLS
 #define UCT_UD_MIN_TIMER_TIMER_BACKOFF 1.0
 
 
+/* Must be less then peer_timeout to avoid false positive errors taking into
+ * account timer resolution and not too small to avoid performance degradation
+ */
+#define UCT_UD_SLOW_TIMER_MAX_TICK(_iface)  ((_iface)->config.peer_timeout / 3)
+
+
 /** @file ud_iface.h */
 
 enum {
@@ -193,6 +199,7 @@ struct uct_ud_iface {
     struct {
         ucs_time_t           linger_timeout;
         ucs_time_t           peer_timeout;
+        ucs_time_t           keepalive_interval;
         ucs_time_t           min_poke_time;
         unsigned             tx_qp_len;
         unsigned             rx_qp_len;

test/gtest/uct/ib/test_ud.cc

@@ -164,6 +164,22 @@ class test_ud : public ud_base_test {
         EXPECT_EQ(3, ep(m_e1)->tx.acked_psn);
     }
 
+    uct_ud_ep_t *find_private_ep(entity *e, uint32_t dest_ep_id) {
+        unsigned index;
+        void *elem;
+
+        ucs_ptr_array_for_each(elem, index, &iface(e)->eps) {
+            uct_ud_ep_t *ud_ep = static_cast<uct_ud_ep_t*>(elem);
+
+            if ((ud_ep->flags & UCT_UD_EP_FLAG_PRIVATE) &&
+                (ud_ep->dest_ep_id == dest_ep_id)) {
+                return ud_ep;
+            }
+        }
+
+        return NULL;
+    }
+
 
     static volatile uint32_t     rx_ack_count;
     static volatile uint32_t     rx_drop_count;
@@ -914,6 +930,72 @@ UCS_TEST_P(test_ud, stale_crep_on_reused_ep_id, "UD_LINGER_TIMEOUT=1s") {
     EXPECT_FALSE(ucs_queue_is_empty(&ep(m_e1)->tx.window));
 }
 
+/* Stale private endpoint must be released after its peer is gone, so that it is
+ * not reused by a connection which is matched by the same conn_sn. */
+UCS_TEST_P(test_ud, stale_private_ep_reuse, "UD_LINGER_TIMEOUT=1s",
+           "UD_TIMEOUT=1s") {
+    void *ud_ep GTEST_ATTRIBUTE_UNUSED_;
+
+    m_e1->connect_to_iface(0, *m_e2);
+    flush();
+
+    /* m_e2 has a private endpoint which is connected to the endpoint of m_e1 */
+    uint32_t old_id         = ep(m_e1)->ep_id;
+    uct_ud_ep_t *private_ep = find_private_ep(m_e2, old_id);
+    ASSERT_TRUE(private_ep != NULL);
+
+    uint32_t private_index = uct_ud_ep_id_to_index(private_ep->ep_id);
+    m_e1->destroy_ep(0);
+
+    /* The second connection makes m_e2 hold 2 private endpoints, where the
+     * stale one is matched by the next conn_sn of m_e2 */
+    m_e1->connect_to_iface(0, *m_e2);
+    flush();
+    ASSERT_NE(old_id, ep(m_e1)->ep_id);
+
+    wait_for_ep_destroyed(iface(m_e1), uct_ud_ep_id_to_index(old_id));
+
+    /* The peer of the stale private endpoint is gone, so it must be released */
+    wait_for_ep_destroyed(iface(m_e2), private_index);
+    EXPECT_FALSE(ucs_ptr_array_lookup(&iface(m_e2)->eps, private_index, ud_ep));
+
+    /* The private endpoint of the live connection must be kept */
+    EXPECT_TRUE(find_private_ep(m_e2, ep(m_e1)->ep_id) != NULL);
+
+    m_e2->connect_to_iface(0, *m_e1);
+    flush();
+
+    EXPECT_NE(UCT_UD_EP_NULL_ID, ep(m_e2)->dest_ep_id);
+    EXPECT_NE(old_id, ep(m_e2)->dest_ep_id);
+}
+
+/* Private endpoint must be kept as long as its peer is alive. */
+UCS_TEST_P(test_ud, private_ep_keepalive, "UD_TIMEOUT=1s") {
+    ucs_time_t deadline = ucs_get_time() + ucs_time_from_sec(3);
+    void *ud_ep GTEST_ATTRIBUTE_UNUSED_;
+
+    m_e1->connect_to_iface(0, *m_e2);
+    flush();
+
+    uct_ud_ep_t *private_ep = find_private_ep(m_e2, ep(m_e1)->ep_id);
+    ASSERT_TRUE(private_ep != NULL);
+
+    uint32_t private_index = uct_ud_ep_id_to_index(private_ep->ep_id);
+
+    /* Stay idle for several peer timeouts, the peer is alive and must be
+     * detected as such */
+    while (ucs_get_time() < deadline) {
+        short_progress_loop();
+    }
+
+    ASSERT_TRUE(ucs_ptr_array_lookup(&iface(m_e2)->eps, private_index, ud_ep));
+    EXPECT_EQ(ep(m_e1)->ep_id, private_ep->dest_ep_id);
+
+    /* The connection is still usable */
+    EXPECT_UCS_OK(tx(m_e1));
+    flush();
+}
+
 #if UCT_UD_EP_DEBUG_HOOKS
 /* Simulate loss of ctl packets during simultaneous CREQs.
  * Use-case: CREQ and CREP packets from m_e2 to m_e1 are lost.

test/gtest/uct/ib/ud_base.cc

@@ -100,5 +100,5 @@ void ud_base_test::wait_for_ep_destroyed(uct_ud_iface_t *iface, uint32_t ep_id)
     wait_for_cond([iface, ep_id]() {
         void *ud_ep GTEST_ATTRIBUTE_UNUSED_;
         return !ucs_ptr_array_lookup(&iface->eps, ep_id, ud_ep);
-    }, []() { usleep(1000); }, 60.0);
+    }, [this]() { short_progress_loop(); }, 60.0);
 }