UCP/PROTO: Applied changes from PR openucx#10395

Author: guy-ealey-morag

src/ucp/core/ucp_context.c

@@ -491,6 +491,7 @@ static ucs_config_field_t ucp_context_config_table[] = {
    "Enable printing protocols information. The value is interpreted as follows:\n"
    " 'y'          : Print information for all protocols\n"
    " 'n'          : Do not print any protocol information\n"
+   " used         : Print information for used protocols\n"
    " glob_pattern : Print information for operations matching the glob pattern.\n"
    "                For example: '*tag*gpu*', '*put*fast*host*'",
    ucs_offsetof(ucp_context_config_t, proto_info), UCS_CONFIG_TYPE_STRING},
@@ -2397,6 +2398,10 @@ static ucs_status_t ucp_fill_config(ucp_context_h context,
     context->config.progress_wrapper_enabled =
             ucs_log_is_enabled(UCS_LOG_LEVEL_TRACE_REQ) ||
             ucp_context_usage_tracker_enabled(context);
+
+    context->config.trace_used_proto_selections =
+            !strcasecmp(context->config.ext.proto_info, "used");
+
     return UCS_OK;
 
 err_free_key_list:

src/ucp/core/ucp_context.h

@@ -460,6 +460,9 @@ typedef struct ucp_context {
         /* Progress wrapper enabled */
         int                       progress_wrapper_enabled;
 
+        /* Indicate whether tracing for used protocol selections is enabled */
+        int                       trace_used_proto_selections;
+
         struct {
            unsigned               count;
            size_t                 *sizes;

src/ucp/core/ucp_request.c

@@ -729,19 +729,43 @@ void ucp_request_purge_enqueue_cb(uct_pending_req_t *self, void *arg)
     ucs_queue_push(queue, (ucs_queue_elem_t*)&req->send.uct.priv);
 }
 
+ucs_status_t ucp_request_progress_counter(uct_pending_req_t *self)
+{
+    ucp_request_t *req       = ucs_container_of(self, ucp_request_t, send.uct);
+    ucp_proto_config_t *conf = ucs_const_cast(ucp_proto_config_t *,
+                                              req->send.proto_config);
+    const ucp_proto_t *proto = conf->proto;
+    ucs_status_t status;
+
+    status = proto->progress[UCP_PROTO_STAGE_START](self);
+    if (status != UCS_OK) {
+        return status;
+    }
+
+    if (++conf->selections == UCP_PROTO_SELECTIONS_COUNT_MAX) {
+        ucp_trace_req(req, "protocol %s was selected %u times, stop tracing",
+                      proto->name, conf->selections);
+        memcpy(conf->progress_wrapper, proto->progress,
+               sizeof(conf->progress_wrapper));
+    }
+
+    return UCS_OK;
+}
+
 ucs_status_t ucp_request_progress_wrapper(uct_pending_req_t *self)
 {
     ucp_request_t *req       = ucs_container_of(self, ucp_request_t, send.uct);
-    const ucp_proto_t *proto = req->send.proto_config->proto;
+    ucp_proto_config_t *conf = ucs_const_cast(ucp_proto_config_t *,
+                                              req->send.proto_config);
+    const ucp_proto_t *proto = conf->proto;
     uct_pending_callback_t progress_cb;
     ucs_status_t status;
 
     progress_cb = proto->progress[req->send.proto_stage];
     ucp_trace_req(req,
                   "progress %s {%s} ep_cfg[%d] rkey_cfg[%d] offset %zu/%zu",
                   proto->name, ucs_debug_get_symbol_name(progress_cb),
-                  req->send.proto_config->ep_cfg_index,
-                  req->send.proto_config->rkey_cfg_index,
+                  conf->ep_cfg_index, conf->rkey_cfg_index,
                   req->send.state.dt_iter.offset,
                   req->send.state.dt_iter.length);
 
@@ -753,9 +777,39 @@ ucs_status_t ucp_request_progress_wrapper(uct_pending_req_t *self)
         ucp_trace_req(req, "progress protocol %s returned: %s lane %d",
                       proto->name, ucs_status_string(status), req->send.lane);
     } else {
+        if (req->send.proto_stage == UCP_PROTO_STAGE_START) {
+            ++conf->selections;
+        }
+
         ucp_trace_req(req, "progress protocol %s returned: %s", proto->name,
                       ucs_status_string(status));
     }
     ucs_log_indent(-1);
     return status;
 }
+
+void ucp_request_init_progress_wrapper(ucp_worker_h worker,
+                                       ucp_proto_config_t *proto_config,
+                                       int internal)
+{
+    uint8_t stage;
+
+    if (worker->context->config.progress_wrapper_enabled) {
+        for (stage = UCP_PROTO_STAGE_START; stage < UCP_PROTO_STAGE_LAST;
+             ++stage) {
+            proto_config->progress_wrapper[stage] =
+                                                ucp_request_progress_wrapper;
+        }
+        return;
+    }
+
+    /* Set wrappers pointing to the original protocol functions */
+    memcpy(proto_config->progress_wrapper, proto_config->proto->progress,
+           sizeof(proto_config->progress_wrapper));
+
+    if (worker->context->config.trace_used_proto_selections) {
+        /* TODO: should we disable progress wrapper for internal protocols? */
+        proto_config->progress_wrapper[UCP_PROTO_STAGE_START] =
+                                                ucp_request_progress_counter;
+    }
+}

src/ucp/core/ucp_request.h

@@ -587,4 +587,8 @@ void ucp_request_purge_enqueue_cb(uct_pending_req_t *self, void *arg);
 
 ucs_status_t ucp_request_progress_wrapper(uct_pending_req_t *self);
 
+void ucp_request_init_progress_wrapper(ucp_worker_h worker,
+                                       ucp_proto_config_t *proto_config,
+                                       int internal);
+
 #endif

src/ucp/core/ucp_worker.c

@@ -2267,6 +2267,21 @@ static void ucp_worker_keepalive_reset(ucp_worker_h worker)
     worker->keepalive.round_count = 0;
 }
 
+static void ucp_worker_trace_configs(ucp_worker_h worker)
+{
+    ucp_ep_config_t *ep_config;
+    ucp_rkey_config_t *rkey_config;
+
+    ucs_array_for_each(ep_config, &worker->ep_config) {
+        ucp_proto_select_trace(worker, &ep_config->proto_select);
+    }
+
+    ucs_carray_for_each(rkey_config, worker->rkey_config,
+                        worker->rkey_config_count) {
+        ucp_proto_select_trace(worker, &rkey_config->proto_select);
+    }
+}
+
 static void ucp_worker_destroy_configs(ucp_worker_h worker)
 {
     ucp_ep_config_t *ep_config;
@@ -2921,6 +2936,9 @@ static void ucp_worker_destroy_eps(ucp_worker_h worker,
 void ucp_worker_destroy(ucp_worker_h worker)
 {
     ucs_debug("destroy worker %p", worker);
+    if (worker->context->config.trace_used_proto_selections) {
+        ucp_worker_trace_configs(worker);
+    }
 
     UCS_ASYNC_BLOCK(&worker->async);
     uct_worker_progress_unregister_safe(worker->uct, &worker->keepalive.cb_id);

src/ucp/proto/proto.h

@@ -141,6 +141,9 @@ typedef struct {
 
     /* Map of used lanes */
     ucp_lane_map_t lane_map;
+
+    /* Selections count */
+    unsigned       selections;
 } ucp_proto_query_attr_t;
 
 

src/ucp/proto/proto_common.inl

@@ -175,11 +175,7 @@ ucp_proto_request_set_stage(ucp_request_t *req, uint8_t proto_stage)
     req->send.proto_stage = proto_stage;
 
     /* Set pointer to progress function */
-    if (req->send.ep->worker->context->config.progress_wrapper_enabled) {
-        req->send.uct.func = ucp_request_progress_wrapper;
-    } else {
-        req->send.uct.func = proto->progress[proto_stage];
-    }
+    req->send.uct.func = req->send.proto_config->progress_wrapper[proto_stage];
 }
 
 /* Select protocol for the request and initialize protocol-related fields */

src/ucp/proto/proto_debug.c

@@ -64,6 +64,7 @@ struct ucp_proto_perf_node {
 
 /* Protocol information table */
 typedef struct {
+    char counter_str[32];
     char range_str[32];
     char desc[UCP_PROTO_DESC_STR_MAX];
     char config[UCP_PROTO_CONFIG_STR_MAX];
@@ -156,41 +157,84 @@ static void ucp_proto_table_row_separator(ucs_string_buffer_t *strb,
 }
 
 static int ucp_proto_debug_is_info_enabled(ucp_context_h context,
-                                           const char *select_param_str)
+                                           const char *select_param_str,
+                                           int show_used)
 {
     const char *proto_info_config = context->config.ext.proto_info;
     int bool_value;
 
+    if (show_used) {
+        return context->config.trace_used_proto_selections;
+    }
+
     if (ucs_config_sscanf_bool(proto_info_config, &bool_value, NULL)) {
         return bool_value;
     }
 
     return fnmatch(proto_info_config, select_param_str, FNM_CASEFOLD) == 0;
 }
 
-void ucp_proto_select_elem_info(ucp_worker_h worker,
-                                ucp_worker_cfg_index_t ep_cfg_index,
-                                ucp_worker_cfg_index_t rkey_cfg_index,
-                                const ucp_proto_select_param_t *select_param,
-                                const ucp_proto_select_elem_t *select_elem,
-                                int show_all, ucs_string_buffer_t *strb)
+static inline int
+ucp_proto_select_elem_has_selections(const ucp_proto_select_elem_t *select_elem)
+{
+    const ucp_proto_threshold_elem_t *thresh_elem = select_elem->thresholds;
+
+    do {
+        if (thresh_elem->proto_config.selections > 0) {
+            return 1;
+        }
+    } while ((thresh_elem++)->max_msg_length < SIZE_MAX);
+
+    return 0;
+}
+
+static void
+ucp_proto_selections_dump(const ucp_proto_query_attr_t *proto_attr,
+                          char *counter_str, size_t size, int show_used)
+{
+    unsigned max_selections;
+
+    if (!show_used) {
+        *counter_str = '\0';
+        return;
+    }
+
+    max_selections = (strstr(proto_attr->desc, "short") != NULL) ?
+                            1 : UCP_PROTO_SELECTIONS_COUNT_MAX;
+
+    ucs_snprintf_safe(counter_str, size, "%u%s  ", proto_attr->selections,
+                      (proto_attr->selections >= max_selections ? "+" : ""));
+}
+
+void
+ucp_proto_select_elem_info(ucp_worker_h worker,
+                           ucp_worker_cfg_index_t ep_cfg_index,
+                           ucp_worker_cfg_index_t rkey_cfg_index,
+                           const ucp_proto_select_param_t *select_param,
+                           const ucp_proto_select_elem_t *select_elem,
+                           int show_all, int show_used, ucs_string_buffer_t *strb)
 {
     UCS_STRING_BUFFER_ONSTACK(ep_cfg_strb, UCP_PROTO_CONFIG_STR_MAX);
     UCS_STRING_BUFFER_ONSTACK(sel_param_strb, UCP_PROTO_CONFIG_STR_MAX);
-    static const char *info_row_fmt = "| %*s | %-*s | %-*s |\n";
+    static const char *info_row_fmt = "| %s%*s | %-*s | %-*s |\n";
     ucp_proto_info_table_t table;
     int hdr_col_width[2], col_width[3];
     ucp_proto_query_attr_t proto_attr;
     ucp_proto_info_row_t *row_elem;
     size_t range_start, range_end;
     int proto_valid;
 
+    if (show_used && !ucp_proto_select_elem_has_selections(select_elem)) {
+        return;
+    }
+
     ucp_proto_select_param_dump(worker, ep_cfg_index, rkey_cfg_index,
                                 select_param, ucp_operation_descs, &ep_cfg_strb,
                                 &sel_param_strb);
     if (!show_all &&
         !ucp_proto_debug_is_info_enabled(
-                worker->context, ucs_string_buffer_cstr(&sel_param_strb))) {
+                worker->context, ucs_string_buffer_cstr(&sel_param_strb),
+                show_used)) {
         return;
     }
 
@@ -221,7 +265,11 @@ void ucp_proto_select_elem_info(ucp_worker_h worker,
         ucs_memunits_range_str(range_start, range_end, row_elem->range_str,
                                sizeof(row_elem->range_str));
 
-        col_width[0] = ucs_max(col_width[0], strlen(row_elem->range_str));
+        ucp_proto_selections_dump(&proto_attr, row_elem->counter_str,
+                                  sizeof(row_elem->counter_str), show_used);
+
+        col_width[0] = ucs_max(col_width[0], strlen(row_elem->counter_str) +
+                                             strlen(row_elem->range_str));
         col_width[1] = ucs_max(col_width[1], strlen(row_elem->desc));
         col_width[2] = ucs_max(col_width[2], strlen(row_elem->config));
     } while (range_end != SIZE_MAX);
@@ -243,7 +291,8 @@ void ucp_proto_select_elem_info(ucp_worker_h worker,
     /* Print contents */
     ucp_proto_table_row_separator(strb, col_width, 3);
     ucs_array_for_each(row_elem, &table) {
-        ucs_string_buffer_appendf(strb, info_row_fmt, col_width[0],
+        ucs_string_buffer_appendf(strb, info_row_fmt, row_elem->counter_str,
+                                  col_width[0] - strlen(row_elem->counter_str),
                                   row_elem->range_str, col_width[1],
                                   row_elem->desc, col_width[2],
                                   row_elem->config);
@@ -264,7 +313,7 @@ void ucp_proto_select_info(ucp_worker_h worker,
 
     kh_foreach(proto_select->hash, key.u64, select_elem,
                ucp_proto_select_elem_info(worker, ep_cfg_index, rkey_cfg_index,
-                                          &key.param, &select_elem, show_all,
+                                          &key.param, &select_elem, show_all, 0,
                                           strb);
                ucs_string_buffer_appendf(strb, "\n"))
 }
@@ -966,7 +1015,7 @@ ucp_proto_select_write_info(ucp_worker_h worker,
                                 ucp_operation_names, &ep_cfg_strb,
                                 &sel_param_strb);
     if (!ucp_proto_debug_is_info_enabled(
-                worker->context, ucs_string_buffer_cstr(&sel_param_strb))) {
+                worker->context, ucs_string_buffer_cstr(&sel_param_strb), 0)) {
         goto out;
     }
 
@@ -1031,17 +1080,19 @@ ucp_proto_select_write_info(ucp_worker_h worker,
 }
 
 void ucp_proto_select_elem_trace(ucp_worker_h worker,
-                                 ucp_worker_cfg_index_t ep_cfg_index,
-                                 ucp_worker_cfg_index_t rkey_cfg_index,
                                  const ucp_proto_select_param_t *select_param,
-                                 ucp_proto_select_elem_t *select_elem)
+                                 const ucp_proto_select_elem_t *select_elem,
+                                 int show_used)
 {
-    ucs_string_buffer_t strb = UCS_STRING_BUFFER_INITIALIZER;
+    const ucp_proto_config_t *proto_config = &select_elem->thresholds[0].proto_config;
+    ucp_worker_cfg_index_t ep_cfg_index    = proto_config->ep_cfg_index;
+    ucp_worker_cfg_index_t rkey_cfg_index  = proto_config->rkey_cfg_index;
+    ucs_string_buffer_t strb               = UCS_STRING_BUFFER_INITIALIZER;
     char *line;
 
     /* Print human-readable protocol selection table to the log */
     ucp_proto_select_elem_info(worker, ep_cfg_index, rkey_cfg_index,
-                               select_param, select_elem, 0, &strb);
+                               select_param, select_elem, 0, show_used, &strb);
     ucs_string_buffer_for_each_token(line, &strb, "\n") {
         ucs_log_print_compact(line);
     }

src/ucp/proto/proto_debug.h

@@ -157,14 +157,14 @@ void ucp_proto_select_elem_info(ucp_worker_h worker,
                                 ucp_worker_cfg_index_t rkey_cfg_index,
                                 const ucp_proto_select_param_t *select_param,
                                 const ucp_proto_select_elem_t *select_elem,
-                                int show_all, ucs_string_buffer_t *strb);
+                                int show_all, int show_used,
+                                ucs_string_buffer_t *strb);
 
 
 void ucp_proto_select_elem_trace(ucp_worker_h worker,
-                                 ucp_worker_cfg_index_t ep_cfg_index,
-                                 ucp_worker_cfg_index_t rkey_cfg_index,
                                  const ucp_proto_select_param_t *select_param,
-                                 ucp_proto_select_elem_t *select_elem);
+                                 const ucp_proto_select_elem_t *select_elem,
+                                 int show_used);
 
 
 void ucp_proto_select_write_info(

src/ucp/proto/proto_reconfig.c

@@ -30,6 +30,7 @@ static ucs_status_t ucp_proto_reconfig_select_progress(uct_pending_req_t *self)
         return UCS_ERR_NO_RESOURCE;
     }
 
+    /* coverity[address_free] */
     return req->send.uct.func(&req->send.uct);
 }