rds: ib: Simplify ib_ring and use atomic ops

The original struct rds_ib_work_ring was implemented using double set
of counters / pointers for the consumer and producer of the ring. It
is complicated and error prone, due to the compilers ability to
re-order execution of the statements and moderns CPUs ability to
re-order execution. Hence, use only a single set of free-running
counters and use only atomic ops, which do not rely on explicit memory
barriers.

Without this commit, we will encounter the following BUG_ON() running
uek7/ga (v5.15.0-0.30.9) on ARM:

kernel BUG at net/rds/ib_send.c:886!
Internal error: Oops - BUG: 0 [#1] SMP
[]
rds_ib_xmit+0xa1c/0xb70 [rds_rdma]
rds_send_xmit+0x4a0/0xd50 [rds]
rds_sendmsg+0xe7c/0xfa0 [rds]
sock_sendmsg+0x74/0x80
____sys_sendmsg+0x270/0x29c
___sys_sendmsg+0xb0/0x118
__sys_sendmsg+0x90/0x104
__arm64_sys_sendmsg+0x30/0x50
invoke_syscall+0x50/0x15c

The bug may also manifest itself as:

kernel: RDS/IB: ib_post_send to ::ffff:192.168.0.4 returned -22

or as:

kernel: RDS/IB: rds_ib_send_unmap_op: unexpected opcode 0xdead in WR!

Orabug: 34137460

Reviewed-by: Konrad Rzeszutek Wilk <konrad.wilk@oracle.com>
Fixes: f528efe ("RDS/IB: Ring-handling code.")
Tested-by: Gerald Gibson <gerald.gibson@oracle.com>
Signed-off-by: Håkon Bugge <haakon.bugge@oracle.com>

Author: Hakon-Bugge

net/rds/ib.c

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2006, 2021 Oracle and/or its affiliates.
+ * Copyright (c) 2006, 2022 Oracle and/or its affiliates.
  *
  * This software is available to you under a choice of one of two
  * licenses.  You may choose to be licensed under the terms of the GNU
@@ -904,18 +904,16 @@ static int rds_ib_conn_info_visitor(struct rds_connection *conn,
 		iinfo->max_send_sge = rds_ibdev->max_sge;
 		iinfo->qp_num = ic->i_qp_num;
 		iinfo->dst_qp_num = ic->i_dst_qp_num;
-		iinfo->recv_alloc_ctr = ic->i_recv_ring.w_alloc_ctr;
-		iinfo->recv_free_ctr =
-			(uint32_t)atomic_read(&ic->i_recv_ring.w_free_ctr);
+		iinfo->recv_alloc_ctr = (uint32_t)atomic64_read(&ic->i_recv_ring.w_alloc_ctr);
+		iinfo->recv_free_ctr = (uint32_t)atomic64_read(&ic->i_recv_ring.w_free_ctr);
 		iinfo->flow_ctl_post_credit =
 			IB_GET_POST_CREDITS(atomic_read(&ic->i_credits));
 		iinfo->flow_ctl_send_credit =
 			IB_GET_SEND_CREDITS(atomic_read(&ic->i_credits));
 		rds_ib_get_mr_info(rds_ibdev, iinfo);
 		iinfo->cache_allocs = atomic_read(&ic->i_cache_info.ci_cache_allocs);
-		iinfo->send_alloc_ctr = ic->i_send_ring.w_alloc_ctr;
-		iinfo->send_free_ctr =
-			(uint32_t)atomic_read(&ic->i_send_ring.w_free_ctr);
+		iinfo->send_alloc_ctr = (uint32_t)atomic64_read(&ic->i_send_ring.w_alloc_ctr);
+		iinfo->send_free_ctr = (uint32_t)atomic64_read(&ic->i_send_ring.w_free_ctr);
 		iinfo->send_bytes =
 			(uint64_t)atomic64_read(&conn->c_send_bytes);
 		iinfo->recv_bytes =
@@ -978,18 +976,16 @@ static int rds6_ib_conn_info_visitor(struct rds_connection *conn,
 		iinfo6->max_send_sge = rds_ibdev->max_sge;
 		iinfo6->qp_num = ic->i_qp_num;
 		iinfo6->dst_qp_num = ic->i_dst_qp_num;
-		iinfo6->recv_alloc_ctr = ic->i_recv_ring.w_alloc_ctr;
-		iinfo6->recv_free_ctr =
-			(uint32_t)atomic_read(&ic->i_recv_ring.w_free_ctr);
+		iinfo6->recv_alloc_ctr = (uint32_t)atomic64_read(&ic->i_recv_ring.w_alloc_ctr);
+		iinfo6->recv_free_ctr = (uint32_t)atomic64_read(&ic->i_recv_ring.w_free_ctr);
 		iinfo6->flow_ctl_post_credit =
 			IB_GET_POST_CREDITS(atomic_read(&ic->i_credits));
 		iinfo6->flow_ctl_send_credit =
 			IB_GET_SEND_CREDITS(atomic_read(&ic->i_credits));
 		rds6_ib_get_mr_info(rds_ibdev, iinfo6);
 		iinfo6->cache_allocs = atomic_read(&ic->i_cache_info.ci_cache_allocs);
-		iinfo6->send_alloc_ctr = ic->i_send_ring.w_alloc_ctr;
-		iinfo6->send_free_ctr =
-			(uint32_t)atomic_read(&ic->i_send_ring.w_free_ctr);
+		iinfo6->send_alloc_ctr = (uint32_t)atomic64_read(&ic->i_send_ring.w_alloc_ctr);
+		iinfo6->send_free_ctr =	(uint32_t)atomic64_read(&ic->i_send_ring.w_free_ctr);
 		iinfo6->send_bytes =
 			(uint64_t)atomic64_read(&conn->c_send_bytes);
 		iinfo6->recv_bytes =

net/rds/ib.h

@@ -201,10 +201,8 @@ struct rds_ib_recv_work {
 
 struct rds_ib_work_ring {
 	u32		w_nr;
-	u32		w_alloc_ptr;
-	u32		w_alloc_ctr;
-	u32		w_free_ptr;
-	atomic_t	w_free_ctr;
+	atomic64_t	w_alloc_ctr;
+	atomic64_t	w_free_ctr;
 };
 
 /*

net/rds/ib_ring.c

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2006 Oracle.  All rights reserved.
+ * Copyright (c) 2006, 2022, Oracle and/or its affiliates.
  *
  * This software is available to you under a choice of one of two
  * licenses.  You may choose to be licensed under the terms of the GNU
@@ -30,6 +30,7 @@
  * SOFTWARE.
  *
  */
+#include <linux/atomic.h>
 #include <linux/kernel.h>
 
 #include "rds.h"
@@ -44,18 +45,17 @@
  * Freeing always happens in an interrupt, and hence only
  * races with allocations, but not with other free()s.
  *
- * The interaction between allocation and freeing is that
- * the alloc code has to determine the number of free entries.
- * To this end, we maintain two counters; an allocation counter
- * and a free counter. Both are allowed to run freely, and wrap
- * around.
+ * The interaction between allocation and freeing is that the alloc
+ * code has to determine the number of free entries.  To this end, we
+ * maintain two free-running counters; an allocation counter and a
+ * free counter.
+ *
  * The number of used entries is always (alloc_ctr - free_ctr) % NR.
  *
- * The current implementation makes free_ctr atomic. When the
- * caller finds an allocation fails, it should set an "alloc fail"
- * bit and retry the allocation. The "alloc fail" bit essentially tells
- * the CQ completion handlers to wake it up after freeing some
- * more entries.
+ * When the caller finds an allocation fails, it should set an "alloc
+ * fail" bit and retry the allocation. The "alloc fail" bit
+ * essentially tells the CQ completion handlers to wake it up after
+ * freeing some more entries.
  */
 
 /*
@@ -72,12 +72,7 @@ void rds_ib_ring_init(struct rds_ib_work_ring *ring, u32 nr)
 
 static inline u32 __rds_ib_ring_used(struct rds_ib_work_ring *ring)
 {
-	u32 diff;
-
-	/* This assumes that atomic_t has at least as many bits as u32 */
-	diff = ring->w_alloc_ctr - (u32) atomic_read(&ring->w_free_ctr);
-
-	return diff;
+	return (u64)atomic64_read(&ring->w_alloc_ctr) - (u64)atomic64_read(&ring->w_free_ctr);
 }
 
 void rds_ib_ring_resize(struct rds_ib_work_ring *ring, u32 nr)
@@ -99,31 +94,30 @@ u32 rds_ib_ring_alloc(struct rds_ib_work_ring *ring, u32 val, u32 *pos)
 
 	avail = ring->w_nr - __rds_ib_ring_used(ring);
 
-	rdsdebug("ring %p val %u next %u free %u\n", ring, val,
-		 ring->w_alloc_ptr, avail);
-
 	if (val && avail) {
-		ret = min(val, avail);
-		*pos = ring->w_alloc_ptr;
+		u64 new_ctr;
 
-		ring->w_alloc_ptr = (ring->w_alloc_ptr + ret) % ring->w_nr;
-		ring->w_alloc_ctr += ret;
+		ret = min(val, avail);
+		new_ctr = (u64)atomic64_add_return(ret, &ring->w_alloc_ctr);
+		*pos = (new_ctr - ret) % ring->w_nr;
 	}
 
 	return ret;
 }
 
 void rds_ib_ring_free(struct rds_ib_work_ring *ring, u32 val)
 {
-	ring->w_free_ptr = (ring->w_free_ptr + val) % ring->w_nr;
-	atomic_add(val, &ring->w_free_ctr);
-	smp_mb();
+	smp_mb__before_atomic();
+	atomic64_add(val, &ring->w_free_ctr);
+	smp_mb__after_atomic();
+
 }
 
 void rds_ib_ring_unalloc(struct rds_ib_work_ring *ring, u32 val)
 {
-	ring->w_alloc_ptr = (ring->w_alloc_ptr + ring->w_nr - val) % ring->w_nr;
-	ring->w_alloc_ctr -= val;
+	smp_mb__before_atomic();
+	atomic64_sub(val, &ring->w_alloc_ctr);
+	smp_mb__after_atomic();
 }
 
 int rds_ib_ring_empty(struct rds_ib_work_ring *ring)
@@ -142,7 +136,7 @@ int rds_ib_ring_low(struct rds_ib_work_ring *ring)
  */
 u32 rds_ib_ring_oldest(struct rds_ib_work_ring *ring)
 {
-	return ring->w_free_ptr;
+	return (u64)atomic64_read(&ring->w_free_ctr) % ring->w_nr;
 }
 
 /*