Revert "net/rds: Get rid of tasklets"

This reverts commit db42a07e602d1d2306e8a7026743205abb5ebe0c.

Orabug: 33590962

Signed-off-by: Gerd Rausch <[email protected]>
Reviewed-by: John Donnelly <[email protected]>
Signed-off-by: Brian Maly <[email protected]>

Author: gerd-rausch

net/rds/ib.c

@@ -78,7 +78,6 @@ DEFINE_SPINLOCK(ib_nodev_conns_lock);
 LIST_HEAD(ib_nodev_conns);
 
 struct workqueue_struct *rds_aux_wq;
-struct workqueue_struct *rds_evt_wq;
 
 static struct ib_mr *rds_ib_get_dma_mr(struct ib_pd *pd, int mr_access_flags)
 {
@@ -1349,15 +1348,9 @@ int rds_ib_init(void)
 		goto out_recv;
 	}
 
-	rds_evt_wq = alloc_workqueue("krdsd_evt", WQ_UNBOUND, 0);
-	if (!rds_evt_wq) {
-		pr_err("RDS/IB: failed to create evt workqueue\n");
-		goto out_aux_wq;
-	}
-
 	ret = rds_trans_register(&rds_ib_transport);
 	if (ret)
-		goto out_evt_wq;
+		goto out_aux_wq;
 
 	rds_info_register_func(RDS_INFO_IB_CONNECTIONS, rds_ib_ic_info);
 #if IS_ENABLED(CONFIG_IPV6)
@@ -1375,8 +1368,6 @@ int rds_ib_init(void)
 
 	goto out;
 
-out_evt_wq:
-	destroy_workqueue(rds_evt_wq);
 out_aux_wq:
 	destroy_workqueue(rds_aux_wq);
 out_recv:
@@ -1445,7 +1436,6 @@ void rds_ib_exit(void)
 	rds_ib_sysctl_exit();
 	rds_ib_recv_exit();
 
-	destroy_workqueue(rds_evt_wq);
 	destroy_workqueue(rds_aux_wq);
 	rds_ib_fmr_exit();
 	debugfs_remove_recursive(debugfs_basedir);

net/rds/ib.h

@@ -256,8 +256,8 @@ struct rds_ib_connection {
 	atomic_t		i_fastreg_wrs;
 
 	/* interrupt handling */
-	struct work_struct	i_recv_w;
-	struct work_struct	i_send_w;
+	struct tasklet_struct	i_stasklet;
+	struct tasklet_struct	i_rtasklet;
 
 	/* tx */
 	struct rds_ib_work_ring	i_send_ring;
@@ -270,6 +270,7 @@ struct rds_ib_connection {
 	atomic_t		i_signaled_sends;
 
 	/* rx */
+	struct tasklet_struct	i_recv_tasklet;
 	struct mutex		i_recv_mutex;
 	struct rds_ib_work_ring	i_recv_ring;
 	struct rds_ib_incoming	*i_ibinc;
@@ -432,9 +433,9 @@ struct rds_ib_device {
 	struct ib_cq		*fastreg_cq;
 	struct ib_wc            fastreg_wc[RDS_WC_MAX];
 	struct ib_qp		*fastreg_qp;
+	struct tasklet_struct	fastreg_tasklet;
 	atomic_t		fastreg_wrs;
 	struct rw_semaphore	fastreg_lock;
-	struct work_struct	fastreg_w;
 	struct work_struct	fastreg_reset_w;
 
 	struct ib_mr		*mr;
@@ -614,7 +615,6 @@ static inline __s32 rds_qp_num(struct rds_connection *conn, int dst)
 
 /* ib.c */
 extern struct workqueue_struct *rds_aux_wq;
-extern struct workqueue_struct *rds_evt_wq;
 extern struct rds_transport rds_ib_transport;
 extern void rds_ib_add_one(struct ib_device *device);
 extern void rds_ib_remove_one(struct ib_device *device, void *client_data);
@@ -717,6 +717,7 @@ int rds_ib_inc_copy_to_user(struct rds_incoming *inc, struct iov_iter *to);
 void rds_ib_recv_cqe_handler(struct rds_ib_connection *ic,
 			    struct ib_wc *wc,
 			    struct rds_ib_ack_state *state);
+void rds_ib_recv_tasklet_fn(unsigned long data);
 void rds_ib_recv_init_ring(struct rds_ib_connection *ic);
 void rds_ib_recv_clear_ring(struct rds_ib_connection *ic);
 void rds_ib_recv_init_ack(struct rds_ib_connection *ic);

net/rds/ib_cm.c

@@ -529,8 +529,7 @@ static void rds_ib_cq_comp_handler_fastreg(struct ib_cq *cq, void *context)
 {
 	struct rds_ib_device *rds_ibdev = context;
 
-	queue_work_on(smp_processor_id(),
-		      rds_evt_wq, &rds_ibdev->fastreg_w);
+	tasklet_schedule(&rds_ibdev->fastreg_tasklet);
 }
 
 static void rds_ib_cq_comp_handler_send(struct ib_cq *cq, void *context)
@@ -542,8 +541,7 @@ static void rds_ib_cq_comp_handler_send(struct ib_cq *cq, void *context)
 
 	rds_ib_stats_inc(s_ib_evt_handler_call);
 
-	queue_work_on(smp_processor_id(),
-		      rds_evt_wq, &ic->i_send_w);
+	tasklet_schedule(&ic->i_stasklet);
 }
 
 static void rds_ib_cq_comp_handler_recv(struct ib_cq *cq, void *context)
@@ -555,8 +553,7 @@ static void rds_ib_cq_comp_handler_recv(struct ib_cq *cq, void *context)
 
 	rds_ib_stats_inc(s_ib_evt_handler_call);
 
-	queue_work_on(smp_processor_id(),
-		      rds_evt_wq, &ic->i_recv_w);
+	tasklet_schedule(&ic->i_rtasklet);
 }
 
 static void poll_fcq(struct rds_ib_device *rds_ibdev, struct ib_cq *cq,
@@ -623,22 +620,18 @@ static void poll_rcq(struct rds_ib_connection *ic, struct ib_cq *cq,
 	}
 }
 
-static void rds_ib_cq_comp_handler_fastreg_w(struct work_struct *work)
+static void rds_ib_tasklet_fn_fastreg(unsigned long data)
 {
-	struct rds_ib_device *rds_ibdev = container_of(work,
-						       struct rds_ib_device,
-						       fastreg_w);
+	struct rds_ib_device *rds_ibdev = (struct rds_ib_device *)data;
 
 	poll_fcq(rds_ibdev, rds_ibdev->fastreg_cq, rds_ibdev->fastreg_wc);
 	ib_req_notify_cq(rds_ibdev->fastreg_cq, IB_CQ_NEXT_COMP);
 	poll_fcq(rds_ibdev, rds_ibdev->fastreg_cq, rds_ibdev->fastreg_wc);
 }
 
-void rds_ib_send_w(struct work_struct *work)
+void rds_ib_tasklet_fn_send(unsigned long data)
 {
-	struct rds_ib_connection *ic = container_of(work,
-						    struct rds_ib_connection,
-						    i_send_w);
+	struct rds_ib_connection *ic = (struct rds_ib_connection *) data;
 	struct rds_connection *conn = ic->conn;
 
 	rds_ib_stats_inc(s_ib_tasklet_call);
@@ -712,11 +705,9 @@ static void rds_ib_rx(struct rds_ib_connection *ic)
 	}
 }
 
-void rds_ib_recv_w(struct work_struct *work)
+void rds_ib_tasklet_fn_recv(unsigned long data)
 {
-	struct rds_ib_connection *ic = container_of(work,
-						    struct rds_ib_connection,
-						    i_recv_w);
+	struct rds_ib_connection *ic = (struct rds_ib_connection *) data;
 
 	spin_lock_bh(&ic->i_rx_lock);
 	if (ic->i_rx_wait_for_handler)
@@ -2122,11 +2113,10 @@ void rds_ib_conn_path_shutdown_prepare(struct rds_conn_path *cp)
 		} else if (rds_ib_srq_enabled && ic->rds_ibdev) {
 			/*
 			   wait for the last wqe to complete, then schedule
-			   the recv work to drain the RX CQ.
+			   the recv tasklet to drain the RX CQ.
 			*/
 			wait_for_completion(&ic->i_last_wqe_complete);
-			queue_work_on(smp_processor_id(),
-				      rds_evt_wq, &ic->i_recv_w);
+			tasklet_schedule(&ic->i_rtasklet);
 		}
 	}
 }
@@ -2166,7 +2156,8 @@ void rds_ib_conn_path_shutdown_final(struct rds_conn_path *cp)
 	if (ic->i_cm_id) {
 		cancel_delayed_work_sync(&ic->i_rx_w.work);
 
-		flush_workqueue(rds_evt_wq);
+		tasklet_kill(&ic->i_stasklet);
+		tasklet_kill(&ic->i_rtasklet);
 
 		ic->i_flags &= ~RDS_IB_CQ_ERR;
 
@@ -2279,8 +2270,8 @@ int rds_ib_conn_alloc(struct rds_connection *conn, gfp_t gfp)
 	INIT_LIST_HEAD(&ic->ib_node);
 	init_rwsem(&ic->i_cm_id_free_lock);
 
-	INIT_WORK(&ic->i_send_w, rds_ib_send_w);
-	INIT_WORK(&ic->i_recv_w, rds_ib_recv_w);
+	tasklet_init(&ic->i_stasklet, rds_ib_tasklet_fn_send, (unsigned long) ic);
+	tasklet_init(&ic->i_rtasklet, rds_ib_tasklet_fn_recv, (unsigned long) ic);
 	mutex_init(&ic->i_recv_mutex);
 #ifndef KERNEL_HAS_ATOMIC64
 	spin_lock_init(&ic->i_ack_lock);
@@ -2365,7 +2356,7 @@ void rds_ib_destroy_fastreg(struct rds_ib_device *rds_ibdev)
 	 */
 	WARN_ON(atomic_read(&rds_ibdev->fastreg_wrs) != RDS_IB_DEFAULT_FREG_WR);
 
-	flush_workqueue(rds_evt_wq);
+	tasklet_kill(&rds_ibdev->fastreg_tasklet);
 	if (rds_ibdev->fastreg_qp) {
 		/* Destroy qp */
 		if (ib_destroy_qp(rds_ibdev->fastreg_qp))
@@ -2526,7 +2517,8 @@ int rds_ib_setup_fastreg(struct rds_ib_device *rds_ibdev)
 					   "moved qp to RTS state for device",
 					   0);
 
-	INIT_WORK(&rds_ibdev->fastreg_w, rds_ib_cq_comp_handler_fastreg_w);
+	tasklet_init(&rds_ibdev->fastreg_tasklet, rds_ib_tasklet_fn_fastreg,
+		     (unsigned long)rds_ibdev);
 	atomic_set(&rds_ibdev->fastreg_wrs, RDS_IB_DEFAULT_FREG_WR);
 
 clean_up: