Re: [PATCH v4] locking/rwsem: Make handoff bit handling more consistent

[Peter Zijlstra <peterz@infradead.org>]

On Thu, Nov 11, 2021 at 11:07:53PM -0500, Waiman Long wrote:
> @@ -889,6 +892,20 @@ rwsem_spin_on_owner(struct rw_semaphore *sem)
>  }
>  #endif
>  
> +/*
> + * Common code to handle rwsem flags in out_nolock path with wait_lock held.
> + * If there is more than one waiter in the queue and the HANDOFF bit is set,
> + * the next waiter will inherit it if the first waiter is removed.
> + */
> +static inline void rwsem_out_nolock_clear_flags(struct rw_semaphore *sem,
> +						struct rwsem_waiter *waiter)

I'm going to rename that, it doesn't just clear flags, it dequeues the
waiter.

Argh, rwsem_mark_wake() doesn't clear HANDOFF when list_empty(), and
write_slowpath() is *far* too clever about all of this.


> +{
> +	list_del(&waiter->list);
	if (list_empty(&sem->wait_list)) {
> +		atomic_long_andnot(RWSEM_FLAG_HANDOFF | RWSEM_FLAG_WAITERS,
> +				   &sem->count);
	}
> +}



> @@ -1098,7 +1110,7 @@ rwsem_down_write_slowpath(struct rw_semaphore *sem, int state)
>  		 * In this case, we attempt to acquire the lock again
>  		 * without sleeping.
>  		 */
> -		if (wstate == WRITER_HANDOFF) {
> +		if (waiter.handoff_set) {

I'm thinking this wants to be something like:

		if (rwsem_first_waiter(sem) == &waiter && waiter.handoff_set) {

>  			enum owner_state owner_state;
>  
>  			preempt_disable();

How's this (on top) then?

---
--- a/kernel/locking/rwsem.c
+++ b/kernel/locking/rwsem.c
@@ -104,11 +104,10 @@
  * atomic_long_fetch_add() is used to obtain reader lock, whereas
  * atomic_long_cmpxchg() will be used to obtain writer lock.
  *
- * 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
+ * There are three 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) rwsem_del_waiter()			-- 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
@@ -363,6 +362,31 @@ enum rwsem_wake_type {
  */
 #define MAX_READERS_WAKEUP	0x100
 
+static inline void
+rwsem_add_waiter(struct rw_semaphore *sem, struct rwsem_waiter *waiter)
+{
+	lockdep_assert_held(&sem->wait_lock);
+	list_add_tail(&waiter->list, &sem->wait_list);
+	/* caller will set RWSEM_FLAG_WAITERS */
+}
+
+/*
+ * Remove a waiter from the wait_list and clear flags.
+ *
+ * Both rwsem_mark_wake() and rwsem_try_write_lock() contain a full 'copy' of
+ * this function. Modify with care.
+ */
+static inline void
+rwsem_del_waiter(struct rw_semaphore *sem, struct rwsem_waiter *waiter)
+{
+	lockdep_assert_held(&sem->wait_lock);
+	list_del(&waiter->list);
+	if (likely(!list_empty(&sem->wait_list)))
+		return;
+
+	atomic_long_andnot(RWSEM_FLAG_HANDOFF | RWSEM_FLAG_WAITERS, &sem->count);
+}
+
 /*
  * handle the lock release when processes blocked on it that can now run
  * - if we come here from up_xxxx(), then the RWSEM_FLAG_WAITERS bit must
@@ -374,6 +398,8 @@ enum rwsem_wake_type {
  *   preferably when the wait_lock is released
  * - woken process blocks are discarded from the list after having task zeroed
  * - writers are only marked woken if downgrading is false
+ *
+ * Implies rwsem_del_waiter() for all woken readers.
  */
 static void rwsem_mark_wake(struct rw_semaphore *sem,
 			    enum rwsem_wake_type wake_type,
@@ -488,18 +514,25 @@ static void rwsem_mark_wake(struct rw_se
 
 	adjustment = woken * RWSEM_READER_BIAS - adjustment;
 	lockevent_cond_inc(rwsem_wake_reader, woken);
+
+	oldcount = atomic_long_read(&sem->count);
 	if (list_empty(&sem->wait_list)) {
-		/* hit end of list above */
+		/*
+		 * Combined with list_move_tail() above, this implies
+		 * rwsem_del_waiter().
+		 */
 		adjustment -= RWSEM_FLAG_WAITERS;
+		if (oldcount & RWSEM_FLAG_HANDOFF)
+			adjustment -= RWSEM_FLAG_HANDOFF;
+	} else if (woken) {
+		/*
+		 * When we've woken a reader, we no longer need to force
+		 * writers to give up the lock and we can clear HANDOFF.
+		 */
+		if (oldcount & RWSEM_FLAG_HANDOFF)
+			adjustment -= RWSEM_FLAG_HANDOFF;
 	}
 
-	/*
-	 * When we've woken a reader, we no longer need to force writers
-	 * to give up the lock and we can clear HANDOFF.
-	 */
-	if (woken && (atomic_long_read(&sem->count) & RWSEM_FLAG_HANDOFF))
-		adjustment -= RWSEM_FLAG_HANDOFF;
-
 	if (adjustment)
 		atomic_long_add(adjustment, &sem->count);
 
@@ -529,6 +562,8 @@ static void rwsem_mark_wake(struct rw_se
  * 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.
+ *
+ * Implies rwsem_del_waiter() on success.
  */
 static inline bool rwsem_try_write_lock(struct rw_semaphore *sem,
 					struct rwsem_waiter *waiter)
@@ -575,6 +610,11 @@ static inline bool rwsem_try_write_lock(
 		return false;
 	}
 
+	/*
+	 * Have rwsem_try_write_lock() fully imply rwsem_del_waiter() on
+	 * success.
+	 */
+	list_del(&waiter->list);
 	rwsem_set_owner(sem);
 	return true;
 }
@@ -893,20 +933,6 @@ rwsem_spin_on_owner(struct rw_semaphore
 #endif
 
 /*
- * Common code to handle rwsem flags in out_nolock path with wait_lock held.
- * If there is more than one waiter in the queue and the HANDOFF bit is set,
- * the next waiter will inherit it if the first waiter is removed.
- */
-static inline void rwsem_out_nolock_clear_flags(struct rw_semaphore *sem,
-						struct rwsem_waiter *waiter)
-{
-	list_del(&waiter->list);
-	if (list_empty(&sem->wait_list))
-		atomic_long_andnot(RWSEM_FLAG_HANDOFF | RWSEM_FLAG_WAITERS,
-				   &sem->count);
-}
-
-/*
  * Wait for the read lock to be granted
  */
 static struct rw_semaphore __sched *
@@ -973,7 +999,7 @@ rwsem_down_read_slowpath(struct rw_semap
 		}
 		adjustment += RWSEM_FLAG_WAITERS;
 	}
-	list_add_tail(&waiter.list, &sem->wait_list);
+	rwsem_add_waiter(sem, &waiter);
 
 	/* we're now waiting on the lock, but no longer actively locking */
 	count = atomic_long_add_return(adjustment, &sem->count);
@@ -1019,7 +1045,7 @@ rwsem_down_read_slowpath(struct rw_semap
 	return sem;
 
 out_nolock:
-	rwsem_out_nolock_clear_flags(sem, &waiter);
+	rwsem_del_waiter(sem, &waiter);
 	raw_spin_unlock_irq(&sem->wait_lock);
 	__set_current_state(TASK_RUNNING);
 	lockevent_inc(rwsem_rlock_fail);
@@ -1034,7 +1060,6 @@ rwsem_down_write_slowpath(struct rw_sema
 {
 	long count;
 	struct rwsem_waiter waiter;
-	struct rw_semaphore *ret = sem;
 	DEFINE_WAKE_Q(wake_q);
 
 	/* do optimistic spinning and steal lock if possible */
@@ -1053,7 +1078,7 @@ rwsem_down_write_slowpath(struct rw_sema
 	waiter.handoff_set = false;
 
 	raw_spin_lock_irq(&sem->wait_lock);
-	list_add_tail(&waiter.list, &sem->wait_list);
+	rwsem_add_waiter(sem, &waiter);
 
 	/* we're now waiting on the lock */
 	if (rwsem_first_waiter(sem) != &waiter) {
@@ -1110,7 +1135,7 @@ rwsem_down_write_slowpath(struct rw_sema
 		 * In this case, we attempt to acquire the lock again
 		 * without sleeping.
 		 */
-		if (waiter.handoff_set) {
+		if (rwsem_first_waiter(sem) == &waiter && waiter.handoff_set) {
 			enum owner_state owner_state;
 
 			preempt_disable();
@@ -1128,16 +1153,14 @@ rwsem_down_write_slowpath(struct rw_sema
 		raw_spin_lock_irq(&sem->wait_lock);
 	}
 	__set_current_state(TASK_RUNNING);
-	list_del(&waiter.list);
 	raw_spin_unlock_irq(&sem->wait_lock);
 	lockevent_inc(rwsem_wlock);
-
-	return ret;
+	return sem;
 
 out_nolock:
 	__set_current_state(TASK_RUNNING);
 	raw_spin_lock_irq(&sem->wait_lock);
-	rwsem_out_nolock_clear_flags(sem, &waiter);
+	rwsem_del_waiter(sem, &waiter);
 	if (!list_empty(&sem->wait_list))
 		rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);
 	raw_spin_unlock_irq(&sem->wait_lock);