[PATCH 1/2] futex/requeue: Fix rtmutex schedule preparation for requeue PI

[Yao Kai <yaokai34@huawei.com>]

A waiter requeued onto a PI futex can reach
rt_mutex_wait_proxy_lock() without rtmutex schedule preparation:

  WARNING: CPU: 0 PID: 293 at kernel/sched/core.c:7606
  RIP: rt_mutex_schedule+0x43/0x50
  Call Trace:
   rt_mutex_slowlock_block.constprop.0+0x5b/0x320
   rt_mutex_wait_proxy_lock+0x3e/0x80
   futex_wait_requeue_pi+0x3ba/0x590
   do_futex+0x171/0x1f0

rt_mutex_schedule() requires current->sched_rt_mutex to be set. Normally,
rt_mutex_pre_schedule() sets it and submits pending work before the waiter
is enqueued. With requeue PI, another task can enqueue the waiter after its
futex_q becomes visible:

        waiter                          requeue task
        ------                          ------------
futex_wait_requeue_pi()
  futex_wait_setup()
    futex_queue(&q)
                                        futex_requeue()
                                          rt_mutex_start_proxy_lock()
                                            enqueue rt_waiter
                                            install pi_blocked_on
                                          requeue_futex()
                                            plist_del(&q->list)
  futex_do_wait()
    plist_node_empty(&q->list)
    skip schedule()
                                            plist_add(&q->list)
                                          futex_requeue_pi_complete()
                                            IN_PROGRESS -> DONE
  futex_requeue_pi_wakeup_sync() // DONE
  rt_mutex_wait_proxy_lock()
    rt_mutex_schedule()

futex_do_wait() mistakes the temporary removal for a wakeup and skips
schedule(). The waiter consequently enters rt_mutex_schedule() with
current->sched_rt_mutex clear. Its block plug remains unflushed, and
workqueue or io-wq users are not notified that the worker is going to
sleep.

rt_mutex_pre_schedule() cannot be used directly. Calling it before q is
published would set current->sched_rt_mutex while futex_do_wait() can still
call regular schedule(). Calling it after DONE would flush the block plug
after current->pi_blocked_on is installed, which can recurse into rtmutex
waiter setup.

Split the preparation instead. Flush the plug before futex_wait_setup()
publishes q. After futex_requeue_pi_wakeup_sync() returns DONE, enter the
rtmutex scheduling state and notify workqueue and io-wq users without
flushing the plug again. Split sched_submit_work() to support this ordering
and leave the scheduling state after the waiter has acquired the lock or
has been removed.

Fixes: d14f9e930b90 ("locking/rtmutex: Use rt_mutex specific scheduler helpers")
Cc: stable@vger.kernel.org
Signed-off-by: Yao Kai <yaokai34@huawei.com>
---
 include/linux/sched/rt.h |  2 ++
 kernel/futex/requeue.c   | 15 +++++++++++++-
 kernel/sched/core.c      | 45 +++++++++++++++++++++++++++++++++-------
 3 files changed, 53 insertions(+), 9 deletions(-)

diff --git a/include/linux/sched/rt.h b/include/linux/sched/rt.h
index 4e3338103654..3d8cbdbfe14b 100644
--- a/include/linux/sched/rt.h
+++ b/include/linux/sched/rt.h
@@ -51,7 +51,9 @@ static inline bool rt_or_dl_task_policy(struct task_struct *tsk)
 }
 
 #ifdef CONFIG_RT_MUTEXES
+void rt_mutex_pre_schedule_flush_plug(void);
 extern void rt_mutex_pre_schedule(void);
+void rt_mutex_pre_schedule_pi_blocked(void);
 extern void rt_mutex_schedule(void);
 extern void rt_mutex_post_schedule(void);
 
diff --git a/kernel/futex/requeue.c b/kernel/futex/requeue.c
index 79823ad13683..abc652b5b2dd 100644
--- a/kernel/futex/requeue.c
+++ b/kernel/futex/requeue.c
@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: GPL-2.0-or-later
 
 #include <linux/plist.h>
+#include <linux/sched/rt.h>
 #include <linux/sched/signal.h>
 
 #include "futex.h"
@@ -820,6 +821,13 @@ int futex_wait_requeue_pi(u32 __user *uaddr, unsigned int flags,
 	q.rt_waiter = &rt_waiter;
 	q.requeue_pi_key = &key2;
 
+	/*
+	 * Once q is visible, requeue can enqueue rt_waiter and install
+	 * current->pi_blocked_on on our behalf. Flush plugged I/O before
+	 * that can happen.
+	 */
+	rt_mutex_pre_schedule_flush_plug();
+
 	/*
 	 * Prepare to wait on uaddr. On success, it holds hb->lock and q
 	 * is initialized.
@@ -865,6 +873,11 @@ int futex_wait_requeue_pi(u32 __user *uaddr, unsigned int flags,
 	case Q_REQUEUE_PI_DONE:
 		/* Requeue completed. Current is 'pi_blocked_on' the rtmutex */
 		pi_mutex = &q.pi_state->pi_mutex;
+		/*
+		 * The rt_waiter is already enqueued and the plug is flushed.
+		 * Enter schedule state and notify worker users.
+		 */
+		rt_mutex_pre_schedule_pi_blocked();
 		ret = rt_mutex_wait_proxy_lock(pi_mutex, to, &rt_waiter);
 
 		/*
@@ -875,6 +888,7 @@ int futex_wait_requeue_pi(u32 __user *uaddr, unsigned int flags,
 
 		futex_q_lockptr_lock(&q);
 		debug_rt_mutex_free_waiter(&rt_waiter);
+		rt_mutex_post_schedule();
 		/*
 		 * Fixup the pi_state owner and possibly acquire the lock if we
 		 * haven't already.
@@ -915,4 +929,3 @@ int futex_wait_requeue_pi(u32 __user *uaddr, unsigned int flags,
 	}
 	return ret;
 }
-
diff --git a/kernel/sched/core.c b/kernel/sched/core.c
index 96226707c2f6..f481e49e16b8 100644
--- a/kernel/sched/core.c
+++ b/kernel/sched/core.c
@@ -7257,17 +7257,12 @@ void __noreturn do_task_dead(void)
 		cpu_relax();
 }
 
-static inline void sched_submit_work(struct task_struct *tsk)
+static DEFINE_WAIT_OVERRIDE_MAP(sched_map, LD_WAIT_CONFIG);
+
+static inline void sched_submit_work_notifiers(struct task_struct *tsk)
 {
-	static DEFINE_WAIT_OVERRIDE_MAP(sched_map, LD_WAIT_CONFIG);
 	unsigned int task_flags;
 
-	/*
-	 * Establish LD_WAIT_CONFIG context to ensure none of the code called
-	 * will use a blocking primitive -- which would lead to recursion.
-	 */
-	lock_map_acquire_try(&sched_map);
-
 	task_flags = tsk->flags;
 	/*
 	 * If a worker goes to sleep, notify and ask workqueue whether it
@@ -7277,7 +7272,10 @@ static inline void sched_submit_work(struct task_struct *tsk)
 		wq_worker_sleeping(tsk);
 	else if (task_flags & PF_IO_WORKER)
 		io_wq_worker_sleeping(tsk);
+}
 
+static inline void sched_submit_work_plug(struct task_struct *tsk)
+{
 	/*
 	 * spinlock and rwlock must not flush block requests.  This will
 	 * deadlock if the callback attempts to acquire a lock which is
@@ -7290,6 +7288,18 @@ static inline void sched_submit_work(struct task_struct *tsk)
 	 * make sure to submit it to avoid deadlocks.
 	 */
 	blk_flush_plug(tsk->plug, true);
+}
+
+static inline void sched_submit_work(struct task_struct *tsk)
+{
+	/*
+	 * Establish LD_WAIT_CONFIG context to ensure none of the code called
+	 * will use a blocking primitive -- which would lead to recursion.
+	 */
+	lock_map_acquire_try(&sched_map);
+
+	sched_submit_work_notifiers(tsk);
+	sched_submit_work_plug(tsk);
 
 	lock_map_release(&sched_map);
 }
@@ -7595,12 +7605,31 @@ const struct sched_class *__setscheduler_class(int policy, int prio)
  */
 #define fetch_and_set(x, v) ({ int _x = (x); (x) = (v); _x; })
 
+void rt_mutex_pre_schedule_flush_plug(void)
+{
+	lockdep_assert(!current->pi_blocked_on);
+	lock_map_acquire_try(&sched_map);
+	sched_submit_work_plug(current);
+	lock_map_release(&sched_map);
+}
+
 void rt_mutex_pre_schedule(void)
 {
 	lockdep_assert(!fetch_and_set(current->sched_rt_mutex, 1));
 	sched_submit_work(current);
 }
 
+void rt_mutex_pre_schedule_pi_blocked(void)
+{
+	lockdep_assert(current->pi_blocked_on);
+	lockdep_assert(!fetch_and_set(current->sched_rt_mutex, 1));
+
+	/* The plug was flushed before current->pi_blocked_on was installed. */
+	lock_map_acquire_try(&sched_map);
+	sched_submit_work_notifiers(current);
+	lock_map_release(&sched_map);
+}
+
 void rt_mutex_schedule(void)
 {
 	lockdep_assert(current->sched_rt_mutex);
-- 
2.43.0