diff --git a/src/cart/crt_corpc.c b/src/cart/crt_corpc.c index 2a7dba5f2c9..5ffd83a1186 100644 --- a/src/cart/crt_corpc.c +++ b/src/cart/crt_corpc.c @@ -684,16 +684,18 @@ crt_corpc_reply_hdlr(const struct crt_cb_info *cb_info) &co_info->co_replied_rpcs, crp_parent_link) { D_ASSERT(tmp_rpc_priv != parent_rpc_priv); - D_ASSERT(co_ops->co_aggregate != NULL); - rc = co_ops->co_aggregate(&tmp_rpc_priv->crp_pub, - &parent_rpc_priv->crp_pub, - co_info->co_priv); - if (rc != 0) { - D_ERROR("co_ops->co_aggregate(opc: %#x) " - "failed: "DF_RC"\n", - child_req->cr_opc, DP_RC(rc)); - if (co_info->co_rc == 0) - co_info->co_rc = rc; + + if (co_ops && co_ops->co_aggregate) { + rc = co_ops->co_aggregate(&tmp_rpc_priv->crp_pub, + &parent_rpc_priv->crp_pub, + co_info->co_priv); + if (rc != 0) { + D_ERROR("co_ops->co_aggregate(opc: %#x) " + "failed: " DF_RC "\n", + child_req->cr_opc, DP_RC(rc)); + if (co_info->co_rc == 0) + co_info->co_rc = rc; + } } co_info->co_child_ack_num++; D_DEBUG(DB_NET, "parent rpc %p, child rpc %p, " @@ -723,16 +725,16 @@ crt_corpc_reply_hdlr(const struct crt_cb_info *cb_info) memset(child_rpc_priv->crp_pub.cr_output, 0, child_rpc_priv->crp_pub.cr_output_size); } else { - D_ASSERT(co_ops->co_aggregate != NULL); - rc = co_ops->co_aggregate(child_req, - &parent_rpc_priv->crp_pub, - co_info->co_priv); - if (rc != 0) { - D_ERROR("co_ops->co_aggregate(opc: %#x)" - " failed: "DF_RC"\n", - child_req->cr_opc, DP_RC(rc)); - if (co_info->co_rc == 0) - co_info->co_rc = rc; + if (co_ops && co_ops->co_aggregate) { + rc = co_ops->co_aggregate( + child_req, &parent_rpc_priv->crp_pub, co_info->co_priv); + if (rc != 0) { + D_ERROR("co_ops->co_aggregate(opc: %#x)" + " failed: " DF_RC "\n", + child_req->cr_opc, DP_RC(rc)); + if (co_info->co_rc == 0) + co_info->co_rc = rc; + } } } } diff --git a/src/cart/crt_hg.c b/src/cart/crt_hg.c index b83e5777ff0..60e4e5c41ad 100644 --- a/src/cart/crt_hg.c +++ b/src/cart/crt_hg.c @@ -1079,6 +1079,14 @@ crt_hg_ctx_init(struct crt_hg_context *hg_ctx, crt_provider_t provider, int idx, D_GOTO(error, rc = crt_hgret_2_der(hg_ret)); } + /* cache self address */ + hg_ret = HG_Addr_self(hg_ctx->chc_hgcla, &hg_ctx->chc_self_addr); + if (hg_ret != HG_SUCCESS) { + hg_ctx->chc_self_addr = HG_ADDR_NULL; + D_ERROR("HG_Addr_self() failed; hg_ret: %d\n", hg_ret); + D_GOTO(error, rc = crt_hgret_2_der(hg_ret)); + } + rc = crt_hg_pool_init(hg_ctx); if (rc != 0) { D_ERROR("crt_hg_pool_init() failed, context idx %d hg_ctx %p, " @@ -1122,6 +1130,17 @@ crt_hg_ctx_fini(struct crt_hg_context *hg_ctx) crt_hg_pool_fini(hg_ctx); + /* Free the cached self address while the HG class is still valid */ + if (hg_ctx->chc_self_addr != HG_ADDR_NULL) { + D_ASSERT(hg_ctx->chc_hgcla != NULL); + + hg_ret = HG_Addr_free(hg_ctx->chc_hgcla, hg_ctx->chc_self_addr); + if (hg_ret != HG_SUCCESS) + D_WARN("HG_Addr_free(self) failed, hg_ret: " DF_HG_RC "\n", + DP_HG_RC(hg_ret)); + hg_ctx->chc_self_addr = HG_ADDR_NULL; + } + if (hg_ctx->chc_epfd > 0) { rc = close(hg_ctx->chc_epfd); if (rc != 0) { @@ -2249,15 +2268,38 @@ crt_hg_bulk_transfer(struct crt_bulk_desc *bulk_desc, crt_bulk_cb_t verify_cb, hg_ctx->chc_hgctx, crt_hg_bulk_transfer_cb, bulk_cbinfo, hg_bulk_op, remote_bulk, bulk_desc->bd_remote_off, local_bulk, bulk_desc->bd_local_off, bulk_desc->bd_len, opid != NULL ? (hg_op_id_t *)opid : HG_OP_ID_IGNORE); - else - hg_ret = HG_Bulk_transfer_id( - hg_ctx->chc_hgctx, crt_hg_bulk_transfer_cb, bulk_cbinfo, hg_bulk_op, - rpc_priv->crp_hg_addr, HG_Get_info(rpc_priv->crp_hg_hdl)->context_id, - remote_bulk, bulk_desc->bd_remote_off, local_bulk, bulk_desc->bd_local_off, - bulk_desc->bd_len, opid != NULL ? (hg_op_id_t *)opid : HG_OP_ID_IGNORE); + else { + hg_addr_t remote_addr = rpc_priv->crp_hg_addr; + int remote_ctx_idx; + + if (rpc_priv->crp_hg_hdl != HG_HANDLE_NULL) { + remote_ctx_idx = HG_Get_info(rpc_priv->crp_hg_hdl)->context_id; + remote_addr = rpc_priv->crp_hg_addr; + } else { + /* + * CoRPC executing on the root node (self) does not have a mercury handle. + * Use cached self address instead in such cases + **/ + + /* regular RPCs must have a valid handle */ + if (!rpc_priv->crp_coll) { + RPC_ERROR(rpc_priv, "Unexpected crp_hg_hdl=NULL\n"); + D_ASSERT(0); + } + + remote_ctx_idx = ctx->cc_idx; + remote_addr = hg_ctx->chc_self_addr; + } + + hg_ret = HG_Bulk_transfer_id(hg_ctx->chc_hgctx, crt_hg_bulk_transfer_cb, + bulk_cbinfo, hg_bulk_op, remote_addr, remote_ctx_idx, + remote_bulk, bulk_desc->bd_remote_off, local_bulk, + bulk_desc->bd_local_off, bulk_desc->bd_len, + opid != NULL ? (hg_op_id_t *)opid : HG_OP_ID_IGNORE); + } + if (hg_ret != HG_SUCCESS) { - D_ERROR("HG_Bulk_(bind)transfer failed, hg_ret: " DF_HG_RC "\n", - DP_HG_RC(hg_ret)); + D_ERROR("HG_Bulk_(bind)transfer failed, hg_ret: " DF_HG_RC "\n", DP_HG_RC(hg_ret)); D_FREE(bulk_cbinfo); D_FREE(bulk_desc_dup); rc = crt_hgret_2_der(hg_ret); diff --git a/src/cart/crt_hg.h b/src/cart/crt_hg.h index 41a738a8ad0..7ff992e5443 100644 --- a/src/cart/crt_hg.h +++ b/src/cart/crt_hg.h @@ -148,6 +148,7 @@ struct crt_hg_context { struct crt_hg_progress_multi chc_progress_multi; /* multi progress */ hg_class_t *chc_hgcla; /* HG class */ hg_context_t *chc_hgctx; /* HG context */ + hg_addr_t chc_self_addr; /* cached self address */ uint64_t chc_diag_pub_ts; /* time of last diagnostics pub */ int chc_provider; /* provider */ int chc_epfd; /* epoll fd */ diff --git a/src/tests/ftest/cart/SConscript b/src/tests/ftest/cart/SConscript index 35336add865..4e307d9e148 100644 --- a/src/tests/ftest/cart/SConscript +++ b/src/tests/ftest/cart/SConscript @@ -1,5 +1,6 @@ # /* # * (C) Copyright 2016-2023 Intel Corporation. +# * (C) Copyright 2026 Hewlett Packard Enterprise Development LP # * # * SPDX-License-Identifier: BSD-2-Clause-Patent # */ @@ -10,7 +11,7 @@ import os SIMPLE_TEST_SRC = ['threaded_client.c', 'dual_iface_server.c', 'no_pmix_multi_ctx.c', 'threaded_server.c', 'test_corpc_prefwd.c', - 'test_corpc_exclusive.c', 'dump_errnos.c', + 'test_corpc_exclusive.c', 'test_corpc_bulks.c', 'dump_errnos.c', 'test_proto_server.c', 'test_proto_client.c', 'test_multisend_server.c', 'test_multisend_client.c', 'test_no_timeout.c', 'test_ep_cred_server.c', diff --git a/src/tests/ftest/cart/corpc/corpc_one_node.yaml b/src/tests/ftest/cart/corpc/corpc_one_node.yaml index eb777f8f044..f4482ce78b0 100644 --- a/src/tests/ftest/cart/corpc/corpc_one_node.yaml +++ b/src/tests/ftest/cart/corpc/corpc_one_node.yaml @@ -36,3 +36,15 @@ tests: !mux test_servers_arg: "-e no_pmix_corpc_errors" test_servers_env: "" test_servers_ppn: "8" + corpc_bulk_inline: + name: corpc_bulk_inline + test_servers_bin: crt_launch + test_servers_arg: "-e test_corpc_bulks -i" + test_servers_env: "" + test_servers_ppn: "5" + corpc_bulk_explicit: + name: corpc_bulk_explicit + test_servers_bin: crt_launch + test_servers_arg: "-e test_corpc_bulks" + test_servers_env: "" + test_servers_ppn: "5" diff --git a/src/tests/ftest/cart/test_corpc_bulks.c b/src/tests/ftest/cart/test_corpc_bulks.c new file mode 100644 index 00000000000..f7ecb47abd9 --- /dev/null +++ b/src/tests/ftest/cart/test_corpc_bulks.c @@ -0,0 +1,485 @@ +/* + * (C) Copyright 2018-2022 Intel Corporation. + * (C) Copyright 2026 Hewlett Packard Enterprise Development LP + * + * SPDX-License-Identifier: BSD-2-Clause-Patent + */ +/** + * CORPC test with bulk transfer. Assumes 5 instances are running + * + * Rank 0 creates a source bulk handle backed by a static buffer and + * sends a CORPC to all ranks, including self. By default bulk is sent + * as an input param to the CORPC. Each rank then performs BULK_GET + * transfer and verifies resultant contents. + * + * If -i (inline) option is passed, bulk is passed inline during CORPC + * creation, and is accessed via crt_bulk_access() by all ranks, bypassing + * manual bulk transfer. + * + */ + +#include "crt_utils.h" +#include + +#include +#include +#include +#include +#include +#include + +#define TEST_CORPC_BULKS_BASE 0x010000000 +#define TEST_CORPC_BULKS_VER 0 +#define TEST_CORPC_BULK_SIZE (20 * 1024) +#define TEST_CORPC_BULK_PATTERN 0xae + +static bool g_corpc_hdlr_called; +static bool g_inline_bulk; +static d_rank_t g_my_rank; +static crt_bulk_t g_source_bulk = CRT_BULK_NULL; +static uint8_t g_source_buf[TEST_CORPC_BULK_SIZE]; + +enum { + TEST_OPC_BULK_TEST = CRT_PROTO_OPC(TEST_CORPC_BULKS_BASE, TEST_CORPC_BULKS_VER, 0), + TEST_OPC_SHUTDOWN +} test_corpc_bulks_opc_t; + +/* clang-format off */ +#define CRT_ISEQ_BULK_TEST /* input fields */ \ + ((crt_bulk_t) (bulk_hdl) CRT_VAR) \ + ((uint64_t) (bulk_size) CRT_VAR) + +#define CRT_OSEQ_BULK_TEST /* output fields */ \ + ((uint32_t) (unused) CRT_VAR) + +#define CRT_ISEQ_SHUTDOWN /* input fields */ \ + ((uint32_t) (unused) CRT_VAR) + +#define CRT_OSEQ_SHUTDOWN /* output fields */ \ + ((uint32_t) (unused) CRT_VAR) +/* clang-format on */ + +#define TEST_CORPC_BULKS_RPC(name, in_seq, out_seq) \ + CRT_RPC_DECLARE(name, in_seq, out_seq) \ + CRT_RPC_DEFINE(name, in_seq, out_seq) + +TEST_CORPC_BULKS_RPC(bulk_test, CRT_ISEQ_BULK_TEST, CRT_OSEQ_BULK_TEST); +TEST_CORPC_BULKS_RPC(shutdown, CRT_ISEQ_SHUTDOWN, CRT_OSEQ_SHUTDOWN); + +static void +__error_exit(int line, const char *fn) +{ + D_ERROR("Failed in %s on line %d\n", fn, line); + assert(0); +} + +#define error_exit() __error_exit(__LINE__, __func__); + +static void +verify_bulk_pattern(const uint8_t *buf, size_t len) +{ + size_t i; + + for (i = 0; i < len; i++) { + if (buf[i] != TEST_CORPC_BULK_PATTERN) { + D_ERROR("bulk data mismatch at %zu: expected %#x got %#x\n", i, + TEST_CORPC_BULK_PATTERN, buf[i]); + error_exit(); + } + } +} + +static void +verify_implicit_bulk(crt_bulk_t bulk_hdl, uint64_t expected_size) +{ + d_sg_list_t sgl = {0}; + d_iov_t *iovs = NULL; + uint32_t seg_num; + size_t total_size = 0; + uint32_t i; + int rc; + + rc = crt_bulk_access(bulk_hdl, &sgl); + if (rc != -DER_TRUNC) { + DL_ERROR(rc, "crt_bulk_access() probe failed"); + error_exit(); + } + + seg_num = sgl.sg_nr_out; + if (seg_num == 0) { + D_ERROR("implicit bulk has no segments\n"); + error_exit(); + } + + D_ALLOC_ARRAY(iovs, seg_num); + if (iovs == NULL) + error_exit(); + + sgl.sg_nr = seg_num; + sgl.sg_iovs = iovs; + rc = crt_bulk_access(bulk_hdl, &sgl); + if (rc != 0) { + DL_ERROR(rc, "crt_bulk_access() read failed"); + D_FREE(iovs); + error_exit(); + } + + for (i = 0; i < seg_num; i++) { + D_ASSERTF(iovs[i].iov_buf != NULL, "implicit bulk segment %u is NULL\n", i); + verify_bulk_pattern(iovs[i].iov_buf, iovs[i].iov_len); + total_size += iovs[i].iov_len; + } + + D_FREE(iovs); + D_ASSERTF(total_size == expected_size, + "implicit bulk size mismatch: expected=" DF_U64 " got=%zu\n", expected_size, + total_size); +} + +static int +bulk_transfer_done_cb(const struct crt_bulk_cb_info *info) +{ + uint8_t *dst_buf; + int rc; + + if (info == NULL || info->bci_bulk_desc == NULL) { + D_ERROR("bulk completion info is invalid\n"); + error_exit(); + } + + if (info->bci_rc != 0) { + DL_ERROR(info->bci_rc, "Bulk transfer failed"); + error_exit(); + } + + dst_buf = info->bci_arg; + verify_bulk_pattern(dst_buf, info->bci_bulk_desc->bd_len); + + rc = crt_reply_send(info->bci_bulk_desc->bd_rpc); + if (rc != 0) { + DL_ERROR(rc, "Failed to send bulk reply"); + error_exit(); + } + + crt_bulk_free(info->bci_bulk_desc->bd_local_hdl); + D_FREE(dst_buf); + RPC_PUB_DECREF(info->bci_bulk_desc->bd_rpc); + + return 0; +} + +static void +corpc_hdlr(crt_rpc_t *rpc) +{ + struct bulk_test_in *input; + crt_bulk_t remote_bulk; + int rc; + + DBG_PRINT("corpc handler called on rank %d\n", g_my_rank); + g_corpc_hdlr_called = true; + + input = crt_req_get(rpc); + D_ASSERTF(input != NULL, "bulk corpc input is NULL\n"); + D_ASSERTF(input->bulk_size == TEST_CORPC_BULK_SIZE, "unexpected bulk size=" DF_U64 "\n", + input->bulk_size); + + remote_bulk = g_inline_bulk ? rpc->cr_co_bulk_hdl : input->bulk_hdl; + D_ASSERTF(remote_bulk != CRT_BULK_NULL, "bulk handle is not set\n"); + + if (g_inline_bulk) { + verify_implicit_bulk(remote_bulk, input->bulk_size); + rc = crt_reply_send(rpc); + D_ASSERTF(rc == 0, "implicit bulk reply failed\n"); + return; + } + + /* Issue a bulk transfer to get the data */ + { + struct crt_bulk_desc bulk_desc; + crt_bulk_t dst_bulk; + d_sg_list_t sgl; + uint8_t *dst_buf; + + D_ALLOC_ARRAY(dst_buf, input->bulk_size); + if (dst_buf == NULL) + error_exit(); + + rc = d_sgl_init(&sgl, 1); + if (rc != 0) + error_exit(); + + sgl.sg_iovs[0].iov_buf = dst_buf; + sgl.sg_iovs[0].iov_buf_len = input->bulk_size; + sgl.sg_iovs[0].iov_len = input->bulk_size; + + rc = crt_bulk_create(rpc->cr_ctx, &sgl, CRT_BULK_RW, &dst_bulk); + if (rc != 0) + error_exit(); + + RPC_PUB_ADDREF(rpc); + bulk_desc.bd_rpc = rpc; + bulk_desc.bd_bulk_op = CRT_BULK_GET; + bulk_desc.bd_remote_hdl = remote_bulk; + bulk_desc.bd_remote_off = 0; + bulk_desc.bd_local_hdl = dst_bulk; + bulk_desc.bd_local_off = 0; + bulk_desc.bd_len = input->bulk_size; + + rc = crt_bulk_transfer(&bulk_desc, bulk_transfer_done_cb, dst_buf, NULL); + if (rc != 0) { + DL_ERROR(rc, "bulk transfer failed"); + error_exit(); + } + } +} + +static void +shutdown_hdlr(crt_rpc_t *rpc) +{ + int rc; + + DBG_PRINT("shutdown handler called\n"); + + rc = crt_reply_send(rpc); + D_ASSERT(rc == 0); + + crtu_progress_stop(); +} + +static void +corpc_response_hdlr(const struct crt_cb_info *info) +{ + sem_t *sem; + + D_ASSERTF(info != NULL, "cb_info is null\n"); + D_ASSERTF(info->cci_rc == 0, "CORPC completed with an error\n"); + + sem = (sem_t *)info->cci_arg; + sem_post(sem); +} + +static void +shutdown_resp_hdlr(const struct crt_cb_info *info) +{ + sem_t *sem; + + D_ASSERTF(info != NULL, "cb_info is null\n"); + D_ASSERTF(info->cci_rc == 0, "Shutdown RPC completed with an error\n"); + + sem = (sem_t *)info->cci_arg; + sem_post(sem); +} + +static struct crt_proto_rpc_format proto_rpc_fmt[] = {{ + .prf_flags = 0, + .prf_req_fmt = &CQF_bulk_test, + .prf_hdlr = corpc_hdlr, + .prf_co_ops = NULL, + }, + { + .prf_flags = 0, + .prf_req_fmt = &CQF_shutdown, + .prf_hdlr = shutdown_hdlr, + .prf_co_ops = NULL, + }}; + +static struct crt_proto_format my_proto = { + .cpf_name = "my-proto-corpc-bulks", + .cpf_ver = TEST_CORPC_BULKS_VER, + .cpf_count = ARRAY_SIZE(proto_rpc_fmt), + .cpf_prf = &proto_rpc_fmt[0], + .cpf_base = TEST_CORPC_BULKS_BASE, +}; + +static void +show_usage(const char *prog) +{ + printf("Usage: %s [-b]\n", prog); + printf("Options:\n"); + printf("-i: Use inline bulk for CORPC\n"); +} + +static int +parse_args(int argc, char **argv) +{ + int c; + + g_inline_bulk = false; + + while ((c = getopt(argc, argv, "i")) != -1) { + switch (c) { + case 'i': + g_inline_bulk = true; + break; + default: + show_usage(argv[0]); + return -1; + } + } + + if (optind < argc) { + show_usage(argv[0]); + return -1; + } + + return 0; +} + +static void +init_source_bulk(crt_context_t ctx) +{ + d_sg_list_t sgl; + int rc; + + memset(g_source_buf, TEST_CORPC_BULK_PATTERN, sizeof(g_source_buf)); + + rc = d_sgl_init(&sgl, 1); + D_ASSERTF(rc == 0, "d_sgl_init() failed; rc=%d\n", rc); + + sgl.sg_iovs[0].iov_buf = g_source_buf; + sgl.sg_iovs[0].iov_buf_len = sizeof(g_source_buf); + sgl.sg_iovs[0].iov_len = sizeof(g_source_buf); + + rc = crt_bulk_create(ctx, &sgl, CRT_BULK_RO, &g_source_bulk); + D_ASSERTF(rc == 0, "crt_bulk_create() for source bulk failed; rc=%d\n", rc); +} + +int +main(int argc, char **argv) +{ + int rc; + crt_context_t ctx; + d_rank_list_t *rank_list; + crt_rpc_t *rpc; + uint32_t grp_size; + crt_group_t *grp; + char *env_self_rank; + char *grp_cfg_file; + pthread_t progress_thread; + sem_t sem; + crt_endpoint_t server_ep; + int i; + static d_rank_t my_rank; + struct bulk_test_in *input; + + rc = parse_args(argc, argv); + if (rc != 0) + return rc; + + /* get self rank from the env that crt_launch prepares */ + d_agetenv_str(&env_self_rank, "CRT_L_RANK"); + my_rank = atoi(env_self_rank); + g_my_rank = my_rank; + d_freeenv_str(&env_self_rank); + + rc = sem_init(&sem, 0, 0); + D_ASSERTF(rc == 0, "sem_init() failed.\n"); + + /* rank, num_attach_retries, is_server, D_ASSERT_on_error */ + crtu_test_init(my_rank, 20, true, true); + crtu_set_shutdown_delay(0); + + rc = d_log_init(); + D_ASSERT(rc == 0); + + rc = crt_init(NULL, CRT_FLAG_BIT_SERVER | CRT_FLAG_BIT_AUTO_SWIM_DISABLE); + D_ASSERTF(rc == 0, "crt_init() failed\n"); + + rc = crt_proto_register(&my_proto); + D_ASSERTF(rc == 0, "crt_proto_register() failed\n"); + + rc = crt_context_create(&ctx); + D_ASSERTF(rc == 0, "crt_context_create() failed\n"); + + d_agetenv_str(&grp_cfg_file, "CRT_L_GRP_CFG"); + + rc = crt_rank_self_set(my_rank, 1 /* group_version_min */); + D_ASSERTF(rc == 0, "crt_rank_self_set(%d) failed\n", my_rank); + + grp = crt_group_lookup(NULL); + D_ASSERTF(grp != NULL, "Failed to lookup group\n"); + + /* load group info from a config file and delete file upon return */ + rc = crtu_load_group_from_file(grp_cfg_file, ctx, grp, my_rank, true); + d_freeenv_str(&grp_cfg_file); + D_ASSERTF(rc == 0, "crtu_load_group_from_file() failed; rc=%d\n", rc); + + /* test requires 5 ranks */ + rc = crt_group_size(grp, &grp_size); + D_ASSERTF(rc == 0, "crt_group_size() failed\n"); + D_ASSERTF(grp_size == 5, "This test requires 5 ranks\n"); + + rc = crt_group_ranks_get(grp, &rank_list); + D_ASSERTF(rc == 0, "crt_group_ranks_get() failed; rc=%d\n", rc); + + rc = pthread_create(&progress_thread, 0, crtu_progress_fn, &ctx); + D_ASSERTF(rc == 0, "pthread_create() failed; rc=%d\n", rc); + + if (my_rank == 0) + init_source_bulk(ctx); + + /* rank=0 is initiator of the test, the rest of ranks wait for rpcs */ + if (my_rank != 0) + D_GOTO(wait_for_rpcs, 0); + + /* Wait for all ranks to come up, 5 seconds per ping, 100 seconds max */ + rc = crtu_wait_for_ranks(ctx, grp, rank_list, 0, 1, 5, 100.0); + D_ASSERTF(rc == 0, "wait_for_ranks() failed; rc=%d\n", rc); + + d_rank_list_free(rank_list); + rank_list = NULL; + + rc = crt_corpc_req_create(ctx, grp, NULL, TEST_OPC_BULK_TEST, + g_inline_bulk ? g_source_bulk : CRT_BULK_NULL, NULL, 0, + crt_tree_topo(CRT_TREE_KNOMIAL, 4), &rpc); + D_ASSERTF(rc == 0, "crt_corpc_req_create() failed\n"); + + DBG_PRINT("Sending CORPC with %s bulk\n", g_inline_bulk ? "inline" : "explicit"); + + input = crt_req_get(rpc); + D_ASSERTF(input != NULL, "bulk corpc input is NULL\n"); + + input->bulk_hdl = g_inline_bulk ? CRT_BULK_NULL : g_source_bulk; + input->bulk_size = TEST_CORPC_BULK_SIZE; + + rc = crt_req_send(rpc, corpc_response_hdlr, &sem); + D_ASSERT(rc == 0); + + /* wait for corpc completion */ + crtu_sem_timedwait(&sem, 61, __LINE__); + + /* Send shutdown RPCs to all ranks */ + server_ep.ep_grp = NULL; + server_ep.ep_tag = 0; + for (i = 1; i < grp_size; i++) { + server_ep.ep_rank = i; + + rc = crt_req_create(ctx, &server_ep, TEST_OPC_SHUTDOWN, &rpc); + D_ASSERTF(rc == 0, "crt_req_create() TEST_OPC_SHUTDOWN failed\n"); + + rc = crt_req_send(rpc, shutdown_resp_hdlr, &sem); + D_ASSERTF(rc == 0, "crt_req_send() TEST_OPC_SHUTDOWN failed\n"); + crtu_sem_timedwait(&sem, 61, __LINE__); + } + + if (g_source_bulk != CRT_BULK_NULL) + crt_bulk_free(g_source_bulk); + + crtu_progress_stop(); + +wait_for_rpcs: + /* Wait until progress thread exits */ + pthread_join(progress_thread, NULL); + + D_ASSERTF(g_corpc_hdlr_called == true, "bulk corpc_handler was not called\n"); + DBG_PRINT("All tests done\n"); + + rc = sem_destroy(&sem); + D_ASSERTF(rc == 0, "sem_destroy() failed\n"); + + rc = crt_finalize(); + D_ASSERTF(rc == 0, "crt_finalize() failed\n"); + + d_log_fini(); + return 0; +} diff --git a/src/tests/ftest/cart/test_corpc_exclusive.c b/src/tests/ftest/cart/test_corpc_exclusive.c index dd3f4ed4ddf..3b0ba796803 100644 --- a/src/tests/ftest/cart/test_corpc_exclusive.c +++ b/src/tests/ftest/cart/test_corpc_exclusive.c @@ -20,16 +20,6 @@ static bool corpc_hdlr_called = false; -static int -corpc_aggregate(crt_rpc_t *src, crt_rpc_t *result, void *priv) -{ - return 0; -} - -struct crt_corpc_ops corpc_set_ivns_ops = { - .co_aggregate = corpc_aggregate, -}; - static void corpc_hdlr(crt_rpc_t *rpc) { @@ -101,7 +91,7 @@ static struct crt_proto_rpc_format proto_rpc_fmt[] = {{ .prf_flags = 0, .prf_req_fmt = &CQF_basic_corpc, .prf_hdlr = corpc_hdlr, - .prf_co_ops = &corpc_set_ivns_ops, + .prf_co_ops = NULL, }, { .prf_flags = 0, diff --git a/src/tests/ftest/cart/test_corpc_prefwd.c b/src/tests/ftest/cart/test_corpc_prefwd.c index 8aa9480476e..e87bdcead3a 100644 --- a/src/tests/ftest/cart/test_corpc_prefwd.c +++ b/src/tests/ftest/cart/test_corpc_prefwd.c @@ -1,5 +1,6 @@ /* * (C) Copyright 2018-2022 Intel Corporation. + * (C) Copyright 2026 Hewlett Packard Enterprise Development LP * * SPDX-License-Identifier: BSD-2-Clause-Patent */ @@ -48,10 +49,10 @@ corpc_post_reply(crt_rpc_t *rpc, void *arg) return 0; } -struct crt_corpc_ops corpc_set_ivns_ops = { - .co_aggregate = corpc_aggregate, - .co_pre_forward = corpc_pre_forward, - .co_post_reply = corpc_post_reply, +struct crt_corpc_ops corpc_ops = { + .co_aggregate = corpc_aggregate, + .co_pre_forward = corpc_pre_forward, + .co_post_reply = corpc_post_reply, }; static void @@ -91,14 +92,12 @@ corpc_response_hdlr(const struct crt_cb_info *info) crtu_progress_stop(); } -static struct crt_proto_rpc_format my_proto_rpc_fmt_basic_corpc[] = { - { - .prf_flags = 0, - .prf_req_fmt = &CQF_basic_corpc, - .prf_hdlr = test_basic_corpc_hdlr, - .prf_co_ops = &corpc_set_ivns_ops, - } -}; +static struct crt_proto_rpc_format my_proto_rpc_fmt_basic_corpc[] = {{ + .prf_flags = 0, + .prf_req_fmt = &CQF_basic_corpc, + .prf_hdlr = test_basic_corpc_hdlr, + .prf_co_ops = &corpc_ops, +}}; static struct crt_proto_format my_proto_fmt_basic_corpc = { .cpf_name = "my-proto-basic_corpc",