[PATCH net-next] net/rds: replace tasklets with workqueue

[PATCH net-next] net/rds: replace tasklets with workqueue

[Sunny Tiwari]

tasklet_struct is deprecated. Convert the send and receive completion
handlers in net/rds/ib to work items on a dedicated high-priority
workqueue.

Allocate per-CPU bound rds_ib_wq with WQ_HIGHPRI | WQ_MEM_RECLAIM in
rds_ib_init() and destroy it in rds_ib_exit(). Per-CPU binding preserves
cache locality with the completion vector, matching original tasklet
affinity without cross-core migration.

Replace i_send_tasklet and i_recv_tasklet with struct work_struct. Order
quiescence via atomic_set_release() and atomic_read_acquire() prior to
cancel_work_sync() during connection shutdown.

Assisted-by: Gemini
Signed-off-by: Sunny Tiwari <sunnytiwari@google.com>
---
 net/rds/ib.c    | 22 +++++++++++++++++++++-
 net/rds/ib.h    |  7 +++++--
 net/rds/ib_cm.c | 39 +++++++++++++++++++++++----------------
 3 files changed, 49 insertions(+), 19 deletions(-)

diff --git a/net/rds/ib.c b/net/rds/ib.c
index 786f39169..3e3551c6a 100644
--- a/net/rds/ib.c
+++ b/net/rds/ib.c
@@ -50,6 +50,7 @@ static unsigned int rds_ib_mr_1m_pool_size = RDS_MR_1M_POOL_SIZE;
 static unsigned int rds_ib_mr_8k_pool_size = RDS_MR_8K_POOL_SIZE;
 unsigned int rds_ib_retry_count = RDS_IB_DEFAULT_RETRY_COUNT;
 static atomic_t rds_ib_unloading;
+struct workqueue_struct *rds_ib_wq;
 
 module_param(rds_ib_mr_1m_pool_size, int, 0444);
 MODULE_PARM_DESC(rds_ib_mr_1m_pool_size, " Max number of 1M mr per HCA");
@@ -533,6 +534,8 @@ static bool rds_ib_is_unloading(struct rds_connection *conn)
 
 void rds_ib_exit(void)
 {
+	struct workqueue_struct *wq = rds_ib_wq;
+
 	rds_ib_set_unloading();
 	synchronize_rcu();
 	rds_info_deregister_func(RDS_INFO_IB_CONNECTIONS, rds_ib_ic_info);
@@ -545,6 +548,10 @@ void rds_ib_exit(void)
 	rds_ib_recv_exit();
 	rds_trans_unregister(&rds_ib_transport);
 	rds_ib_mr_exit();
+
+	rds_ib_wq = NULL;
+	if (wq)
+		destroy_workqueue(wq);
 }
 
 static u8 rds_ib_get_tos_map(u8 tos)
@@ -587,6 +594,7 @@ struct rds_transport rds_ib_transport = {
 
 int rds_ib_init(void)
 {
+	struct workqueue_struct *wq;
 	int ret;
 
 	INIT_LIST_HEAD(&rds_ib_devices);
@@ -595,9 +603,16 @@ int rds_ib_init(void)
 	if (ret)
 		goto out;
 
+	rds_ib_wq = alloc_workqueue("rds_ib_wq",
+				    WQ_HIGHPRI | WQ_MEM_RECLAIM, 0);
+	if (!rds_ib_wq) {
+		ret = -ENOMEM;
+		goto out_mr_exit;
+	}
+
 	ret = ib_register_client(&rds_ib_client);
 	if (ret)
-		goto out_mr_exit;
+		goto out_wq;
 
 	ret = rds_ib_sysctl_init();
 	if (ret)
@@ -620,6 +635,11 @@ int rds_ib_init(void)
 	rds_ib_sysctl_exit();
 out_ibreg:
 	rds_ib_unregister_client();
+out_wq:
+	wq = rds_ib_wq;
+	rds_ib_wq = NULL;
+	if (wq)
+		destroy_workqueue(wq);
 out_mr_exit:
 	rds_ib_mr_exit();
 out:
diff --git a/net/rds/ib.h b/net/rds/ib.h
index 5ff346a1e..5f617f66e 100644
--- a/net/rds/ib.h
+++ b/net/rds/ib.h
@@ -159,8 +159,8 @@ struct rds_ib_connection {
 	atomic_t		i_fastreg_inuse_count;
 
 	/* interrupt handling */
-	struct tasklet_struct	i_send_tasklet;
-	struct tasklet_struct	i_recv_tasklet;
+	struct work_struct	i_send_work;
+	struct work_struct	i_recv_work;
 
 	/* tx */
 	struct rds_ib_work_ring	i_send_ring;
@@ -276,6 +276,7 @@ struct rds_ib_statistics {
 	uint64_t	s_ib_connect_raced;
 	uint64_t	s_ib_listen_closed_stale;
 	uint64_t	s_ib_evt_handler_call;
+	/* Retained as s_ib_tasklet_call for userspace rds-info -I ABI */
 	uint64_t	s_ib_tasklet_call;
 	uint64_t	s_ib_tx_cq_event;
 	uint64_t	s_ib_tx_ring_full;
@@ -452,4 +453,6 @@ extern unsigned long rds_ib_sysctl_max_unsig_bytes;
 extern unsigned long rds_ib_sysctl_max_recv_allocation;
 extern unsigned int rds_ib_sysctl_flow_control;
 
+extern struct workqueue_struct *rds_ib_wq;
+
 #endif
diff --git a/net/rds/ib_cm.c b/net/rds/ib_cm.c
index 6e3110a04..976a9d7e8 100644
--- a/net/rds/ib_cm.c
+++ b/net/rds/ib_cm.c
@@ -253,7 +253,10 @@ static void rds_ib_cq_comp_handler_recv(struct ib_cq *cq, void *context)
 
 	rds_ib_stats_inc(s_ib_evt_handler_call);
 
-	tasklet_schedule(&ic->i_recv_tasklet);
+	if (unlikely(!rds_ib_wq || atomic_read_acquire(&ic->i_cq_quiesce)))
+		return;
+
+	queue_work(rds_ib_wq, &ic->i_recv_work);
 }
 
 static void poll_scq(struct rds_ib_connection *ic, struct ib_cq *cq,
@@ -279,9 +282,10 @@ static void poll_scq(struct rds_ib_connection *ic, struct ib_cq *cq,
 	}
 }
 
-static void rds_ib_tasklet_fn_send(unsigned long data)
+static void rds_ib_send_worker(struct work_struct *work)
 {
-	struct rds_ib_connection *ic = (struct rds_ib_connection *)data;
+	struct rds_ib_connection *ic =
+		container_of(work, struct rds_ib_connection, i_send_work);
 	struct rds_connection *conn = ic->conn;
 
 	rds_ib_stats_inc(s_ib_tasklet_call);
@@ -319,9 +323,10 @@ static void poll_rcq(struct rds_ib_connection *ic, struct ib_cq *cq,
 	}
 }
 
-static void rds_ib_tasklet_fn_recv(unsigned long data)
+static void rds_ib_recv_worker(struct work_struct *work)
 {
-	struct rds_ib_connection *ic = (struct rds_ib_connection *)data;
+	struct rds_ib_connection *ic =
+		container_of(work, struct rds_ib_connection, i_recv_work);
 	struct rds_connection *conn = ic->conn;
 	struct rds_ib_device *rds_ibdev = ic->rds_ibdev;
 	struct rds_ib_ack_state state;
@@ -381,7 +386,10 @@ static void rds_ib_cq_comp_handler_send(struct ib_cq *cq, void *context)
 
 	rds_ib_stats_inc(s_ib_evt_handler_call);
 
-	tasklet_schedule(&ic->i_send_tasklet);
+	if (unlikely(!rds_ib_wq || atomic_read_acquire(&ic->i_cq_quiesce)))
+		return;
+
+	queue_work(rds_ib_wq, &ic->i_send_work);
 }
 
 static inline int ibdev_get_unused_vector(struct rds_ib_device *rds_ibdev)
@@ -1099,14 +1107,15 @@ void rds_ib_conn_path_shutdown(struct rds_conn_path *cp)
 		while (!wait_event_timeout(rds_ib_ring_empty_wait,
 					   rds_ib_conn_path_shutdown_check_wait(cp) == 0,
 					   msecs_to_jiffies(1000))) {
-			tasklet_schedule(&ic->i_send_tasklet);
-			tasklet_schedule(&ic->i_recv_tasklet);
+			if (rds_ib_wq) {
+				queue_work(rds_ib_wq, &ic->i_send_work);
+				queue_work(rds_ib_wq, &ic->i_recv_work);
+			}
 		}
 
-		tasklet_kill(&ic->i_send_tasklet);
-		tasklet_kill(&ic->i_recv_tasklet);
-
-		atomic_set(&ic->i_cq_quiesce, 1);
+		atomic_set_release(&ic->i_cq_quiesce, 1);
+		cancel_work_sync(&ic->i_send_work);
+		cancel_work_sync(&ic->i_recv_work);
 
 		/* first destroy the ib state that generates callbacks */
 		if (ic->i_cm_id->qp)
@@ -1234,10 +1243,8 @@ int rds_ib_conn_alloc(struct rds_connection *conn, gfp_t gfp)
 	}
 
 	INIT_LIST_HEAD(&ic->ib_node);
-	tasklet_init(&ic->i_send_tasklet, rds_ib_tasklet_fn_send,
-		     (unsigned long)ic);
-	tasklet_init(&ic->i_recv_tasklet, rds_ib_tasklet_fn_recv,
-		     (unsigned long)ic);
+	INIT_WORK(&ic->i_send_work, rds_ib_send_worker);
+	INIT_WORK(&ic->i_recv_work, rds_ib_recv_worker);
 	mutex_init(&ic->i_recv_mutex);
 #ifndef KERNEL_HAS_ATOMIC64
 	spin_lock_init(&ic->i_ack_lock);

[syzbot ci] Re: net/rds: replace tasklets with workqueue

[syzbot ci]

syzbot ci has tested the following series

[v1] net/rds: replace tasklets with workqueue
https://lore.kernel.org/all/6b07af04b3ba66a09009e1a0a7b98ec5ec85d3a0.1790109248.git.sunnytiwari@google.com
* [PATCH net-next] net/rds: replace tasklets with workqueue

and found the following issue:
WARNING in __alloc_workqueue

Full report is available here:
https://ci.syzbot.org/series/e7102860-1b25-4e10-b653-afe8e9820ef8

***

WARNING in __alloc_workqueue

tree:      net-next
URL:       https://kernel.googlesource.com/pub/scm/linux/kernel/git/netdev/net-next.git
base:      42a9fb3382fc2573e92f41d203b095d9a372cfc9
arch:      amd64
compiler:  Debian clang version 22.1.8 (++20260613092233+e80beda6e255-1~exp1~20260613092250.77), Debian LLD 22.1.8
config:    https://ci.syzbot.org/builds/220e9c3a-eede-449e-8bb1-d34ad95dc867/config

gre: GRE over IPv4 demultiplexer driver
ip_gre: GRE over IPv4 tunneling driver
IPv4 over IPsec tunneling driver
Initializing XFRM netlink socket
IPsec XFRM device driver
xfrm_iptfs: IPsec IP-TFS tunnel mode module
NET: Registered PF_INET6 protocol family
Segment Routing with IPv6
RPL Segment Routing with IPv6
In-situ OAM (IOAM) with IPv6
mip6: Mobile IPv6
sit: IPv6, IPv4 and MPLS over IPv4 tunneling driver
ip6_gre: GRE over IPv6 tunneling driver
NET: Registered PF_PACKET protocol family
PFKEY is deprecated and scheduled to be removed in 2027, please contact the netdev mailing list
NET: Registered PF_KEY protocol family
Bridge firewalling registered
NET: Registered PF_X25 protocol family
X25: Linux Version 0.2
can: controller area network core
NET: Registered PF_CAN protocol family
can: raw protocol
can: broadcast manager protocol
can: netlink gateway - max_hops=1
can: SAE J1939
can: isotp protocol (max_pdu_size 8300)
Bluetooth: RFCOMM TTY layer initialized
Bluetooth: RFCOMM socket layer initialized
Bluetooth: RFCOMM ver 1.11
Bluetooth: BNEP (Ethernet Emulation) ver 1.3
Bluetooth: BNEP filters: protocol multicast
Bluetooth: BNEP socket layer initialized
Bluetooth: HIDP (Human Interface Emulation) ver 1.2
Bluetooth: HIDP socket layer initialized
NET: Registered PF_RXRPC protocol family
Key type rxrpc registered
Key type rxrpc_s registered
NET: Registered PF_KCM protocol family
l2tp_core: L2TP core driver, V2.0
l2tp_ppp: PPPoL2TP kernel driver, V2.0
l2tp_ip: L2TP IP encapsulation support (L2TPv3)
l2tp_netlink: L2TP netlink interface
l2tp_eth: L2TP ethernet pseudowire support (L2TPv3)
l2tp_ip6: L2TP IP encapsulation support for IPv6 (L2TPv3)
NET: Registered PF_PHONET protocol family
8021q: 802.1Q VLAN Support v1.8
sctp: Hash tables configured (bind 32/56)
NET: Registered PF_RDS protocol family
------------[ cut here ]------------
workqueue: rds_ib_wq is using neither WQ_PERCPU or WQ_UNBOUND. Setting WQ_PERCPU.
WARNING: kernel/workqueue.c:5939 at __alloc_workqueue+0x1cd2/0x1fe0, CPU#1: swapper/0/1
Modules linked in:
CPU: 1 UID: 0 PID: 1 Comm: swapper/0 Not tainted syzkaller #0 PREEMPT(full) 
Hardware name: QEMU Standard PC (Q35 + ICH9, 2009), BIOS 1.16.2-debian-1.16.2-1 04/01/2014
RIP: 0010:__alloc_workqueue+0x1cd5/0x1fe0
Code: 0b 90 e9 6d fc ff ff e8 89 02 39 00 e9 7f fb ff ff e8 7f 02 39 00 e9 85 fb ff ff e8 75 02 39 00 48 8d 3d ce 68 ee 0e 4c 89 f6 <67> 48 0f b9 3a 41 81 cf 00 01 00 00 e9 10 e6 ff ff e8 55 02 39 00
RSP: 0000:ffffc900000676a8 EFLAGS: 00010293
RAX: ffffffff818ed98b RBX: 0000000000000000 RCX: ffff888102adda00
RDX: 0000000000000000 RSI: ffff888173f5d170 RDI: ffffffff907d4260
RBP: ffffffff8d43a5a0 R08: ffff888102adda00 R09: 0000000000000002
R10: 0000000000000102 R11: 0000000000000000 R12: ffff888173f5d000
R13: ffff888173f5d000 R14: ffff888173f5d170 R15: 0000000000000018
FS:  0000000000000000(0000) GS:ffff8882a8cc6000(0000) knlGS:0000000000000000
CS:  0010 DS: 0000 ES: 0000 CR0: 0000000080050033
CR2: 0000000000000000 CR3: 000000000eb48000 CR4: 00000000000006f0
Call Trace:
 <TASK>
 alloc_workqueue_noprof+0xe3/0x210
 rds_ib_init+0x54/0x190
 rds_rdma_init+0x74/0x170
 do_one_initcall+0x250/0x870
 do_initcall_level+0x10a/0x1a0
 do_initcalls+0x59/0xa0
 kernel_init_freeable+0x29d/0x3e0
 kernel_init+0x22/0x1d0
 ret_from_fork+0x514/0xb70
 ret_from_fork_asm+0x1a/0x30
 </TASK>


***

If these findings have caused you to resend the series or submit a
separate fix, please add the following tag to your commit message:
  Tested-by: syzbot@syzkaller.appspotmail.com

---
This report is generated by a bot. It may contain errors.
syzbot ci engineers can be reached at syzkaller@googlegroups.com.

To test a fix for this bug, please reply with `#syz test`
(on a separate line) and attach the patch to the email.

Notes:
- The patch will be applied on top of the tested series (as an
  incremental fix).
- To test a new version of the whole series, please send it directly
  to syzbot@lists.linux.dev.
- Arguments like custom git repos and branches are not supported.