net/rds: Reduce RDS headers de-allocation time

After commit 10b43f1 ("net/rds: Use DMA memory pool allocation
for rds_header"), the time to free the RDS headers at connection tear
down becomes longer.  At connection fail over time, hundreds of
connections are shut down sequentially.  A little extra time spent for
each connection shutdown adds up.  To reduce this fail over time, the
headers free work is offloaded to a workqueue.

Orabug: 30434704

Fixes: 10b43f1 ("net/rds: Use DMA memory pool allocation for
rds_header")

Signed-off-by: Ka-Cheong Poon <[email protected]>
Reviewed-by: Håkon Bugge <[email protected]>
Tested-by: Håkon Bugge <[email protected]>
Signed-off-by: Somasundaram Krishnasamy <[email protected]>

Author: kcp-git

net/rds/ib.c

@@ -131,6 +131,11 @@ static void rds_ib_dev_free_dev(struct rds_ib_device *rds_ibdev)
 		ib_dereg_mr(rds_ibdev->mr);
 	if (rds_ibdev->pd)
 		ib_dealloc_pd(rds_ibdev->pd);
+	/* Flush and destroy the header pool clean up work queue before
+	 * destroying the pool.
+	 */
+	if (rds_ibdev->rid_hdrs_pool_wq)
+		destroy_workqueue(rds_ibdev->rid_hdrs_pool_wq);
 	if (rds_ibdev->rid_hdrs_pool)
 		dma_pool_destroy(rds_ibdev->rid_hdrs_pool);
 out:
@@ -593,6 +598,13 @@ void rds_ib_add_one(struct ib_device *device)
 	if (!rds_ibdev->rid_hdrs_pool)
 		goto put_dev;
 
+	rds_ibdev->rid_hdrs_pool_wq = alloc_workqueue("rds_pool_%s",
+						      WQ_UNBOUND |
+						      WQ_MEM_RECLAIM, 0,
+						      device->name);
+	if (!rds_ibdev->rid_hdrs_pool_wq)
+		goto put_dev;
+
 	rds_ibdev->vector_load = kzalloc(sizeof(int) *
 					device->num_comp_vectors, GFP_KERNEL);
 	if (!rds_ibdev->vector_load) {

net/rds/ib.h

@@ -184,6 +184,26 @@ struct rds_ib_rx_work {
 	struct rds_ib_connection        *ic;
 };
 
+/* Structure for the DMA headers pool deallocation work */
+struct rds_dma_hdrs_free_wk {
+	struct work_struct	rdhf_work;
+	struct dma_pool		*rdhf_pool;	/* Pool for the headers. */
+
+	/* Info of headers for sending. */
+	struct rds_header	**rdhf_snd_hdrs;
+	dma_addr_t		*rdhf_snd_dma_addrs;
+	u32			rdhf_snd_num_hdrs;
+
+	/* Info of headers for receiving. */
+	struct rds_header	**rdhf_rcv_hdrs;
+	dma_addr_t		*rdhf_rcv_dma_addrs;
+	u32			rdhf_rcv_num_hdrs;
+
+	/* Info of the ACK header. */
+	struct rds_header	*rdhf_ack_rds_hdr;
+	dma_addr_t		rdhf_ack_dma_addr;
+};
+
 struct rds_ib_connection {
 
 	struct list_head	ib_node;
@@ -316,11 +336,6 @@ struct rds_ib_srq {
 };
 
 
-struct rds_ib_conn_drop_work {
-	struct delayed_work             work;
-	struct rds_connection          *conn;
-};
-
 struct rds_ib_conn_destroy_work {
 	struct delayed_work             work;
 	struct rds_connection          *conn;
@@ -338,6 +353,7 @@ struct rds_ib_device {
 	struct ib_device	*dev;
 	struct ib_pd		*pd;
 	struct dma_pool		*rid_hdrs_pool; /* RDS headers DMA pool */
+	struct workqueue_struct	*rid_hdrs_pool_wq; /* Hdrs pool clean up wq */
 
 	bool			use_fastreg;
 	int			fastreg_cq_vector;

net/rds/ib_cm.c

@@ -804,6 +804,82 @@ void rds_dma_hdrs_free(struct dma_pool *pool, struct rds_header **hdrs,
 	kvfree(dma_addrs);
 }
 
+/* Worker function for freeing the DMA pool headers for sending and
+ * receiving.
+ */
+static void rds_dma_hdrs_cleanup_worker(struct work_struct *work)
+{
+	struct rds_dma_hdrs_free_wk *wk;
+	struct dma_pool *pool;
+
+	wk = container_of(work, struct rds_dma_hdrs_free_wk, rdhf_work);
+
+	pool = wk->rdhf_pool;
+	if (wk->rdhf_snd_hdrs)
+		rds_dma_hdrs_free(pool, wk->rdhf_snd_hdrs,
+				  wk->rdhf_snd_dma_addrs,
+				  wk->rdhf_snd_num_hdrs);
+	if (wk->rdhf_rcv_hdrs)
+		rds_dma_hdrs_free(pool, wk->rdhf_rcv_hdrs,
+				  wk->rdhf_rcv_dma_addrs,
+				  wk->rdhf_rcv_num_hdrs);
+	if (wk->rdhf_ack_rds_hdr)
+		dma_pool_free(pool, wk->rdhf_ack_rds_hdr,
+			      wk->rdhf_ack_dma_addr);
+
+	kfree(wk);
+}
+
+/* Free all the DMA pool headers associated with an RDS RDMA connection.
+ *
+ * @ic: the RDS RDMA connection
+ */
+static void rds_dma_hdrs_free_all(struct rds_ib_connection *ic)
+{
+	struct rds_dma_hdrs_free_wk *wk;
+	struct dma_pool *pool;
+
+	pool = ic->rds_ibdev->rid_hdrs_pool;
+
+	wk = kmalloc(sizeof(*wk), GFP_ATOMIC);
+	if (wk) {
+		wk->rdhf_pool = pool;
+
+		wk->rdhf_snd_hdrs = ic->i_send_hdrs;
+		wk->rdhf_snd_dma_addrs = ic->i_send_hdrs_dma;
+		wk->rdhf_snd_num_hdrs = ic->i_send_ring.w_nr;
+
+		wk->rdhf_rcv_hdrs = ic->i_recv_hdrs;
+		wk->rdhf_rcv_dma_addrs = ic->i_recv_hdrs_dma;
+		wk->rdhf_rcv_num_hdrs = ic->i_recv_ring.w_nr;
+
+		wk->rdhf_ack_rds_hdr = ic->i_ack;
+		wk->rdhf_ack_dma_addr = ic->i_ack_dma;
+
+		INIT_WORK(&wk->rdhf_work, rds_dma_hdrs_cleanup_worker);
+		queue_work(ic->rds_ibdev->rid_hdrs_pool_wq,  &wk->rdhf_work);
+	} else {
+		if (ic->i_send_hdrs)
+			rds_dma_hdrs_free(pool, ic->i_send_hdrs,
+					  ic->i_send_hdrs_dma,
+					  ic->i_send_ring.w_nr);
+		if (ic->i_recv_hdrs)
+			rds_dma_hdrs_free(pool, ic->i_recv_hdrs,
+					  ic->i_recv_hdrs_dma,
+					  ic->i_recv_ring.w_nr);
+		if (ic->i_ack)
+			dma_pool_free(pool, ic->i_ack, ic->i_ack_dma);
+	}
+
+	ic->i_send_hdrs = NULL;
+	ic->i_send_hdrs_dma = NULL;
+
+	ic->i_recv_hdrs = NULL;
+	ic->i_recv_hdrs_dma = NULL;
+
+	ic->i_ack = NULL;
+}
+
 /*
  * This needs to be very careful to not leave IS_ERR pointers around for
  * cleanup to trip over.
@@ -1586,32 +1662,9 @@ void rds_ib_conn_path_shutdown(struct rds_conn_path *cp)
 		if (ic->i_cm_id->qp)
 			rdma_destroy_qp(ic->i_cm_id);
 
+		/* then free the resources that ib callbacks use */
 		if (ic->rds_ibdev) {
-			struct dma_pool *pool;
-
-			pool = ic->rds_ibdev->rid_hdrs_pool;
-
-			/* then free the resources that ib callbacks use */
-			if (ic->i_send_hdrs) {
-				rds_dma_hdrs_free(pool, ic->i_send_hdrs,
-						  ic->i_send_hdrs_dma,
-						  ic->i_send_ring.w_nr);
-				ic->i_send_hdrs = NULL;
-				ic->i_send_hdrs_dma = NULL;
-			}
-
-			if (ic->i_recv_hdrs) {
-				rds_dma_hdrs_free(pool, ic->i_recv_hdrs,
-						  ic->i_recv_hdrs_dma,
-						  ic->i_recv_ring.w_nr);
-				ic->i_recv_hdrs = NULL;
-				ic->i_recv_hdrs_dma = NULL;
-			}
-
-			if (ic->i_ack) {
-				dma_pool_free(pool, ic->i_ack, ic->i_ack_dma);
-				ic->i_ack = NULL;
-			}
+			rds_dma_hdrs_free_all(ic);
 		} else {
 			WARN_ON(ic->i_send_hdrs);
 			WARN_ON(ic->i_send_hdrs_dma);