Re: [RFC] Avoid mutex starvation when optimistic spinning is disabled

[Imre Deak]

On ti, 2016-07-19 at 21:39 -0700, Jason Low wrote:
> On Tue, 2016-07-19 at 16:04 -0700, Jason Low wrote:
> > Hi Imre,
> > 
> > Here is a patch which prevents a thread from spending too much "time"
> > waiting for a mutex in the !CONFIG_MUTEX_SPIN_ON_OWNER case.
> > 
> > Would you like to try this out and see if this addresses the mutex
> > starvation issue you are seeing in your workload when optimistic
> > spinning is disabled?
> 
> Although it looks like it didn't take care of the 'lock stealing' case
> in the slowpath. Here is the updated fixed version:

This also got rid of the problem, I only needed to change the ww
functions accordingly. Also, imo mutex_trylock() needs the same
handling and lock->yield_to_waiter should be reset only in the waiter
thread that has set it.

--Imre

> ---
> Signed-off-by: Jason Low <jason.low2@xxxxxxx>
> ---
> Âinclude/linux/mutex.hÂÂ|ÂÂ2 ++
> Âkernel/locking/mutex.c | 65 ++++++++++++++++++++++++++++++++++++++++++++------
> Â2 files changed, 60 insertions(+), 7 deletions(-)
> 
> diff --git a/include/linux/mutex.h b/include/linux/mutex.h
> index 2cb7531..c1ca68d 100644
> --- a/include/linux/mutex.h
> +++ b/include/linux/mutex.h
> @@ -57,6 +57,8 @@ struct mutex {
> Â#endif
> Â#ifdef CONFIG_MUTEX_SPIN_ON_OWNER
> Â	struct optimistic_spin_queue osq; /* Spinner MCS lock */
> +#else
> +	bool yield_to_waiter; /* Prevent starvation when spinning disabled */
> Â#endif
> Â#ifdef CONFIG_DEBUG_MUTEXES
> Â	void			*magic;
> diff --git a/kernel/locking/mutex.c b/kernel/locking/mutex.c
> index a70b90d..6c915ca 100644
> --- a/kernel/locking/mutex.c
> +++ b/kernel/locking/mutex.c
> @@ -55,6 +55,8 @@ __mutex_init(struct mutex *lock, const char *name, struct lock_class_key *key)
> Â	mutex_clear_owner(lock);
> Â#ifdef CONFIG_MUTEX_SPIN_ON_OWNER
> Â	osq_lock_init(&lock->osq);
> +#else
> +	lock->yield_to_waiter = false;
> Â#endif
> Â
> Â	debug_mutex_init(lock, name, key);
> @@ -71,6 +73,9 @@ EXPORT_SYMBOL(__mutex_init);
> Â */
> Â__visible void __sched __mutex_lock_slowpath(atomic_t *lock_count);
> Â
> +
> +static inline bool need_yield_to_waiter(struct mutex *lock);
> +
> Â/**
> Â * mutex_lock - acquire the mutex
> Â * @lock: the mutex to be acquired
> @@ -95,11 +100,15 @@ __visible void __sched __mutex_lock_slowpath(atomic_t *lock_count);
> Âvoid __sched mutex_lock(struct mutex *lock)
> Â{
> Â	might_sleep();
> +
> Â	/*
> Â	Â* The locking fastpath is the 1->0 transition from
> Â	Â* 'unlocked' into 'locked' state.
> Â	Â*/
> -	__mutex_fastpath_lock(&lock->count, __mutex_lock_slowpath);
> +	if (!need_yield_to_waiter(lock))
> +		__mutex_fastpath_lock(&lock->count, __mutex_lock_slowpath);
> +	else
> +		__mutex_lock_slowpath(&lock->count);
> Â	mutex_set_owner(lock);
> Â}
> Â
> @@ -398,12 +407,39 @@ done:
> Â
> Â	return false;
> Â}
> +
> +static inline void do_yield_to_waiter(struct mutex *lock, int loops)
> +{
> +	return;
> +}
> +
> +static inline bool need_yield_to_waiter(struct mutex *lock)
> +{
> +	return false;
> +}
> +
> Â#else
> Âstatic bool mutex_optimistic_spin(struct mutex *lock,
> Â				ÂÂstruct ww_acquire_ctx *ww_ctx, const bool use_ww_ctx)
> Â{
> Â	return false;
> Â}
> +
> +#define MUTEX_MAX_WAIT 32
> +
> +static inline void do_yield_to_waiter(struct mutex *lock, int loops)
> +{
> +	if (loops < MUTEX_MAX_WAIT)
> +		return;
> +
> +	if (lock->yield_to_waiter != true)
> +		lock->yield_to_waiter =true;
> +}
> +
> +static inline bool need_yield_to_waiter(struct mutex *lock)
> +{
> +	return lock->yield_to_waiter;
> +}
> Â#endif
> Â
> Â__visible __used noinline
> @@ -510,6 +546,7 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
> Â	struct mutex_waiter waiter;
> Â	unsigned long flags;
> Â	int ret;
> +	int loop = 0;
> Â
> Â	if (use_ww_ctx) {
> Â		struct ww_mutex *ww = container_of(lock, struct ww_mutex, base);
> @@ -532,7 +569,7 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
> Â	Â* Once more, try to acquire the lock. Only try-lock the mutex if
> Â	Â* it is unlocked to reduce unnecessary xchg() operations.
> Â	Â*/
> -	if (!mutex_is_locked(lock) &&
> +	if (!need_yield_to_waiter(lock) && !mutex_is_locked(lock) &&
> Â	ÂÂÂÂ(atomic_xchg_acquire(&lock->count, 0) == 1))
> Â		goto skip_wait;
> Â
> @@ -546,6 +583,8 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
> Â	lock_contended(&lock->dep_map, ip);
> Â
> Â	for (;;) {
> +		loop++;
> +
> Â		/*
> Â		Â* Lets try to take the lock again - this is needed even if
> Â		Â* we get here for the first time (shortly after failing to
> @@ -556,7 +595,8 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
> Â		Â* other waiters. We only attempt the xchg if the count is
> Â		Â* non-negative in order to avoid unnecessary xchg operations:
> Â		Â*/
> -		if (atomic_read(&lock->count) >= 0 &&
> +		if ((!need_yield_to_waiter(lock) || loop > 1) &&
> +		ÂÂÂÂatomic_read(&lock->count) >= 0 &&
> Â		ÂÂÂÂ(atomic_xchg_acquire(&lock->count, -1) == 1))
> Â			break;
> Â
> @@ -581,6 +621,7 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
> Â		spin_unlock_mutex(&lock->wait_lock, flags);
> Â		schedule_preempt_disabled();
> Â		spin_lock_mutex(&lock->wait_lock, flags);
> +		do_yield_to_waiter(lock, loop);
> Â	}
> Â	__set_task_state(task, TASK_RUNNING);
> Â
> @@ -590,6 +631,10 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
> Â		atomic_set(&lock->count, 0);
> Â	debug_mutex_free_waiter(&waiter);
> Â
> +#ifndef CONFIG_MUTEX_SPIN_ON_OWNER
> +	lock->yield_to_waiter = false;
> +#endif
> +
> Âskip_wait:
> Â	/* got the lock - cleanup and rejoice! */
> Â	lock_acquired(&lock->dep_map, ip);
> @@ -789,10 +834,13 @@ __mutex_lock_interruptible_slowpath(struct mutex *lock);
> Â */
> Âint __sched mutex_lock_interruptible(struct mutex *lock)
> Â{
> -	int ret;
> +	int ret = 1;
> Â
> Â	might_sleep();
> -	ret =ÂÂ__mutex_fastpath_lock_retval(&lock->count);
> +
> +	if (!need_yield_to_waiter(lock))
> +		ret =ÂÂ__mutex_fastpath_lock_retval(&lock->count);
> +
> Â	if (likely(!ret)) {
> Â		mutex_set_owner(lock);
> Â		return 0;
> @@ -804,10 +852,13 @@ EXPORT_SYMBOL(mutex_lock_interruptible);
> Â
> Âint __sched mutex_lock_killable(struct mutex *lock)
> Â{
> -	int ret;
> +	int ret = 1;
> Â
> Â	might_sleep();
> -	ret = __mutex_fastpath_lock_retval(&lock->count);
> +
> +	if (!need_yield_to_waiter(lock))
> +		ret = __mutex_fastpath_lock_retval(&lock->count);
> +
> Â	if (likely(!ret)) {
> Â		mutex_set_owner(lock);
> Â		return 0;