DAOS-16952 cart: Ignore expired bulks and rpcs

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;
 }

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);

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);

src/cart/crt_hg.c

@@ -1103,7 +1103,20 @@ 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);
+	/* 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);
+
+		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 +1748,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 +1786,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 +1807,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 +1853,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;

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 <gurt/list.h>
 
@@ -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__ */