Skip to content

DAOS-14166 cart: Add versioning support to RPC headers #19277

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

Open
frostedcmos wants to merge 6 commits into
base: master
Choose a base branch
from
Open

DAOS-14166 cart: Add versioning support to RPC headers #19277

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
1692a73
DAOS-14166 cart: Add versioning support to RPC headers
frostedcmos Oct 14, 2026
998cbbc
- Added copyright header
frostedcmos Oct 14, 2026
0171b9e
- Adding spaceholder for bulk changes required for multiprovider
frostedcmos Oct 16, 2026
7aa384d
- Reduce deadline size back to 32bit for now, to see perf impact
frostedcmos Oct 19, 2026
15b5dc7
Merge branch 'master' into frostedcmos/rpc-header-refactor-3.0
frostedcmos Oct 20, 2026
a61c82e
Merge branch 'master' into frostedcmos/rpc-header-refactor-3.0
frostedcmos Oct 22, 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
12 changes: 7 additions & 5 deletions src/cart/crt_bulk.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,16 +253,18 @@ crt_bulk_free_common(struct crt_bulk *bulk)
static inline bool
crt_bulk_desc_expired(struct crt_bulk_desc *bulk_desc)
{
struct timespec now;
uint64_t now_sec;
struct crt_rpc_priv *rpc_priv;

clock_gettime(CLOCK_REALTIME_COARSE, &now);
now_sec = crt_realtime_sec();
rpc_priv = container_of(bulk_desc->bd_rpc, struct crt_rpc_priv, crp_pub);

/* Deadline expired */
if (now.tv_sec > rpc_priv->crp_deadline_sec) {
RPC_TRACE(DB_NET, rpc_priv, "Deadline expired for bulk. Deadline=%d, now=%ld\n",
rpc_priv->crp_deadline_sec, now.tv_sec);
if (now_sec > rpc_priv->crp_deadline_sec) {
RPC_TRACE(DB_NET, rpc_priv,
"Deadline expired for bulk. Deadline=" DF_U64 ", now=" DF_U64
" seconds\n",
rpc_priv->crp_deadline_sec, now_sec);
return true;
}

Expand Down
31 changes: 17 additions & 14 deletions src/cart/crt_context.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -513,7 +513,7 @@ crt_rpc_complete_and_unlock(struct crt_rpc_priv *rpc_priv, int rc)
cbinfo.cci_rc = rc;

if (cbinfo.cci_rc == 0)
cbinfo.cci_rc = rpc_priv->crp_reply_hdr.cch_rc;
cbinfo.cci_rc = rpc_priv->crp_reply_hdr.crh_status;

if (cbinfo.cci_rc != 0)
RPC_CWARN(crt_quiet_error(cbinfo.cci_rc), DB_NET, rpc_priv, "failed, " DF_RC "\n",
Expand Down Expand Up @@ -1051,7 +1051,6 @@ crt_req_timeout_track(struct crt_rpc_priv *rpc_priv)
if (rpc_priv->crp_in_binheap == 1)
D_GOTO(out, rc = 0);

/* Set the deadline as RPC is starting to be tracked */
rpc_priv->crp_deadline_sec = crt_timeout_to_deadline(rpc_priv->crp_timeout_sec);

/* add to binheap for timeout tracking */
Expand Down Expand Up @@ -1274,19 +1273,23 @@ crt_context_timeout_check(struct crt_context *crt_ctx)
if (!print_once) {
print_once = true;

RPC_WARN(rpc_priv,
"ctx_id %d, (status: %#x) timed out (%d seconds) [deadline: %d], "
"target (%d:%d)\n",
crt_ctx->cc_idx, rpc_priv->crp_state, rpc_priv->crp_timeout_sec,
rpc_priv->crp_deadline_sec, rpc_priv->crp_pub.cr_ep.ep_rank,
rpc_priv->crp_pub.cr_ep.ep_tag);
RPC_WARN(
rpc_priv,
"ctx_id %d, (status: %#x) timed out (%d seconds) [deadline_sec: " DF_U64
"], "
"target (%d:%d)\n",
crt_ctx->cc_idx, rpc_priv->crp_state, rpc_priv->crp_timeout_sec,
rpc_priv->crp_deadline_sec, rpc_priv->crp_pub.cr_ep.ep_rank,
rpc_priv->crp_pub.cr_ep.ep_tag);
} else {
RPC_INFO(rpc_priv,
"ctx_id %d, (status: %#x) timed out (%d seconds) [deadline: %d], "
"target (%d:%d)\n",
crt_ctx->cc_idx, rpc_priv->crp_state, rpc_priv->crp_timeout_sec,
rpc_priv->crp_deadline_sec, rpc_priv->crp_pub.cr_ep.ep_rank,
rpc_priv->crp_pub.cr_ep.ep_tag);
RPC_INFO(
rpc_priv,
"ctx_id %d, (status: %#x) timed out (%d seconds) [deadline_sec: " DF_U64
"], "
"target (%d:%d)\n",
crt_ctx->cc_idx, rpc_priv->crp_state, rpc_priv->crp_timeout_sec,
rpc_priv->crp_deadline_sec, rpc_priv->crp_pub.cr_ep.ep_rank,
rpc_priv->crp_pub.cr_ep.ep_tag);
}

crt_req_timeout_hdlr(rpc_priv);
Expand Down
13 changes: 8 additions & 5 deletions src/cart/crt_corpc.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -522,7 +522,7 @@ corpc_del_child_rpc_locked(struct crt_rpc_priv *parent_rpc_priv,
static inline void
crt_corpc_fail_parent_rpc(struct crt_rpc_priv *parent_rpc_priv, int failed_rc)
{
parent_rpc_priv->crp_reply_hdr.cch_rc = failed_rc;
parent_rpc_priv->crp_reply_hdr.crh_status = failed_rc;
parent_rpc_priv->crp_corpc_info->co_rc = failed_rc;
if (failed_rc != -DER_GRPVER)
RPC_ERROR(parent_rpc_priv,
Expand Down Expand Up @@ -662,9 +662,9 @@ crt_corpc_reply_hdlr(const struct crt_cb_info *cb_info)
co_info->co_rc = rc;
}
/* propagate failure rc to parent */
if (child_rpc_priv->crp_reply_hdr.cch_rc != 0)
if (child_rpc_priv->crp_reply_hdr.crh_status != 0)
crt_corpc_fail_parent_rpc(parent_rpc_priv,
child_rpc_priv->crp_reply_hdr.cch_rc);
child_rpc_priv->crp_reply_hdr.crh_status);

co_ops = opc_info->coi_co_ops;
if (co_ops == NULL) {
Expand Down Expand Up @@ -900,8 +900,11 @@ crt_corpc_req_hdlr(struct crt_rpc_priv *rpc_priv)

/* Set child's deadline same as parent and calculate timeout left from it */
child_rpc_priv->crp_deadline_sec = rpc_priv->crp_deadline_sec;
child_rpc_priv->crp_timeout_sec =
crt_deadline_to_timeout(rpc_priv->crp_deadline_sec);
{
int64_t timeout_sec = crt_deadline_to_timeout(rpc_priv->crp_deadline_sec);

child_rpc_priv->crp_timeout_sec = timeout_sec > 0 ? timeout_sec : 1;
}

rc = crt_req_send(child_rpc, crt_corpc_reply_hdlr, rpc_priv);
if (rc != 0) {
Expand Down
25 changes: 11 additions & 14 deletions src/cart/crt_hg.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1253,11 +1253,12 @@ crt_rpc_handler_common(hg_handle_t hg_hdl)
rc = crt_hg_unpack_header(hg_hdl, &rpc_tmp, &proc);
if (unlikely(rc != 0)) {
D_ERROR("crt_hg_unpack_header failed, rc: %d.\n", rc);
D_GOTO(reply_error, rc = -DER_MISC);
HG_Destroy(rpc_tmp.crp_hg_hdl);
return HG_PROTOCOL_ERROR;
}

D_ASSERT(proc != NULL);
opc = rpc_tmp.crp_req_hdr.cch_opc;
opc = rpc_tmp.crp_req_hdr.crh_opc;
rpc_tmp.crp_pub.cr_opc = opc;

/* allocate rpc struct for a given opcode; in/out size will vary per opc */
Expand All @@ -1276,14 +1277,11 @@ crt_rpc_handler_common(hg_handle_t hg_hdl)
opc_info = rpc_priv->crp_opc_info;
rpc_pub = &rpc_priv->crp_pub;

/* populate rpc_priv based on raw header values in rpc_tmp.
* perform conversion to either a timeout or deadline based on header
* feature flag and set rpc_priv fields accordingly
* returns error if an rpc is determined to be expired already
*/
/* Populate RPC state and reject requests whose absolute deadline has expired. */
rc = crt_hg_process_header(&rpc_tmp, rpc_priv);
if (unlikely(rc != 0)) {
RPC_WARN(rpc_priv, "RPC expired. Deadline was %d\n", rpc_priv->crp_deadline_sec);
RPC_WARN(rpc_priv, "RPC expired. Deadline was " DF_U64 " seconds\n",
rpc_priv->crp_deadline_sec);

crt_hg_unpack_cleanup(proc);
crt_rpc_priv_free(rpc_priv);
Expand All @@ -1296,8 +1294,8 @@ crt_rpc_handler_common(hg_handle_t hg_hdl)
}

rpc_priv->crp_fail_hlc = rpc_tmp.crp_fail_hlc;
rpc_pub->cr_ep.ep_rank = rpc_priv->crp_req_hdr.cch_dst_rank;
rpc_pub->cr_ep.ep_tag = rpc_priv->crp_req_hdr.cch_dst_tag;
rpc_pub->cr_ep.ep_rank = crt_req_dst_rank(&rpc_priv->crp_req_hdr);
rpc_pub->cr_ep.ep_tag = crt_req_dst_tag(&rpc_priv->crp_req_hdr);

RPC_TRACE(DB_ALL, rpc_priv, "(opc: %#x rpc_pub: %p) allocated per RPC request received.\n",
rpc_priv->crp_opc_info->coi_opc, &rpc_priv->crp_pub);
Expand All @@ -1313,8 +1311,7 @@ crt_rpc_handler_common(hg_handle_t hg_hdl)
rc = crt_hg_unpack_body(rpc_priv, proc);
if (rc == 0) {
rpc_priv->crp_input_got = 1;
rpc_pub->cr_ep.ep_grp = NULL;
/* TODO lookup by rpc_priv->crp_req_hdr.cch_grp_id */
rpc_pub->cr_ep.ep_grp = NULL;
} else {
DHL_ERROR(rpc_priv, rc, "crt_hg_unpack_body failed, opc: %#x",
rpc_pub->cr_opc);
Expand Down Expand Up @@ -1522,7 +1519,7 @@ crt_hg_req_send_cb(const struct hg_cb_info *hg_cbinfo)
hg_ret = HG_Get_output(hg_cbinfo->info.forward.handle, &rpc_pub->cr_output);
if (hg_ret == HG_SUCCESS) {
rpc_priv->crp_output_got = 1;
rc = rpc_priv->crp_reply_hdr.cch_rc;
rc = rpc_priv->crp_reply_hdr.crh_status;
} else {
rc = crt_hgret_2_der(hg_ret);
RPC_ERROR(rpc_priv, "HG_Get_output failed, hg_ret: " DF_HG_RC "\n",
Expand Down Expand Up @@ -1708,7 +1705,7 @@ crt_hg_reply_error_send(struct crt_rpc_priv *rpc_priv, int error_code)
D_ASSERT(error_code != 0);

hg_out_struct = &rpc_priv->crp_pub.cr_output;
rpc_priv->crp_reply_hdr.cch_rc = error_code;
rpc_priv->crp_reply_hdr.crh_status = error_code;
hg_ret = HG_Respond(rpc_priv->crp_hg_hdl, NULL, NULL, hg_out_struct);
if (hg_ret != HG_SUCCESS) {
CRT_METRIC_INC(crt_ctx, CM_RPC_REPLY_FAILED);
Expand Down
12 changes: 9 additions & 3 deletions src/cart/crt_hg.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,8 @@
#include <stdatomic.h>

/** the shared HG RPC ID used for all CRT opc */
#define CRT_HG_RPCID (0xDA036868)
#define CRT_HG_ONEWAY_RPCID (0xDA036869)
#define CRT_HG_RPCID (0xDA03686A)
#define CRT_HG_ONEWAY_RPCID (0xDA03686B)

/** MAX number of HG handles in pool */
#define CRT_HG_POOL_MAX_NUM (512)
Expand All @@ -45,9 +45,15 @@
#define CRT_UCX_STR "ucx"

struct crt_rpc_priv;
struct crt_common_hdr;
struct crt_req_hdr;
struct crt_reply_hdr;
struct crt_corpc_hdr;

int
crt_proc_req_hdr(crt_proc_t proc, struct crt_req_hdr *hdr);
int
crt_proc_reply_hdr(crt_proc_t proc, struct crt_reply_hdr *hdr);

/**
* Enumeration specifying providers supported by the library
*/
Expand Down
Loading
Loading