net/rds: use different workqueue for base_conn

RDS uses rds_wq for various operations. During ADDR_CHANGE event,
each connection queues at least two tasks into this single-threaded
workqueue. Furthermore, the TOS connections have dependency on its
base connection. Thus, a separate workqueue is created specifically
for the base connections to speed up base connection establishment.

Orabug: 25521901

Signed-off-by: Wei Lin Guay <[email protected]>
Reviewed-by: Håkon Bugge <[email protected]>
Reviewed-by: Ajaykumar Hotchandani <[email protected]>
Tested-by: Dib Chatterjee <[email protected]>
Tested-by: Rosa Isela Lopez Romero <[email protected]>

Author: Wei Lin Guay

net/rds/connection.c

@@ -243,7 +243,6 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 	}
 
 	conn->c_trans = trans;
-
 	init_waitqueue_head(&conn->c_hs_waitq);
 	for (i = 0; i < RDS_MPATH_WORKERS; i++) {
 		struct rds_conn_path *cp;
@@ -254,7 +253,7 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 		if (conn->c_loopback)
 			cp->cp_wq = rds_local_wq;
 		else
-			cp->cp_wq = rds_wq;
+			cp->cp_wq = tos ? rds_wq : rds_tos_wq;
 	}
 	ret = trans->conn_alloc(conn, gfp);
 	if (ret) {

net/rds/ib.c

@@ -2099,6 +2099,8 @@ static void rds_ib_unregister_client(void)
 	ib_unregister_client(&rds_ib_client);
 	/* wait for rds_ib_dev_free() to complete */
 	flush_workqueue(rds_ip_wq);
+	flush_workqueue(rds_wq);
+	flush_workqueue(rds_tos_wq);
 	flush_workqueue(rds_local_wq);
 }
 

net/rds/rds.h

@@ -1136,6 +1136,7 @@ extern unsigned int  rds_sysctl_shutdown_trace_end_time;
 int rds_threads_init(void);
 void rds_threads_exit(void);
 extern struct workqueue_struct *rds_wq;
+extern struct workqueue_struct *rds_tos_wq;
 extern struct workqueue_struct *rds_local_wq;
 void rds_queue_reconnect(struct rds_conn_path *cp);
 void rds_connect_worker(struct work_struct *);

net/rds/threads.c

@@ -74,6 +74,8 @@ MODULE_PARM_DESC(rds_conn_hb_timeout, " Connection heartbeat timeout");
  */
 struct workqueue_struct *rds_wq;
 EXPORT_SYMBOL_GPL(rds_wq);
+struct workqueue_struct *rds_tos_wq;
+EXPORT_SYMBOL_GPL(rds_tos_wq);
 struct workqueue_struct *rds_local_wq;
 EXPORT_SYMBOL_GPL(rds_local_wq);
 
@@ -352,6 +354,7 @@ void rds_shutdown_worker(struct work_struct *work)
 void rds_threads_exit(void)
 {
 	destroy_workqueue(rds_wq);
+	destroy_workqueue(rds_tos_wq);
 	destroy_workqueue(rds_local_wq);
 }
 
@@ -361,6 +364,10 @@ int rds_threads_init(void)
 	if (!rds_wq)
 		return -ENOMEM;
 
+	rds_tos_wq = create_singlethread_workqueue("krdsd_tos");
+	if (!rds_wq)
+		return -ENOMEM;
+
 	rds_local_wq = create_singlethread_workqueue("krdsd_local");
 	if (!rds_local_wq)
 		return -ENOMEM;