locking/rwsem: Rwsem rearchitecture part 2

+#define RWSEM_COUNT_LOCKED(c) ((c) & RWSEM_LOCK_MASK)

Quoted 1 line · Waiman Long

+#define RWSEM_COUNT_LOCKED(c) ((c) & RWSEM_LOCK_MASK)

The above doesn't seem to make it more readable or shorter.

Quoted 3 lines · Peter Zijlstra

Quoted 1 line · Waiman Long

+#define RWSEM_COUNT_LOCKED(c) ((c) & RWSEM_LOCK_MASK)

The above doesn't seem to make it more readable or shorter.

Fair enough. I can remove that macro.

Quoted 6 lines · Waiman Long

Quoted 3 lines · Peter Zijlstra

Quoted 1 line · Waiman Long

+#define RWSEM_COUNT_LOCKED(c) ((c) & RWSEM_LOCK_MASK)

The above doesn't seem to make it more readable or shorter.

Fair enough. I can remove that macro.

I did the same for the HANDOFF patch but seem to have misplaced the delta and have already refreshed the patch.

kernel/locking/rwsem.c static inline int __down_read_trylock(st · line 776

Quoted 11 lines · Peter Zijlstra

kernel/locking/rwsem.c +4 -3

@@ -794,34 +770,38 @@ static inline int __down_read_trylock(st
  */
 static inline void __down_write(struct rw_semaphore *sem)
 {
-	if (unlikely(atomic_long_cmpxchg_acquire(&sem->count, 0,
-						 RWSEM_WRITER_LOCKED)))
-		rwsem_down_write_failed(sem);
+	long tmp = RWSEM_UNLOCKED_VALUE;
+
+	if (unlikely(atomic_long_try_cmpxchg_acquire(&sem->count, &tmp,
+						     RWSEM_WRITER_LOCKED)))

!

kernel/locking/rwsem.c static inline int __down_read_trylock(st · line 786

Quoted 13 lines

kernel/locking/rwsem.c +5 -3

+		rwsem_down_write_slow(sem, TASK_UNINTERRUPTIBLE);
 	rwsem_set_owner(sem);
 }
 
 static inline int __down_write_killable(struct rw_semaphore *sem)
 {
-	if (unlikely(atomic_long_cmpxchg_acquire(&sem->count, 0,
-						 RWSEM_WRITER_LOCKED)))
-		if (IS_ERR(rwsem_down_write_failed_killable(sem)))
+	long tmp = RWSEM_UNLOCKED_VALUE;
+
+	if (unlikely(atomic_long_try_cmpxchg_acquire(&sem->count, &tmp,
+						     RWSEM_WRITER_LOCKED))) {

also !

kernel/locking/rwsem.c static inline int __down_read_trylock(st · line 792

Quoted 6 lines

kernel/locking/rwsem.c +2 -0

+		if (IS_ERR(rwsem_down_write_slow(sem, TASK_KILLABLE)))
 			return -EINTR;
+	}
 	rwsem_set_owner(sem);
 	return 0;
 }

I'm having a great day it seems, it's like back in uni, trying to find all the missing - signs in this page-long DE.

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

Quoted 5 lines · Waiman Long

kernel/locking/rwsem.c +5 -0

+/*
+ * The typical HZ value is either 250 or 1000. So set the minimum waiting
+ * time to 4ms in the wait queue before initiating the handoff protocol.
+ */
+#define RWSEM_WAIT_TIMEOUT	(HZ/250)

That seems equally unfortunate. For HZ=100 that results in 0ms, and for HZ=300 that results in 3 1/3-rd ms.

(and this is not considering Alpha,ARM and MIPS, who all have various other 'creative' HZ values)

In general aiming for sub 10ms timing using jiffies seems 'optimistic'.

Quoted 9 lines · Peter Zijlstra

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

Quoted 5 lines · Waiman Long

kernel/locking/rwsem.c +5 -0

+/*
+ * The typical HZ value is either 250 or 1000. So set the minimum waiting
+ * time to 4ms in the wait queue before initiating the handoff protocol.
+ */
+#define RWSEM_WAIT_TIMEOUT	(HZ/250)

That seems equally unfortunate. For HZ=100 that results in 0ms, and for HZ=300 that results in 3 1/3-rd ms.

Quoted 4 lines

(and this is not considering Alpha,ARM and MIPS, who all have various other 'creative' HZ values)

In general aiming for sub 10ms timing using jiffies seems 'optimistic'.

I see your point. I will change it to use sched_clock() instead.

Thinking about it again. I think I will just change its definition to "((HZ + 249)/250)" for now to make sure that it is at least 1. The

Quoted 2 lines · Waiman Long

Thinking about it again. I think I will just change its definition to "((HZ + 249)/250)" for now to make sure that it is at least 1. The

DIV_ROUND_UP()

kernel/locking/rwsem.c +2 -0

@@ -245,6 +274,8 @@ static void __rwsem_mark_wake(struct rw_semaphore *sem,
 	struct rwsem_waiter *waiter, *tmp;
 	long oldcount, woken = 0, adjustment = 0;
 
+	lockdep_assert_held(&sem->wait_lock);
+
 	/*
 	 * Take a peek at the queue head waiter such that we can determine
 	 * the wakeup(s) to perform.

@@ -276,6 +307,15 @@ static void __rwsem_mark_wake(struct rw_semaphore *sem,
 		adjustment = RWSEM_READER_BIAS;

kernel/locking/rwsem.c +3 -0

+	if (woken && RWSEM_COUNT_HANDOFF(atomic_long_read(&sem->count)))
+		adjustment -= RWSEM_FLAG_HANDOFF;
+
 	if (adjustment)
 		atomic_long_add(adjustment, &sem->count);
 }

@@ -332,22 +378,42 @@ 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

kernel/locking/rwsem.c +16 -0

+retry:
+	if (RWSEM_COUNT_LOCKED(count)) {
+		if (RWSEM_COUNT_HANDOFF(count) || (wstate != WRITER_HANDOFF))
+			return false;
+		/*
+		 * The lock may become free just before setting handoff bit.
+		 * It will be simpler if atomic_long_or_return() is available.
+		 */
+		atomic_long_or(RWSEM_FLAG_HANDOFF, &sem->count);
+		count = atomic_long_read(&sem->count);
+		goto retry;
+	}
+
+	if ((wstate == WRITER_NOT_FIRST) && RWSEM_COUNT_HANDOFF(count))
 		return false;
 
+	new = (count & ~RWSEM_FLAG_HANDOFF) + RWSEM_WRITER_LOCKED -
+	      (list_is_singular(&sem->wait_list) ? RWSEM_FLAG_WAITERS : 0);
 

kernel/locking/rwsem.c +8 -0

+}
+
 /*
  * Wait for the read lock to be granted
  */

@@ -510,16 +586,18 @@ __rwsem_down_read_failed_common(struct rw_semaphore *sem, int state)
 
 	waiter.task = current;
 	waiter.type = RWSEM_WAITING_FOR_READ;
+	waiter.timeout = jiffies + RWSEM_WAIT_TIMEOUT;
 
 	raw_spin_lock_irq(&sem->wait_lock);
 	if (list_empty(&sem->wait_list)) {
 		/*
 		 * In case the wait queue is empty and the lock isn't owned
+		 * by a writer or has the handoff bit set, this reader can
+		 * exit the slowpath and return immediately as its
+		 * RWSEM_READER_BIAS has already been set in the count.
 		 */
+		if (!(atomic_long_read(&sem->count) &
+		     (RWSEM_WRITER_MASK | RWSEM_FLAG_HANDOFF))) {
 			raw_spin_unlock_irq(&sem->wait_lock);
 			rwsem_set_reader_owned(sem);
 			lockevent_inc(rwsem_rlock_fast);

kernel/locking/rwsem.c +6 -0

 	raw_spin_unlock_irq(&sem->wait_lock);
 	__set_current_state(TASK_RUNNING);
 	lockevent_inc(rwsem_rlock_fail);

@@ -593,7 +672,7 @@ static inline struct rw_semaphore *
 __rwsem_down_write_failed_common(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);

@@ -608,56 +687,63 @@ __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state)
 	 */
 	waiter.task = current;
 	waiter.type = RWSEM_WAITING_FOR_WRITE;
+	waiter.timeout = jiffies + RWSEM_WAIT_TIMEOUT;
 
 	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) {
 		count = atomic_long_read(&sem->count);
 
 		/*
+		 * If there were already threads queued before us and:
+		 *  1) there are no no active locks, wake the front

kernel/locking/rwsem.c +6 -0

+		wake_q_init(&wake_q);
 	} else {
 		count = atomic_long_add_return(RWSEM_FLAG_WAITERS, &sem->count);
 	}
 
+wait:
 	/* wait until we successfully acquire the lock */
 	set_current_state(state);
 	while (true) {
+		if (rwsem_try_write_lock(count, sem, wstate))
 			break;
+
 		raw_spin_unlock_irq(&sem->wait_lock);
 
 		/* Block until there are no active lockers. */
+		for (;;) {
 			if (signal_pending_state(state, current))
 				goto out_nolock;
 

@@ -665,9 +751,34 @@ __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state)
 			lockevent_inc(rwsem_sleep_writer);
 			set_current_state(state);
 			count = atomic_long_read(&sem->count);
+

kernel/locking/rwsem.c +4 -0

+				break;
+			}
+		}
 
 		raw_spin_lock_irq(&sem->wait_lock);
+		count = atomic_long_read(&sem->count);
 	}
 	__set_current_state(TASK_RUNNING);
 	list_del(&waiter.list);

(quoted diff) +0 -0

 	else

See, if you first write that function in the form:

	long new;

	do {
		new = count | RWSEM_WRITER_LOCKED;

		if (count & RWSEM_LOCK_MASK)
			return false;

		if (list_is_singular(&sem->wait_list))
			new &= ~RWSEM_FLAG_WAITERS;

	} while (atomic_long_try_cmpxchg_acquire(&sem->count, &count, new));

	rwsem_set_owner(sem);
	return true;

And then add the HANDOFF bits like:

	long new;

	do {

(quoted diff) +2 -0

+		bool has_handoff = !!(count & RWSEM_FLAG_HANDOFF);
 
+		new = (count | RWSEM_WRITER_LOCKED) & ~RWSEM_FLAG_HANDOFF;

		if (count & RWSEM_LOCK_MASK) {

(quoted diff) +2 -0

+			if (has_handoff && wstate != WRITER_HANDOFF)
+				return false;

			new |= RWSEM_FLAG_HANDOFF;
		}

(quoted diff) +2 -0

+		if (has_handoff && wstate == WRITER_NOT_FIRST)
+			return false;

		if (list_is_singular(&sem->wait_list))
			new &= ~RWSEM_FLAG_WAITERS;

	} while (atomic_long_try_cmpxchg_acquire(&sem->count, &count, new));

	rwsem_set_owner(sem);
	return true;

it almost looks like sensible code.

Quoted 33 lines · Peter Zijlstra

See, if you first write that function in the form:

	long new;

	do {
		new = count | RWSEM_WRITER_LOCKED;

		if (count & RWSEM_LOCK_MASK)
			return false;

		if (list_is_singular(&sem->wait_list))
			new &= ~RWSEM_FLAG_WAITERS;

	} while (atomic_long_try_cmpxchg_acquire(&sem->count, &count, new));

	rwsem_set_owner(sem);
	return true;

And then add the HANDOFF bits like:

	long new;

	do {

(quoted diff) +2 -0

+		bool has_handoff = !!(count & RWSEM_FLAG_HANDOFF);
 
+		new = (count | RWSEM_WRITER_LOCKED) & ~RWSEM_FLAG_HANDOFF;

		if (count & RWSEM_LOCK_MASK) {

(quoted diff) +2 -0

+			if (has_handoff && wstate != WRITER_HANDOFF)
+				return false;

			new |= RWSEM_FLAG_HANDOFF;
		}

Quoted 7 lines

(quoted diff) +2 -0

+		if (has_handoff && wstate == WRITER_NOT_FIRST)
+			return false;

		if (list_is_singular(&sem->wait_list))
			new &= ~RWSEM_FLAG_WAITERS;

	} while (atomic_long_try_cmpxchg_acquire(&sem->count, &count, new));

obviously that should be !

kernel/locking/rwsem.c line 108

Quoted 3 lines · Waiman Long

kernel/locking/rwsem.c +3 -0

+#define RWSEM_COUNT_HANDOFF(c)	((c) & RWSEM_FLAG_HANDOFF)
+#define RWSEM_COUNT_LOCKED_OR_HANDOFF(c)	\
+	((c) & (RWSEM_LOCK_MASK|RWSEM_FLAG_HANDOFF))

Like said before, I also made these go away.

Quoted 12 lines

kernel/locking/rwsem.c +2 -0

@@ -245,6 +274,8 @@ static void __rwsem_mark_wake(struct rw_semaphore *sem,
 	struct rwsem_waiter *waiter, *tmp;
 	long oldcount, woken = 0, adjustment = 0;
 
+	lockdep_assert_held(&sem->wait_lock);
+
 	/*
 	 * Take a peek at the queue head waiter such that we can determine
 	 * the wakeup(s) to perform.

@@ -276,6 +307,15 @@ static void __rwsem_mark_wake(struct rw_semaphore *sem,
 		adjustment = RWSEM_READER_BIAS;

kernel/locking/rwsem.c static void __rwsem_mark_wake(struct rw_semaphore *sem, · line 317

Quoted 10 lines

kernel/locking/rwsem.c +8 -0

 		oldcount = atomic_long_fetch_add(adjustment, &sem->count);
 		if (unlikely(oldcount & RWSEM_WRITER_MASK)) {
+			/*
+			 * Initiate handoff to reader, if applicable.
+			 */
+			if (!(oldcount & RWSEM_FLAG_HANDOFF) &&
+			    time_after(jiffies, waiter->timeout)) {
+				adjustment -= RWSEM_FLAG_HANDOFF;
+				lockevent_inc(rwsem_rlock_handoff);
+			}

/*
* When we've been waiting 'too' long (for
* writers to give up the lock) request a
* HANDOFF to force the issue. */

?

kernel/locking/rwsem.c static void __rwsem_mark_wake(struct rw_semaphore *sem, · line 319

Quoted 3 lines

kernel/locking/rwsem.c +1 -0

+
 			atomic_long_sub(adjustment, &sem->count);

Can we change this to: atomic_long_add() please? The below loop that wakes all remaining readers does use add(), so it is a bit 'weird' to have the adjustment being negated on handover.

kernel/locking/rwsem.c static void __rwsem_mark_wake(struct rw_semaphore *sem, · line 369

Quoted 9 lines

kernel/locking/rwsem.c +3 -0

 			return;
 		}

@@ -324,6 +364,12 @@ static void __rwsem_mark_wake(struct rw_semaphore *sem,
 		adjustment -= RWSEM_FLAG_WAITERS;
 	}
 
+	/*
+	 * Clear the handoff flag
+	 */

Right, but that is a trivial comment in the 'increment i' style, it clearly states what the code does, but completely fails to elucidate the code.

Maybe:

	/*
	 * When we've woken a reader, we no longer need to force writers
	 * to give up the lock and we can clear HANDOFF.
	 */

And I suppose this is required if we were the pickup of the handoff set above, but is there a guarantee that the HANDOFF was not set by a writer?

Quoted 13 lines

kernel/locking/rwsem.c +5 -0

+	if (woken && RWSEM_COUNT_HANDOFF(atomic_long_read(&sem->count)))
+		adjustment -= RWSEM_FLAG_HANDOFF;
+
 	if (adjustment)
 		atomic_long_add(adjustment, &sem->count);
 }

@@ -332,22 +378,42 @@ 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

kernel/locking/rwsem.c static void __rwsem_mark_wake(struct rw_semaphore *sem, · line 389

Quoted 7 lines

kernel/locking/rwsem.c +3 -0

+ * bit is set or the lock is acquired.
  */
+static inline bool rwsem_try_write_lock(long count, struct rw_semaphore *sem,
+					enum writer_wait_state wstate)
 {
 	long new;
 

	lockdep_assert_held(&sem->wait_lock);

Quoted 20 lines

kernel/locking/rwsem.c +16 -0

+retry:
+	if (RWSEM_COUNT_LOCKED(count)) {
+		if (RWSEM_COUNT_HANDOFF(count) || (wstate != WRITER_HANDOFF))
+			return false;
+		/*
+		 * The lock may become free just before setting handoff bit.
+		 * It will be simpler if atomic_long_or_return() is available.
+		 */
+		atomic_long_or(RWSEM_FLAG_HANDOFF, &sem->count);
+		count = atomic_long_read(&sem->count);
+		goto retry;
+	}
+
+	if ((wstate == WRITER_NOT_FIRST) && RWSEM_COUNT_HANDOFF(count))
 		return false;
 
+	new = (count & ~RWSEM_FLAG_HANDOFF) + RWSEM_WRITER_LOCKED -
+	      (list_is_singular(&sem->wait_list) ? RWSEM_FLAG_WAITERS : 0);
 

kernel/locking/rwsem.c static void __rwsem_mark_wake(struct rw_semaphore *sem, · line 418

Quoted 10 lines

kernel/locking/rwsem.c +3 -0

 	if (atomic_long_try_cmpxchg_acquire(&sem->count, &count, new)) {
 		rwsem_set_owner(sem);
 		return true;
 	}
 
+	if (unlikely((wstate == WRITER_HANDOFF) && !RWSEM_COUNT_HANDOFF(count)))
+		goto retry;
+
 	return false;
 }

This function gives me heartburn. Don't you just feel something readable trying to struggle free from that?

See, if you first write that function in the form:

	long new;

	do {
		new = count | RWSEM_WRITER_LOCKED;

		if (count & RWSEM_LOCK_MASK)
			return false;

		if (list_is_singular(&sem->wait_list))
			new &= ~RWSEM_FLAG_WAITERS;

	} while (atomic_long_try_cmpxchg_acquire(&sem->count, &count, new));

	rwsem_set_owner(sem);
	return true;

And then add the HANDOFF bits like:

	long new;

	do {

(quoted diff) +2 -0

+		bool has_handoff = !!(count & RWSEM_FLAG_HANDOFF);
 
+		new = (count | RWSEM_WRITER_LOCKED) & ~RWSEM_FLAG_HANDOFF;

		if (count & RWSEM_LOCK_MASK) {

(quoted diff) +2 -0

+			if (has_handoff && wstate != WRITER_HANDOFF)
+				return false;

			new |= RWSEM_FLAG_HANDOFF;
		}

(quoted diff) +2 -0

+		if (has_handoff && wstate == WRITER_NOT_FIRST)
+			return false;

		if (list_is_singular(&sem->wait_list))
			new &= ~RWSEM_FLAG_WAITERS;

	} while (atomic_long_try_cmpxchg_acquire(&sem->count, &count, new));

	rwsem_set_owner(sem);
	return true;

it almost looks like sensible code.

kernel/locking/rwsem.c static inline bool rwsem_try_write_lock_unqueued(struct rw_semaphore *sem) · line 430

Quoted 9 lines

kernel/locking/rwsem.c +1 -1

@@ -359,7 +425,7 @@ static inline bool rwsem_try_write_lock_unqueued(struct rw_semaphore *sem)
 {
 	long count = atomic_long_read(&sem->count);
 
-	while (!RWSEM_COUNT_LOCKED(count)) {
+	while (!RWSEM_COUNT_LOCKED_OR_HANDOFF(count)) {
 		if (atomic_long_try_cmpxchg_acquire(&sem->count, &count,
 					count + RWSEM_WRITER_LOCKED)) {

RWSEM_WRITER_LOCKED really should be RWSEM_FLAG_WRITER or something like that, and since it is a flag, that really should've been | not +.

kernel/locking/rwsem.c static bool rwsem_optimistic_spin(struct rw_semaphore *sem) · line 574

Quoted 13 lines

kernel/locking/rwsem.c +8 -0

 			rwsem_set_owner(sem);

@@ -498,6 +564,16 @@ static bool rwsem_optimistic_spin(struct rw_semaphore *sem)
 }
 #endif
 
+/*
+ * This is safe to be called without holding the wait_lock.
+ */
+static inline bool
+rwsem_waiter_is_first(struct rw_semaphore *sem, struct rwsem_waiter *waiter)
+{
+	return list_first_entry(&sem->wait_list, struct rwsem_waiter, list)
+			== waiter;

Just bust the line limit on that, this is silly. If you feel strongly about the 80 char thing, we could do:

#define rwsem_first_waiter(sem) \
	list_first_entry(&sem->wait_list, struct rwsem_waiter, list)

and use that in both locations. (and one could even write the list_for_each_entry_safe() loop in the form:

	while (!list_empty(&sem->wait_list)) {
		entry = rwsem_first_waiter(sem);

		...

		list_del();

		...
	}

Although I suppose that gets you confused later on where you want to wake more readers still... I'll get there,.. eventually.

Quoted 23 lines

kernel/locking/rwsem.c +8 -0

+}
+
 /*
  * Wait for the read lock to be granted
  */

@@ -510,16 +586,18 @@ __rwsem_down_read_failed_common(struct rw_semaphore *sem, int state)
 
 	waiter.task = current;
 	waiter.type = RWSEM_WAITING_FOR_READ;
+	waiter.timeout = jiffies + RWSEM_WAIT_TIMEOUT;
 
 	raw_spin_lock_irq(&sem->wait_lock);
 	if (list_empty(&sem->wait_list)) {
 		/*
 		 * In case the wait queue is empty and the lock isn't owned
+		 * by a writer or has the handoff bit set, this reader can
+		 * exit the slowpath and return immediately as its
+		 * RWSEM_READER_BIAS has already been set in the count.
 		 */
+		if (!(atomic_long_read(&sem->count) &
+		     (RWSEM_WRITER_MASK | RWSEM_FLAG_HANDOFF))) {
 			raw_spin_unlock_irq(&sem->wait_lock);
 			rwsem_set_reader_owned(sem);

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

Quoted 7 lines

kernel/locking/rwsem.c +2 -0

 			lockevent_inc(rwsem_rlock_fast);

@@ -567,7 +645,8 @@ __rwsem_down_read_failed_common(struct rw_semaphore *sem, int state)
 out_nolock:
 	list_del(&waiter.list);
 	if (list_empty(&sem->wait_list))
+		atomic_long_andnot(RWSEM_FLAG_WAITERS|RWSEM_FLAG_HANDOFF,
+				   &sem->count);

If you split the line, this wants { }.

Quoted 28 lines

kernel/locking/rwsem.c +4 -0

 	raw_spin_unlock_irq(&sem->wait_lock);
 	__set_current_state(TASK_RUNNING);
 	lockevent_inc(rwsem_rlock_fail);

@@ -593,7 +672,7 @@ static inline struct rw_semaphore *
 __rwsem_down_write_failed_common(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);

@@ -608,56 +687,63 @@ __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state)
 	 */
 	waiter.task = current;
 	waiter.type = RWSEM_WAITING_FOR_WRITE;
+	waiter.timeout = jiffies + RWSEM_WAIT_TIMEOUT;
 
 	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) {
 		count = atomic_long_read(&sem->count);
 

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

Quoted 13 lines

kernel/locking/rwsem.c +10 -0

 		/*
+		 * If there were already threads queued before us and:
+		 *  1) there are no no active locks, wake the front
+		 *     queued process(es) as the handoff bit might be set.
+		 *  2) there are no active writers and some readers, the lock
+		 *     must be read owned; so we try to wake any read lock
+		 *     waiters that were queued ahead of us.
 		 */
+		if (!RWSEM_COUNT_LOCKED(count))
+			__rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);
+		else if (!(count & RWSEM_WRITER_MASK) &&
+			  (count & RWSEM_READER_MASK))
 			__rwsem_mark_wake(sem, RWSEM_WAKE_READERS, &wake_q);

That RWSEM_WRITER_MASK is another layer of obfustaction we can do without.

Does the above want to be something like:

		if (!(count & RWSEM_WRITER_LOCKED)) {
			__rwsem_mark_wake(sem, (count & RWSEM_READER_MASK) ?
					       RWSEM_WAKE_READERS :
					       RWSEM_WAKE_ANY, &wake_q);
		}

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

Quoted 11 lines

kernel/locking/rwsem.c +10 -0

+		else
+			goto wait;
 
+		/*
+		 * The wakeup is normally called _after_ the wait_lock
+		 * is released, but given that we are proactively waking
+		 * readers we can deal with the wake_q overhead as it is
+		 * similar to releasing and taking the wait_lock again
+		 * for attempting rwsem_try_write_lock().
+		 */
+		wake_up_q(&wake_q);

Hurmph.. the reason we do wake_up_q() outside of wait_lock is such that those tasks don't bounce on wait_lock. Also, it removes a great deal of hold-time from wait_lock.

So I'm not sure I buy your argument here.

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

Quoted 3 lines

kernel/locking/rwsem.c +3 -0

+		/*
+		 * Reinitialize wake_q after use.
+		 */

Or:

		/* we need wake_q again below, reinitialize */

Quoted 28 lines

kernel/locking/rwsem.c +10 -0

+		wake_q_init(&wake_q);
 	} else {
 		count = atomic_long_add_return(RWSEM_FLAG_WAITERS, &sem->count);
 	}
 
+wait:
 	/* wait until we successfully acquire the lock */
 	set_current_state(state);
 	while (true) {
+		if (rwsem_try_write_lock(count, sem, wstate))
 			break;
+
 		raw_spin_unlock_irq(&sem->wait_lock);
 
 		/* Block until there are no active lockers. */
+		for (;;) {
 			if (signal_pending_state(state, current))
 				goto out_nolock;
 

@@ -665,9 +751,34 @@ __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state)
 			lockevent_inc(rwsem_sleep_writer);
 			set_current_state(state);
 			count = atomic_long_read(&sem->count);
+
+			if ((wstate == WRITER_NOT_FIRST) &&
+			    rwsem_waiter_is_first(sem, &waiter))
+				wstate = WRITER_FIRST;
+

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

Quoted 6 lines

kernel/locking/rwsem.c +6 -0

+			if (!RWSEM_COUNT_LOCKED(count))
+				break;
+
+			/*
+			 * An RT task sets the HANDOFF bit immediately.
+			 * Non-RT task will wait a while before doing so.

Again, this describes what we already read the code to do; but doesn't add anything.

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

Quoted 11 lines

kernel/locking/rwsem.c +11 -0

+			 *
+			 * 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 out to call rwsem_try_write_lock().
+				 */

Another exceedingly useful comment.

Quoted 10 lines

kernel/locking/rwsem.c +4 -0

+				break;
+			}
+		}
 
 		raw_spin_lock_irq(&sem->wait_lock);
+		count = atomic_long_read(&sem->count);
 	}
 	__set_current_state(TASK_RUNNING);
 	list_del(&waiter.list);

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

Quoted 9 lines

kernel/locking/rwsem.c +6 -0

@@ -680,6 +791,12 @@ __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state)
 	__set_current_state(TASK_RUNNING);
 	raw_spin_lock_irq(&sem->wait_lock);
 	list_del(&waiter.list);
+	/*
+	 * If handoff bit has been set by this waiter, make sure that the
+	 * clearing of it is seen by others before proceeding.
+	 */
+	if (unlikely(wstate == WRITER_HANDOFF))
+		atomic_long_add_return(-RWSEM_FLAG_HANDOFF,  &sem->count);

_AGAIN_ no explanation what so ff'ing ever.

And why add_return() if you ignore the return value.

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

Quoted 2 lines

kernel/locking/rwsem.c +0 -0

 	if (list_empty(&sem->wait_list))
 		atomic_long_andnot(RWSEM_FLAG_WAITERS, &sem->count);

And you could've easily combined the two flags in a single andnot op.

Quoted 9 lines

Quoted 9 lines

kernel/locking/rwsem.c +5 -0

@@ -665,9 +751,34 @@ __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state)
 			lockevent_inc(rwsem_sleep_writer);
 			set_current_state(state);
 			count = atomic_long_read(&sem->count);
+
+			if ((wstate == WRITER_NOT_FIRST) &&
+			    rwsem_waiter_is_first(sem, &waiter))
+				wstate = WRITER_FIRST;
+

Quoted 8 lines

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

Quoted 6 lines

kernel/locking/rwsem.c +6 -0

+			if (!RWSEM_COUNT_LOCKED(count))
+				break;
+
+			/*
+			 * An RT task sets the HANDOFF bit immediately.
+			 * Non-RT task will wait a while before doing so.

Again, this describes what we already read the code to do; but doesn't add anything.

Will remove that.

Quoted 34 lines

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

Quoted 11 lines

kernel/locking/rwsem.c +11 -0

+			 *
+			 * 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 out to call rwsem_try_write_lock().
+				 */

Another exceedingly useful comment.

Quoted 10 lines

kernel/locking/rwsem.c +4 -0

+				break;
+			}
+		}
 
 		raw_spin_lock_irq(&sem->wait_lock);
+		count = atomic_long_read(&sem->count);
 	}
 	__set_current_state(TASK_RUNNING);
 	list_del(&waiter.list);

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

Quoted 9 lines

kernel/locking/rwsem.c +6 -0

@@ -680,6 +791,12 @@ __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state)
 	__set_current_state(TASK_RUNNING);
 	raw_spin_lock_irq(&sem->wait_lock);
 	list_del(&waiter.list);
+	/*
+	 * If handoff bit has been set by this waiter, make sure that the
+	 * clearing of it is seen by others before proceeding.
+	 */
+	if (unlikely(wstate == WRITER_HANDOFF))
+		atomic_long_add_return(-RWSEM_FLAG_HANDOFF,  &sem->count);

_AGAIN_ no explanation what so ff'ing ever.

Quoted 2 lines

And why add_return() if you ignore the return value.

OK, will remove those.

Quoted 18 lines

kernel/locking/rwsem.c static void __rwsem_mark_wake(struct rw_semaphore *sem, · line 369

Quoted 7 lines

kernel/locking/rwsem.c +3 -0

@@ -324,6 +364,12 @@ static void __rwsem_mark_wake(struct rw_semaphore *sem,
 		adjustment -= RWSEM_FLAG_WAITERS;
 	}
 
+	/*
+	 * Clear the handoff flag
+	 */

Right, but that is a trivial comment in the 'increment i' style, it clearly states what the code does, but completely fails to elucidate the code.

Maybe:

	/*
	 * When we've woken a reader, we no longer need to force writers
	 * to give up the lock and we can clear HANDOFF.
	 */

Quoted 3 lines

And I suppose this is required if we were the pickup of the handoff set above, but is there a guarantee that the HANDOFF was not set by a writer?

I can change the comment. The handoff bit is always cleared in rwsem_try_write_lock() when the lock is successfully acquire. Will add a comment to document that.

Quoted 18 lines · Waiman Long

Quoted 18 lines

kernel/locking/rwsem.c static void __rwsem_mark_wake(struct rw_semaphore *sem, · line 369

Quoted 7 lines

kernel/locking/rwsem.c +3 -0

@@ -324,6 +364,12 @@ static void __rwsem_mark_wake(struct rw_semaphore *sem,
 		adjustment -= RWSEM_FLAG_WAITERS;
 	}
 
+	/*
+	 * Clear the handoff flag
+	 */

Right, but that is a trivial comment in the 'increment i' style, it clearly states what the code does, but completely fails to elucidate the code.

Maybe:

	/*
	 * When we've woken a reader, we no longer need to force writers
	 * to give up the lock and we can clear HANDOFF.
	 */

Quoted 6 lines

Quoted 3 lines

And I suppose this is required if we were the pickup of the handoff set above, but is there a guarantee that the HANDOFF was not set by a writer?

I can change the comment. The handoff bit is always cleared in rwsem_try_write_lock() when the lock is successfully acquire. Will add a comment to document that.

That doesn't help much, because it drops ->wait_lock between setting it and acquiring it. So the read-acquire can interleave.

I _think_ it works, but I'm having trouble explaining how exactly. I think because readers don't spin yet and thus wakeups abide by queue order.

And the other way around should have (write) spinners terminate the moment they see HANDOFF set by a readers, but I'm not immediately seeing that either.

I'll continue staring at that.

Quoted 33 lines

Quoted 20 lines

(quoted diff) +5 -0

@@ -608,56 +687,63 @@ __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state)
 	 */
 	waiter.task = current;
 	waiter.type = RWSEM_WAITING_FOR_WRITE;
+	waiter.timeout = jiffies + RWSEM_WAIT_TIMEOUT;
 
 	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) {
 		count = atomic_long_read(&sem->count);
 
 		/*
+		 * If there were already threads queued before us and:
+		 *  1) there are no no active locks, wake the front

Review __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state) · line 715

Quoted 10 lines

(quoted diff) +8 -0

+		 *     queued process(es) as the handoff bit might be set.
+		 *  2) there are no active writers and some readers, the lock
+		 *     must be read owned; so we try to wake any read lock
+		 *     waiters that were queued ahead of us.
 		 */
+		if (!RWSEM_COUNT_LOCKED(count))
+			__rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);
+		else if (!(count & RWSEM_WRITER_MASK) &&
+			  (count & RWSEM_READER_MASK))
 			__rwsem_mark_wake(sem, RWSEM_WAKE_READERS, &wake_q);

Does the above want to be something like:

Quoted 5 lines

		if (!(count & RWSEM_WRITER_LOCKED)) {
			__rwsem_mark_wake(sem, (count & RWSEM_READER_MASK) ?
					       RWSEM_WAKE_READERS :
					       RWSEM_WAKE_ANY, &wake_q);
		}

Yes.

Quoted 15 lines

Review __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state) · line 726

Quoted 11 lines

(quoted diff) +10 -0

+		else
+			goto wait;
 
+		/*
+		 * The wakeup is normally called _after_ the wait_lock
+		 * is released, but given that we are proactively waking
+		 * readers we can deal with the wake_q overhead as it is
+		 * similar to releasing and taking the wait_lock again
+		 * for attempting rwsem_try_write_lock().
+		 */
+		wake_up_q(&wake_q);

Hurmph.. the reason we do wake_up_q() outside of wait_lock is such that those tasks don't bounce on wait_lock. Also, it removes a great deal of hold-time from wait_lock.

Quoted 2 lines

So I'm not sure I buy your argument here.

Actually, we don't want to release the wait_lock, do wake_up_q() and acquire the wait_lock again as the state would have been changed. I didn't change the comment on this patch, but will reword it to discuss that.

Quoted 54 lines · Waiman Long

Quoted 32 lines

Quoted 20 lines

(quoted diff) +5 -0

@@ -608,56 +687,63 @@ __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state)
 	 */
 	waiter.task = current;
 	waiter.type = RWSEM_WAITING_FOR_WRITE;
+	waiter.timeout = jiffies + RWSEM_WAIT_TIMEOUT;
 
 	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) {
 		count = atomic_long_read(&sem->count);
 
 		/*
+		 * If there were already threads queued before us and:
+		 *  1) there are no no active locks, wake the front

Review __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state) · line 715

Quoted 10 lines

(quoted diff) +8 -0

+		 *     queued process(es) as the handoff bit might be set.
+		 *  2) there are no active writers and some readers, the lock
+		 *     must be read owned; so we try to wake any read lock
+		 *     waiters that were queued ahead of us.
 		 */
+		if (!RWSEM_COUNT_LOCKED(count))
+			__rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);
+		else if (!(count & RWSEM_WRITER_MASK) &&
+			  (count & RWSEM_READER_MASK))
 			__rwsem_mark_wake(sem, RWSEM_WAKE_READERS, &wake_q);

Does the above want to be something like:

Quoted 5 lines

		if (!(count & RWSEM_WRITER_LOCKED)) {
			__rwsem_mark_wake(sem, (count & RWSEM_READER_MASK) ?
					       RWSEM_WAKE_READERS :
					       RWSEM_WAKE_ANY, &wake_q);
		}

Yes.

Quoted 15 lines

Review __rwsem_down_write_failed_common(struct rw_semaphore *sem, int state) · line 726

Quoted 11 lines

(quoted diff) +10 -0

+		else
+			goto wait;
 
+		/*
+		 * The wakeup is normally called _after_ the wait_lock
+		 * is released, but given that we are proactively waking
+		 * readers we can deal with the wake_q overhead as it is
+		 * similar to releasing and taking the wait_lock again
+		 * for attempting rwsem_try_write_lock().
+		 */
+		wake_up_q(&wake_q);

Hurmph.. the reason we do wake_up_q() outside of wait_lock is such that those tasks don't bounce on wait_lock. Also, it removes a great deal of hold-time from wait_lock.

Quoted 5 lines

Quoted 2 lines

So I'm not sure I buy your argument here.

Actually, we don't want to release the wait_lock, do wake_up_q() and acquire the wait_lock again as the state would have been changed. I didn't change the comment on this patch, but will reword it to discuss that.

I don't understand, we've queued ourselves, we're on the list, we're not first. How would dropping the lock to try and kick waiters before us be a problem?

Sure, once we re-acquire the lock we have to re-avaluate @wstate to see if we're first now or not, but we need to do that anyway.

So what is wrong with the below?

include/linux/sched/wake_q.h +5 -0

@@ -51,6 +51,11 @@ static inline void wake_q_init(struct wa
 	head->lastp = &head->first;
 }
 
+static inline bool wake_q_empty(struct wake_q_head *head)
+{
+	return head->first == WAKE_Q_TAIL;
+}
+
 extern void wake_q_add(struct wake_q_head *head, struct task_struct *task);
 extern void wake_q_add_safe(struct wake_q_head *head, struct task_struct *task);
 extern void wake_up_q(struct wake_q_head *head);

kernel/locking/rwsem.c +8 -18

@@ -700,25 +700,22 @@ __rwsem_down_write_failed_common(struct
 		 *     must be read owned; so we try to wake any read lock
 		 *     waiters that were queued ahead of us.
 		 */
-		if (!(count & RWSEM_LOCKED_MASK))
-			__rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);
-		else if (!(count & RWSEM_WRITER_MASK) &&
-				(count & RWSEM_READER_MASK))
-			__rwsem_mark_wake(sem, RWSEM_WAKE_READERS, &wake_q);
-		else
+		if (count & RWSEM_WRITER_LOCKED)
 			goto wait;
-		/*
-		 * The wakeup is normally called _after_ the wait_lock
-		 * is released, but given that we are proactively waking
-		 * readers we can deal with the wake_q overhead as it is
-		 * similar to releasing and taking the wait_lock again
-		 * for attempting rwsem_try_write_lock().
-		 */
-		wake_up_q(&wake_q);
-		/*
-		 * Reinitialize wake_q after use.
-		 */
-		wake_q_init(&wake_q);
+
+		__rwsem_mark_wake(sem, (count & RWSEM_READER_MASK) ?
+				RWSEM_WAKE_READERS :
+				RWSEM_WAKE_ANY, &wake_q);
+
+		if (!wake_q_empty(&wake_q)) {
+			raw_spin_unlock_irq(&sem->wait_lock);

Quoted 56 lines · Peter Zijlstra

So what is wrong with the below?

include/linux/sched/wake_q.h +5 -0

@@ -51,6 +51,11 @@ static inline void wake_q_init(struct wa
 	head->lastp = &head->first;
 }
 
+static inline bool wake_q_empty(struct wake_q_head *head)
+{
+	return head->first == WAKE_Q_TAIL;
+}
+
 extern void wake_q_add(struct wake_q_head *head, struct task_struct *task);
 extern void wake_q_add_safe(struct wake_q_head *head, struct task_struct *task);
 extern void wake_up_q(struct wake_q_head *head);

kernel/locking/rwsem.c +15 -18

@@ -700,25 +700,22 @@ __rwsem_down_write_failed_common(struct
 		 *     must be read owned; so we try to wake any read lock
 		 *     waiters that were queued ahead of us.
 		 */
-		if (!(count & RWSEM_LOCKED_MASK))
-			__rwsem_mark_wake(sem, RWSEM_WAKE_ANY, &wake_q);
-		else if (!(count & RWSEM_WRITER_MASK) &&
-				(count & RWSEM_READER_MASK))
-			__rwsem_mark_wake(sem, RWSEM_WAKE_READERS, &wake_q);
-		else
+		if (count & RWSEM_WRITER_LOCKED)
 			goto wait;
-		/*
-		 * The wakeup is normally called _after_ the wait_lock
-		 * is released, but given that we are proactively waking
-		 * readers we can deal with the wake_q overhead as it is
-		 * similar to releasing and taking the wait_lock again
-		 * for attempting rwsem_try_write_lock().
-		 */
-		wake_up_q(&wake_q);
-		/*
-		 * Reinitialize wake_q after use.
-		 */
-		wake_q_init(&wake_q);
+
+		__rwsem_mark_wake(sem, (count & RWSEM_READER_MASK) ?
+				RWSEM_WAKE_READERS :
+				RWSEM_WAKE_ANY, &wake_q);
+
+		if (!wake_q_empty(&wake_q)) {
+			raw_spin_unlock_irq(&sem->wait_lock);
+			wake_up_q(&wake_q);
+			/* used again, reinit */
+			wake_q_init(&wake_q);
+			raw_spin_lock_irq(&sem->wait_lock);
+			if (rwsem_waiter_is_first(sem, &waiter))
+				wstate = WRITER_FIRST;
+		}

kernel/locking/rwsem.c __rwsem_down_write_failed_common(struct · line 721

Quoted 3 lines

kernel/locking/rwsem.c +0 -0

 	} else {
 		count = atomic_long_add_return(RWSEM_FLAG_WAITERS, &sem->count);
 	}

Yes, we can certainly do that. My point is that I haven't changed the existing logic regarding that wakeup, I only move it around in the patch. As it is not related to lock handoff, we can do it as a separate patch.

+/*
+ * This is safe to be called without holding the wait_lock.

+ */

kernel/locking/rwsem.c +6 -0

+static inline bool
+rwsem_waiter_is_first(struct rw_semaphore *sem, struct rwsem_waiter *waiter)
+{
+	return list_first_entry(&sem->wait_list, struct rwsem_waiter, list)
+			== waiter;
+}

This patch modifies rwsem_spin_on_owner() to return four possible values to better reflect the state of lock holder which enables us to make a better decision of what to do next.

In the special case that there is no active lock and the handoff bit is set, optimistic spinning has to be stopped.

Signed-off-by Waiman Long <[email protected]>

---
 kernel/locking/rwsem.c | 45 +++++++++++++++++++++++++++++++++++-------
 1 file changed, 38 insertions(+), 7 deletions(-)

kernel/locking/rwsem.c +0 -0

index aaab546a890d..2d6850c3e77b 100644

Quoted 20 lines · Waiman Long

This patch modifies rwsem_spin_on_owner() to return four possible values to better reflect the state of lock holder which enables us to make a better decision of what to do next.

In the special case that there is no active lock and the handoff bit is set, optimistic spinning has to be stopped.

Signed-off-by Waiman Long <[email protected]>

---
 kernel/locking/rwsem.c | 45 +++++++++++++++++++++++++++++++++++-------
 1 file changed, 38 insertions(+), 7 deletions(-)

kernel/locking/rwsem.c +0 -0

index aaab546a890d..2d6850c3e77b 100644
@@ -156,6 +156,11 @@ static inline bool is_rwsem_owner_spinnable(struct task_struct *owner)
 	return !((unsigned long)owner & RWSEM_ANONYMOUSLY_OWNED);
 }
 

kernel/locking/rwsem.c static inline bool is_rwsem_owner_spinnable(struct task_struct *owner) · line 162

Quoted 4 lines

kernel/locking/rwsem.c +4 -0

+static inline bool is_rwsem_owner_reader(struct task_struct *owner)
+{
+	return (unsigned long)owner & RWSEM_READER_OWNED;
+}

Move this and the surrounding helpers into the RWSEM_SPIN_ON_OWNER block, it is only used there and that way all the code is together.

In the special case that there is no active lock and the handoff bit is set, optimistic spinning has to be stopped.

Quoted 2 lines · Waiman Long

In the special case that there is no active lock and the handoff bit is set, optimistic spinning has to be stopped.

This makes me think this should've been _before_ you added the handoff bit. So that when you introduce the handoff, everything is solid.

In the special case that there is no active lock and the handoff bit is set, optimistic spinning has to be stopped.

kernel/locking/rwsem.c +3 -2

@@ -500,9 +521,19 @@ static noinline bool rwsem_spin_on_owner(struct rw_semaphore *sem)
 
 	/*
 	 * If there is a new owner or the owner is not set, we continue
-	 * spinning.
+	 * spinning except when here is no active locks and the handoff bit
+	 * is set. In this case, we have to stop spinning.
 	 */
-	return is_rwsem_owner_spinnable(READ_ONCE(sem->owner));
+	owner = READ_ONCE(sem->owner);

Quoted 13 lines · Waiman Long

In the special case that there is no active lock and the handoff bit is set, optimistic spinning has to be stopped.

kernel/locking/rwsem.c +3 -2

@@ -500,9 +521,19 @@ static noinline bool rwsem_spin_on_owner(struct rw_semaphore *sem)
 
 	/*
 	 * If there is a new owner or the owner is not set, we continue
-	 * spinning.
+	 * spinning except when here is no active locks and the handoff bit
+	 * is set. In this case, we have to stop spinning.
 	 */
-	return is_rwsem_owner_spinnable(READ_ONCE(sem->owner));
+	owner = READ_ONCE(sem->owner);

kernel/locking/rwsem.c static noinline bool rwsem_spin_on_owner(struct rw_semaphore *sem) · line 537

Quoted 10 lines

kernel/locking/rwsem.c +9 -0

+	if (!is_rwsem_owner_spinnable(owner))
+		return OWNER_NONSPINNABLE;
+	if (owner && !is_rwsem_owner_reader(owner))
+		return OWNER_WRITER;
+
+	count = atomic_long_read(&sem->count);
+	if (RWSEM_COUNT_HANDOFF(count) && !RWSEM_COUNT_LOCKED(count))
+		return OWNER_NONSPINNABLE;
+	return !owner ? OWNER_NULL : OWNER_READER;
 }

So this fixes a straight up bug in the last patch (and thus should be done before so the bug never exists), and creates unreadable code while at it.

Also, I think only checking HANDOFF after the loop is wrong; the moment HANDOFF happens you have to terminate the loop, irrespective of what @owner does.

Does something like so work?

---

enum owner_state {
	OWNER_NULL		= 1 << 0,
	OWNER_WRITER		= 1 << 1,
	OWNER_READER		= 1 << 2,
	OWNER_NONSPINNABLE	= 1 << 3,
};

#define OWNER_SPINNABLE (OWNER_NULL | OWNER_WRITER)

static inline enum owner_state rwsem_owner_state(unsigned long owner)

{
	if (!owner)
		return OWNER_NULL;

	if (owner & RWSEM_ANONYMOUSLY_OWNED)
		return OWNER_NONSPINNABLE;

	if (owner & RWSEM_READER_OWNER)
		return OWNER_READER;

	return OWNER_WRITER;
}

static noinline enum owner_state rwsem_spin_on_owner(struct rw_semaphore *sem)

{
	struct task_struct *tmp, *owner = READ_ONCE(sem->owner);
	enum owner_state state;

	rcu_read_lock();
	for (;;) {
		state = rwsem_owner_state((unsigned long)owner);
		if (!(state & OWNER_SPINNABLE))
			break;

		if (atomic_long_read(&sem->count) & RWSEM_FLAG_HANDOFF) {
			state = OWNER_NONSPINNABLE;
			break;
		}

		tmp = READ_ONCE(sem->owner);
		if (tmp != owner) {
			state = rwsem_owner_state((unsigned long)tmp);
			break;
		}

		/*
		 * Ensure we emit the owner->on_cpu, dereference _after_
		 * checking sem->owner still matches owner, if that fails,
		 * owner might point to free()d memory, if it still matches,
		 * the rcu_read_lock() ensures the memory stays valid.
		 */
		barrier();

		if (need_resched() || !owner_on_cpu(owner)) {
			state = OWNER_NONSPINNABLE;
			break;
		}

		cpu_relax();
	}
	rcu_read_unlock();

	return state;
}

Quoted 35 lines · Peter Zijlstra

Quoted 12 lines · Waiman Long

In the special case that there is no active lock and the handoff bit is set, optimistic spinning has to be stopped.

kernel/locking/rwsem.c +3 -2

@@ -500,9 +521,19 @@ static noinline bool rwsem_spin_on_owner(struct rw_semaphore *sem)
 
 	/*
 	 * If there is a new owner or the owner is not set, we continue
-	 * spinning.
+	 * spinning except when here is no active locks and the handoff bit
+	 * is set. In this case, we have to stop spinning.
 	 */
-	return is_rwsem_owner_spinnable(READ_ONCE(sem->owner));
+	owner = READ_ONCE(sem->owner);

kernel/locking/rwsem.c static noinline bool rwsem_spin_on_owner(struct rw_semaphore *sem) · line 537

Quoted 10 lines

kernel/locking/rwsem.c +9 -0

+	if (!is_rwsem_owner_spinnable(owner))
+		return OWNER_NONSPINNABLE;
+	if (owner && !is_rwsem_owner_reader(owner))
+		return OWNER_WRITER;
+
+	count = atomic_long_read(&sem->count);
+	if (RWSEM_COUNT_HANDOFF(count) && !RWSEM_COUNT_LOCKED(count))
+		return OWNER_NONSPINNABLE;
+	return !owner ? OWNER_NULL : OWNER_READER;
 }

So this fixes a straight up bug in the last patch (and thus should be done before so the bug never exists), and creates unreadable code while at it.

Also, I think only checking HANDOFF after the loop is wrong; the moment HANDOFF happens you have to terminate the loop, irrespective of what @owner does.

Does something like so work?

---

Quoted 7 lines

enum owner_state {
	OWNER_NULL		= 1 << 0,
	OWNER_WRITER		= 1 << 1,
	OWNER_READER		= 1 << 2,
	OWNER_NONSPINNABLE	= 1 << 3,
};

#define OWNER_SPINNABLE (OWNER_NULL | OWNER_WRITER)

Hmm, we should not spin on OWNER_NULL. Or at least not mixed in with the patch that changes the shape of all this. That should go in the RT thingy patch, which comes after this.

Quoted 8 lines · Peter Zijlstra

Quoted 8 lines

enum owner_state {
	OWNER_NULL		= 1 << 0,
	OWNER_WRITER		= 1 << 1,
	OWNER_READER		= 1 << 2,
	OWNER_NONSPINNABLE	= 1 << 3,
};

#define OWNER_SPINNABLE (OWNER_NULL | OWNER_WRITER)

Quoted 3 lines

Hmm, we should not spin on OWNER_NULL. Or at least not mixed in with the patch that changes the shape of all this. That should go in the RT thingy patch, which comes after this.

We do spin on OWNER_NULL right now, not in rwsem_spin_on_owner() but in the main rwsem_optimistic_spin() function.

RT task will quit if owner is NULL.

Quoted 13 lines · Waiman Long

In the special case that there is no active lock and the handoff bit is set, optimistic spinning has to be stopped.

kernel/locking/rwsem.c +3 -2

@@ -500,9 +521,19 @@ static noinline bool rwsem_spin_on_owner(struct rw_semaphore *sem)
 
 	/*
 	 * If there is a new owner or the owner is not set, we continue
-	 * spinning.
+	 * spinning except when here is no active locks and the handoff bit
+	 * is set. In this case, we have to stop spinning.
 	 */
-	return is_rwsem_owner_spinnable(READ_ONCE(sem->owner));
+	owner = READ_ONCE(sem->owner);

kernel/locking/rwsem.c static noinline bool rwsem_spin_on_owner(struct rw_semaphore *sem) · line 537

Quoted 10 lines

kernel/locking/rwsem.c +9 -0

+	if (!is_rwsem_owner_spinnable(owner))
+		return OWNER_NONSPINNABLE;
+	if (owner && !is_rwsem_owner_reader(owner))
+		return OWNER_WRITER;
+
+	count = atomic_long_read(&sem->count);
+	if (RWSEM_COUNT_HANDOFF(count) && !RWSEM_COUNT_LOCKED(count))
+		return OWNER_NONSPINNABLE;
+	return !owner ? OWNER_NULL : OWNER_READER;
 }

So this fixes a straight up bug in the last patch (and thus should be done before so the bug never exists), and creates unreadable code while at it.

Also, I think only checking HANDOFF after the loop is wrong; the moment HANDOFF happens you have to terminate the loop, irrespective of what @owner does.

Does something like so work?

---

Quoted 27 lines · Peter Zijlstra

Quoted 12 lines · Waiman Long

In the special case that there is no active lock and the handoff bit is set, optimistic spinning has to be stopped.

kernel/locking/rwsem.c +3 -2

@@ -500,9 +521,19 @@ static noinline bool rwsem_spin_on_owner(struct rw_semaphore *sem)
 
 	/*
 	 * If there is a new owner or the owner is not set, we continue
-	 * spinning.
+	 * spinning except when here is no active locks and the handoff bit
+	 * is set. In this case, we have to stop spinning.
 	 */
-	return is_rwsem_owner_spinnable(READ_ONCE(sem->owner));
+	owner = READ_ONCE(sem->owner);

kernel/locking/rwsem.c static noinline bool rwsem_spin_on_owner(struct rw_semaphore *sem) · line 537

Quoted 10 lines

kernel/locking/rwsem.c +9 -0

+	if (!is_rwsem_owner_spinnable(owner))
+		return OWNER_NONSPINNABLE;
+	if (owner && !is_rwsem_owner_reader(owner))
+		return OWNER_WRITER;
+
+	count = atomic_long_read(&sem->count);
+	if (RWSEM_COUNT_HANDOFF(count) && !RWSEM_COUNT_LOCKED(count))
+		return OWNER_NONSPINNABLE;
+	return !owner ? OWNER_NULL : OWNER_READER;
 }

So this fixes a straight up bug in the last patch (and thus should be done before so the bug never exists), and creates unreadable code while at it.

Quoted 7 lines

Also, I think only checking HANDOFF after the loop is wrong; the moment HANDOFF happens you have to terminate the loop, irrespective of what @owner does.

Does something like so work?

---

enum owner_state {
	OWNER_NULL		= 1 << 0,
	OWNER_WRITER		= 1 << 1,
	OWNER_READER		= 1 << 2,
	OWNER_NONSPINNABLE	= 1 << 3,
};

#define OWNER_SPINNABLE (OWNER_NULL | OWNER_WRITER)

static inline enum owner_state rwsem_owner_state(unsigned long owner)

{
	if (!owner)
		return OWNER_NULL;

	if (owner & RWSEM_ANONYMOUSLY_OWNED)
		return OWNER_NONSPINNABLE;

	if (owner & RWSEM_READER_OWNER)
		return OWNER_READER;

	return OWNER_WRITER;
}

static noinline enum owner_state rwsem_spin_on_owner(struct rw_semaphore *sem)

{
	struct task_struct *tmp, *owner = READ_ONCE(sem->owner);
	enum owner_state state = rwsem_owner_state((unsigned long)owner);

	if (state != OWNER_WRITER)
		return state;

	rcu_read_lock();
	for (;;) {
		if (atomic_long_read(&sem->count) & RWSEM_FLAG_HANDOFF) {
			state = OWNER_NONSPINNABLE;
			break;
		}

		tmp = READ_ONCE(sem->owner);
		if (tmp != owner) {
			state = rwsem_owner_state((unsigned long)tmp);
			break;
		}

		/*
		 * Ensure we emit the owner->on_cpu, dereference _after_
		 * checking sem->owner still matches owner, if that fails,
		 * owner might point to free()d memory, if it still matches,
		 * the rcu_read_lock() ensures the memory stays valid.
		 */
		barrier();

		if (need_resched() || !owner_on_cpu(owner)) {
			state = OWNER_NONSPINNABLE;
			break;

		}

		cpu_relax();
	}
	rcu_read_unlock();

	return state;
}

An RT task can do optimistic spinning only if the lock holder is actually running. If the state of the lock holder isn't known, there is a possibility that high priority of the RT task may block forward progress of the lock holder if it happens to reside on the same CPU. This will lead to deadlock. So we have to make sure that an RT task will not spin on a reader-owned rwsem.

When the owner is temporarily set to NULL, it is more tricky to decide if an RT task should stop spinning as it may be a temporary state where another writer may have just stolen the lock which then failed the task's trylock attempt. So one more retry is allowed to make sure that the lock is not spinnable by an RT task.

When testing on a 8-socket IvyBridge-EX system, the one additional retry seems to improve locking performance of RT write locking threads under heavy contentions. The table below shows the locking rates (in kops/s) with various write locking threads before and after the patch.

    Locking threads     Pre-patch     Post-patch
    ---------------     ---------     -----------
            4             2,753          2,608
            8             2,529          2,520
           16             1,727          1,918
           32             1,263          1,956
           64               889          1,343

Signed-off-by Waiman Long <[email protected]>

---
 kernel/locking/rwsem.c | 36 +++++++++++++++++++++++++++++-------
 1 file changed, 29 insertions(+), 7 deletions(-)

kernel/locking/rwsem.c +0 -0

index 2d6850c3e77b..8e19b5141595 100644

kernel/locking/rwsem.c +21 -7

+	int prev_owner_state = OWNER_NULL;
 
 	preempt_disable();
 

@@ -556,7 +558,12 @@ static bool rwsem_optimistic_spin(struct rw_semaphore *sem)
 	 *  2) readers own the lock as we can't determine if they are
 	 *     actively running or not.
 	 */
-	while (rwsem_spin_on_owner(sem) & OWNER_SPINNABLE) {
+	for (;;) {
+		enum owner_state owner_state = rwsem_spin_on_owner(sem);
+
+		if (!(owner_state & OWNER_SPINNABLE))
+			break;
+
 		/*
 		 * Try to acquire the lock
 		 */

@@ -566,13 +573,28 @@ static bool rwsem_optimistic_spin(struct rw_semaphore *sem)
 		}
 
 		/*
-		 * When there's no owner, we might have preempted between the
-		 * owner acquiring the lock and setting the owner field. If
-		 * we're an RT task that will live-lock because we won't let
-		 * the owner complete.
+		 * An RT task cannot do optimistic spinning if it cannot
+		 * be sure the lock holder is running or live-lock may
+		 * happen if the current task and the lock holder happen
+		 * to run in the same CPU.
+		 *
+		 * When there's no owner or is reader-owned, an RT task
+		 * will stop spinning if the owner state is not a writer
+		 * at the previous iteration of the loop. This allows the
+		 * RT task to recheck if the task that steals the lock is
+		 * a spinnable writer. If so, it can keeps on spinning.
+		 *
+		 * If the owner is a writer, the need_resched() check is
+		 * done inside rwsem_spin_on_owner(). If the owner is not
+		 * a writer, need_resched() check needs to be done here.
 		 */
-		if (!sem->owner && (need_resched() || rt_task(current)))
-			break;