Merge pull request #6891 from dmitrygx/topic/ucp/ka_iter

dmitrygx · web-flow · commit 034b2389a94b · 2021-06-10T13:25:25.000+03:00
UCP/CORE: Ensure that KA isn't done several times for the same EP during the round
diff --git a/src/ucp/core/ucp_ep.c b/src/ucp/core/ucp_ep.c
@@ -164,23 +164,23 @@ ucs_status_t ucp_ep_create_base(ucp_worker_h worker, const char *peer_name,
         goto err_free_ep;
     }
 
-    ep->refcount                         = 1;
-    ep->cfg_index                        = UCP_WORKER_CFG_INDEX_NULL;
-    ep->worker                           = worker;
-    ep->am_lane                          = UCP_NULL_LANE;
-    ep->flags                            = 0;
-    ep->conn_sn                          = UCP_EP_MATCH_CONN_SN_MAX;
+    ep->refcount                          = 1;
+    ep->cfg_index                         = UCP_WORKER_CFG_INDEX_NULL;
+    ep->worker                            = worker;
+    ep->am_lane                           = UCP_NULL_LANE;
+    ep->flags                             = 0;
+    ep->conn_sn                           = UCP_EP_MATCH_CONN_SN_MAX;
 #if UCS_ENABLE_ASSERT
-    ep->flush_iter_refcount              = 0;
-    ep->discard_refcount                 = 0;
+    ep->flush_iter_refcount               = 0;
+    ep->discard_refcount                  = 0;
 #endif
-    ucp_ep_ext_gen(ep)->user_data        = NULL;
-    ucp_ep_ext_control(ep)->cm_idx       = UCP_NULL_RESOURCE;
-    ucp_ep_ext_control(ep)->err_cb       = NULL;
-    ucp_ep_ext_control(ep)->local_ep_id  = UCS_PTR_MAP_KEY_INVALID;
-    ucp_ep_ext_control(ep)->remote_ep_id = UCS_PTR_MAP_KEY_INVALID;
+    ucp_ep_ext_gen(ep)->user_data         = NULL;
+    ucp_ep_ext_control(ep)->cm_idx        = UCP_NULL_RESOURCE;
+    ucp_ep_ext_control(ep)->err_cb        = NULL;
+    ucp_ep_ext_control(ep)->local_ep_id   = UCS_PTR_MAP_KEY_INVALID;
+    ucp_ep_ext_control(ep)->remote_ep_id  = UCS_PTR_MAP_KEY_INVALID;
 #if UCS_ENABLE_ASSERT
-    ucp_ep_ext_control(ep)->ka_count     = 0;
+    ucp_ep_ext_control(ep)->ka_last_round = 0;
 #endif
 
     UCS_STATIC_ASSERT(sizeof(ucp_ep_ext_gen(ep)->ep_match) >=
@@ -2754,7 +2754,7 @@ ucs_status_t ucp_ep_do_uct_ep_keepalive(ucp_ep_h ucp_ep, uct_ep_h uct_ep,
     return (packed_len > 0) ? UCS_OK : (ucs_status_t)packed_len;
 }
 
-int ucp_ep_do_keepalive(ucp_ep_h ep)
+int ucp_ep_do_keepalive(ucp_ep_h ep, ucs_time_t now)
 {
     ucp_worker_h worker = ep->worker;
     ucp_lane_index_t lane;
@@ -2763,16 +2763,8 @@ int ucp_ep_do_keepalive(ucp_ep_h ep)
 
     UCP_WORKER_THREAD_CS_CHECK_IS_BLOCKED(ep->worker);
 
-    if (ep->flags & UCP_EP_FLAG_FAILED) {
-        worker->keepalive.lane_map = 0;
-        return 1;
-    }
-
-    /* Take updated ep_check_map, in case ep configuration has changed */
-    worker->keepalive.lane_map &= ucp_ep_config(ep)->key.ep_check_map;
-    if (worker->keepalive.lane_map == 0) {
-        return 1;
-    }
+    ucs_assert(!(ep->flags & UCP_EP_FLAG_FAILED));
+    ucs_assert(worker->keepalive.lane_map != 0);
 
     ucs_for_each_bit(lane, worker->keepalive.lane_map) {
         ucs_assert(lane < UCP_MAX_LANES);
@@ -2793,23 +2785,23 @@ int ucp_ep_do_keepalive(ucp_ep_h ep)
         worker->keepalive.lane_map &= ~UCS_BIT(lane);
     }
 
-    if (worker->keepalive.lane_map == 0) {
+    /* Keepalive round is still not finished for this ep - skip ka_last_round
+     * update */
+    if (worker->keepalive.lane_map != 0) {
+        return 0;
+    }
+
 #if UCS_ENABLE_ASSERT
-        ucp_ep_ext_control(ep)->ka_count++;
-        /* Difference between the number of EP keepalive rounds and total number
-         * of KA rounds done on Worker must be <= 2, because keepalive could be
-         * done twice for the same EP in case of number of EPs was decreased */
-        ucs_assertv((ucp_ep_ext_control(ep)->ka_count /
-                             (worker->keepalive.round_count + 1)) <= 2,
-                    "ep %p: keepalive.round_count=%" PRIu64
-                    " ep.ka_count=%" PRIu64,
-                    ep, worker->keepalive.round_count,
-                    ucp_ep_ext_control(ep)->ka_count);
+    ucs_assertv((now - ucp_ep_ext_control(ep)->ka_last_round) >=
+                        worker->context->config.ext.keepalive_interval,
+                "ep %p: now=%" PRIu64 " ka_last_round=%" PRIu64
+                " ka_interval=%" PRIu64,
+                ep, now, ucp_ep_ext_control(ep)->ka_last_round,
+                worker->context->config.ext.keepalive_interval);
+    ucp_ep_ext_control(ep)->ka_last_round = now;
 #endif
-        return 1;
-    }
 
-    return 0;
+    return 1;
 }
 
 static void ucp_ep_req_purge(ucp_ep_h ucp_ep, ucp_request_t *req,
diff --git a/src/ucp/core/ucp_ep.h b/src/ucp/core/ucp_ep.h
@@ -433,7 +433,7 @@ typedef struct {
     ucp_err_handler_cb_t     err_cb; /* Error handler */
     ucp_ep_close_proto_req_t close_req; /* Close protocol request */
 #if UCS_ENABLE_ASSERT
-    size_t                   ka_count; /* Number of KA rounds done */
+    ucs_time_t               ka_last_round; /* Time of last KA round done */
 #endif
 } ucp_ep_ext_control_t;
 
@@ -671,10 +671,11 @@ ucs_status_t ucp_ep_do_uct_ep_keepalive(ucp_ep_h ucp_ep, uct_ep_h uct_ep,
  * @brief Do keepalive operation.
  *
  * @param [in] ep    UCP Endpoint object to operate keepalive.
+ * @param [in] now   Current time when keepalive started.
  *
  * @return Indication whether keepalive was fully done for UCP Endpoint or not.
  */
-int ucp_ep_do_keepalive(ucp_ep_h ep);
+int ucp_ep_do_keepalive(ucp_ep_h ep, ucs_time_t now);
 
 /**
  * @brief Purge flush and protocol requests scheduled on a given UCP endpoint.
diff --git a/src/ucp/core/ucp_worker.c b/src/ucp/core/ucp_worker.c
@@ -2055,6 +2055,7 @@ static UCS_F_ALWAYS_INLINE void ucp_worker_keepalive_reset(ucp_worker_h worker)
     worker->keepalive.ep_count    = 0;
     worker->keepalive.iter_count  = 0;
     worker->keepalive.iter        = &worker->all_eps;
+    worker->keepalive.iter_begin  = worker->keepalive.iter;
     worker->keepalive.round_count = 0;
 }
 
@@ -2908,42 +2909,50 @@ void ucp_worker_print_info(ucp_worker_h worker, FILE *stream)
 }
 
 static UCS_F_ALWAYS_INLINE ucp_ep_h
-ucp_worker_keepalive_current_ep(ucp_worker_h worker)
-{
-    ucp_ep_ext_gen_t *ep_ext;
-
-    ucs_assert(worker->keepalive.iter != &worker->all_eps);
-    ep_ext = ucs_container_of(worker->keepalive.iter, ucp_ep_ext_gen_t,
-                              ep_list);
-    return ucp_ep_from_ext_gen(ep_ext);
-}
-
-static UCS_F_ALWAYS_INLINE void
 ucp_worker_keepalive_next_ep(ucp_worker_h worker)
 {
     ucp_ep_h ep;
 
-    worker->keepalive.iter = worker->keepalive.iter->next;
-    if (worker->keepalive.iter == &worker->all_eps) {
-        /* if next list item points to all_eps then step one more time */
+    if (worker->keepalive.lane_map == 0) {
         worker->keepalive.iter = worker->keepalive.iter->next;
+        if (worker->keepalive.iter == &worker->all_eps) {
+            return NULL;
+        }
+
+        worker->keepalive.lane_map = UCS_MASK(UCP_MAX_LANES);
     }
 
     ucs_assert(worker->keepalive.iter != &worker->all_eps);
-    ep                         = ucp_worker_keepalive_current_ep(worker);
-    worker->keepalive.lane_map = ((ep->cfg_index != UCP_WORKER_CFG_INDEX_NULL) &&
-                                  !(ep->flags & UCP_EP_FLAG_FAILED)) ?
-                                 ucp_ep_config(ep)->key.ep_check_map : 0;
+    ep = ucp_ep_from_ext_gen(ucs_container_of(worker->keepalive.iter,
+                                              ucp_ep_ext_gen_t, ep_list));
+
+    if ((ep->cfg_index == UCP_WORKER_CFG_INDEX_NULL) ||
+        (ep->flags & UCP_EP_FLAG_FAILED)) {
+        worker->keepalive.lane_map = 0;
+        return NULL;
+    }
+
+    /* Take updated ep_check_map, in case endpoint configuration has changed
+     * before continuing this round */
+    worker->keepalive.lane_map &= ucp_ep_config(ep)->key.ep_check_map;
+    if (worker->keepalive.lane_map == 0) {
+        return NULL;
+    }
+
+    return ep;
 }
 
 static UCS_F_NOINLINE unsigned
 ucp_worker_do_keepalive_progress(ucp_worker_h worker)
 {
     unsigned progress_count = 0;
-    unsigned max_ep_count;
+    unsigned max_ep_count   = worker->context->config.ext.keepalive_num_eps;
     ucs_time_t now;
     ucp_ep_h ep;
 
+    ucs_assertv(worker->keepalive.ep_count < max_ep_count,
+                "worker %p: ep_count=%u max_ep_count=%u",
+                worker, worker->keepalive.ep_count, max_ep_count);
     ucs_assert(worker->context->config.ext.keepalive_num_eps != 0);
 
     now = ucs_get_time();
@@ -2967,22 +2976,19 @@ ucp_worker_do_keepalive_progress(ucp_worker_h worker)
         goto out_unblock;
     }
 
-    if (ucs_unlikely(worker->keepalive.iter == &worker->all_eps)) {
-        ucp_worker_keepalive_next_ep(worker);
-    }
-
-    max_ep_count = ucs_min(worker->context->config.ext.keepalive_num_eps,
-                           worker->num_all_eps);
-
     /* Use own loop for elements because standard for_each skips
      * head element */
     /* TODO: use more optimal algo to enumerate EPs to keepalive
      * (linked list) */
-    while (worker->keepalive.ep_count < max_ep_count) {
-        ep = ucp_worker_keepalive_current_ep(worker);
+    do {
+        ep = ucp_worker_keepalive_next_ep(worker);
+        if (ep == NULL) {
+            continue;
+        }
+
         ucs_trace_func("worker %p: do keepalive on ep %p lane_map 0x%x", worker,
                        ep, worker->keepalive.lane_map);
-        if (!ucp_ep_do_keepalive(ep)) {
+        if (!ucp_ep_do_keepalive(ep, now)) {
             /* In case if EP has no resources to send keepalive message
              * then just return without update of last_round timestamp,
              * on next progress iteration we will continue from this point */
@@ -2991,11 +2997,12 @@ ucp_worker_do_keepalive_progress(ucp_worker_h worker)
 
         progress_count++;
         worker->keepalive.ep_count++;
-        ucp_worker_keepalive_next_ep(worker);
-    }
+    } while ((worker->keepalive.ep_count < max_ep_count) &&
+             (worker->keepalive.iter != worker->keepalive.iter_begin));
 
     ucs_trace("worker %p: sent keepalive on %u endpoints",
               worker, worker->keepalive.ep_count);
+    worker->keepalive.iter_begin = worker->keepalive.iter;
     worker->keepalive.last_round = now;
     worker->keepalive.ep_count   = 0;
     worker->keepalive.round_count++;
@@ -3053,13 +3060,14 @@ void ucp_worker_keepalive_remove_ep(ucp_ep_h ep)
         return;
     }
 
-    if (ucs_list_is_only(&worker->all_eps, &ucp_ep_ext_gen(ep)->ep_list)) {
-        /* this is the last EP in worker */
-        worker->keepalive.iter = &worker->all_eps;
-    } else if (worker->keepalive.iter == &ucp_ep_ext_gen(ep)->ep_list) {
-        /* if iterator points into EP to be removed - then
-         * step to next EP */
+    if (worker->keepalive.iter == &ucp_ep_ext_gen(ep)->ep_list) {
+        /* Set lane_map=0 to make sure the endpoint won't be selected again */
+        worker->keepalive.lane_map = 0;
+
         ucp_worker_keepalive_next_ep(worker);
+        ucs_assert(worker->keepalive.iter != &ucp_ep_ext_gen(ep)->ep_list);
+
+        worker->keepalive.iter_begin = worker->keepalive.iter;
     }
 }
 
diff --git a/src/ucp/core/ucp_worker.h b/src/ucp/core/ucp_worker.h
@@ -302,6 +302,8 @@ typedef struct ucp_worker {
         uct_worker_cb_id_t           cb_id;               /* Keepalive callback id */
         ucs_time_t                   last_round;          /* Last round timestamp */
         ucs_list_link_t              *iter;               /* Last EP processed keepalive */
+        ucs_list_link_t              *iter_begin;         /* First EP processed keepalive in the
+                                                           * current round */
         ucp_lane_map_t               lane_map;            /* Lane map used to retry after no-resources */
         unsigned                     ep_count;            /* Number of EPs processed in current time slot */
         unsigned                     iter_count;          /* Number of progress iterations to skip,
