From 15d9e13c6512a2799e8dce1b67555ba50e3d7f57 Mon Sep 17 00:00:00 2001 From: Alexander Oganezov Date: Tue, 16 Sep 2025 01:51:28 -0700 Subject: [PATCH 1/3] DAOS-16952 cart: Ignore expired bulks and rpcs (#14971) * DAOS-16251 cart: Ignore expired bulks and rpcs - Switch rpc headers to transfer deadline instead of a timeout - Add checks at the start and end of bulk transfer to ensure deadline has not expired. - Add deadline expiration checks in all places where rpc_priv timeout is initialized Signed-off-by: Alexander A Oganezov --- src/cart/crt_bulk.c | 63 ++++++++++++++++++---- src/cart/crt_context.c | 13 +++-- src/cart/crt_corpc.c | 58 +++++++++------------ src/cart/crt_hg.c | 32 ++++++++---- src/cart/crt_hg.h | 16 +++--- src/cart/crt_hg_proc.c | 66 ++++++++++++++++++------ src/cart/crt_rpc.c | 20 ++++--- src/cart/crt_rpc.h | 36 +++++++++++-- src/include/cart/types.h | 1 + src/include/daos_errno.h | 1 - src/object/srv_obj_remote.c | 2 + src/tests/ftest/cart/test_proto_client.c | 3 +- 12 files changed, 217 insertions(+), 94 deletions(-) diff --git a/src/cart/crt_bulk.c b/src/cart/crt_bulk.c index 27c4a70f310..872356e48b2 100644 --- a/src/cart/crt_bulk.c +++ b/src/cart/crt_bulk.c @@ -1,5 +1,5 @@ /* - * (C) Copyright 2016-2022 Intel Corporation. + * (C) Copyright 2016-2024 Intel Corporation. * (C) Copyright 2025 Google LLC * (C) Copyright 2025 Hewlett Packard Enterprise Development LP * @@ -241,41 +241,84 @@ crt_bulk_free(crt_bulk_t crt_bulk) return rc; } +/* Helper function to check for bulk expiration */ +static inline bool +crt_bulk_desc_expired(struct crt_bulk_desc *bulk_desc) +{ + struct timespec now; + struct crt_rpc_priv *rpc_priv; + + clock_gettime(CLOCK_REALTIME_COARSE, &now); + 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); + return true; + } + + return false; +} + +int +verify_complete_cb(const struct crt_bulk_cb_info *cb_info) +{ + struct crt_bulk_cb_info actual_info; + crt_bulk_cb_t complete_cb; + + actual_info = *cb_info; + + /* If successful transfer, check for expired deadline */ + if (cb_info->bci_rc == 0) { + if (crt_bulk_desc_expired(cb_info->bci_bulk_desc)) + actual_info.bci_rc = -DER_TIMEDOUT; + } + + complete_cb = (crt_bulk_cb_t)cb_info->bci_complete_cb; + D_ASSERT(complete_cb != NULL); + + return complete_cb(&actual_info); +} + int crt_bulk_transfer(struct crt_bulk_desc *bulk_desc, crt_bulk_cb_t complete_cb, void *arg, crt_bulk_opid_t *opid) { - int rc = 0; + int rc = 0; if (!crt_bulk_desc_valid(bulk_desc)) { D_ERROR("invalid parameter of bulk_desc.\n"); D_GOTO(out, rc = -DER_INVAL); } - rc = crt_hg_bulk_transfer(bulk_desc, complete_cb, arg, opid, false); + if (crt_bulk_desc_expired(bulk_desc)) + D_GOTO(out, rc = -DER_TIMEDOUT); + + rc = crt_hg_bulk_transfer(bulk_desc, verify_complete_cb, complete_cb, arg, opid, false); if (rc != 0) DL_ERROR(rc, "crt_hg_bulk_transfer() failed"); - out: return rc; } int -crt_bulk_bind_transfer(struct crt_bulk_desc *bulk_desc, - crt_bulk_cb_t complete_cb, void *arg, +crt_bulk_bind_transfer(struct crt_bulk_desc *bulk_desc, crt_bulk_cb_t complete_cb, void *arg, crt_bulk_opid_t *opid) { - int rc = 0; + int rc = 0; if (!crt_bulk_desc_valid(bulk_desc)) { D_ERROR("invalid parameter, bulk_desc not valid.\n"); D_GOTO(out, rc = -DER_INVAL); } - rc = crt_hg_bulk_transfer(bulk_desc, complete_cb, arg, opid, true); - if (rc != 0) - D_ERROR("crt_hg_bulk_transfer() failed, rc: %d.\n", rc); + if (crt_bulk_desc_expired(bulk_desc)) + D_GOTO(out, rc = -DER_TIMEDOUT); + rc = crt_hg_bulk_transfer(bulk_desc, verify_complete_cb, complete_cb, arg, opid, true); + if (rc != 0) + DL_ERROR(rc, "crt_hg_bulk_transfer() failed"); out: return rc; } diff --git a/src/cart/crt_context.c b/src/cart/crt_context.c index 6ddd3707138..6f745a6e19d 100644 --- a/src/cart/crt_context.c +++ b/src/cart/crt_context.c @@ -1017,6 +1017,9 @@ 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 */ RPC_ADDREF(rpc_priv); /* decref in crt_req_timeout_untrack */ rc = d_binheap_insert(&crt_ctx->cc_bh_timeout, @@ -1230,16 +1233,18 @@ crt_context_timeout_check(struct crt_context *crt_ctx) print_once = true; RPC_WARN(rpc_priv, - "ctx_id %d, (status: %#x) timed out (%d seconds), " + "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_pub.cr_ep.ep_rank, rpc_priv->crp_pub.cr_ep.ep_tag); + 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), " + "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_pub.cr_ep.ep_rank, rpc_priv->crp_pub.cr_ep.ep_tag); + 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); diff --git a/src/cart/crt_corpc.c b/src/cart/crt_corpc.c index f6b995ad3f1..a8aeb27e829 100644 --- a/src/cart/crt_corpc.c +++ b/src/cart/crt_corpc.c @@ -94,8 +94,7 @@ crt_corpc_initiate(struct crt_rpc_priv *rpc_priv) { struct crt_grp_gdata *grp_gdata; struct crt_grp_priv *grp_priv; - struct crt_corpc_hdr *co_hdr; - int src_timeout; + struct crt_corpc_hdr *co_hdr; bool grp_ref_taken = false; int rc = 0; @@ -123,12 +122,6 @@ crt_corpc_initiate(struct crt_rpc_priv *rpc_priv) } } - /* Inherit a timeout from a source */ - src_timeout = rpc_priv->crp_req_hdr.cch_src_timeout; - - if (src_timeout != 0) - rpc_priv->crp_timeout_sec = src_timeout; - rc = crt_corpc_info_init(rpc_priv, grp_priv, grp_ref_taken, co_hdr->coh_filter_ranks, co_hdr->coh_grp_ver /* grp_ver */, @@ -268,7 +261,8 @@ crt_corpc_common_hdlr(struct crt_rpc_priv *rpc_priv) struct crt_bulk_desc bulk_desc; int rc = 0; - D_ASSERT(rpc_priv != NULL && (rpc_priv->crp_flags & CRT_RPC_FLAG_COLL)); + D_ASSERT(rpc_priv != NULL); + D_ASSERT(rpc_priv->crp_flags & CRT_RPC_FLAG_COLL); if (!crt_initialized()) { D_ERROR("CaRT not initialized yet.\n"); @@ -795,8 +789,7 @@ crt_corpc_req_hdlr(struct crt_rpc_priv *rpc_priv) /* Invoke pre-forward callback first if it is registered */ if (co_ops && co_ops->co_pre_forward) { - rc = co_ops->co_pre_forward(&rpc_priv->crp_pub, - co_info->co_priv); + rc = co_ops->co_pre_forward(&rpc_priv->crp_pub, co_info->co_priv); if (rc != 0) { RPC_ERROR(rpc_priv, "co_pre_forward(group %s) failed: "DF_RC"\n", @@ -835,8 +828,7 @@ crt_corpc_req_hdlr(struct crt_rpc_priv *rpc_priv) D_GOTO(forward_done, rc); } - co_info->co_child_num = (children_rank_list == NULL) ? 0 : - children_rank_list->rl_nr; + co_info->co_child_num = (children_rank_list == NULL) ? 0 : children_rank_list->rl_nr; co_info->co_child_ack_num = 0; D_DEBUG(DB_TRACE, "group %s grp_rank %d, co_info->co_child_num: %d.\n", @@ -864,35 +856,34 @@ crt_corpc_req_hdlr(struct crt_rpc_priv *rpc_priv) rpc_priv->crp_pub.cr_opc, true /* forward */, &child_rpc); if (rc != 0) { - RPC_ERROR(rpc_priv, - "crt_req_create(tgt_ep: %d) failed: " - DF_RC"\n", tgt_ep.ep_rank, DP_RC(rc)); - crt_corpc_fail_child_rpc(rpc_priv, - co_info->co_child_num - i, rc); + RPC_ERROR(rpc_priv, "crt_req_create(tgt_ep: %d) failed: " DF_RC "\n", + tgt_ep.ep_rank, DP_RC(rc)); + crt_corpc_fail_child_rpc(rpc_priv, co_info->co_child_num - i, rc); D_GOTO(forward_done, rc); } D_ASSERT(child_rpc != NULL); - D_ASSERT(child_rpc->cr_output_size == - rpc_priv->crp_pub.cr_output_size); - D_ASSERT(child_rpc->cr_output_size == 0 || - child_rpc->cr_output != NULL); + D_ASSERT(child_rpc->cr_output_size == rpc_priv->crp_pub.cr_output_size); + D_ASSERT(child_rpc->cr_output_size == 0 || child_rpc->cr_output != NULL); D_ASSERT(child_rpc->cr_input_size == 0); D_ASSERT(child_rpc->cr_input == NULL); - child_rpc_priv = container_of(child_rpc, struct crt_rpc_priv, - crp_pub); - - child_rpc_priv->crp_timeout_sec = rpc_priv->crp_timeout_sec; + child_rpc_priv = container_of(child_rpc, struct crt_rpc_priv, crp_pub); corpc_add_child_rpc(rpc_priv, child_rpc_priv); - child_rpc_priv->crp_grp_priv = co_info->co_grp_priv; RPC_ADDREF(rpc_priv); + + D_ASSERT(rpc_priv->crp_deadline_sec != 0); + + /* 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); + rc = crt_req_send(child_rpc, crt_corpc_reply_hdlr, rpc_priv); if (rc != 0) { - RPC_ERROR(rpc_priv, - "crt_req_send(tgt_ep: %d) failed: " - DF_RC"\n", tgt_ep.ep_rank, DP_RC(rc)); + RPC_ERROR(rpc_priv, "crt_req_send(tgt_ep: %d) failed: " DF_RC "\n", + tgt_ep.ep_rank, DP_RC(rc)); RPC_DECREF(rpc_priv); /* @@ -901,8 +892,8 @@ crt_corpc_req_hdlr(struct crt_rpc_priv *rpc_priv) * to fail rest child rpcs */ if (i != (co_info->co_child_num - 1)) - crt_corpc_fail_child_rpc(rpc_priv, - co_info->co_child_num - i - 1, rc); + crt_corpc_fail_child_rpc(rpc_priv, co_info->co_child_num - i - 1, + rc); D_GOTO(forward_done, rc); } } @@ -928,8 +919,7 @@ crt_corpc_req_hdlr(struct crt_rpc_priv *rpc_priv) /* invoke RPC handler on local node */ rc = crt_rpc_common_hdlr(rpc_priv); if (rc != 0) { - RPC_ERROR(rpc_priv, "crt_rpc_common_hdlr failed: "DF_RC"\n", - DP_RC(rc)); + RPC_ERROR(rpc_priv, "crt_rpc_common_hdlr failed: " DF_RC "\n", DP_RC(rc)); crt_corpc_fail_child_rpc(rpc_priv, 1, rc); D_SPIN_LOCK(&rpc_priv->crp_lock); diff --git a/src/cart/crt_hg.c b/src/cart/crt_hg.c index b8f39a7a563..ac00815e9c0 100644 --- a/src/cart/crt_hg.c +++ b/src/cart/crt_hg.c @@ -1103,7 +1103,15 @@ crt_rpc_handler_common(hg_handle_t hg_hdl) opc_info = rpc_priv->crp_opc_info; rpc_pub = &rpc_priv->crp_pub; - crt_hg_header_copy(&rpc_tmp, rpc_priv); + 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); + + crt_hg_reply_error_send(&rpc_tmp, -DER_TIMEDOUT); + crt_hg_unpack_cleanup(proc); + HG_Destroy(rpc_tmp.crp_hg_hdl); + D_GOTO(out, hg_ret = HG_SUCCESS); + } if (rpc_priv->crp_flags & CRT_RPC_FLAG_COLL) { is_coll_req = true; @@ -1735,7 +1743,8 @@ crt_hg_bulk_access(hg_bulk_t hg_bulk_hdl, d_sg_list_t *sgl) struct crt_hg_bulk_cbinfo { struct crt_bulk_desc *bci_desc; - crt_bulk_cb_t bci_cb; + crt_bulk_cb_t bci_complete_cb; + crt_bulk_cb_t bci_verify_cb; void *bci_arg; }; @@ -1772,17 +1781,19 @@ crt_hg_bulk_transfer_cb(const struct hg_cb_info *hg_cbinfo) } } - if (bulk_cbinfo->bci_cb == NULL) { + if (bulk_cbinfo->bci_verify_cb == NULL || bulk_cbinfo->bci_complete_cb == NULL) { D_DEBUG(DB_NET, "No bulk completion callback registered.\n"); D_GOTO(out, hg_ret); } + crt_bulk_cbinfo.bci_arg = bulk_cbinfo->bci_arg; crt_bulk_cbinfo.bci_rc = rc; crt_bulk_cbinfo.bci_bulk_desc = bulk_desc; + crt_bulk_cbinfo.bci_complete_cb = bulk_cbinfo->bci_complete_cb; - rc = bulk_cbinfo->bci_cb(&crt_bulk_cbinfo); + rc = bulk_cbinfo->bci_verify_cb(&crt_bulk_cbinfo); if (rc != 0) - D_ERROR("bulk_cbinfo->bci_cb failed, rc: %d.\n", rc); + D_ERROR("bulk_cbinfo->bci_verify_cb failed, rc: %d.\n", rc); out: D_FREE(bulk_cbinfo); @@ -1791,8 +1802,8 @@ crt_hg_bulk_transfer_cb(const struct hg_cb_info *hg_cbinfo) } int -crt_hg_bulk_transfer(struct crt_bulk_desc *bulk_desc, crt_bulk_cb_t complete_cb, - void *arg, crt_bulk_opid_t *opid, bool bind) +crt_hg_bulk_transfer(struct crt_bulk_desc *bulk_desc, crt_bulk_cb_t verify_cb, + crt_bulk_cb_t complete_cb, void *arg, crt_bulk_opid_t *opid, bool bind) { struct crt_context *ctx; struct crt_hg_context *hg_ctx; @@ -1837,9 +1848,10 @@ crt_hg_bulk_transfer(struct crt_bulk_desc *bulk_desc, crt_bulk_cb_t complete_cb, } crt_bulk_desc_dup(bulk_desc_dup, bulk_desc); - bulk_cbinfo->bci_desc = bulk_desc_dup; - bulk_cbinfo->bci_cb = complete_cb; - bulk_cbinfo->bci_arg = arg; + bulk_cbinfo->bci_desc = bulk_desc_dup; + bulk_cbinfo->bci_arg = arg; + bulk_cbinfo->bci_verify_cb = verify_cb; + bulk_cbinfo->bci_complete_cb = complete_cb; hg_bulk_op = (bulk_desc->bd_bulk_op == CRT_BULK_PUT) ? HG_BULK_PUSH : HG_BULK_PULL; diff --git a/src/cart/crt_hg.h b/src/cart/crt_hg.h index 58e98c7dccc..f886714d92f 100644 --- a/src/cart/crt_hg.h +++ b/src/cart/crt_hg.h @@ -8,8 +8,8 @@ /** * This file is part of CaRT. It is the header file of bridging to mercury. */ -#ifndef __CRT_MERCURY_H__ -#define __CRT_MERCURY_H__ +#ifndef __CRT_HG_H__ +#define __CRT_HG_H__ #include @@ -148,7 +148,8 @@ int crt_rpc_handler_common(hg_handle_t hg_hdl); /* crt_hg_proc.c */ int crt_hg_unpack_header(hg_handle_t hg_hdl, struct crt_rpc_priv *rpc_priv, crt_proc_t *proc); -void crt_hg_header_copy(struct crt_rpc_priv *in, struct crt_rpc_priv *out); +int + crt_hg_process_header(struct crt_rpc_priv *tmp_in, struct crt_rpc_priv *out); void crt_hg_unpack_cleanup(crt_proc_t proc); int crt_hg_unpack_body(struct crt_rpc_priv *rpc_priv, crt_proc_t proc); int crt_proc_in_common(crt_proc_t proc, crt_rpc_input_t *data); @@ -248,7 +249,10 @@ crt_hg_bulk_bind(hg_bulk_t bulk_hdl, struct crt_hg_context *hg_ctx); int crt_hg_bulk_access(hg_bulk_t bulk_hdl, d_sg_list_t *sgl); int -crt_hg_bulk_transfer(struct crt_bulk_desc *bulk_desc, crt_bulk_cb_t complete_cb, void *arg, - crt_bulk_opid_t *opid, bool bind); +crt_hg_bulk_transfer(struct crt_bulk_desc *bulk_desc, crt_bulk_cb_t verify_cb, + crt_bulk_cb_t complete_cb, void *arg, crt_bulk_opid_t *opid, bool bind); + +void +crt_hg_republish_diags(struct crt_hg_context *hg_ctx); -#endif /* __CRT_MERCURY_H__ */ +#endif /* __CRT_HG_H__ */ diff --git a/src/cart/crt_hg_proc.c b/src/cart/crt_hg_proc.c index 197b3741929..412c01aa527 100644 --- a/src/cart/crt_hg_proc.c +++ b/src/cart/crt_hg_proc.c @@ -1,5 +1,5 @@ /* - * (C) Copyright 2016-2023 Intel Corporation. + * (C) Copyright 2016-2024 Intel Corporation. * (C) Copyright 2025 Google LLC * (C) Copyright 2025 Hewlett Packard Enterprise Development LP * @@ -576,22 +576,52 @@ crt_hg_unpack_header(hg_handle_t handle, struct crt_rpc_priv *rpc_priv, return rc; } -/* Copy the RPC header from one descriptor to another */ -void -crt_hg_header_copy(struct crt_rpc_priv *in, struct crt_rpc_priv *out) +/* Process header */ +int +crt_hg_process_header(struct crt_rpc_priv *in, struct crt_rpc_priv *out) { - out->crp_hg_addr = in->crp_hg_addr; - out->crp_hg_hdl = in->crp_hg_hdl; - out->crp_pub.cr_ctx = in->crp_pub.cr_ctx; - out->crp_flags = in->crp_flags; - - out->crp_req_hdr = in->crp_req_hdr; + int src_val; + int timeout; + int rc = 0; + + out->crp_hg_addr = in->crp_hg_addr; + out->crp_hg_hdl = in->crp_hg_hdl; + out->crp_pub.cr_ctx = in->crp_pub.cr_ctx; + out->crp_flags = in->crp_flags; + out->crp_req_hdr = in->crp_req_hdr; out->crp_reply_hdr.cch_hlc = in->crp_reply_hdr.cch_hlc; + src_val = in->crp_req_hdr.cch_src_deadline_sec; + + /* Treat header value as a deadline or a timeout depending on the feat flag */ + if (in->crp_flags & CRT_RPC_FLAG_DEADLINES_USED) { + out->crp_deadline_sec = src_val; + + timeout = crt_deadline_to_timeout(src_val); + /* + * TODO: need a better way in future to handle an edge case where a client + * can be running ahead of the server, but within hlc_epsilon allowed. Also + * need to account for 1 second granularity of deadlines. + */ + if (timeout == 0) + timeout = 1; + + if (timeout > 0) + out->crp_timeout_sec = timeout; + else + D_GOTO(exit, rc = -DER_TIMEDOUT); + } else { + out->crp_deadline_sec = crt_timeout_to_deadline(src_val); + out->crp_timeout_sec = src_val; + } + if (!(out->crp_flags & CRT_RPC_FLAG_COLL)) - return; + return 0; out->crp_coreq_hdr = in->crp_coreq_hdr; + +exit: + return rc; } void @@ -669,14 +699,18 @@ crt_proc_in_common(crt_proc_t proc, crt_rpc_input_t *data) if (ENCODING(proc_op)) { hdr = &rpc_priv->crp_req_hdr; - hdr->cch_flags = rpc_priv->crp_flags; + hdr->cch_flags = rpc_priv->crp_flags; hdr->cch_dst_rank = crt_grp_priv_get_primary_rank( - rpc_priv->crp_grp_priv, - rpc_priv->crp_pub.cr_ep.ep_rank - ); + rpc_priv->crp_grp_priv, rpc_priv->crp_pub.cr_ep.ep_rank); hdr->cch_dst_tag = rpc_priv->crp_pub.cr_ep.ep_tag; - hdr->cch_src_timeout = rpc_priv->crp_timeout_sec; + if (rpc_priv->crp_flags & CRT_RPC_FLAG_DEADLINES_USED) { + hdr->cch_src_deadline_sec = rpc_priv->crp_deadline_sec; + } else { + /* Support forwarding of rpc timeout for deprecated clients */ + hdr->cch_src_deadline_sec = rpc_priv->crp_timeout_sec; + } + if (crt_is_service()) { hdr->cch_src_rank = crt_grp_priv_get_primary_rank( diff --git a/src/cart/crt_rpc.c b/src/cart/crt_rpc.c index abd3c40d1a1..480aa9e1f0e 100644 --- a/src/cart/crt_rpc.c +++ b/src/cart/crt_rpc.c @@ -739,6 +739,7 @@ crt_req_set_timeout(crt_rpc_t *req, uint32_t timeout_sec) rpc_priv = container_of(req, struct crt_rpc_priv, crp_pub); rpc_priv->crp_timeout_sec = timeout_sec; + RPC_TRACE(DB_NET, rpc_priv, "Caller set explicit timeout to %d\n", timeout_sec); out: return rc; } @@ -1490,6 +1491,11 @@ crt_req_send(crt_rpc_t *req, crt_cb_t complete_cb, void *arg) rpc_priv->crp_complete_cb = complete_cb; rpc_priv->crp_arg = arg; + /* Set deadline based on the timeout for collective calls. + * Deadline is updated in crt_req_timeout_track() + */ + rpc_priv->crp_deadline_sec = crt_timeout_to_deadline(rpc_priv->crp_timeout_sec); + if (rpc_priv->crp_coll) { rc = crt_corpc_req_hdlr(rpc_priv); if (rc != 0) @@ -1721,7 +1727,7 @@ crt_common_hdr_init(struct crt_rpc_priv *rpc_priv, crt_opcode_t opc) void crt_rpc_priv_init(struct crt_rpc_priv *rpc_priv, crt_context_t crt_ctx, bool srv_flag) { - crt_opcode_t opc = rpc_priv->crp_opc_info->coi_opc; + crt_opcode_t opc = rpc_priv->crp_opc_info->coi_opc; struct crt_context *ctx = crt_ctx; D_INIT_LIST_HEAD(&rpc_priv->crp_epi_link); @@ -1739,7 +1745,7 @@ crt_rpc_priv_init(struct crt_rpc_priv *rpc_priv, crt_context_t crt_ctx, bool srv rpc_priv->crp_hdl_reuse = NULL; rpc_priv->crp_srv = srv_flag; rpc_priv->crp_ul_retry = 0; - + rpc_priv->crp_flags |= CRT_RPC_FLAG_DEADLINES_USED; if (srv_flag) { rpc_priv->crp_src_is_primary = ctx->cc_primary; @@ -1757,11 +1763,11 @@ crt_rpc_priv_init(struct crt_rpc_priv *rpc_priv, crt_context_t crt_ctx, bool srv crt_rpc_inout_buff_init(rpc_priv); - if (srv_flag && rpc_priv->crp_req_hdr.cch_src_timeout != 0) - rpc_priv->crp_timeout_sec = rpc_priv->crp_req_hdr.cch_src_timeout; - else + /* server timeout set based on header unpack info. see crt_hg_process_header() */ + if (!srv_flag) { rpc_priv->crp_timeout_sec = (ctx->cc_timeout_sec == 0 ? crt_gdata.cg_timeout : ctx->cc_timeout_sec); + } } void @@ -2000,7 +2006,7 @@ crt_req_dst_tag_get(crt_rpc_t *rpc, uint32_t *tag) int crt_req_src_timeout_get(crt_rpc_t *rpc, uint32_t *timeout) { - struct crt_rpc_priv *rpc_priv; + struct crt_rpc_priv *rpc_priv; int rc = 0; if (rpc == NULL || timeout == NULL) { @@ -2009,7 +2015,7 @@ crt_req_src_timeout_get(crt_rpc_t *rpc, uint32_t *timeout) } rpc_priv = container_of(rpc, struct crt_rpc_priv, crp_pub); - *timeout = rpc_priv->crp_req_hdr.cch_src_timeout; + *timeout = rpc_priv->crp_timeout_sec; out: return rc; } diff --git a/src/cart/crt_rpc.h b/src/cart/crt_rpc.h index 654d602fc95..c73e336b69b 100644 --- a/src/cart/crt_rpc.h +++ b/src/cart/crt_rpc.h @@ -34,9 +34,11 @@ void crt_hdlr_memb_sample(crt_rpc_t *rpc_req); */ enum crt_rpc_flags_internal { /* flag of collective RPC (bcast) */ - CRT_RPC_FLAG_COLL = (1U << 16), + CRT_RPC_FLAG_COLL = (1U << 16), /* flag of targeting primary group */ - CRT_RPC_FLAG_PRIMARY_GRP = (1U << 17), + CRT_RPC_FLAG_PRIMARY_GRP = (1U << 17), + /* flag of using deadlines instead of timeouts in RPC headers */ + CRT_RPC_FLAG_DEADLINES_USED = (1U << 18), }; struct crt_corpc_hdr { @@ -76,11 +78,30 @@ struct crt_common_hdr { /* used in crp_reply_hdr to propagate rpc failure back to sender */ /* TODO: workaround for DAOS-13973 */ union { - uint32_t cch_src_timeout; + uint32_t cch_src_deadline_sec; uint32_t cch_rc; }; }; +static inline int32_t +crt_deadline_to_timeout(uint32_t deadline_sec) +{ + struct timespec now; + + clock_gettime(CLOCK_REALTIME_COARSE, &now); + return deadline_sec - now.tv_sec - 1; +} + +static inline uint32_t +crt_timeout_to_deadline(int timeout_sec) +{ + struct timespec now; + + clock_gettime(CLOCK_REALTIME_COARSE, &now); + + /* Deadline is the next second after timeout */ + return now.tv_sec + timeout_sec + 1; +} typedef enum { RPC_STATE_INITED = 0x36, @@ -144,6 +165,8 @@ struct crt_rpc_priv { struct d_binheap_node crp_timeout_bp_node; /* the timeout in seconds set by user */ uint32_t crp_timeout_sec; + /* the deadline corresponding to crp_timeout_sec */ + uint32_t crp_deadline_sec; /* time stamp to be timeout, the key of timeout binheap */ uint64_t crp_timeout_ts; crt_cb_t crp_complete_cb; @@ -191,7 +214,9 @@ struct crt_rpc_priv { /* RPC originated from a primary provider */ crp_src_is_primary : 1, /* release input buffer early */ - crp_release_input_early : 1; + crp_release_input_early : 1, + /* rpc expired */ + crp_expired : 1; struct crt_opc_info *crp_opc_info; /* corpc info, only valid when (crp_coll == 1) */ @@ -668,7 +693,8 @@ crt_rpc_cb_customized(struct crt_context *crt_ctx, int crt_rpc_priv_alloc(crt_opcode_t opc, struct crt_rpc_priv **priv_allocated, bool forward); void crt_rpc_priv_free(struct crt_rpc_priv *rpc_priv); -void crt_rpc_priv_init(struct crt_rpc_priv *rpc_priv, crt_context_t crt_ctx, bool srv_flag); +void + crt_rpc_priv_init(struct crt_rpc_priv *rpc_priv, crt_context_t crt_ctx, bool srv_flag); void crt_rpc_priv_fini(struct crt_rpc_priv *rpc_priv); int crt_req_create_internal(crt_context_t crt_ctx, crt_endpoint_t *tgt_ep, crt_opcode_t opc, bool forward, crt_rpc_t **req); diff --git a/src/include/cart/types.h b/src/include/cart/types.h index b9bdcfbd612..0cb9493902e 100644 --- a/src/include/cart/types.h +++ b/src/include/cart/types.h @@ -350,6 +350,7 @@ struct crt_bulk_cb_info { struct crt_bulk_desc *bci_bulk_desc; /**< bulk descriptor */ void *bci_arg; /**< User passed in arg */ int bci_rc; /**< return code */ + void *bci_complete_cb; /**< completion callback */ }; /** diff --git a/src/include/daos_errno.h b/src/include/daos_errno.h index 8bb0ffdbde1..4d7b3a5c0e6 100644 --- a/src/include/daos_errno.h +++ b/src/include/daos_errno.h @@ -121,7 +121,6 @@ extern "C" { ACTION(DER_HG_FATAL, Fatal transport layer mercury error) \ /** Quota limit reached on the requested resource */ \ ACTION(DER_QUOTA_LIMIT, Quota limit reached) - /** TODO: add more error numbers */ /** Preprocessor macro defining DAOS errno values and internal definition of d_errstr */ #define D_FOREACH_DAOS_ERR(ACTION) \ diff --git a/src/object/srv_obj_remote.c b/src/object/srv_obj_remote.c index f64d851e5b4..0ec92d22ec8 100644 --- a/src/object/srv_obj_remote.c +++ b/src/object/srv_obj_remote.c @@ -1,5 +1,6 @@ /** * (C) Copyright 2019-2024 Intel Corporation. + * (C) Copyright 2025 Hewlett Packard Enterprise Development LP * * SPDX-License-Identifier: BSD-2-Clause-Patent */ @@ -73,6 +74,7 @@ obj_inherit_timeout(crt_rpc_t *parent, crt_rpc_t *child) rc = crt_req_src_timeout_get(parent, &timeout); D_ASSERT(rc == 0); + rc = crt_req_set_timeout(child, timeout); D_ASSERT(rc == 0); } diff --git a/src/tests/ftest/cart/test_proto_client.c b/src/tests/ftest/cart/test_proto_client.c index ffcec7461e0..ae6b5ad6bf1 100644 --- a/src/tests/ftest/cart/test_proto_client.c +++ b/src/tests/ftest/cart/test_proto_client.c @@ -1,5 +1,6 @@ /* * (C) Copyright 2018-2024 Intel Corporation. + * (C) Copyright 2025 Hewlett Packard Enterprise Development LP * * SPDX-License-Identifier: BSD-2-Clause-Patent */ @@ -120,7 +121,7 @@ test_run() server_ep.ep_rank = 0; DBG_PRINT("proto query\n"); - timeout = 1; + timeout = 3; do { rc = crt_proto_query(&server_ep, OPC_MY_PROTO, my_ver_array, 7, timeout++, query_cb, &s_high_ver); From 76a9cb1ab8d4830f85311cc6a4296a0af92473db Mon Sep 17 00:00:00 2001 From: Alexander A Oganezov Date: Wed, 17 Sep 2025 20:43:01 +0000 Subject: [PATCH 2/3] - deadlines fix Signed-off-by: Alexander A Oganezov --- src/cart/crt_rpc.c | 2 ++ src/cart/crt_rpc.h | 4 +++- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/src/cart/crt_rpc.c b/src/cart/crt_rpc.c index 480aa9e1f0e..080ee5e2b2f 100644 --- a/src/cart/crt_rpc.c +++ b/src/cart/crt_rpc.c @@ -738,6 +738,7 @@ crt_req_set_timeout(crt_rpc_t *req, uint32_t timeout_sec) rpc_priv = container_of(req, struct crt_rpc_priv, crp_pub); rpc_priv->crp_timeout_sec = timeout_sec; + rpc_priv->crp_deadline_sec = crt_timeout_to_deadline(timeout_sec); RPC_TRACE(DB_NET, rpc_priv, "Caller set explicit timeout to %d\n", timeout_sec); out: @@ -1767,6 +1768,7 @@ crt_rpc_priv_init(struct crt_rpc_priv *rpc_priv, crt_context_t crt_ctx, bool srv if (!srv_flag) { rpc_priv->crp_timeout_sec = (ctx->cc_timeout_sec == 0 ? crt_gdata.cg_timeout : ctx->cc_timeout_sec); + rpc_priv->crp_deadline_sec = crt_timeout_to_deadline(rpc_priv->crp_timeout_sec); } } diff --git a/src/cart/crt_rpc.h b/src/cart/crt_rpc.h index c73e336b69b..b16d286df32 100644 --- a/src/cart/crt_rpc.h +++ b/src/cart/crt_rpc.h @@ -717,8 +717,10 @@ crt_set_timeout(struct crt_rpc_priv *rpc_priv) { D_ASSERT(rpc_priv != NULL); - if (rpc_priv->crp_timeout_sec == 0) + if (rpc_priv->crp_timeout_sec == 0) { rpc_priv->crp_timeout_sec = crt_gdata.cg_timeout; + rpc_priv->crp_deadline_sec = crt_timeout_to_deadline(rpc_priv->crp_timeout_sec); + } rpc_priv->crp_timeout_ts = d_timeus_secdiff(rpc_priv->crp_timeout_sec); } From 9d730f8a98d9fdab92053de5be7cb4460cdade59 Mon Sep 17 00:00:00 2001 From: Alexander Oganezov Date: Mon, 22 Sep 2025 10:23:35 -0700 Subject: [PATCH 3/3] DAOS-17990 cart: Fix deadlines (#16878) Deadlines were not set for some internal RPCs due to conversion from timeout to deadline happening in an rpc tracking code. Some internal RPCs bypass rpc tracking Signed-off-by: Alexander A Oganezov --- src/cart/crt_hg.c | 5 +++++ src/cart/crt_hg_proc.c | 10 +++++++++- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/src/cart/crt_hg.c b/src/cart/crt_hg.c index ac00815e9c0..134d8498446 100644 --- a/src/cart/crt_hg.c +++ b/src/cart/crt_hg.c @@ -1103,6 +1103,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 + */ 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); diff --git a/src/cart/crt_hg_proc.c b/src/cart/crt_hg_proc.c index 412c01aa527..0e8f5edd7aa 100644 --- a/src/cart/crt_hg_proc.c +++ b/src/cart/crt_hg_proc.c @@ -576,7 +576,15 @@ crt_hg_unpack_header(hg_handle_t handle, struct crt_rpc_priv *rpc_priv, return rc; } -/* Process header */ +/* Process header. + * Populate 'out' fields of rpc_priv struct based on incoming rpc header 'in' + * + * Treat incoming 'cch_src_deadline_sec' header value as either a timeout or a + * deadline based on a feature flag set in the RPC header. + * + * + * Returns 0 on success or -DER_TIMEDOUT if RPC is determined to be expired already. + */ int crt_hg_process_header(struct crt_rpc_priv *in, struct crt_rpc_priv *out) {