Messages in this thread | | | Date | Thu, 11 Nov 2021 21:26:47 +0100 | From | Peter Zijlstra <> | Subject | Re: [BUG]locking/rwsem: only clean RWSEM_FLAG_HANDOFF when already set |
| |
On Thu, Nov 11, 2021 at 02:36:52PM -0500, Waiman Long wrote:
> @@ -434,6 +430,7 @@ static void rwsem_mark_wake(struct rw_semaphore *sem, > if (!(oldcount & RWSEM_FLAG_HANDOFF) && > time_after(jiffies, waiter->timeout)) { > adjustment -= RWSEM_FLAG_HANDOFF; > + waiter->handoff_set = true; > lockevent_inc(rwsem_rlock_handoff); > } >
Do we really need this flag? Wouldn't it be the same as waiter-is-first AND sem-has-handoff ?
> static inline bool rwsem_try_write_lock(struct rw_semaphore *sem, > + struct rwsem_waiter *waiter) > { > long count, new; > + bool first = rwsem_first_waiter(sem) == waiter;
flip those lines for reverse xmas please
> > lockdep_assert_held(&sem->wait_lock); > > @@ -546,13 +541,14 @@ static inline bool rwsem_try_write_lock(struct rw_semaphore *sem, > do { > bool has_handoff = !!(count & RWSEM_FLAG_HANDOFF); > > - if (has_handoff && wstate == WRITER_NOT_FIRST) > + if (has_handoff && !first) > return false; > > new = count; > > if (count & RWSEM_LOCK_MASK) { > - if (has_handoff || (wstate != WRITER_HANDOFF)) > + if (has_handoff || (!waiter->rt_task && > + !time_after(jiffies, waiter->timeout)))
Does ->rt_task really help over rt_task(current) ? I suppose there's an argument for locality, but that should be pretty much it, no?
> return false; > > new |= RWSEM_FLAG_HANDOFF; > @@ -889,6 +888,24 @@ rwsem_spin_on_owner(struct rw_semaphore *sem) > } > #endif > > +/* > + * Common code to handle rwsem flags in out_nolock path with wait_lock held. > + */ > +static inline void rwsem_out_nolock_clear_flags(struct rw_semaphore *sem, > + struct rwsem_waiter *waiter) > +{ > + long flags = 0; > + > + list_del(&waiter->list); > + if (list_empty(&sem->wait_list)) > + flags = RWSEM_FLAG_HANDOFF | RWSEM_FLAG_WAITERS; > + else if (waiter->handoff_set) > + flags = RWSEM_FLAG_HANDOFF; > + > + if (flags) > + atomic_long_andnot(flags, &sem->count); > +}
Right, so I like sharing this between the two _slowpath functions, that makes sense.
The difference between this and my approach is that I unconditionally clear HANDOFF when @waiter was the first. Because if it was set, it must've been ours, and if it wasn't set, clearing it doesn't really hurt much. This is an unlikely path, I don't think the potentially extra atomic is an issue here.
> + > /* > * Wait for the read lock to be granted > */ > @@ -936,6 +953,7 @@ rwsem_down_read_slowpath(struct rw_semaphore *sem, long count, unsigned int stat > waiter.task = current; > waiter.type = RWSEM_WAITING_FOR_READ; > waiter.timeout = jiffies + RWSEM_WAIT_TIMEOUT; > + waiter.handoff_set = false;
Forgot to set rt_task
> > raw_spin_lock_irq(&sem->wait_lock); > if (list_empty(&sem->wait_list)) {
> @@ -1038,16 +1051,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;
Forget to set handoff_set
> + waiter.rt_task = rt_task(current); > > raw_spin_lock_irq(&sem->wait_lock);
Again, I'm not convinced we need these variables.
> @@ -1083,13 +1093,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 +1111,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();
Does it matter much if we spin-wait for every first or only for handoff?
Either way around, I think spin-wait ought to terminate on sigpending (same for mutex I suppose).
> @@ -1109,40 +1122,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); > }
Nice.
| |