net/rds: serialize up+down-work to relax strict ordering

Serialize calls to rds_connect_worker() and rds_shutdown_worker()
by taking advantage of the workqueue-API only executing a single
running callback per "struct work_struct":
We simply en-queue a combined rds_up_or_down_worker(),
that calls either rds_shutdown_worker() or rds_connect_worker()
depending on the state of "RDS_SHUTDOWN_WORK_QUEUED" or
"RDS_RECONNECT_PENDING".

Ideally, we wouldn't have such a mess and complexity to
deal with state transitions, but rather have a proper
state-machine with trigger- and action-definitions,
all dispatched by a single worker:
That way we'd know that parallelism was taken care of
properly, and would also be able to understand
which events in a given context lead to what actions.

But this commit doesn't aim at rewriting all of this,
but rather kick the can down the road by simply
combining up+down-work to avoid race conditions
and allow "cp->cp_wq" to no longer be strictly ordered.

For that we introduce module parameter "rds_wq_strictly_ordered",
which is "true" by default, until we convince ourselves,
that all race-conditions in the code triggered by no
longer using strictly-ordered workqueues are taken care of.

Orabug: 35094723

Signed-off-by: Gerd Rausch <[email protected]>
Reviewed-by: Håkon Bugge <[email protected]>
Signed-off-by: Brian Maly <[email protected]>

Author: gerd-rausch

net/rds/connection.c

@@ -46,6 +46,9 @@
 
 #define RDS_CONN_FADDR_HASH_ENTRIES 256
 
+static bool rds_wq_strictly_ordered = true;
+module_param(rds_wq_strictly_ordered, bool, 0644);
+
 /* converting this to RCU is a chore for another day.. */
 static DEFINE_SPINLOCK(rds_conn_lock);
 static struct hlist_head rds_conn_hash[RDS_CONNECTION_HASH_ENTRIES];
@@ -215,9 +218,8 @@ static void __rds_conn_path_init(struct rds_connection *conn,
 
 	INIT_DELAYED_WORK(&cp->cp_send_w, rds_send_worker);
 	INIT_DELAYED_WORK(&cp->cp_recv_w, rds_recv_worker);
-	INIT_DELAYED_WORK(&cp->cp_conn_w, rds_connect_worker);
 	INIT_DELAYED_WORK(&cp->cp_hb_w, rds_hb_worker);
-	INIT_WORK(&cp->cp_down_w, rds_shutdown_worker);
+	INIT_DELAYED_WORK(&cp->cp_up_or_down_w, rds_up_or_down_worker);
 	INIT_DELAYED_WORK(&cp->cp_down_wait_w, rds_conn_shutdown_check_wait);
 	mutex_init(&cp->cp_cm_lock);
 	atomic_set(&cp->cp_rdma_map_pending, 0);
@@ -359,7 +361,12 @@ static struct rds_connection *__rds_conn_create(struct net *net,
 		__rds_conn_path_init(conn, cp, is_outgoing);
 		cp->cp_index = i;
 
-		cp->cp_wq = alloc_ordered_workqueue("krds_cp_wq#%d/%d", 0,
+		if (rds_wq_strictly_ordered)
+			cp->cp_wq = alloc_ordered_workqueue("krds_cp_wq#%d/%d", 0,
+							    atomic_read(&conn->c_trans->t_conn_count), i);
+		else
+			cp->cp_wq = alloc_workqueue("krds_cp_wq#%d/%d", 0,
+						    RDS_CP_WQ_MAX_ACTIVE,
 						    atomic_read(&conn->c_trans->t_conn_count), i);
 		if (!cp->cp_wq) {
 			while (--i >= 0) {
@@ -494,7 +501,6 @@ static void rds_conn_shutdown_final(struct rds_conn_path *cp)
 	 * Otherwise, the reconnect is always triggered by the active peer.
 	 */
 
-	rds_queue_cancel_work(cp, &cp->cp_conn_w, "final shutdown");
 	rds_clear_reconnect_pending_work_bit(cp);
 
 	rcu_read_lock();
@@ -678,7 +684,7 @@ static void rds_conn_path_destroy(struct rds_conn_path *cp, int shutdown)
 
 	cp->cp_shutdown_final = &shutdown_final;
 	rds_conn_path_drop(cp, DR_CONN_DESTROY, 0);
-	rds_queue_flush_work(cp, &cp->cp_down_w, "conn path destroy down work");
+	flush_delayed_work(&cp->cp_up_or_down_w);
 	wait_for_completion(&shutdown_final);
 
 	/* tear down queued messages */

net/rds/rds.h

@@ -107,6 +107,8 @@ enum {
 #define RDS_CONG_MAP_PAGES	(PAGE_ALIGN(RDS_CONG_MAP_BYTES) / RDS_CONG_PAGE_SIZE)
 #define RDS_CONG_MAP_PAGE_BITS	(RDS_CONG_PAGE_SIZE * 8)
 
+#define RDS_CP_WQ_MAX_ACTIVE	4
+
 struct rds_cong_map {
 	struct rb_node		m_rb_node;
 	struct in6_addr		m_addr;
@@ -266,9 +268,8 @@ struct rds_conn_path {
 	unsigned long		cp_reconnect_jiffies;
 	struct delayed_work	cp_send_w;
 	struct delayed_work	cp_recv_w;
-	struct delayed_work	cp_conn_w;
 	struct delayed_work     cp_hb_w;
-	struct work_struct	cp_down_w;
+	struct delayed_work	cp_up_or_down_w;
 	struct delayed_work	cp_down_wait_w;
 	struct mutex		cp_cm_lock;	/* protect cp_state & cm */
 	wait_queue_head_t	cp_waitq;
@@ -1129,6 +1130,10 @@ void rds_queue_delayed_work_on(struct rds_conn_path *cp, int cpu,
 			       struct workqueue_struct *wq,
 			       struct delayed_work *dwork,
 			       unsigned long delay, char *reason);
+void rds_mod_delayed_work(struct rds_conn_path *cp,
+			  struct workqueue_struct *wq,
+			  struct delayed_work *dwork,
+			  unsigned long delay, char *reason);
 
 static inline void rds_conn_path_state_change(struct rds_conn_path *cp,
 					      int new, int reason, int err)
@@ -1142,8 +1147,8 @@ static inline void rds_conn_path_state_change(struct rds_conn_path *cp,
 static inline void rds_cond_queue_shutdown_work(struct rds_conn_path *cp)
 {
 	if (!test_and_set_bit(RDS_SHUTDOWN_WORK_QUEUED, &cp->cp_flags))
-		rds_queue_work(cp, cp->cp_wq, &cp->cp_down_w,
-			       "queue shutdown work");
+		rds_mod_delayed_work(cp, cp->cp_wq, &cp->cp_up_or_down_w,
+				     0, "queue shutdown work");
 }
 
 static inline void rds_clear_shutdown_pending_work_bit(struct rds_conn_path *cp)
@@ -1158,11 +1163,12 @@ static inline void rds_clear_shutdown_pending_work_bit(struct rds_conn_path *cp)
 static inline bool rds_cond_queue_reconnect_work(struct rds_conn_path *cp, unsigned long delay)
 {
 	if (!test_and_set_bit(RDS_RECONNECT_PENDING, &cp->cp_flags)) {
-		rds_queue_delayed_work(cp, cp->cp_wq, &cp->cp_conn_w, delay,
-				       "reconnect work");
+		rds_queue_delayed_work(cp, cp->cp_wq, &cp->cp_up_or_down_w,
+				       delay, "reconnect work");
 		return true;
-	} else
+	} else {
 		return false;
+	}
 }
 
 static inline void
@@ -1460,8 +1466,7 @@ int rds_threads_init(void);
 void rds_threads_exit(void);
 extern struct workqueue_struct *rds_wq;
 void rds_queue_reconnect(struct rds_conn_path *cp, bool immediate);
-void rds_connect_worker(struct work_struct *);
-void rds_shutdown_worker(struct work_struct *);
+void rds_up_or_down_worker(struct work_struct *);
 void rds_send_worker(struct work_struct *);
 void rds_recv_worker(struct work_struct *);
 void rds_hb_worker(struct work_struct *);

net/rds/rds_single_path.h

@@ -21,8 +21,7 @@
 #define c_reconnect_jiffies	c_path[0].cp_reconnect_jiffies
 #define c_send_w		c_path[0].cp_send_w
 #define c_recv_w		c_path[0].cp_recv_w
-#define c_conn_w		c_path[0].cp_conn_w
-#define c_down_w		c_path[0].cp_down_w
+#define c_up_or_down_w		c_path[0].cp_up_or_down_w
 #define c_down_wait_w		c_path[0].cp_down_wait_w
 #define c_cm_lock		c_path[0].cp_cm_lock
 #define c_waitq			c_path[0].cp_waitq

net/rds/threads.c

@@ -144,6 +144,25 @@ void rds_queue_delayed_work_on(struct rds_conn_path *cp,
 }
 EXPORT_SYMBOL_GPL(rds_queue_delayed_work_on);
 
+void rds_mod_delayed_work(struct rds_conn_path *cp,
+			  struct workqueue_struct *wq,
+			  struct delayed_work *dwork,
+			  unsigned long delay,
+			  char *reason)
+{
+	int cpu;
+
+	trace_rds_queue_work(cp ? cp->cp_conn : NULL, cp, wq, &dwork->work,
+			     delay, reason);
+
+	if (cp && cp->cp_conn->c_trans->conn_preferred_cpu) {
+		cpu = cp->cp_conn->c_trans->conn_preferred_cpu(cp->cp_conn, false);
+		mod_delayed_work_on(cpu, wq, dwork, delay);
+	} else
+		mod_delayed_work(wq, dwork, delay);
+}
+EXPORT_SYMBOL_GPL(rds_mod_delayed_work);
+
 void rds_queue_cancel_work(struct rds_conn_path *cp,
 			   struct delayed_work *dwork, char *reason)
 {
@@ -236,11 +255,9 @@ void rds_queue_reconnect(struct rds_conn_path *cp, bool immediate)
 					       rds_sysctl_reconnect_max_jiffies);
 }
 
-void rds_connect_worker(struct work_struct *work)
+static void rds_connect_worker(struct rds_conn_path *cp,
+			       struct work_struct *work)
 {
-	struct rds_conn_path *cp = container_of(work,
-						struct rds_conn_path,
-						cp_conn_w.work);
 	struct rds_connection *conn = cp->cp_conn;
 	int ret;
 	bool is_tcp = conn->c_trans->t_type == RDS_TRANS_TCP;
@@ -394,11 +411,9 @@ void rds_hb_worker(struct work_struct *work)
 	}
 }
 
-void rds_shutdown_worker(struct work_struct *work)
+static void rds_shutdown_worker(struct rds_conn_path *cp,
+				struct work_struct *work)
 {
-	struct rds_conn_path *cp = container_of(work,
-						struct rds_conn_path,
-						cp_down_w);
 	time64_t now = ktime_get_real_seconds();
 	bool is_tcp = cp->cp_conn->c_trans->t_type == RDS_TRANS_TCP;
 	struct rds_connection *conn = cp->cp_conn;
@@ -420,6 +435,25 @@ void rds_shutdown_worker(struct work_struct *work)
 	rds_conn_init_shutdown(cp);
 }
 
+void rds_up_or_down_worker(struct work_struct *work)
+{
+	struct rds_conn_path *cp = container_of(work,
+						struct rds_conn_path,
+						cp_up_or_down_w.work);
+
+	/* shutdown-work always takes precedence over reconnect-work */
+	if (test_bit(RDS_SHUTDOWN_WORK_QUEUED, &cp->cp_flags)) {
+		/* reconnect is always re-scheduled in
+		 * rds_conn_shutdown_final(), if necessary
+		 */
+		rds_clear_reconnect_pending_work_bit(cp);
+
+		rds_shutdown_worker(cp, work);
+	} else if (test_bit(RDS_RECONNECT_PENDING, &cp->cp_flags)) {
+		rds_connect_worker(cp, work);
+	}
+}
+
 void rds_threads_exit(void)
 {
 	destroy_workqueue(rds_wq);