| lists.openwall.net | lists / announce owl-users owl-dev john-users john-dev passwdqc-users yescrypt popa3d-users / oss-security kernel-hardening musl sabotage tlsify passwords / crypt-dev xvendor / Bugtraq Full-Disclosure linux-kernel linux-netdev linux-ext4 linux-hardening linux-cve-announce PHC | |
|
Open Source and information security mailing list archives
| ||
|
Date: Thu, 11 Nov 2021 21:26:47 +0100
From: Peter Zijlstra <peterz@...radead.org>
To: Waiman Long <longman@...hat.com>
Cc: Hillf Danton <hdanton@...a.com>,
马振华 <mazhenhua@...omi.com>,
mingo <mingo@...hat.com>, will <will@...nel.org>,
"boqun.feng" <boqun.feng@...il.com>,
linux-kernel <linux-kernel@...r.kernel.org>
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.
Powered by blists - more mailing lists