diff --git a/src/ucp/core/ucp_ep.c b/src/ucp/core/ucp_ep.c index a822b85d580..d2869d7926f 100644 --- a/src/ucp/core/ucp_ep.c +++ b/src/ucp/core/ucp_ep.c @@ -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, @@ -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); @@ -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) { @@ -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); + } } } @@ -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); @@ -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 */ @@ -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); @@ -4740,8 +4748,8 @@ 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; @@ -4749,13 +4757,11 @@ static void ucp_ep_config_proto_init(ucp_worker_h worker, 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; @@ -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) diff --git a/src/ucp/core/ucp_ep.h b/src/ucp/core/ucp_ep.h index 6df732b7a72..e58b2408145 100644 --- a/src/ucp/core/ucp_ep.h +++ b/src/ucp/core/ucp_ep.h @@ -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; @@ -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, diff --git a/src/ucp/core/ucp_worker.c b/src/ucp/core/ucp_worker.c index 45ced6b3524..a8231bc9ff3 100644 --- a/src/ucp/core/ucp_worker.c +++ b/src/ucp/core/ucp_worker.c @@ -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); @@ -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 */ @@ -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 */ @@ -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 */ @@ -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 @@ -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 { @@ -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); @@ -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); } @@ -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) { @@ -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"); } } @@ -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); @@ -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); @@ -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. */ @@ -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; } @@ -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; diff --git a/src/ucp/core/ucp_worker.h b/src/ucp/core/ucp_worker.h index 3f917ea63a1..b7f68c585af 100644 --- a/src/ucp/core/ucp_worker.h +++ b/src/ucp/core/ucp_worker.h @@ -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 */ }; @@ -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 */ @@ -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); diff --git a/src/ucp/proto/proto_select.c b/src/ucp/proto/proto_select.c index ad19d6fb3f3..7279ec7bf33 100644 --- a/src/ucp/proto/proto_select.c +++ b/src/ucp/proto/proto_select.c @@ -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)) { diff --git a/src/ucp/proto/proto_select.inl b/src/ucp/proto/proto_select.inl index 2a9f09c7ae6..598a107b86f 100644 --- a/src/ucp/proto/proto_select.inl +++ b/src/ucp/proto/proto_select.inl @@ -84,6 +84,10 @@ ucp_proto_select_lookup(ucp_worker_h worker, ucp_proto_select_t *proto_select, ucp_proto_select_key_t key; khiter_t khiter; + /* Protocol selection should not be triggered from async context, since + it may race with protocol table updates from main thread */ + ucs_assert(!ucs_async_is_from_async(&worker->async)); + UCS_STATIC_ASSERT(sizeof(key.param) == sizeof(key.u64)); key.param = *select_param; @@ -104,6 +108,7 @@ ucp_proto_select_lookup(ucp_worker_h worker, ucp_proto_select_t *proto_select, } } + /* FIXME proto_select could get invalidated by adding rkey config */ proto_select->cache.key = key.u64; proto_select->cache.value = select_elem; } diff --git a/src/ucp/wireup/wireup.c b/src/ucp/wireup/wireup.c index 42c4f024c26..1257fe498ae 100644 --- a/src/ucp/wireup/wireup.c +++ b/src/ucp/wireup/wireup.c @@ -566,6 +566,7 @@ unsigned ucp_wireup_eps_progress(void *arg) } /* Replay pending requests */ + ucs_trace("ep %p: replaying pending requests", ucp_ep); ucp_wireup_replay_pending_requests(ucp_ep, &tmp_pending_queue); UCS_ASYNC_UNBLOCK(&ucp_ep->worker->async); return 1; @@ -1212,16 +1213,33 @@ uct_ep_h ucp_wireup_extract_lane(ucp_ep_h ep, ucp_lane_index_t lane) static void ucp_wireup_replay_pending_request(uct_pending_req_t *self, ucp_ep_h ucp_ep) { - ucp_request_t *req = ucs_container_of(self, ucp_request_t, send.uct); + ucp_request_t *req = ucs_container_of(self, ucp_request_t, send.uct); + ucp_lane_index_t am_lane = ucp_ep_get_am_lane(ucp_ep); + int pending_added; ucs_assert(req->send.ep == ucp_ep); if ((req->flags & UCP_REQUEST_FLAG_PROTO_SEND) && ((ucp_ep->cfg_index != req->send.proto_config->ep_cfg_index) || ucp_ep->worker->context->config.ext.proto_request_reset)) { - ucp_trace_req(req, "replay proto %s", - req->send.proto_config->proto->name); - ucp_proto_request_restart(req); + if (ucp_wireup_ep_test(ucp_ep_get_lane(ucp_ep, am_lane))) { + ucs_debug("ep %p: adding request %p to pending of lane %d", ucp_ep, + req, am_lane); + req->send.lane = am_lane; + pending_added = ucp_request_pending_add(req); + + /* Pending add should always succeed, since am_lane is a wireup_ep */ + ucs_assertv_always(pending_added, "ep=%p req=%p am_lane=%d", ucp_ep, + req, am_lane); + } else { + /* This path should not be reachable from async thread, because only + * wireup protocol can trigger an async thread callback, and in this + * case there would be a wireup ep on am_lane. + */ + ucp_trace_req(req, "replay proto %s", + req->send.proto_config->proto->name); + ucp_proto_request_restart(req); + } } else { ucp_trace_req(req, "replay proto %s lane %d", (req->flags & UCP_REQUEST_FLAG_PROTO_SEND) ? diff --git a/test/gtest/ucp/test_ucp_proto.cc b/test/gtest/ucp/test_ucp_proto.cc index 88a754ea810..5d4f6b94f7a 100644 --- a/test/gtest/ucp/test_ucp_proto.cc +++ b/test/gtest/ucp/test_ucp_proto.cc @@ -558,6 +558,33 @@ UCS_TEST_P(test_ucp_proto, dump_protocols) { ucp_ep_print_info(sender().ep(), stdout); } +UCS_TEST_P(test_ucp_proto, ep_config_proto_init_on_send) +{ + static constexpr ucp_tag_t tag = 0xdeadbeef; + ucp_request_param_t param = {}; + uint64_t send_data = 0x0123456789abcdef; + uint64_t recv_data = 0; + ucp_ep_config_t *ep_config; + void *send_req, *recv_req; + + recv_req = ucp_tag_recv_nbx(receiver().worker(), &recv_data, + sizeof(recv_data), tag, UINT64_MAX, ¶m); + + ep_config = &ucs_array_elem(&worker()->ep_config, sender().ep()->cfg_index); + EXPECT_FALSE(ep_config->proto_init_flags & + UCP_EP_PROTO_SHORT_INITIALIZED); + + /* No protocol can be cached while the initialization flag is clear */ + send_req = ucp_tag_send_nbx(sender().ep(), &send_data, sizeof(send_data), + tag, ¶m); + + /* Expect short-circuit path initialization to be done after first send */ + EXPECT_TRUE(ep_config->proto_init_flags & UCP_EP_PROTO_SHORT_INITIALIZED); + + ASSERT_UCS_OK(requests_wait({send_req, recv_req})); + EXPECT_EQ(send_data, recv_data); +} + UCS_TEST_P(test_ucp_proto, rkey_config) { ucp_rkey_config_key_t rkey_config_key = create_rkey_config_key(0); ucs_status_t status; diff --git a/test/gtest/ucp/test_ucp_sockaddr.cc b/test/gtest/ucp/test_ucp_sockaddr.cc index 6bdef6e9fb6..911a28f8b4c 100644 --- a/test/gtest/ucp/test_ucp_sockaddr.cc +++ b/test/gtest/ucp/test_ucp_sockaddr.cc @@ -3315,10 +3315,9 @@ class test_ucp_sockaddr_iface_activate : public test_ucp_sockaddr { return false; } - void client_connect_disconnect() + void client_connect_disconnect(bool communicate) { /* Check state before connection establishment */ - EXPECT_FALSE(is_any_interface_activated(receiver())); int receiver_ep_count = receiver().get_num_eps(); /* Connect sender and receiver */ @@ -3330,16 +3329,21 @@ class test_ucp_sockaddr_iface_activate : public test_ucp_sockaddr { /* Check state after connection establishment */ EXPECT_EQ(0, sender().get_err_num()); EXPECT_TRUE(is_any_interface_activated(sender())); - EXPECT_TRUE(is_any_interface_activated(receiver())); + + /* Check state after communication */ + if (communicate) { + for (int i = 0; i < 10; ++i) { + send_recv(sender(), receiver(), SEND_RECV_TAG, false, + cb_type()); + progress(); + } + EXPECT_TRUE(is_any_interface_activated(receiver())); + } one_sided_disconnect(sender()); one_sided_disconnect(receiver(), 0, receiver_ep_count); - EXPECT_FALSE(is_any_interface_activated(receiver())); receiver().reset_err_num(); - - /* TODO: check why receiver interfaces are not deactivated on one-sided - * client disconnect */ } }; @@ -3347,8 +3351,16 @@ UCS_TEST_SKIP_COND_P(test_ucp_sockaddr_iface_activate, iface_activate_count, !is_proto_enabled()) { listen(cb_type()); - client_connect_disconnect(); - client_connect_disconnect(); + EXPECT_FALSE(is_any_interface_activated(receiver())); + + client_connect_disconnect(false); + client_connect_disconnect(false); + EXPECT_FALSE(is_any_interface_activated(receiver())); + + client_connect_disconnect(true); + + /* Receiver stays activated to progress incoming messages */ + EXPECT_TRUE(is_any_interface_activated(receiver())); } UCP_INSTANTIATE_TEST_CASE_TLS(test_ucp_sockaddr_iface_activate, tcp, "tcp")