Re: [PATCH v4] locking/rwsem: Make handoff bit handling more consistent
From: Waiman Long
Date: Mon Nov 15 2021 - 11:59:09 EST
On 11/15/21 10:45, Aiqun(Maria) Yu wrote:
On 11/12/2021 8:10 PM, Peter Zijlstra wrote:
On Thu, Nov 11, 2021 at 11:07:53PM -0500, Waiman Long wrote:
@@ -889,6 +892,20 @@ rwsem_spin_on_owner(struct rw_semaphore *sem)
}
#endif
+/*
+ * Common code to handle rwsem flags in out_nolock path with
wait_lock held.
+ * If there is more than one waiter in the queue and the HANDOFF
bit is set,
+ * the next waiter will inherit it if the first waiter is removed.
+ */
+static inline void rwsem_out_nolock_clear_flags(struct rw_semaphore
*sem,
+ struct rwsem_waiter *waiter)
I'm going to rename that, it doesn't just clear flags, it dequeues the
waiter.
Argh, rwsem_mark_wake() doesn't clear HANDOFF when list_empty(), and
write_slowpath() is *far* too clever about all of this.
+{
+ list_del(&waiter->list);
if (list_empty(&sem->wait_list)) {
+ atomic_long_andnot(RWSEM_FLAG_HANDOFF | RWSEM_FLAG_WAITERS,
+ &sem->count);
}
+}
@@ -1098,7 +1110,7 @@ rwsem_down_write_slowpath(struct rw_semaphore
*sem, int state)
* In this case, we attempt to acquire the lock again
* without sleeping.
*/
- if (wstate == WRITER_HANDOFF) {
+ if (waiter.handoff_set) {
I'm thinking this wants to be something like:
if (rwsem_first_waiter(sem) == &waiter && waiter.handoff_set) {
enum owner_state owner_state;
preempt_disable();
How's this (on top) then?
---
--- a/kernel/locking/rwsem.c
+++ b/kernel/locking/rwsem.c
@@ -104,11 +104,10 @@
* atomic_long_fetch_add() is used to obtain reader lock, whereas
* atomic_long_cmpxchg() will be used to obtain writer lock.
*
- * There are four places where the lock handoff bit may be set or
cleared.
- * 1) rwsem_mark_wake() for readers -- set, clear
- * 2) rwsem_try_write_lock() for writers -- set, clear
- * 3) Error path of rwsem_down_write_slowpath() -- clear
- * 4) Error path of rwsem_down_read_slowpath() -- clear
+ * There are three places where the lock handoff bit may be set or
cleared.
+ * 1) rwsem_mark_wake() for readers -- set, clear
+ * 2) rwsem_try_write_lock() for writers -- set, clear
+ * 3) rwsem_del_waiter() -- clear
*
* For all the above cases, wait_lock will be held. A writer must also
* be the first one in the wait_list to be eligible for setting the
handoff
@@ -363,6 +362,31 @@ enum rwsem_wake_type {
*/
#define MAX_READERS_WAKEUP 0x100
+static inline void
+rwsem_add_waiter(struct rw_semaphore *sem, struct rwsem_waiter *waiter)
+{
+ lockdep_assert_held(&sem->wait_lock);
+ list_add_tail(&waiter->list, &sem->wait_list);
+ /* caller will set RWSEM_FLAG_WAITERS */
/* 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.
+}
+
+/*
+ * 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);
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.
+ 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);
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.
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