diff --git a/src/uct/ib/ud/base/ud_ep.c b/src/uct/ib/ud/base/ud_ep.c index 5c19f38d638..245421bf522 100644 --- a/src/uct/ib/ud/base/ud_ep.c +++ b/src/uct/ib/ud/base/ud_ep.c @@ -22,11 +22,6 @@ #include -/* 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,7 @@ 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)); + uct_ud_ep_purge(ep, UCS_ERR_ENDPOINT_TIMEOUT); uct_ep_destroy(&ep->super.super); goto out; } @@ -330,6 +327,11 @@ static void uct_ud_ep_handle_timeout(uct_ud_ep_t *ep) uct_ud_iface_t *iface = ucs_derived_of(ep->super.super.iface, uct_ud_iface_t); + if (ep->flags & UCT_UD_EP_FLAG_PRIVATE) { + /* The peer is gone, stop matching the endpoint by new connections */ + uct_ud_iface_cep_remove_ep(iface, ep); + } + ucs_callbackq_add_oneshot(&iface->super.super.worker->super.progress_q, ep, uct_ud_ep_deferred_timeout_handler, ep); if (iface->async.event_cb != NULL) { @@ -338,6 +340,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 +381,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; } diff --git a/src/uct/ib/ud/base/ud_ep.h b/src/uct/ib/ud/base/ud_ep.h index 6b24cd16c30..430a1073df7 100644 --- a/src/uct/ib/ud/base/ud_ep.h +++ b/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) diff --git a/src/uct/ib/ud/base/ud_iface.c b/src/uct/ib/ud/base/ud_iface.c index ad43fb5e7e8..ec4b16b911e 100644 --- a/src/uct/ib/ud/base/ud_iface.c +++ b/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); diff --git a/src/uct/ib/ud/base/ud_iface.h b/src/uct/ib/ud/base/ud_iface.h index 4578789ec8c..61490701ef2 100644 --- a/src/uct/ib/ud/base/ud_iface.h +++ b/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 than 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; diff --git a/test/gtest/uct/ib/test_ud.cc b/test/gtest/uct/ib/test_ud.cc index ada6cae0f73..ea74c70da4f 100644 --- a/test/gtest/uct/ib/test_ud.cc +++ b/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(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. diff --git a/test/gtest/uct/ib/ud_base.cc b/test/gtest/uct/ib/ud_base.cc index 4436a05ef6b..88b35e0f07c 100644 --- a/test/gtest/uct/ib/ud_base.cc +++ b/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); }