Date:   Thu, 11 Nov 2021 16:08:23 +0100
From:   Peter Zijlstra <peterz@...radead.org>
To:     Waiman Long <longman@...hat.com>
Cc:     Hillf Danton <hdanton@...a.com>,
        马振华 <mazhenhua@...omi.com>,
        mingo <mingo@...hat.com>, will <will@...nel.org>,
        "boqun.feng" <boqun.feng@...il.com>,
        linux-kernel <linux-kernel@...r.kernel.org>
Subject: Re: [BUG]locking/rwsem: only clean RWSEM_FLAG_HANDOFF when already
 set

On Wed, Nov 10, 2021 at 10:38:55PM +0100, Peter Zijlstra wrote:
> On Sun, Nov 07, 2021 at 02:52:36PM -0500, Waiman Long wrote:
> > > 
> > > I did have a tentative patch to address this issue which is somewhat
> > > similar to your approach. However, I would like to further investigate
> > > the exact mechanics of the race condition to make sure that I won't miss
> > > a latent bug somewhere else in the rwsem code.
> > 
> > I still couldn't figure how this race condition can happen. However, I do
> > discover that it is possible to leave rwsem with no waiter but handoff bit
> > set if we kill or interrupt all the waiters in the wait queue. I have just
> > sent out a patch to address that concern, but it should be able to handle
> > this race condition as well if it really happens.
> 
> The comment above RWSEM_WRITER_LOCKED seems wrong/out-dated in that
> there's a 4th place that modifies the HANDOFF bit namely
> rwsem_down_read_slowpath() in the out_nolock: case.
> 
> Now the thing I'm most worried about is that rwsem_down_write_slowpath()
> modifies the HANDOFF bit depending on wstate, and wstate itself it not
> determined under the same ->wait_lock section, so there could be a race
> there.
> 
> Another thing is that once wstate==HANDOFF, we rely on spin_on_owner()
> to return OWNER_NULL such that it goes to trylock_again, however if it
> returns anything else then we're at signal_pending_state() and the
> observed race can happen.
> 
> Now, spin_on_owner() *can* in fact return something else, consider
> need_resched() being set for instance.
> 
> Combined I think the observed race is valid.
> 
> Now before we go make things more complicated, I think we should see if
> we can make things simpler. Also I think perhaps the HANDOFF name here
> is a misnomer.
> 
> I agree that using _andnot() will fix this issue; I also agree with
> folding it with the existing _andnot() already there. But let me stare a
> little more at this code, something isn't making sense...

I think I want to see WRITER_HANDOFF go away. And preferably all of
wstate.

Something like the *completely* untested below, might set fire to your
pet, eat your granny, etc..

Also, perhaps s/HANDOFF/PHASE_CHANGE/ ?

Waiman, did I overlook something fundamental here?

---
 kernel/locking/rwsem.c | 85 +++++++++++++++++++-------------------------------
 1 file changed, 32 insertions(+), 53 deletions(-)

diff --git a/kernel/locking/rwsem.c b/kernel/locking/rwsem.c
index c51387a43265..bc5da05346e2 100644
--- a/kernel/locking/rwsem.c
+++ b/kernel/locking/rwsem.c
@@ -104,14 +104,17 @@
  * atomic_long_fetch_add() is used to obtain reader lock, whereas
  * atomic_long_cmpxchg() will be used to obtain writer lock.
  *
- * There are three places where the lock handoff bit may be set or cleared.
- * 1) rwsem_mark_wake() for readers.
- * 2) rwsem_try_write_lock() for writers.
- * 3) Error path of rwsem_down_write_slowpath().
+ * There are four places where the lock handoff bit may be set or cleared.
+ * 1) rwsem_mark_wake() for readers		-- set, clear
+ * 2) rwsem_try_write_lock() for writers	-- set, clear
+ * 3) Error path of rwsem_down_write_slowpath() -- clear
+ * 4) Error path of rwsem_down_read_slowpath()  -- clear
  *
  * For all the above cases, wait_lock will be held. A writer must also
  * be the first one in the wait_list to be eligible for setting the handoff
  * bit. So concurrent setting/clearing of handoff bit is not possible.
+ *
+ * XXX handoff is a misnomer here, all it does it force a phase change
  */
 #define RWSEM_WRITER_LOCKED	(1UL << 0)
 #define RWSEM_FLAG_WAITERS	(1UL << 1)
@@ -344,12 +347,6 @@ enum rwsem_wake_type {
 	RWSEM_WAKE_READ_OWNED	/* Waker thread holds the read lock */
 };
 
-enum writer_wait_state {
-	WRITER_NOT_FIRST,	/* Writer is not first in wait list */
-	WRITER_FIRST,		/* Writer is first in wait list     */
-	WRITER_HANDOFF		/* Writer is first & handoff needed */
-};
-
 /*
  * The typical HZ value is either 250 or 1000. So set the minimum waiting
  * time to at least 4ms or 1 jiffy (if it is higher than 4ms) in the wait
@@ -531,13 +528,12 @@ static void rwsem_mark_wake(struct rw_semaphore *sem,
  * This function must be called with the sem->wait_lock held to prevent
  * race conditions between checking the rwsem wait list and setting the
  * sem->count accordingly.
- *
- * If wstate is WRITER_HANDOFF, it will make sure that either the handoff
- * bit is set or the lock is acquired with handoff bit cleared.
  */
 static inline bool rwsem_try_write_lock(struct rw_semaphore *sem,
-					enum writer_wait_state wstate)
+					struct rwsem_waiter *waiter)
 {
+	bool first = rwsem_first_waiter(sem) == waiter;
+	bool timo = time_after(jiffies, waiter->timeout);
 	long count, new;
 
 	lockdep_assert_held(&sem->wait_lock);
@@ -546,13 +542,13 @@ static inline bool rwsem_try_write_lock(struct rw_semaphore *sem,
 	do {
 		bool has_handoff = !!(count & RWSEM_FLAG_HANDOFF);
 
-		if (has_handoff && wstate == WRITER_NOT_FIRST)
+		if (has_handoff && !first)
 			return false;
 
 		new = count;
 
 		if (count & RWSEM_LOCK_MASK) {
-			if (has_handoff || (wstate != WRITER_HANDOFF))
+			if (has_handoff || !(first && timo))
 				return false;
 
 			new |= RWSEM_FLAG_HANDOFF;
@@ -707,6 +703,8 @@ rwsem_spin_on_owner(struct rw_semaphore *sem)
 			break;
 		}
 
+		// XXX also terminate on signal_pending_state(current->__state, current) ?
+
 		cpu_relax();
 	}
 
@@ -1019,11 +1017,10 @@ rwsem_down_read_slowpath(struct rw_semaphore *sem, long count, unsigned int stat
 static struct rw_semaphore *
 rwsem_down_write_slowpath(struct rw_semaphore *sem, int state)
 {
-	long count;
-	enum writer_wait_state wstate;
 	struct rwsem_waiter waiter;
 	struct rw_semaphore *ret = sem;
 	DEFINE_WAKE_Q(wake_q);
+	long count, flags = 0;
 
 	/* do optimistic spinning and steal lock if possible */
 	if (rwsem_can_spin_on_owner(sem) && rwsem_optimistic_spin(sem)) {
@@ -1041,13 +1038,10 @@ rwsem_down_write_slowpath(struct rw_semaphore *sem, int state)
 
 	raw_spin_lock_irq(&sem->wait_lock);
 
-	/* account for this before adding a new element to the list */
-	wstate = list_empty(&sem->wait_list) ? WRITER_FIRST : WRITER_NOT_FIRST;
-
 	list_add_tail(&waiter.list, &sem->wait_list);
 
 	/* we're now waiting on the lock */
-	if (wstate == WRITER_NOT_FIRST) {
+	if (rwsem_first_waiter(sem) != &waiter) {
 		count = atomic_long_read(&sem->count);
 
 		/*
@@ -1083,22 +1077,14 @@ rwsem_down_write_slowpath(struct rw_semaphore *sem, int state)
 	/* wait until we successfully acquire the lock */
 	set_current_state(state);
 	for (;;) {
-		if (rwsem_try_write_lock(sem, wstate)) {
+		if (rwsem_try_write_lock(sem, &waiter)) {
 			/* rwsem_try_write_lock() implies ACQUIRE on success */
 			break;
 		}
 
 		raw_spin_unlock_irq(&sem->wait_lock);
 
-		/*
-		 * After setting the handoff bit and failing to acquire
-		 * the lock, attempt to spin on owner to accelerate lock
-		 * transfer. If the previous owner is a on-cpu writer and it
-		 * has just released the lock, OWNER_NULL will be returned.
-		 * In this case, we attempt to acquire the lock again
-		 * without sleeping.
-		 */
-		if (wstate == WRITER_HANDOFF) {
+		if (rwsem_first_waiter(sem) == &waiter) {
 			enum owner_state owner_state;
 
 			preempt_disable();
@@ -1117,28 +1103,15 @@ rwsem_down_write_slowpath(struct rw_semaphore *sem, int state)
 			schedule();
 			lockevent_inc(rwsem_sleep_writer);
 			set_current_state(state);
-			/*
-			 * If HANDOFF bit is set, unconditionally do
-			 * a trylock.
-			 */
-			if (wstate == WRITER_HANDOFF)
-				break;
 
-			if ((wstate == WRITER_NOT_FIRST) &&
-			    (rwsem_first_waiter(sem) == &waiter))
-				wstate = WRITER_FIRST;
+			if (rwsem_first_waiter(sem) != &waiter)
+				continue;
 
 			count = atomic_long_read(&sem->count);
 			if (!(count & RWSEM_LOCK_MASK))
 				break;
 
-			/*
-			 * The setting of the handoff bit is deferred
-			 * until rwsem_try_write_lock() is called.
-			 */
-			if ((wstate == WRITER_FIRST) && (rt_task(current) ||
-			    time_after(jiffies, waiter.timeout))) {
-				wstate = WRITER_HANDOFF;
+			if (rt_task(current) || time_after(jiffies, waiter.timeout)) {
 				lockevent_inc(rwsem_wlock_handoff);
 				break;
 			}
@@ -1156,15 +1129,21 @@ rwsem_down_write_slowpath(struct rw_semaphore *sem, int state)
 out_nolock:
 	__set_current_state(TASK_RUNNING);
 	raw_spin_lock_irq(&sem->wait_lock);
-	list_del(&waiter.list);
 
-	if (unlikely(wstate == WRITER_HANDOFF))
-		atomic_long_add(-RWSEM_FLAG_HANDOFF,  &sem->count);
+	if (rwsem_first_waiter(sem) == &waiter)
+		flags |= RWSEM_FLAG_HANDOFF;
+
+	list_del(&waiter.list);
 
 	if (list_empty(&sem->wait_list))
-		atomic_long_andnot(RWSEM_FLAG_WAITERS, &sem->count);
-	else
+		flags |= RWSEM_FLAG_WAITERS | RWSEM_FLAG_HANDOFF;
+
+	if (flags)
+		atomic_long_andnot(flags, &sem->count);
+
+	if (!list_empty(&sem->wait_list))
 		rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);
+
 	raw_spin_unlock_irq(&sem->wait_lock);
 	wake_up_q(&wake_q);
 	lockevent_inc(rwsem_wlock_fail);

Powered by blists - more mailing lists