Re: [RFC PATCH-queue/locking/rfc 2/2] locking/mutex: Enable optimistic spinning of woken waiter

[Waiman Long <waiman.long@hpe.com>]

On 08/30/2016 11:08 AM, Peter Zijlstra wrote:
> On Fri, Aug 26, 2016 at 07:35:09PM -0400, Waiman Long wrote:
>
>> @@ -624,13 +649,24 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
>>   		/* didn't get the lock, go to sleep: */
>>   		spin_unlock_mutex(&lock->wait_lock, flags);
>>   		schedule_preempt_disabled();
>>
>> +		/*
>> +		 * Both __mutex_trylock() and __mutex_waiter_is_first()
>> +		 * can be done without the protection of wait_lock.
>> +		 */
> True, but it took me a little while to figure out why
> __mutex_waiter_is_first() is safe without the lock :-)

Yes, if you are the first waiter, the condition will not be changed even 
when new waiter is being added to the tail of the list.

>
>> +		acquired = __mutex_trylock(lock);
>>
>> +		if (!acquired&&  __mutex_waiter_is_first(lock,&waiter)) {
>>   			__mutex_set_flag(lock, MUTEX_FLAG_HANDOFF);
>> +			/*
>> +			 * Wait until the lock is handed off or the owner
>> +			 * sleeps.
>> +			 */
>> +			acquired = mutex_optimistic_spin(lock, ww_ctx,
>> +							 use_ww_ctx, true);
>> +		}
> That said; I think there's a few problems with this. Since we now poke
> at the loop termination conditions outside of the wait_lock, it becomes
> important where we do the task->state vs wakeup bits.
>
> Specifically, since we still have state==RUNNING here, its possible
> we'll fail to acquire the lock _and_ miss the wakeup from
> mutex_unlock(). Leaving us stuck forever more.
>
> Also, we should do the __mutex_trylock _after_ we set the handoff,
> otherwise its possible we get the lock handed (miss the wakeup as per
> the above) and fail to notice, again going back to sleep forever more.
>

Yes, you are right. I am less familiar with the intricacy of the 
sleep-wakeup interaction.

> @@ -638,7 +636,8 @@ __mutex_lock_common(struct mutex *lock,
>
>   	lock_contended(&lock->dep_map, ip);
>
> -	for (acquired = false; !acquired; ) {
> +	set_task_state(task, state);
> +	for (;;) {
>   		/*
>   		 * got a signal? (This code gets eliminated in the
>   		 * TASK_UNINTERRUPTIBLE case.)
> @@ -654,30 +653,23 @@ __mutex_lock_common(struct mutex *lock,
>   				goto err;
>   		}
>
> -		__set_task_state(task, state);
> -
> -		/* didn't get the lock, go to sleep: */
>   		spin_unlock_mutex(&lock->wait_lock, flags);
>   		schedule_preempt_disabled();
>
> -		/*
> -		 * Both __mutex_trylock() and __mutex_waiter_is_first()
> -		 * can be done without the protection of wait_lock.
> -		 */
> -		acquired = __mutex_trylock(lock, true);
> +		set_task_state(task, state);
>
> -		if (!acquired&&  __mutex_waiter_is_first(lock,&waiter)) {
> +		if (__mutex_waiter_is_first(lock,&waiter)) {
>   			__mutex_set_flag(lock, MUTEX_FLAG_HANDOFF);
> -			/*
> -			 * Wait until the lock is handed off or the owner
> -			 * sleeps.
> -			 */
> -			acquired = mutex_optimistic_spin(lock, ww_ctx,
> -							 use_ww_ctx, true);
> +			if (mutex_optimistic_spin(lock, ww_ctx, use_ww_ctx, true))
> +				break;
>   		}
>
> +		if (__mutex_trylock(lock, true))
> +			break;
> +

I think the set _task_state() can be moved to just before 
__mutex_trylock(). In this way, we can save a smp_mb() if we can get the 
lock in the optspin loop. Other than that, I am fine with the other changes.

Cheers,
Longman