Skip to contentSkip to conversation

locking/rwsem: Make handoff bit handling more consistent

PATCH Superseded Waiman Long 3 participants 9 messages Nov 12, 2021 – Nov 16, 2021 Locking primitives

v4 of 5 Superseded Latest is v5 →

lore
Share & tools
lore

Waiman Long

author reviewer
#0 lore raw

PATCH v4 1/1 locking/rwsem: Make handoff bit handling more consistent

There are some inconsistency in the way that the handoff bit is being handled in readers and writers.

Firstly, when a queue head writer set the handoff bit, it will clear it when the writer is being killed or interrupted on its way out without acquiring the lock. That is not the case for a queue head reader. The handoff bit will simply be inherited by the next waiter.

Secondly, in the out_nolock path of rwsem_down_read_slowpath(), both the waiter and handoff bits are cleared if the wait queue becomes empty. For rwsem_down_write_slowpath(), however, the handoff bit is not checked and cleared if the wait queue is empty. This can potentially make the handoff bit set with empty wait queue.

To make the handoff bit handling more consistent and robust, extract out the rwsem flags handling code into a common rwsem_out_nolock_clear_flags() function and call it from both the reader and writer's out_nolock paths. The common function will only use atomic_long_andnot() to clear bits to avoid possible race condition. It will also let the next waiter inherit the handoff bit if it had been set.

This will elminate the handoff bit set with empty wait queue case as well as the possible race condition that may screw up the count value.

While at it, simplify the trylock for loop in rwsem_down_write_slowpath() to make it easier to read.

Fixes 4f23dbc1e657 ("locking/rwsem: Implement lock handoff to prevent lock starvation")
Suggested-by Peter Zijlstra <[email protected]>
Signed-off-by Waiman Long <[email protected]>
---
 kernel/locking/rwsem.c | 120 ++++++++++++++++-------------------------
 1 file changed, 47 insertions(+), 73 deletions(-)
kernel/locking/rwsem.c +47 -73
index c51387a43265..f376e6eeb36a 100644
@@ -104,10 +104,11 @@
  * 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

@@ -334,6 +335,9 @@ struct rwsem_waiter {
 	struct task_struct *task;
 	enum rwsem_waiter_type type;
 	unsigned long timeout;
+
+	/* Writer only, not initialized in reader */
+	bool handoff_set;
 };
 #define rwsem_first_waiter(sem) \
 	list_first_entry(&sem->wait_list, struct rwsem_waiter, list)

@@ -344,12 +348,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 +529,11 @@ 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;
 	long count, new;
 
 	lockdep_assert_held(&sem->wait_lock);

@@ -546,13 +542,17 @@ 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))
+			/* May have inherited a previously set handoff bit */
+			waiter->handoff_set = has_handoff;
+
+			if (has_handoff || (!rt_task(waiter->task) &&
+					    !time_after(jiffies, waiter->timeout)))
 				return false;
 
 			new |= RWSEM_FLAG_HANDOFF;

@@ -569,8 +569,11 @@ static inline bool rwsem_try_write_lock(struct rw_semaphore *sem,
 	 * We have either acquired the lock with handoff bit cleared or
 	 * set the handoff bit.
 	 */
-	if (new & RWSEM_FLAG_HANDOFF)
+	if (new & RWSEM_FLAG_HANDOFF) {
+		waiter->handoff_set = true;
+		lockevent_inc(rwsem_wlock_handoff);
 		return false;
+	}
 
 	rwsem_set_owner(sem);
 	return true;

@@ -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)
+{
+	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
  */

@@ -1002,11 +1019,7 @@ rwsem_down_read_slowpath(struct rw_semaphore *sem, long count, unsigned int stat
 	return sem;
 
 out_nolock:
-	list_del(&waiter.list);
-	if (list_empty(&sem->wait_list)) {
-		atomic_long_andnot(RWSEM_FLAG_WAITERS|RWSEM_FLAG_HANDOFF,
-				   &sem->count);
-	}
+	rwsem_out_nolock_clear_flags(sem, &waiter);
 	raw_spin_unlock_irq(&sem->wait_lock);
 	__set_current_state(TASK_RUNNING);
 	lockevent_inc(rwsem_rlock_fail);

@@ -1020,7 +1033,6 @@ 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);

@@ -1038,16 +1050,13 @@ rwsem_down_write_slowpath(struct rw_semaphore *sem, int state)
 	waiter.task = current;
 	waiter.type = RWSEM_WAITING_FOR_WRITE;
 	waiter.timeout = jiffies + RWSEM_WAIT_TIMEOUT;
+	waiter.handoff_set = false;
 
 	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,13 +1092,16 @@ 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);
 
+		if (signal_pending_state(state, current))
+			goto out_nolock;
+
 		/*
 		 * After setting the handoff bit and failing to acquire
 		 * the lock, attempt to spin on owner to accelerate lock

@@ -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) {
 			enum owner_state owner_state;
 
 			preempt_disable();

@@ -1109,40 +1121,9 @@ rwsem_down_write_slowpath(struct rw_semaphore *sem, int state)
 				goto trylock_again;
 		}
 
-		/* Block until there are no active lockers. */
-		for (;;) {
-			if (signal_pending_state(state, current))
-				goto out_nolock;
-
-			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;
-
-			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;
-				lockevent_inc(rwsem_wlock_handoff);
-				break;
-			}
-		}
+		schedule();
+		lockevent_inc(rwsem_sleep_writer);
+		set_current_state(state);
 trylock_again:
 		raw_spin_lock_irq(&sem->wait_lock);
 	}

@@ -1156,19 +1137,12 @@ 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 (list_empty(&sem->wait_list))
-		atomic_long_andnot(RWSEM_FLAG_WAITERS, &sem->count);
-	else
+	rwsem_out_nolock_clear_flags(sem, &waiter);
+	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);
-
 	return ERR_PTR(-EINTR);
 }
Signature · 1 line
2.27.0
maintainer
#1 lore raw

Review rwsem_spin_on_owner(struct rw_semaphore *sem) · line 901

Quoted 11 lines · Waiman Long
(quoted diff) +7 -0
@@ -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.

Review rwsem_spin_on_owner(struct rw_semaphore *sem) · line 903

Quoted 2 lines
(quoted diff) +2 -0
+{
+	list_del(&waiter->list);

	if (list_empty(&sem->wait_list)) {

Review rwsem_spin_on_owner(struct rw_semaphore *sem) · line 905

Quoted 2 lines
(quoted diff) +2 -0
+		atomic_long_andnot(RWSEM_FLAG_HANDOFF | RWSEM_FLAG_WAITERS,
+				   &sem->count);

	}

Review rwsem_down_write_slowpath(struct rw_semaphore *sem, int state) · line 1113

Quoted 10 lines
(quoted diff) +2 -1
+}
 
 
 

@@ -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) {

Review rwsem_down_write_slowpath(struct rw_semaphore *sem, int state) · line 1116

Quoted 3 lines
(quoted diff) +0 -0
 			enum owner_state owner_state;
 
 			preempt_disable();

How's this (on top) then?

---

kernel/locking/rwsem.c +59 -36
@@ -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);
author reviewer
#2 lore raw
Quoted 15 lines · Peter Zijlstra

Review rwsem_spin_on_owner(struct rw_semaphore *sem) · line 901

Quoted 11 lines · Waiman Long
(quoted diff) +7 -0
@@ -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.

Quoted 2 lines

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

rwsem_mark_wake() does clear the HANDOFF flag if it was set.

Quoted 19 lines

Review rwsem_spin_on_owner(struct rw_semaphore *sem) · line 903

Quoted 4 lines
(quoted diff) +2 -0
+{
+	list_del(&waiter->list);

	if (list_empty(&sem->wait_list)) {

Review rwsem_spin_on_owner(struct rw_semaphore *sem) · line 905

Quoted 2 lines
(quoted diff) +2 -0
+		atomic_long_andnot(RWSEM_FLAG_HANDOFF | RWSEM_FLAG_WAITERS,
+				   &sem->count);

	}

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_semaphore *sem, int state) · line 1113

Quoted 9 lines
kernel/locking/rwsem.c +2 -1
+}
 
 

@@ -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:

Quoted 1 line
		if (rwsem_first_waiter(sem) == &waiter && waiter.handoff_set) {

handoff_set flag is only set when the waiter becomes the first.

Quoted 187 lines

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_sema · line 1141

Quoted 4 lines
kernel/locking/rwsem.c +0 -0
  			enum owner_state owner_state;
 
  			preempt_disable();

How's this (on top) then?

---

kernel/locking/rwsem.c +57 -32
@@ -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();

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_sema · line 1131

Quoted 4 lines
kernel/locking/rwsem.c +0 -1
@@ -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);

+    rwsem_del_waiter(sem, &waiters); ?

Quoted 5 lines
kernel/locking/rwsem.c +0 -2
  	raw_spin_unlock_irq(&sem->wait_lock);
  	lockevent_inc(rwsem_wlock);
-
-	return ret;

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_sema · line 1166

Quoted 10 lines
kernel/locking/rwsem.c +2 -1
+	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);

Sorry for the late reply as I was busy on other works.

I like the idea. I will incorporate in a new patch tomorrow.

Cheers,
Longman

maintainer
#3 lore raw
Quoted 5 lines · Waiman Long
Quoted 2 lines · Peter Zijlstra

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

rwsem_mark_wake() does clear the HANDOFF flag if it was set.

Argh, yeah, I got confused by the whole !woken case, but that case won't ever hit list_empty() either. Perhaps that stuff could use a bit of a reflow too.

Quoted 10 lines
Quoted 9 lines

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_semaphore *sem, int state) · line 1113

Quoted 6 lines
kernel/locking/rwsem.c +1 -1
@@ -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) {

handoff_set flag is only set when the waiter becomes the first.

Yes, but a random waiter can wake up and see it be set and also start spinning.

Quoted 16 lines
Quoted 15 lines
Quoted 3 lines
kernel/locking/rwsem.c +0 -0
  			enum owner_state owner_state;
  			preempt_disable();
kernel/locking/rwsem.c +5 -0
@@ -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;
  }
Quoted 6 lines

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_sema · line 1131

Quoted 5 lines
kernel/locking/rwsem.c +0 -1
@@ -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);

+    rwsem_del_waiter(sem, &waiters); ?

I tried that, but then we get an extra atomic in this path. As is I made try_write_lock() do the full del_waiter, see the hunk above.

Quoted 16 lines
Quoted 4 lines
kernel/locking/rwsem.c +0 -2
  	raw_spin_unlock_irq(&sem->wait_lock);
  	lockevent_inc(rwsem_wlock);
-
-	return ret;

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_sema

Quoted 10 lines
kernel/locking/rwsem.c +2 -1
+	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);

Sorry for the late reply as I was busy on other works.

Quoted 1 line

I like the idea. I will incorporate in a new patch tomorrow.

Thanks!

author reviewer
#4 lore raw
Quoted 8 lines · Peter Zijlstra
Quoted 4 lines · Waiman Long
Quoted 2 lines · Peter Zijlstra

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

rwsem_mark_wake() does clear the HANDOFF flag if it was set.

Argh, yeah, I got confused by the whole !woken case, but that case won't ever hit list_empty() either. Perhaps that stuff could use a bit of a reflow too.

I think your modification already have included the rewrite for that part.

Quoted 9 lines
Quoted 9 lines
Quoted 9 lines

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_semaphore *sem, int state) · line 1113

Quoted 7 lines
kernel/locking/rwsem.c +1 -1
@@ -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:

Quoted 4 lines
Quoted 2 lines
Quoted 1 line
		if (rwsem_first_waiter(sem) == &waiter && waiter.handoff_set) {

handoff_set flag is only set when the waiter becomes the first.

Yes, but a random waiter can wake up and see it be set and also start spinning.

The handoff_set flag can only be true for a first waiter. A random waiter in the middle of a wait queue will never has this flag set.

This flag is set in two places in rwsem_try_write_lock():

1)

               if (has_handoff && !first)
                        return false;
                new = count;

                if (count & RWSEM_LOCK_MASK) {
                        /*
                         * Only the first waiter can inherit a previously set
                         * handoff bit.
                         */
                        waiter->handoff_set = has_handoff;

handoff_set can only be set to true here if first is also true. In that case, it will also return false immediately afterward.

2)

        if (new & RWSEM_FLAG_HANDOFF) { waiter->handoff_set = true; lockevent_inc(rwsem_wlock_handoff); return false;
        }

Again, only first waiter will have a chance of setting the handoff bit and have handoff_set set to true.

Quoted 19 lines
Quoted 11 lines
Quoted 11 lines
Quoted 2 lines
kernel/locking/rwsem.c +0 -0
   			enum owner_state owner_state;
   			preempt_disable();
kernel/locking/rwsem.c +5 -0
@@ -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);
Quoted 8 lines

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_sema · line 1131

Quoted 7 lines
kernel/locking/rwsem.c +0 -1
   	return true;
   }

@@ -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);

+    rwsem_del_waiter(sem, &waiters); ?

Quoted 2 lines

I tried that, but then we get an extra atomic in this path. As is I made try_write_lock() do the full del_waiter, see the hunk above.

You are right. I missed your change in rwsem_try_write_lock().

Thanks,
Longman

maintainer
#5 lore raw
Quoted 5 lines · Waiman Long

The handoff_set flag can only be true for a first waiter. A random waiter in the middle of a wait queue will never has this flag set.

This flag is set in two places in rwsem_try_write_lock():

Bah, I thought it would unconditionally propagate the bit from @count. I missed the early exit :/

author reviewer
#6 lore raw
Quoted 8 lines · Peter Zijlstra
Quoted 4 lines · Waiman Long

The handoff_set flag can only be true for a first waiter. A random waiter in the middle of a wait queue will never has this flag set.

This flag is set in two places in rwsem_try_write_lock():

Bah, I thought it would unconditionally propagate the bit from @count. I missed the early exit :/

I am going to restructure the code there to make it easier to see that only the first waiter will have this bit set to avoid this confusion.

Cheers,
Longman

#7 lore raw
In reply to Peter Zijlstra
Quoted 67 lines · Peter Zijlstra

Review rwsem_spin_on_owner(struct rw_semaphore *sem) · line 901

Quoted 11 lines · Waiman Long
(quoted diff) +7 -0
@@ -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.

Review rwsem_spin_on_owner(struct rw_semaphore *sem) · line 903

Quoted 2 lines
(quoted diff) +2 -0
+{
+	list_del(&waiter->list);

	if (list_empty(&sem->wait_list)) {

Review rwsem_spin_on_owner(struct rw_semaphore *sem) · line 905

Quoted 2 lines
(quoted diff) +2 -0
+		atomic_long_andnot(RWSEM_FLAG_HANDOFF | RWSEM_FLAG_WAITERS,
+				   &sem->count);

	}

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_semaphore *sem, int state) · line 1113

Quoted 10 lines
kernel/locking/rwsem.c +2 -1
+}
 
 
 

@@ -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) {

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_sema · line 1141

Quoted 3 lines
kernel/locking/rwsem.c +0 -0
  			enum owner_state owner_state;
 
  			preempt_disable();

How's this (on top) then?

---

kernel/locking/rwsem.c +4 -5
@@ -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

kernel/locking/rwsem.c enum rwsem_wake_type { · line 370

Quoted 9 lines
kernel/locking/rwsem.c +6 -0
@@ -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 */

	/* each time a waiter is just added in to the list,
	 * handoff_set initialed as false. */
	waiter->handoff_set = false;

kernel/locking/rwsem.c enum rwsem_wake_type { · line 383

Quoted 13 lines
kernel/locking/rwsem.c +13 -0
+}
+
+/*
+ * 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);

what about avoid unnecessary inherit of handoff bit for waiters?

if (waiter->handoff_set)

	atomic_long_andnot(RWSEM_FLAG_HANDOFF, &sem->count);
Quoted 148 lines
kernel/locking/rwsem.c +34 -30
+	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;

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_sema · line 1166

Quoted 10 lines
kernel/locking/rwsem.c +2 -1
+	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);

what about avoid unnecessary inherit of handoff bit for waiters?
1. when set handoff bit also set waiter.handoff_set as true;
2. when clear handoff bit also set waiter.handoff_set as false.
3. And rwsem_add_waiter initial as false;
4. And rwsem_del_waiter also can clear the handoff bit according to
waiter.handoff_set.

Because handoff bit can have better performance if correctly set and cleared.

Signature · 2 lines
Thx and BRs,
Aiqun(Maria) Yu
author reviewer
#8 lore raw
Quoted 75 lines · Aiqun(Maria) Yu
Quoted 74 lines · Peter Zijlstra
Quoted 5 lines · Waiman Long
(quoted diff) +0 -0
@@ -889,6 +892,20 @@ rwsem_spin_on_owner(struct rw_semaphore *sem)
   }
   #endif

  +/*
+ * Common code to handle rwsem flags in out_nolock path with

Quoted 8 lines

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.

Quoted 2 lines

+{
+    list_del(&waiter->list);

    if (list_empty(&sem->wait_list)) {

Quoted 2 lines

+ atomic_long_andnot(RWSEM_FLAG_HANDOFF | RWSEM_FLAG_WAITERS, +                   &sem->count);

    }

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_semaphore · line 1114

Quoted 11 lines
kernel/locking/rwsem.c +2 -1
+}

@@ -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) {

kernel/locking/rwsem.c rwsem_down_write_slowpath(struct rw_sema · line 1141

Quoted 2 lines
kernel/locking/rwsem.c +0 -0
               enum owner_state owner_state;
                 preempt_disable();

How's this (on top) then?

---

kernel/locking/rwsem.c +0 -1
@@ -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

kernel/locking/rwsem.c +0 -0
@@ -363,6 +362,31 @@ enum rwsem_wake_type {
    */
   #define MAX_READERS_WAKEUP    0x100

  +static inline void

Quoted 8 lines

kernel/locking/rwsem.c enum rwsem_wake_type { · line 370

Quoted 5 lines
kernel/locking/rwsem.c +5 -0
+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 */

    /* each time a waiter is just added in to the list,
     * handoff_set initialed as false. */ waiter->handoff_set = false;

waiter initialization is done at entry to the slowpath function. This helper function just insert the waiter into the wait list.

Quoted 16 lines
Quoted 9 lines
kernel/locking/rwsem.c +6 -0
+}
+
+/*
+ * 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.
+ */

kernel/locking/rwsem.c enum rwsem_wake_type { · line 383

Quoted 5 lines
kernel/locking/rwsem.c +5 -0
+static inline void
+rwsem_del_waiter(struct rw_semaphore *sem, struct rwsem_waiter *waiter)
+{
+    lockdep_assert_held(&sem->wait_lock);
+    list_del(&waiter->list);

what about avoid unnecessary inherit of handoff bit for waiters?

Quoted 2 lines

if (waiter->handoff_set) atomic_long_andnot(RWSEM_FLAG_HANDOFF, &sem->count);

We have decided that to let the next waiter inherit the handoff bit. So there is no need to clear it unless there is no more waiter in the queue.

Quoted 157 lines
Quoted 141 lines

+    if (likely(!list_empty(&sem->wait_list)))
+        return;
+
+    atomic_long_andnot(RWSEM_FLAG_HANDOFF | RWSEM_FLAG_WAITERS, &sem->count);

kernel/locking/rwsem.c +2 -0
+}
+

  /*
   * 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

kernel/locking/rwsem.c +17 -1
@@ -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)

kernel/locking/rwsem.c +0 -0
@@ -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;
  }

kernel/locking/rwsem.c +0 -1
@@ -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)

kernel/locking/rwsem.c +0 -8
-{
-    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 *

kernel/locking/rwsem.c +1 -1
@@ -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);

kernel/locking/rwsem.c +3 -6
@@ -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;
Quoted 9 lines

    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);

what about avoid unnecessary inherit of handoff bit for waiters?
    1. when set handoff bit also set waiter.handoff_set as  true;
    2. when clear handoff bit also set waiter.handoff_set as false.
    3. And rwsem_add_waiter initial as false;
    4. And rwsem_del_waiter also can clear the handoff bit according to waiter.handoff_set.

Quoted 2 lines

Because handoff bit can have better performance if correctly set and cleared.

Killing or interrupt a waiter to force it to quit is not considered a fast path operation. It is an exception rather than the rule. So performance consideration is less important here.

Cheers,
Longman