Skip to content

UCP/PROTO: Initialize short thresholds from protocol selection - v1.23.x #11921

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
yosefe wants to merge 2 commits into from
Closed

UCP/PROTO: Initialize short thresholds from protocol selection - v1.23.x #11921

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
61b454d
UCP/WORKER: Fix hang in worker_wait because of race with check_events_do
yosefe Sep 1, 2026
a26bb45
UCP/PROTO: Initialize short thresholds from protocol selection
yosefe Aug 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 23 additions & 18 deletions src/ucp/core/ucp_ep.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1444,7 +1444,7 @@ ucp_ep_config_activate_worker_ifaces(ucp_worker_h worker,
{
ucp_ep_config_t *ep_config = ucp_worker_ep_config(worker, cfg_index);

ucs_trace("activate wifaces worker %p ep config %u ep count %u", worker,
ucs_debug("activate wifaces worker %p ep config %u ep count %u", worker,
cfg_index, ep_config->ep_count);
if (ep_config->ep_count++ == 0) {
ucp_wiface_process_for_each_lane(worker, ep_config,
Expand All @@ -1464,7 +1464,7 @@ ucp_ep_config_deactivate_worker_ifaces(ucp_worker_h worker,
}

ep_config = ucp_worker_ep_config(worker, cfg_index);
ucs_trace("deactivate wifaces worker %p ep config %u ep count %u", worker,
ucs_debug("deactivate wifaces worker %p ep config %u ep count %u", worker,
cfg_index, ep_config->ep_count);
ucs_assertv(ep_config->ep_count > 0, "worker %p ep config %u", worker,
cfg_index);
Expand All @@ -1481,7 +1481,7 @@ ucp_ep_config_reactivate_worker_ifaces(ucp_worker_h worker,
ucp_worker_cfg_index_t old_cfg_index,
ucp_worker_cfg_index_t new_cfg_index)
{
ucs_trace("worker %p: reactivating interfaces deactivate cfg_index %u "
ucs_debug("worker %p: reactivating interfaces deactivate cfg_index %u "
"activate cfg_index %u", worker, old_cfg_index, new_cfg_index);

if (old_cfg_index == new_cfg_index) {
Expand Down Expand Up @@ -3767,9 +3767,12 @@ ucs_status_t ucp_ep_config_init(ucp_worker_h worker, ucp_ep_config_t *config,
&config->tag.rndv.am_thresh);

/* Max Eager short has to be set after Zcopy and RNDV thresholds */
ucp_ep_config_set_memtype_thresh(&config->tag.offload.max_eager_short,
config->tag.eager.max_short,
context->num_mem_type_detect_mds);
if (!context->config.ext.proto_enable) {
ucp_ep_config_set_memtype_thresh(
&config->tag.offload.max_eager_short,
config->tag.eager.max_short,
context->num_mem_type_detect_mds);
}
}
}

Expand Down Expand Up @@ -3825,7 +3828,8 @@ ucs_status_t ucp_ep_config_init(ucp_worker_h worker, ucp_ep_config_t *config,
iface_attr->cap.am.max_short, sizeof(ucp_am_hdr_t),
config->am.zcopy_thresh[0], &config->rndv.am_thresh);

if (iface_attr->cap.am.max_iov >= UCP_AM_SEND_SHORT_MIN_IOV) {
if (!context->config.ext.proto_enable &&
(iface_attr->cap.am.max_iov >= UCP_AM_SEND_SHORT_MIN_IOV)) {
ucp_ep_config_set_memtype_thresh(
&config->am_u.max_eager_short, am_max_eager_short,
context->num_mem_type_detect_mds);
Expand All @@ -3850,9 +3854,12 @@ ucs_status_t ucp_ep_config_init(ucp_worker_h worker, ucp_ep_config_t *config,
config->tag.rndv.rma_thresh = config->rndv.rma_thresh;

/* Max Eager short has to be set after Zcopy and RNDV thresholds */
ucp_ep_config_set_memtype_thresh(&config->tag.max_eager_short,
config->tag.eager.max_short,
context->num_mem_type_detect_mds);
if (!context->config.ext.proto_enable) {
ucp_ep_config_set_memtype_thresh(
&config->tag.max_eager_short,
config->tag.eager.max_short,
context->num_mem_type_detect_mds);
}
}

/* Calculate max short threshold for UCP AM short reply protocol */
Expand All @@ -3862,7 +3869,8 @@ ucs_status_t ucp_ep_config_init(ucp_worker_h worker, ucp_ep_config_t *config,
sizeof(ucp_am_hdr_t) + sizeof(ucp_am_reply_ftr_t),
config->am.zcopy_thresh[0], &config->rndv.am_thresh);

if (iface_attr->cap.am.max_iov >= UCP_AM_SEND_SHORT_MIN_IOV) {
if (!context->config.ext.proto_enable &&
(iface_attr->cap.am.max_iov >= UCP_AM_SEND_SHORT_MIN_IOV)) {
ucp_ep_config_set_memtype_thresh(
&config->am_u.max_reply_eager_short, am_max_eager_short,
context->num_mem_type_detect_mds);
Expand Down Expand Up @@ -4740,22 +4748,20 @@ ucp_ep_select_short_init(ucp_worker_h worker, ucp_worker_cfg_index_t cfg_index,
max_eager_short->memtype_on = proto_short.max_length_host_mem;
}

static void ucp_ep_config_proto_init(ucp_worker_h worker,
ucp_worker_cfg_index_t cfg_index)
void ucp_ep_config_proto_short_lazy_init(ucp_worker_h worker,
ucp_worker_cfg_index_t cfg_index)
{
ucp_ep_config_t *ep_config = ucp_worker_ep_config(worker, cfg_index);
ucp_ep_config_key_t *key = &ep_config->key;
ucp_memtype_thresh_t *tag_max_short;
ucp_lane_index_t tag_exp_lane;
unsigned tag_proto_flags;

/* Do protocol init once per EP config and only for protov2 */
if ((!worker->context->config.ext.proto_enable) ||
(ep_config->proto_init_flags & UCP_EP_PROTO_INITIALIZED)) {
if (ep_config->proto_init_flags & UCP_EP_PROTO_SHORT_INITIALIZED) {
return;
}

ep_config->proto_init_flags |= UCP_EP_PROTO_INITIALIZED;
ep_config->proto_init_flags |= UCP_EP_PROTO_SHORT_INITIALIZED;

if (ucp_ep_config_key_has_tag_lane(key)) {
tag_proto_flags = UCP_PROTO_FLAG_TAG_SHORT;
Expand Down Expand Up @@ -4793,7 +4799,6 @@ void ucp_ep_set_cfg_index(ucp_ep_h ep, ucp_worker_cfg_index_t cfg_index,
ucs_trace("ep %p: set cfg_index %u -> %u", ep, ep->cfg_index, cfg_index);
ep->cfg_index = cfg_index;
ep->am_lane = ucp_ep_config(ep)->key.am_lane;
ucp_ep_config_proto_init(ep->worker, cfg_index);
}

unsigned ucp_ep_err_mode_init_flags(ucp_err_handling_mode_t err_mode)
Expand Down
7 changes: 5 additions & 2 deletions src/ucp/core/ucp_ep.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -367,8 +367,8 @@ KHASH_DECLARE(ucp_ep_peer_mem_hash, uint64_t, ucp_ep_peer_mem_data_t);


typedef enum {
/* Protocol initialization was done */
UCP_EP_PROTO_INITIALIZED = UCS_BIT(0),
/* Protocol short-circuit path initialization was done */
UCP_EP_PROTO_SHORT_INITIALIZED = UCS_BIT(0),
} ucp_ep_init_flags_t;


Expand Down Expand Up @@ -789,6 +789,9 @@ void ucp_ep_cleanup_lanes(ucp_ep_h ep);
ucs_status_t ucp_ep_config_init(ucp_worker_h worker, ucp_ep_config_t *config,
const ucp_ep_config_key_t *key);

void ucp_ep_config_proto_short_lazy_init(ucp_worker_h worker,
ucp_worker_cfg_index_t cfg_index);

void ucp_ep_config_cleanup(ucp_worker_h worker, ucp_ep_config_t *config);

int ucp_ep_config_lane_is_peer_match(const ucp_ep_config_key_t *key1,
Expand Down
92 changes: 74 additions & 18 deletions src/ucp/core/ucp_worker.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -399,12 +399,18 @@ int ucp_worker_iface_get_event_fd(const ucp_worker_iface_t *wiface)
return wiface->event_fd;
}

static UCS_F_ALWAYS_INLINE
void ucp_worker_iface_event_fd_ctl(ucp_worker_iface_t *wiface,
ucp_worker_event_fd_op_t op)
static void ucp_worker_iface_event_fd_ctl(ucp_worker_iface_t *wiface,
ucp_worker_event_fd_op_t op)
{
static const char *const op_str[] = {
[UCP_WORKER_EPFD_OP_ADD] = "add",
[UCP_WORKER_EPFD_OP_DEL] = "delete",
};
ucs_status_t status;

ucs_debug("worker %p: %s fd %d epfd %d on " UCP_WIFACE_FMT, wiface->worker,
op_str[op], ucp_worker_iface_get_event_fd(wiface),
wiface->worker->event_fd, UCP_WIFACE_ARG(wiface));
status = ucp_worker_wakeup_ctl_fd(wiface->worker, op,
ucp_worker_iface_get_event_fd(wiface));
ucs_assert_always(status == UCS_OK);
Expand Down Expand Up @@ -575,13 +581,45 @@ ucp_worker_iface_error_handler(void *arg, uct_ep_h uct_ep, ucs_status_t status)
return status;
}

void ucp_worker_iface_activate(ucp_worker_iface_t *wiface, unsigned uct_flags)
static void ucp_worker_iface_activate_arm_mark_busy(ucp_worker_iface_t *wiface)
{
if (wiface->flags & UCP_WORKER_IFACE_FLAG_ACTIVATE_ARM_BUSY) {
return;
}

wiface->flags |= UCP_WORKER_IFACE_FLAG_ACTIVATE_ARM_BUSY;
++wiface->worker->arm_block_count;
ucs_debug("worker %p: marked " UCP_WIFACE_FMT
" as busy, arm block count %d",
wiface->worker, UCP_WIFACE_ARG(wiface),
wiface->worker->arm_block_count);
}

static void
ucp_worker_iface_activate_arm_unmark_busy(ucp_worker_iface_t *wiface)
{
if (!(wiface->flags & UCP_WORKER_IFACE_FLAG_ACTIVATE_ARM_BUSY)) {
return;
}

wiface->flags &= ~UCP_WORKER_IFACE_FLAG_ACTIVATE_ARM_BUSY;
ucs_assert(wiface->worker->arm_block_count > 0);
--wiface->worker->arm_block_count;
ucs_debug("worker %p: marked " UCP_WIFACE_FMT
" as not-busy, arm block count %d",
wiface->worker, UCP_WIFACE_ARG(wiface),
wiface->worker->arm_block_count);
}

void ucp_worker_iface_activate(ucp_worker_iface_t *wiface, unsigned uct_flags,
const char *reason)
{
ucp_worker_h worker = wiface->worker;

ucs_trace("activate " UCP_WIFACE_FMT " a_count=%u a_ifaces=%u",
UCP_WIFACE_ARG(wiface), wiface->activate_count,
worker->num_active_ifaces);
ucs_debug("worker %p: activate " UCP_WIFACE_FMT
" %s a_count=%u a_ifaces=%u uct_flags=0x%x",
worker, UCP_WIFACE_ARG(wiface), reason, wiface->activate_count,
worker->num_active_ifaces, uct_flags);

if (wiface->activate_count++ > 0) {
return; /* was already activated */
Expand All @@ -593,6 +631,8 @@ void ucp_worker_iface_activate(ucp_worker_iface_t *wiface, unsigned uct_flags)
/* Set default active message handlers */
ucp_worker_set_am_handlers(wiface, 0);

ucp_worker_iface_activate_arm_unmark_busy(wiface);

if (ucp_worker_iface_has_event_notify(wiface)) {
if (ucp_worker_iface_use_event_fd(wiface)) {
/* Add to user wakeup */
Expand Down Expand Up @@ -639,7 +679,7 @@ static ucs_status_t ucp_worker_iface_check_events_do(ucp_worker_iface_t *wiface,
*progress_count = uct_iface_progress(wiface->iface);
if (prev_recv_count != wiface->proxy_recv_count) {
/* Received relevant active messages, activate the interface */
ucp_worker_iface_activate(wiface, 0);
ucp_worker_iface_activate(wiface, 0, "active message");
return UCS_OK;
} else if (*progress_count == 0) {
/* Arm the interface to wait for next event */
Expand All @@ -651,7 +691,9 @@ static ucs_status_t ucp_worker_iface_check_events_do(ucp_worker_iface_t *wiface,
status = uct_iface_event_arm(wiface->iface,
UCT_EVENT_RECV | UCT_EVENT_SEND_COMP);
if (status == UCS_OK) {
ucs_trace("armed iface %p", wiface->iface);
ucs_trace("check_events: armed " UCP_WIFACE_FMT,
UCP_WIFACE_ARG(wiface));
ucp_worker_iface_activate_arm_unmark_busy(wiface);

if (ucp_worker_iface_use_event_fd(wiface)) {
/* re-enable events, which were disabled by
Expand All @@ -666,10 +708,14 @@ static ucs_status_t ucp_worker_iface_check_events_do(ucp_worker_iface_t *wiface,

return UCS_OK;
} else if (status != UCS_ERR_BUSY) {
ucs_fatal("failed to arm iface %p: %s", wiface->iface,
ucs_status_string(status));
ucs_fatal("check_events: failed to arm " UCP_WIFACE_FMT ": %s",
UCP_WIFACE_ARG(wiface), ucs_status_string(status));
} else {
ucs_trace("arm iface %p returned BUSY", wiface->iface);
ucs_debug("check_events: arm " UCP_WIFACE_FMT " returned BUSY",
UCP_WIFACE_ARG(wiface));
/* Prevent ucp_worker_arm() from returning OK until the interface is
either activated or armed, so we will not block-wait forever. */
ucp_worker_iface_activate_arm_mark_busy(wiface);
return UCS_ERR_BUSY;
}
} else {
Expand Down Expand Up @@ -733,7 +779,7 @@ static void ucp_worker_iface_deactivate(ucp_worker_iface_t *wiface, int force)
{
ucp_worker_h worker = wiface->worker;

ucs_trace("deactivate " UCP_WIFACE_FMT " force=%d a_count=%u a_ifaces=%u",
ucs_debug("deactivate " UCP_WIFACE_FMT " force=%d a_count=%u a_ifaces=%u",
UCP_WIFACE_ARG(wiface), force, wiface->activate_count,
worker->num_active_ifaces);

Expand Down Expand Up @@ -775,7 +821,7 @@ void ucp_worker_iface_progress_ep(ucp_worker_iface_t *wiface)
/* This function may be called from progress thread (such as when processing
* wireup messages), so ask UCT to be thread-safe.
*/
ucp_worker_iface_activate(wiface, UCT_PROGRESS_THREAD_SAFE);
ucp_worker_iface_activate(wiface, UCT_PROGRESS_THREAD_SAFE, "progress ep");

UCS_ASYNC_UNBLOCK(&wiface->worker->async);
}
Expand Down Expand Up @@ -895,7 +941,7 @@ ucp_worker_iface_async_fd_event(int fd, ucs_event_set_types_t events, void *arg)
ucs_status_t status;

ucs_assertv(fd == event_fd, "fd=%d vs wiface::event_fd=%d", fd, event_fd);
ucs_trace_func("fd=%d iface=%p", event_fd, wiface->iface);
ucs_trace_func(UCP_WIFACE_FMT " fd=%d", UCP_WIFACE_ARG(wiface), event_fd);

status = ucs_async_modify_handler(event_fd, 0);
if (status != UCS_OK) {
Expand Down Expand Up @@ -1597,7 +1643,7 @@ ucs_status_t ucp_worker_iface_init(ucp_worker_h worker, ucp_rsc_index_t tl_id,
UCP_WIREUP_UCT_EVENT_CAP_FLAGS)) {
ucp_worker_iface_deactivate(wiface, 1);
} else {
ucp_worker_iface_activate(wiface, 0);
ucp_worker_iface_activate(wiface, 0, "event handler");
}
}

Expand All @@ -1611,6 +1657,7 @@ ucs_status_t ucp_worker_iface_init(ucp_worker_h worker, ucp_rsc_index_t tl_id,

void ucp_worker_iface_cleanup(ucp_worker_iface_t *wiface)
{
ucp_worker_iface_activate_arm_unmark_busy(wiface);
uct_worker_progress_unregister_safe(wiface->worker->uct,
&wiface->check_events_id);
ucp_worker_iface_disarm(wiface);
Expand Down Expand Up @@ -2620,6 +2667,7 @@ ucs_status_t ucp_worker_create(ucp_context_h context,
worker->flush_ops_count = 0;
worker->fence_seq = 0;
worker->inprogress = 0;
worker->arm_block_count = 0;
worker->num_active_ifaces = 0;
worker->num_ifaces = 0;
worker->am_message_id = ucs_generate_uuid(0);
Expand Down Expand Up @@ -3313,6 +3361,12 @@ ucs_status_t ucp_worker_arm(ucp_worker_h worker)
UCP_CONTEXT_CHECK_FEATURE_FLAGS(worker->context, UCP_FEATURE_WAKEUP,
return UCS_ERR_INVALID_PARAM);

if (worker->arm_block_count > 0) {
ucs_debug("worker %p: arm block count %d - returning busy", worker,
worker->arm_block_count);
return UCS_ERR_BUSY;
}

/* Read from event pipe. If some events are found, return BUSY, otherwise -
* continue to arm the transport interfaces.
*/
Expand Down Expand Up @@ -3348,8 +3402,8 @@ ucs_status_t ucp_worker_arm(ucp_worker_h worker)
ucs_list_for_each(wiface, &worker->arm_ifaces, arm_list) {
ucs_assert(wiface->activate_count > 0);
status = uct_iface_event_arm(wiface->iface, worker->uct_events);
ucs_trace_data("arm iface %p returned %s", wiface->iface,
ucs_status_string(status));
ucs_trace_data("arm " UCP_WIFACE_FMT " returned %s",
UCP_WIFACE_ARG(wiface), ucs_status_string(status));
if (status != UCS_OK) {
break;
}
Expand Down Expand Up @@ -3415,7 +3469,9 @@ ucs_status_t ucp_worker_wait(ucp_worker_h worker)
* because of using the same descriptor in multiple threads.
*/
for (;;) {
ucs_trace_data("worker %p: poll waiting on %ld fds", worker, nfds);
ret = poll(pfd, nfds, -1);
ucs_trace_data("worker %p: poll returned %d", worker, ret);
if (ret >= 0) {
ucs_assertv(ret == 1, "ret=%d", ret);
status = UCS_OK;
Expand Down
11 changes: 9 additions & 2 deletions src/ucp/core/ucp_worker.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,7 +124,12 @@ enum {
of arm_ifaces list, so
it needs to be armed
in ucp_worker_arm(). */
UCP_WORKER_IFACE_FLAG_UNUSED = UCS_BIT(2) /**< There is another UCP iface
UCP_WORKER_IFACE_FLAG_ACTIVATE_ARM_BUSY = UCS_BIT(2), /**< When tried to arm the interface
for activation events, the interface
is already busy, so need to keep
checking events for it in the
main progress loop. */
UCP_WORKER_IFACE_FLAG_UNUSED = UCS_BIT(3) /**< There is another UCP iface
with the same caps, but
with better performance */
};
Expand Down Expand Up @@ -325,6 +330,7 @@ typedef struct ucp_worker {
int eventfd; /* Event fd to support signal() calls */
unsigned uct_events; /* UCT arm events */
ucs_list_link_t arm_ifaces; /* List of interfaces to arm */
int arm_block_count; /* If >0, ucp_worker_arm() should return UCS_ERR_BUSY */

void *user_data; /* User-defined data */
ucs_strided_alloc_t ep_alloc; /* Endpoint allocator */
Expand Down Expand Up @@ -441,7 +447,8 @@ void ucp_worker_iface_unprogress_ep(ucp_worker_iface_t *wiface);

void ucp_worker_signal_internal(ucp_worker_h worker);

void ucp_worker_iface_activate(ucp_worker_iface_t *wiface, unsigned uct_flags);
void ucp_worker_iface_activate(ucp_worker_iface_t *wiface, unsigned uct_flags,
const char *reason);

int ucp_worker_iface_is_activated(const ucp_worker_iface_t *wiface);

Expand Down
5 changes: 5 additions & 0 deletions src/ucp/proto/proto_select.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -550,6 +550,11 @@ ucp_proto_select_lookup_slow(ucp_worker_h worker,
khiter_t khiter;
int khret;

ucs_assert(!ucs_async_is_from_async(&worker->async));

/* Initialize short-circuit thresholds on the relevant endpoint config */
ucp_ep_config_proto_short_lazy_init(worker, ep_cfg_index);

key.param = *select_param;
khiter = kh_get(ucp_proto_select_hash, proto_select->hash, key.u64);
if (khiter != kh_end(proto_select->hash)) {
Expand Down
Loading
Loading