[PATCH for-next 0/5] Fix incorrect atomic retry behavior

[PATCH for-next 0/5] Fix incorrect atomic retry behavior

[Bob Pearson]

Restructure the design of the atomic retries in rxe_resp.c modeled
on the design of RDMA read reply. This fixes failures which
occur when an atomic ack packet is lost as observed in the
pyverbs test suite with a patch to randomly drop packets.

Link: https://lore.kernel.org/linux-rdma/84aaf934-9cac-30ae-0fa5-1e40819a519e@gmail.com/
Signed-off-by: Bob Pearson <rpearsonhpe@gmail.com>
---

Bob Pearson (5):
  RDMA/rxe: Move code to rxe_prepare_atomic_res()
  RDMA/rxe: Add a responder state for atomic reply
  RDMA/rxe: Move atomic responder res to atomic_reply
  RDMA/rxe: Move atomic original value to res
  RDMA/rxe: Merge normal and retry atomic flows

 drivers/infiniband/sw/rxe/rxe_qp.c    |   2 -
 drivers/infiniband/sw/rxe/rxe_resp.c  | 132 ++++++++++++++++----------
 drivers/infiniband/sw/rxe/rxe_verbs.h |   3 +-
 3 files changed, 81 insertions(+), 56 deletions(-)


base-commit: f2906aa863381afb0015a9eb7fefad885d4e5a56
-- 
2.34.1

[PATCH 1/5] RDMA/rxe: Move code to rxe_prepare_atomic_res()

[Bob Pearson]

Separate the code that prepares the atomic responder resource
into a subroutine. This is preparation for merging the normal
and retry atomic responder flows.

Signed-off-by: Bob Pearson <rpearsonhpe@gmail.com>
---
 drivers/infiniband/sw/rxe/rxe_resp.c | 37 ++++++++++++++++++----------
 1 file changed, 24 insertions(+), 13 deletions(-)

diff --git a/drivers/infiniband/sw/rxe/rxe_resp.c b/drivers/infiniband/sw/rxe/rxe_resp.c
index f4f6ee5d81fe..69723bc1a071 100644
--- a/drivers/infiniband/sw/rxe/rxe_resp.c
+++ b/drivers/infiniband/sw/rxe/rxe_resp.c
@@ -1019,10 +1019,27 @@ static int send_ack(struct rxe_qp *qp, struct rxe_pkt_info *pkt,
 	return err;
 }
 
+static struct resp_res *rxe_prepare_atomic_res(struct rxe_qp *qp,
+					       struct rxe_pkt_info *pkt)
+{
+	struct resp_res *res;
+
+	res = &qp->resp.resources[qp->resp.res_head];
+	rxe_advance_resp_resource(qp);
+	free_rd_atomic_resource(qp, res);
+
+	res->type = RXE_ATOMIC_MASK;
+	res->first_psn = pkt->psn;
+	res->last_psn = pkt->psn;
+	res->cur_psn = pkt->psn;
+
+	return res;
+}
+
 static int send_atomic_ack(struct rxe_qp *qp, struct rxe_pkt_info *pkt,
 			   u8 syndrome)
 {
-	int rc = 0;
+	int err = 0;
 	struct rxe_pkt_info ack_pkt;
 	struct sk_buff *skb;
 	struct resp_res *res;
@@ -1031,28 +1048,22 @@ static int send_atomic_ack(struct rxe_qp *qp, struct rxe_pkt_info *pkt,
 				 IB_OPCODE_RC_ATOMIC_ACKNOWLEDGE, 0, pkt->psn,
 				 syndrome);
 	if (!skb) {
-		rc = -ENOMEM;
+		err = -ENOMEM;
 		goto out;
 	}
 
-	res = &qp->resp.resources[qp->resp.res_head];
-	free_rd_atomic_resource(qp, res);
-	rxe_advance_resp_resource(qp);
-
 	skb_get(skb);
-	res->type = RXE_ATOMIC_MASK;
+
+	res = rxe_prepare_atomic_res(qp, pkt);
 	res->atomic.skb = skb;
-	res->first_psn = ack_pkt.psn;
-	res->last_psn  = ack_pkt.psn;
-	res->cur_psn   = ack_pkt.psn;
 
-	rc = rxe_xmit_packet(qp, &ack_pkt, skb);
-	if (rc) {
+	err = rxe_xmit_packet(qp, &ack_pkt, skb);
+	if (err) {
 		pr_err_ratelimited("Failed sending ack\n");
 		rxe_put(qp);
 	}
 out:
-	return rc;
+	return err;
 }
 
 static enum resp_states acknowledge(struct rxe_qp *qp,
-- 
2.34.1

[PATCH 2/5] RDMA/rxe: Add a responder state for atomic reply

[Bob Pearson]

Add a responder state for atomic reply similar to read reply and
rename process_atomic() rxe_atomic_reply(). In preparation for
merging the normal and retry atomic responder flows.

Signed-off-by: Bob Pearson <rpearsonhpe@gmail.com>
---
 drivers/infiniband/sw/rxe/rxe_resp.c | 24 ++++++++++++++++++------
 1 file changed, 18 insertions(+), 6 deletions(-)

diff --git a/drivers/infiniband/sw/rxe/rxe_resp.c b/drivers/infiniband/sw/rxe/rxe_resp.c
index 69723bc1a071..4babd6fbfefe 100644
--- a/drivers/infiniband/sw/rxe/rxe_resp.c
+++ b/drivers/infiniband/sw/rxe/rxe_resp.c
@@ -21,6 +21,7 @@ enum resp_states {
 	RESPST_CHK_RKEY,
 	RESPST_EXECUTE,
 	RESPST_READ_REPLY,
+	RESPST_ATOMIC_REPLY,
 	RESPST_COMPLETE,
 	RESPST_ACKNOWLEDGE,
 	RESPST_CLEANUP,
@@ -55,6 +56,7 @@ static char *resp_state_name[] = {
 	[RESPST_CHK_RKEY]			= "CHK_RKEY",
 	[RESPST_EXECUTE]			= "EXECUTE",
 	[RESPST_READ_REPLY]			= "READ_REPLY",
+	[RESPST_ATOMIC_REPLY]			= "ATOMIC_REPLY",
 	[RESPST_COMPLETE]			= "COMPLETE",
 	[RESPST_ACKNOWLEDGE]			= "ACKNOWLEDGE",
 	[RESPST_CLEANUP]			= "CLEANUP",
@@ -552,8 +554,8 @@ static enum resp_states write_data_in(struct rxe_qp *qp,
 /* Guarantee atomicity of atomic operations at the machine level. */
 static DEFINE_SPINLOCK(atomic_ops_lock);
 
-static enum resp_states process_atomic(struct rxe_qp *qp,
-				       struct rxe_pkt_info *pkt)
+static enum resp_states rxe_atomic_reply(struct rxe_qp *qp,
+					 struct rxe_pkt_info *pkt)
 {
 	u64 *vaddr;
 	enum resp_states ret;
@@ -585,7 +587,16 @@ static enum resp_states process_atomic(struct rxe_qp *qp,
 
 	spin_unlock_bh(&atomic_ops_lock);
 
-	ret = RESPST_NONE;
+	qp->resp.msn++;
+
+	/* next expected psn, read handles this separately */
+	qp->resp.psn = (pkt->psn + 1) & BTH_PSN_MASK;
+	qp->resp.ack_psn = qp->resp.psn;
+
+	qp->resp.opcode = pkt->opcode;
+	qp->resp.status = IB_WC_SUCCESS;
+
+	ret = RESPST_ACKNOWLEDGE;
 out:
 	return ret;
 }
@@ -858,9 +869,7 @@ static enum resp_states execute(struct rxe_qp *qp, struct rxe_pkt_info *pkt)
 		qp->resp.msn++;
 		return RESPST_READ_REPLY;
 	} else if (pkt->mask & RXE_ATOMIC_MASK) {
-		err = process_atomic(qp, pkt);
-		if (err)
-			return err;
+		return RESPST_ATOMIC_REPLY;
 	} else {
 		/* Unreachable */
 		WARN_ON_ONCE(1);
@@ -1327,6 +1336,9 @@ int rxe_responder(void *arg)
 		case RESPST_READ_REPLY:
 			state = read_reply(qp, pkt);
 			break;
+		case RESPST_ATOMIC_REPLY:
+			state = rxe_atomic_reply(qp, pkt);
+			break;
 		case RESPST_ACKNOWLEDGE:
 			state = acknowledge(qp, pkt);
 			break;
-- 
2.34.1

[PATCH 3/5] RDMA/rxe: Move atomic responder res to atomic_reply

[Bob Pearson]

Move the allocation of the atomic responder resource up into
rxe_atomic_reply() from send_atomic_ack(). In preparation for
merging the normal and retry atomic responder flows.

Signed-off-by: Bob Pearson <rpearsonhpe@gmail.com>
---
 drivers/infiniband/sw/rxe/rxe_resp.c | 49 +++++++++++++++++-----------
 1 file changed, 30 insertions(+), 19 deletions(-)

diff --git a/drivers/infiniband/sw/rxe/rxe_resp.c b/drivers/infiniband/sw/rxe/rxe_resp.c
index 4babd6fbfefe..4908b9fc0204 100644
--- a/drivers/infiniband/sw/rxe/rxe_resp.c
+++ b/drivers/infiniband/sw/rxe/rxe_resp.c
@@ -554,12 +554,36 @@ static enum resp_states write_data_in(struct rxe_qp *qp,
 /* Guarantee atomicity of atomic operations at the machine level. */
 static DEFINE_SPINLOCK(atomic_ops_lock);
 
+static struct resp_res *rxe_prepare_atomic_res(struct rxe_qp *qp,
+					       struct rxe_pkt_info *pkt)
+{
+	struct resp_res *res;
+
+	res = &qp->resp.resources[qp->resp.res_head];
+	rxe_advance_resp_resource(qp);
+	free_rd_atomic_resource(qp, res);
+
+	res->type = RXE_ATOMIC_MASK;
+	res->first_psn = pkt->psn;
+	res->last_psn = pkt->psn;
+	res->cur_psn = pkt->psn;
+	res->replay = 0;
+
+	return res;
+}
+
 static enum resp_states rxe_atomic_reply(struct rxe_qp *qp,
 					 struct rxe_pkt_info *pkt)
 {
 	u64 *vaddr;
 	enum resp_states ret;
 	struct rxe_mr *mr = qp->resp.mr;
+	struct resp_res *res = qp->resp.res;
+
+	if (!res) {
+		res = rxe_prepare_atomic_res(qp, pkt);
+		qp->resp.res = res;
+	}
 
 	if (mr->state != RXE_MR_STATE_VALID) {
 		ret = RESPST_ERR_RKEY_VIOLATION;
@@ -1028,30 +1052,13 @@ static int send_ack(struct rxe_qp *qp, struct rxe_pkt_info *pkt,
 	return err;
 }
 
-static struct resp_res *rxe_prepare_atomic_res(struct rxe_qp *qp,
-					       struct rxe_pkt_info *pkt)
-{
-	struct resp_res *res;
-
-	res = &qp->resp.resources[qp->resp.res_head];
-	rxe_advance_resp_resource(qp);
-	free_rd_atomic_resource(qp, res);
-
-	res->type = RXE_ATOMIC_MASK;
-	res->first_psn = pkt->psn;
-	res->last_psn = pkt->psn;
-	res->cur_psn = pkt->psn;
-
-	return res;
-}
-
 static int send_atomic_ack(struct rxe_qp *qp, struct rxe_pkt_info *pkt,
 			   u8 syndrome)
 {
 	int err = 0;
 	struct rxe_pkt_info ack_pkt;
 	struct sk_buff *skb;
-	struct resp_res *res;
+	struct resp_res *res = qp->resp.res;
 
 	skb = prepare_ack_packet(qp, pkt, &ack_pkt,
 				 IB_OPCODE_RC_ATOMIC_ACKNOWLEDGE, 0, pkt->psn,
@@ -1063,7 +1070,6 @@ static int send_atomic_ack(struct rxe_qp *qp, struct rxe_pkt_info *pkt,
 
 	skb_get(skb);
 
-	res = rxe_prepare_atomic_res(qp, pkt);
 	res->atomic.skb = skb;
 
 	err = rxe_xmit_packet(qp, &ack_pkt, skb);
@@ -1071,6 +1077,11 @@ static int send_atomic_ack(struct rxe_qp *qp, struct rxe_pkt_info *pkt,
 		pr_err_ratelimited("Failed sending ack\n");
 		rxe_put(qp);
 	}
+
+	/* have to clear this since it is used to trigger
+	 * long read replies
+	 */
+	qp->resp.res = NULL;
 out:
 	return err;
 }
-- 
2.34.1

[PATCH 4/5] RDMA/rxe: Move atomic original value to res

[Bob Pearson]

Move the saved original value to the atomic responder resource.
This replaces saving it in the qp. In preparation for
merging the normal and retry atomic responder flows.

Signed-off-by: Bob Pearson <rpearsonhpe@gmail.com>
---
 drivers/infiniband/sw/rxe/rxe_resp.c  | 13 +++++++------
 drivers/infiniband/sw/rxe/rxe_verbs.h |  2 +-
 2 files changed, 8 insertions(+), 7 deletions(-)

diff --git a/drivers/infiniband/sw/rxe/rxe_resp.c b/drivers/infiniband/sw/rxe/rxe_resp.c
index 4908b9fc0204..320ab7c717cb 100644
--- a/drivers/infiniband/sw/rxe/rxe_resp.c
+++ b/drivers/infiniband/sw/rxe/rxe_resp.c
@@ -579,6 +579,7 @@ static enum resp_states rxe_atomic_reply(struct rxe_qp *qp,
 	enum resp_states ret;
 	struct rxe_mr *mr = qp->resp.mr;
 	struct resp_res *res = qp->resp.res;
+	u64 value;
 
 	if (!res) {
 		res = rxe_prepare_atomic_res(qp, pkt);
@@ -599,16 +600,16 @@ static enum resp_states rxe_atomic_reply(struct rxe_qp *qp,
 	}
 
 	spin_lock_bh(&atomic_ops_lock);
-
-	qp->resp.atomic_orig = *vaddr;
+	res->atomic.orig_val = value = *vaddr;
 
 	if (pkt->opcode == IB_OPCODE_RC_COMPARE_SWAP) {
-		if (*vaddr == atmeth_comp(pkt))
-			*vaddr = atmeth_swap_add(pkt);
+		if (value == atmeth_comp(pkt))
+			value = atmeth_swap_add(pkt);
 	} else {
-		*vaddr += atmeth_swap_add(pkt);
+		value += atmeth_swap_add(pkt);
 	}
 
+	*vaddr = value;
 	spin_unlock_bh(&atomic_ops_lock);
 
 	qp->resp.msn++;
@@ -664,7 +665,7 @@ static struct sk_buff *prepare_ack_packet(struct rxe_qp *qp,
 	}
 
 	if (ack->mask & RXE_ATMACK_MASK)
-		atmack_set_orig(ack, qp->resp.atomic_orig);
+		atmack_set_orig(ack, qp->resp.res->atomic.orig_val);
 
 	err = rxe_prepare(&qp->pri_av, ack, skb);
 	if (err) {
diff --git a/drivers/infiniband/sw/rxe/rxe_verbs.h b/drivers/infiniband/sw/rxe/rxe_verbs.h
index ac464e68c923..5ee0f2599896 100644
--- a/drivers/infiniband/sw/rxe/rxe_verbs.h
+++ b/drivers/infiniband/sw/rxe/rxe_verbs.h
@@ -156,6 +156,7 @@ struct resp_res {
 	union {
 		struct {
 			struct sk_buff	*skb;
+			u64		orig_val;
 		} atomic;
 		struct {
 			u64		va_org;
@@ -189,7 +190,6 @@ struct rxe_resp_info {
 	u32			resid;
 	u32			rkey;
 	u32			length;
-	u64			atomic_orig;
 
 	/* SRQ only */
 	struct {
-- 
2.34.1

[PATCH 5/5] RDMA/rxe: Merge normal and retry atomic flows

[Bob Pearson]

Make the execution of the atomic operation in rxe_atomic_reply()
conditional on res->replay and make duplicate_request() call into
rxe_atomic_reply() to merge the two flows. This is modeled on
the behavior of read reply. Delete the skb from the atomic
responder resource since it is no longer used. Adjust the reference
counting of the qp in send_atomic_ack() for this flow.

Fixes: 8700e3e7c485 ("The software RoCE driver")
Signed-off-by: Bob Pearson <rpearsonhpe@gmail.com>
---
 drivers/infiniband/sw/rxe/rxe_qp.c    |  2 -
 drivers/infiniband/sw/rxe/rxe_resp.c  | 79 ++++++++++++---------------
 drivers/infiniband/sw/rxe/rxe_verbs.h |  1 -
 3 files changed, 36 insertions(+), 46 deletions(-)

diff --git a/drivers/infiniband/sw/rxe/rxe_qp.c b/drivers/infiniband/sw/rxe/rxe_qp.c
index 22e9b85344c3..8355a5b1cb60 100644
--- a/drivers/infiniband/sw/rxe/rxe_qp.c
+++ b/drivers/infiniband/sw/rxe/rxe_qp.c
@@ -129,8 +129,6 @@ static void free_rd_atomic_resources(struct rxe_qp *qp)
 
 void free_rd_atomic_resource(struct rxe_qp *qp, struct resp_res *res)
 {
-	if (res->type == RXE_ATOMIC_MASK)
-		kfree_skb(res->atomic.skb);
 	res->type = 0;
 }
 
diff --git a/drivers/infiniband/sw/rxe/rxe_resp.c b/drivers/infiniband/sw/rxe/rxe_resp.c
index 320ab7c717cb..6888bd9d0bfc 100644
--- a/drivers/infiniband/sw/rxe/rxe_resp.c
+++ b/drivers/infiniband/sw/rxe/rxe_resp.c
@@ -586,40 +586,43 @@ static enum resp_states rxe_atomic_reply(struct rxe_qp *qp,
 		qp->resp.res = res;
 	}
 
-	if (mr->state != RXE_MR_STATE_VALID) {
-		ret = RESPST_ERR_RKEY_VIOLATION;
-		goto out;
-	}
+	if (!res->replay) {
+		if (mr->state != RXE_MR_STATE_VALID) {
+			ret = RESPST_ERR_RKEY_VIOLATION;
+			goto out;
+		}
 
-	vaddr = iova_to_vaddr(mr, qp->resp.va + qp->resp.offset, sizeof(u64));
+		vaddr = iova_to_vaddr(mr, qp->resp.va + qp->resp.offset,
+					sizeof(u64));
 
-	/* check vaddr is 8 bytes aligned. */
-	if (!vaddr || (uintptr_t)vaddr & 7) {
-		ret = RESPST_ERR_MISALIGNED_ATOMIC;
-		goto out;
-	}
+		/* check vaddr is 8 bytes aligned. */
+		if (!vaddr || (uintptr_t)vaddr & 7) {
+			ret = RESPST_ERR_MISALIGNED_ATOMIC;
+			goto out;
+		}
 
-	spin_lock_bh(&atomic_ops_lock);
-	res->atomic.orig_val = value = *vaddr;
+		spin_lock_bh(&atomic_ops_lock);
+		res->atomic.orig_val = value = *vaddr;
 
-	if (pkt->opcode == IB_OPCODE_RC_COMPARE_SWAP) {
-		if (value == atmeth_comp(pkt))
-			value = atmeth_swap_add(pkt);
-	} else {
-		value += atmeth_swap_add(pkt);
-	}
+		if (pkt->opcode == IB_OPCODE_RC_COMPARE_SWAP) {
+			if (value == atmeth_comp(pkt))
+				value = atmeth_swap_add(pkt);
+		} else {
+			value += atmeth_swap_add(pkt);
+		}
 
-	*vaddr = value;
-	spin_unlock_bh(&atomic_ops_lock);
+		*vaddr = value;
+		spin_unlock_bh(&atomic_ops_lock);
 
-	qp->resp.msn++;
+		qp->resp.msn++;
 
-	/* next expected psn, read handles this separately */
-	qp->resp.psn = (pkt->psn + 1) & BTH_PSN_MASK;
-	qp->resp.ack_psn = qp->resp.psn;
+		/* next expected psn, read handles this separately */
+		qp->resp.psn = (pkt->psn + 1) & BTH_PSN_MASK;
+		qp->resp.ack_psn = qp->resp.psn;
 
-	qp->resp.opcode = pkt->opcode;
-	qp->resp.status = IB_WC_SUCCESS;
+		qp->resp.opcode = pkt->opcode;
+		qp->resp.status = IB_WC_SUCCESS;
+	}
 
 	ret = RESPST_ACKNOWLEDGE;
 out:
@@ -1059,7 +1062,6 @@ static int send_atomic_ack(struct rxe_qp *qp, struct rxe_pkt_info *pkt,
 	int err = 0;
 	struct rxe_pkt_info ack_pkt;
 	struct sk_buff *skb;
-	struct resp_res *res = qp->resp.res;
 
 	skb = prepare_ack_packet(qp, pkt, &ack_pkt,
 				 IB_OPCODE_RC_ATOMIC_ACKNOWLEDGE, 0, pkt->psn,
@@ -1069,15 +1071,9 @@ static int send_atomic_ack(struct rxe_qp *qp, struct rxe_pkt_info *pkt,
 		goto out;
 	}
 
-	skb_get(skb);
-
-	res->atomic.skb = skb;
-
 	err = rxe_xmit_packet(qp, &ack_pkt, skb);
-	if (err) {
-		pr_err_ratelimited("Failed sending ack\n");
-		rxe_put(qp);
-	}
+	if (err)
+		pr_err_ratelimited("Failed sending atomic ack\n");
 
 	/* have to clear this since it is used to trigger
 	 * long read replies
@@ -1205,14 +1201,11 @@ static enum resp_states duplicate_request(struct rxe_qp *qp,
 		/* Find the operation in our list of responder resources. */
 		res = find_resource(qp, pkt->psn);
 		if (res) {
-			skb_get(res->atomic.skb);
-			/* Resend the result. */
-			rc = rxe_xmit_packet(qp, pkt, res->atomic.skb);
-			if (rc) {
-				pr_err("Failed resending result. This flow is not handled - skb ignored\n");
-				rc = RESPST_CLEANUP;
-				goto out;
-			}
+			res->replay = 1;
+			res->cur_psn = pkt->psn;
+			qp->resp.res = res;
+			rc = RESPST_ATOMIC_REPLY;
+			goto out;
 		}
 
 		/* Resource not found. Class D error. Drop the request. */
diff --git a/drivers/infiniband/sw/rxe/rxe_verbs.h b/drivers/infiniband/sw/rxe/rxe_verbs.h
index 5ee0f2599896..0c01c7f58d43 100644
--- a/drivers/infiniband/sw/rxe/rxe_verbs.h
+++ b/drivers/infiniband/sw/rxe/rxe_verbs.h
@@ -155,7 +155,6 @@ struct resp_res {
 
 	union {
 		struct {
-			struct sk_buff	*skb;
 			u64		orig_val;
 		} atomic;
 		struct {
-- 
2.34.1

Re: [PATCH for-next 0/5] Fix incorrect atomic retry behavior

[Jason Gunthorpe]

On Mon, Jun 06, 2022 at 09:38:32AM -0500, Bob Pearson wrote:
> Restructure the design of the atomic retries in rxe_resp.c modeled
> on the design of RDMA read reply. This fixes failures which
> occur when an atomic ack packet is lost as observed in the
> pyverbs test suite with a patch to randomly drop packets.

Applied to for-next

Thanks,
Jason

Re: [PATCH for-next 0/5] Fix incorrect atomic retry behavior

[Bob Pearson]

On 6/30/22 13:52, Jason Gunthorpe wrote:
> On Mon, Jun 06, 2022 at 09:38:32AM -0500, Bob Pearson wrote:
>> Restructure the design of the atomic retries in rxe_resp.c modeled
>> on the design of RDMA read reply. This fixes failures which
>> occur when an atomic ack packet is lost as observed in the
>> pyverbs test suite with a patch to randomly drop packets.
> 
> Applied to for-next
> 
> Thanks,
> Jason
Thanks.
I think this will commute with the other fixes I just sent. If it doesn't I can rebase them.
Bob