[PATCH v4] locking/mutex: Prevent lock starvation when spinning is disabled

[PATCH v4] locking/mutex: Prevent lock starvation when spinning is disabled

[Jason Low]

Imre reported an issue where threads are getting starved when trying
to acquire a mutex. Threads acquiring a mutex can get arbitrarily delayed
sleeping on a mutex because other threads can continually steal the lock
in the fastpath and/or through optimistic spinning.

Waiman has developed patches that allow waiters to return to optimistic
spinning, thus reducing the probability that starvation occurs. However,
Imre still sees this starvation problem in the workloads when optimistic
spinning is disabled.

This patch adds an additional boolean to the mutex that gets used in
the CONFIG_SMP && !CONFIG_MUTEX_SPIN_ON_OWNER cases. The flag signifies
whether or not other threads need to yield to a waiter and gets set
when a waiter spends too much time waiting for the mutex. The threshold
is currently set to 16 wakeups, and once the wakeup threshold is exceeded,
other threads must yield to the top waiter. The flag gets cleared
immediately after the top waiter acquires the mutex.

This prevents waiters from getting starved without sacrificing much
much performance, as lock stealing is still allowed and only
temporarily disabled when it is detected that a waiter has been waiting
for too long.

Reported-by: Imre Deak <imre.deak@intel.com>
Signed-off-by: Jason Low <jason.low2@hpe.com>
---
 include/linux/mutex.h  |   2 +
 kernel/locking/mutex.c | 122 +++++++++++++++++++++++++++++++++++++++----------
 2 files changed, 99 insertions(+), 25 deletions(-)

diff --git a/include/linux/mutex.h b/include/linux/mutex.h
index f8e91ad..988c020 100644
--- a/include/linux/mutex.h
+++ b/include/linux/mutex.h
@@ -58,6 +58,8 @@ struct mutex {
 #ifdef CONFIG_MUTEX_SPIN_ON_OWNER
 	struct optimistic_spin_queue osq; /* Spinner MCS lock */
 	int waiter_spinning;
+#elif defined(CONFIG_SMP)
+	int yield_to_waiter;
 #endif
 #ifdef CONFIG_DEBUG_MUTEXES
 	void			*magic;
diff --git a/kernel/locking/mutex.c b/kernel/locking/mutex.c
index 64a0bfa..e078c49 100644
--- a/kernel/locking/mutex.c
+++ b/kernel/locking/mutex.c
@@ -56,6 +56,8 @@ __mutex_init(struct mutex *lock, const char *name, struct lock_class_key *key)
 #ifdef CONFIG_MUTEX_SPIN_ON_OWNER
 	osq_lock_init(&lock->osq);
 	lock->waiter_spinning = false;
+#elif defined(CONFIG_SMP)
+	lock->yield_to_waiter = false;
 #endif
 
 	debug_mutex_init(lock, name, key);
@@ -72,6 +74,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
@@ -100,7 +105,10 @@ void __sched mutex_lock(struct mutex *lock)
 	 * 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);
 }
 
@@ -449,6 +457,49 @@ static bool mutex_optimistic_spin(struct mutex *lock,
 }
 #endif
 
+#if !defined(CONFIG_MUTEX_SPIN_ON_OWNER) && defined(CONFIG_SMP)
+
+#define MUTEX_WAKEUP_THRESHOLD 16
+
+static inline void update_yield_to_waiter(struct mutex *lock, int *wakeups)
+{
+	if (++(*wakeups) > MUTEX_WAKEUP_THRESHOLD && !lock->yield_to_waiter)
+		lock->yield_to_waiter = true;
+}
+
+static inline void clear_yield_to_waiter(struct mutex *lock,
+					 struct mutex_waiter *waiter)
+{
+	/*  Only clear yield_to_waiter if we are the top waiter. */
+	if (lock->wait_list.next == &waiter->list && lock->yield_to_waiter)
+		lock->yield_to_waiter = false;
+}
+
+static inline bool need_yield_to_waiter(struct mutex *lock)
+{
+	return unlikely(lock->yield_to_waiter);
+}
+
+#else /* !yield_to_waiter */
+
+static inline void update_yield_to_waiter(struct mutex *lock, int *wakeups)
+{
+	return;
+}
+
+static inline void clear_yield_to_waiter(struct mutex *lock,
+					 struct mutex_waiter *waiter)
+{
+	return;
+}
+
+static inline bool need_yield_to_waiter(struct mutex *lock)
+{
+	return false;
+}
+
+#endif /* yield_to_waiter */
+
 __visible __used noinline
 void __sched __mutex_unlock_slowpath(atomic_t *lock_count);
 
@@ -541,6 +592,12 @@ __ww_mutex_lock_check_stamp(struct mutex *lock, struct ww_acquire_ctx *ctx)
 	return 0;
 }
 
+static inline bool __mutex_trylock_pending(struct mutex *lock)
+{
+	return atomic_read(&lock->count) >= 0 &&
+	       atomic_xchg_acquire(&lock->count, -1) == 1;
+}
+
 /*
  * Lock a mutex (possibly interruptible), slowpath:
  */
@@ -553,7 +610,7 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
 	struct mutex_waiter waiter;
 	unsigned long flags;
 	bool  acquired = false;	/* True if the lock is acquired */
-	int ret;
+	int ret, wakeups = 0;
 
 	if (use_ww_ctx) {
 		struct ww_mutex *ww = container_of(lock, struct ww_mutex, base);
@@ -576,7 +633,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;
 
@@ -587,24 +644,18 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
 	list_add_tail(&waiter.list, &lock->wait_list);
 	waiter.task = task;
 
+	/*
+	 * If this is the first waiter, mark the lock as having pending
+	 * waiters, if we happen to acquire it while doing so, yay!
+	 */
+	if (list_is_singular(&lock->wait_list) &&
+	    __mutex_trylock_pending(lock))
+		goto remove_waiter;
+
 	lock_contended(&lock->dep_map, ip);
 
 	while (!acquired) {
 		/*
-		 * Lets try to take the lock again - this is needed even if
-		 * we get here for the first time (shortly after failing to
-		 * acquire the lock), to make sure that we get a wakeup once
-		 * it's unlocked. Later on, if we sleep, this is the
-		 * operation that gives us the lock. We xchg it to -1, so
-		 * that when we release the lock, we properly wake up the
-		 * 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 &&
-		    (atomic_xchg_acquire(&lock->count, -1) == 1))
-			break;
-
-		/*
 		 * got a signal? (This code gets eliminated in the
 		 * TASK_UNINTERRUPTIBLE case.)
 		 */
@@ -631,9 +682,21 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
 		acquired = mutex_optimistic_spin(lock, ww_ctx, use_ww_ctx,
 						 true);
 		spin_lock_mutex(&lock->wait_lock, flags);
+
+		update_yield_to_waiter(lock, &wakeups);
+
+		/*
+		 * Try-acquire now that we got woken at the head of the queue
+		 * or we received a signal.
+		 */
+		if (__mutex_trylock_pending(lock))
+			break;
 	}
 	__set_task_state(task, TASK_RUNNING);
 
+	clear_yield_to_waiter(lock, &waiter);
+
+remove_waiter:
 	mutex_remove_waiter(lock, &waiter, task);
 	/* set it to 0 if there are no waiters left: */
 	if (likely(list_empty(&lock->wait_list)))
@@ -659,6 +722,7 @@ unlock:
 	return 0;
 
 err:
+	clear_yield_to_waiter(lock, &waiter);
 	mutex_remove_waiter(lock, &waiter, task);
 	spin_unlock_mutex(&lock->wait_lock, flags);
 	debug_mutex_free_waiter(&waiter);
@@ -843,10 +907,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;
@@ -858,10 +925,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;
@@ -971,11 +1041,12 @@ EXPORT_SYMBOL(mutex_trylock);
 int __sched
 __ww_mutex_lock(struct ww_mutex *lock, struct ww_acquire_ctx *ctx)
 {
-	int ret;
+	int ret = 1;
 
 	might_sleep();
 
-	ret = __mutex_fastpath_lock_retval(&lock->base.count);
+	if (!need_yield_to_waiter(&lock->base))
+		ret = __mutex_fastpath_lock_retval(&lock->base.count);
 
 	if (likely(!ret)) {
 		ww_mutex_set_context_fastpath(lock, ctx);
@@ -989,11 +1060,12 @@ EXPORT_SYMBOL(__ww_mutex_lock);
 int __sched
 __ww_mutex_lock_interruptible(struct ww_mutex *lock, struct ww_acquire_ctx *ctx)
 {
-	int ret;
+	int ret = 1;
 
 	might_sleep();
 
-	ret = __mutex_fastpath_lock_retval(&lock->base.count);
+	if (!need_yield_to_waiter(&lock->base))
+		ret = __mutex_fastpath_lock_retval(&lock->base.count);
 
 	if (likely(!ret)) {
 		ww_mutex_set_context_fastpath(lock, ctx);
-- 
2.1.4

Re: [PATCH v4] locking/mutex: Prevent lock starvation when spinning is disabled

[Jason Low]

On Thu, 2016-08-18 at 17:39 -0700, Jason Low wrote:
> Imre reported an issue where threads are getting starved when trying
> to acquire a mutex. Threads acquiring a mutex can get arbitrarily delayed
> sleeping on a mutex because other threads can continually steal the lock
> in the fastpath and/or through optimistic spinning.
> 
> Waiman has developed patches that allow waiters to return to optimistic
> spinning, thus reducing the probability that starvation occurs. However,
> Imre still sees this starvation problem in the workloads when optimistic
> spinning is disabled.
> 
> This patch adds an additional boolean to the mutex that gets used in
> the CONFIG_SMP && !CONFIG_MUTEX_SPIN_ON_OWNER cases. The flag signifies
> whether or not other threads need to yield to a waiter and gets set
> when a waiter spends too much time waiting for the mutex. The threshold
> is currently set to 16 wakeups, and once the wakeup threshold is exceeded,
> other threads must yield to the top waiter. The flag gets cleared
> immediately after the top waiter acquires the mutex.
> 
> This prevents waiters from getting starved without sacrificing much
> much performance, as lock stealing is still allowed and only
> temporarily disabled when it is detected that a waiter has been waiting
> for too long.

Changes from v3 (in peterz locking/core) -> v4:

1. Fixed patch title. It should be "Prevent lock starvation when spinning is
   disabled" instead of "when spinning is enabled".

2. Call clear_yield_to_waiter() when a top waiter exits in the 'err' case.

3. Only clear yield_to_waiter if the thread is the top waiter and not if it
   is a non-top waiter that received a signal.

Re: [PATCH v4] locking/mutex: Prevent lock starvation when spinning is disabled

[Peter Zijlstra]

On Thu, Aug 18, 2016 at 09:11:16PM -0700, Jason Low wrote:

> 3. Only clear yield_to_waiter if the thread is the top waiter and not if it
>    is a non-top waiter that received a signal.

Ah, and that is an equivalent condition to the singular one I use to
test if we need to set pending, since if we just add waiter and its the
top waiter, it must be the only waiter.

How about something like so on top?

--- a/kernel/locking/mutex.c
+++ b/kernel/locking/mutex.c
@@ -74,7 +74,6 @@ EXPORT_SYMBOL(__mutex_init);
  */
 __visible void __sched __mutex_lock_slowpath(atomic_t *lock_count);
 
-
 static inline bool need_yield_to_waiter(struct mutex *lock);
 
 /**
@@ -457,6 +456,12 @@ static bool mutex_optimistic_spin(struct
 }
 #endif
 
+static inline bool
+__mutex_waiter_is_head(struct mutex *lock, struct mutex_waiter *waiter)
+{
+	return list_first_entry(&lock->wait_list, struct mutex_waiter, list) == waiter;
+}
+
 #if !defined(CONFIG_MUTEX_SPIN_ON_OWNER) && defined(CONFIG_SMP)
 
 #define MUTEX_WAKEUP_THRESHOLD 16
@@ -471,7 +476,7 @@ static inline void clear_yield_to_waiter
 					 struct mutex_waiter *waiter)
 {
 	/*  Only clear yield_to_waiter if we are the top waiter. */
-	if (lock->wait_list.next == &waiter->list && lock->yield_to_waiter)
+	if (lock->yield_to_waiter && __mutex_waiter_is_head(lock, waiter))
 		lock->yield_to_waiter = false;
 }
 
@@ -648,7 +653,7 @@ __mutex_lock_common(struct mutex *lock,
 	 * If this is the first waiter, mark the lock as having pending
 	 * waiters, if we happen to acquire it while doing so, yay!
 	 */
-	if (list_is_singular(&lock->wait_list) &&
+	if (__mutex_waiter_is_head(lock, &waiter) &&
 	    __mutex_trylock_pending(lock))
 		goto remove_waiter;
 

Re: [PATCH v4] locking/mutex: Prevent lock starvation when spinning is disabled

[Waiman Long]

On 08/18/2016 08:39 PM, Jason Low wrote:
> Imre reported an issue where threads are getting starved when trying
> to acquire a mutex. Threads acquiring a mutex can get arbitrarily delayed
> sleeping on a mutex because other threads can continually steal the lock
> in the fastpath and/or through optimistic spinning.
>
> Waiman has developed patches that allow waiters to return to optimistic
> spinning, thus reducing the probability that starvation occurs. However,
> Imre still sees this starvation problem in the workloads when optimistic
> spinning is disabled.
>
> This patch adds an additional boolean to the mutex that gets used in
> the CONFIG_SMP&&  !CONFIG_MUTEX_SPIN_ON_OWNER cases. The flag signifies
> whether or not other threads need to yield to a waiter and gets set
> when a waiter spends too much time waiting for the mutex. The threshold
> is currently set to 16 wakeups, and once the wakeup threshold is exceeded,
> other threads must yield to the top waiter. The flag gets cleared
> immediately after the top waiter acquires the mutex.
>
> This prevents waiters from getting starved without sacrificing much
> much performance, as lock stealing is still allowed and only
> temporarily disabled when it is detected that a waiter has been waiting
> for too long.
>
> Reported-by: Imre Deak<imre.deak@intel.com>
> Signed-off-by: Jason Low<jason.low2@hpe.com>
> ---
>   include/linux/mutex.h  |   2 +
>   kernel/locking/mutex.c | 122 +++++++++++++++++++++++++++++++++++++++----------
>   2 files changed, 99 insertions(+), 25 deletions(-)
>
> diff --git a/include/linux/mutex.h b/include/linux/mutex.h
> index f8e91ad..988c020 100644
> --- a/include/linux/mutex.h
> +++ b/include/linux/mutex.h
> @@ -58,6 +58,8 @@ struct mutex {
>   #ifdef CONFIG_MUTEX_SPIN_ON_OWNER
>   	struct optimistic_spin_queue osq; /* Spinner MCS lock */
>   	int waiter_spinning;
> +#elif defined(CONFIG_SMP)
> +	int yield_to_waiter;
>   #endif
>   #ifdef CONFIG_DEBUG_MUTEXES
>   	void			*magic;
> diff --git a/kernel/locking/mutex.c b/kernel/locking/mutex.c
> index 64a0bfa..e078c49 100644
> --- a/kernel/locking/mutex.c
> +++ b/kernel/locking/mutex.c
> @@ -56,6 +56,8 @@ __mutex_init(struct mutex *lock, const char *name, struct lock_class_key *key)
>   #ifdef CONFIG_MUTEX_SPIN_ON_OWNER
>   	osq_lock_init(&lock->osq);
>   	lock->waiter_spinning = false;
> +#elif defined(CONFIG_SMP)
> +	lock->yield_to_waiter = false;
>   #endif
>
>   	debug_mutex_init(lock, name, key);
> @@ -72,6 +74,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
> @@ -100,7 +105,10 @@ void __sched mutex_lock(struct mutex *lock)
>   	 * 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);
>   }
>
> @@ -449,6 +457,49 @@ static bool mutex_optimistic_spin(struct mutex *lock,
>   }
>   #endif
>
> +#if !defined(CONFIG_MUTEX_SPIN_ON_OWNER)&&  defined(CONFIG_SMP)
> +
> +#define MUTEX_WAKEUP_THRESHOLD 16
> +
> +static inline void update_yield_to_waiter(struct mutex *lock, int *wakeups)
> +{
> +	if (++(*wakeups)>  MUTEX_WAKEUP_THRESHOLD&&  !lock->yield_to_waiter)
> +		lock->yield_to_waiter = true;
> +}
> +
> +static inline void clear_yield_to_waiter(struct mutex *lock,
> +					 struct mutex_waiter *waiter)
> +{
> +	/*  Only clear yield_to_waiter if we are the top waiter. */
> +	if (lock->wait_list.next ==&waiter->list&&  lock->yield_to_waiter)
> +		lock->yield_to_waiter = false;
> +}
> +
> +static inline bool need_yield_to_waiter(struct mutex *lock)
> +{
> +	return unlikely(lock->yield_to_waiter);
> +}
> +
> +#else /* !yield_to_waiter */
> +
> +static inline void update_yield_to_waiter(struct mutex *lock, int *wakeups)
> +{
> +	return;
> +}
> +
> +static inline void clear_yield_to_waiter(struct mutex *lock,
> +					 struct mutex_waiter *waiter)
> +{
> +	return;
> +}
> +
> +static inline bool need_yield_to_waiter(struct mutex *lock)
> +{
> +	return false;
> +}
> +
> +#endif /* yield_to_waiter */
> +
>   __visible __used noinline
>   void __sched __mutex_unlock_slowpath(atomic_t *lock_count);
>
> @@ -541,6 +592,12 @@ __ww_mutex_lock_check_stamp(struct mutex *lock, struct ww_acquire_ctx *ctx)
>   	return 0;
>   }
>
> +static inline bool __mutex_trylock_pending(struct mutex *lock)
> +{
> +	return atomic_read(&lock->count)>= 0&&
> +	       atomic_xchg_acquire(&lock->count, -1) == 1;
> +}
> +

Maybe you can make a more general __mutex_trylock function that is used 
in all three trylock attempts in the slowpath. For example,

static inline bool __mutex_trylock(struct mutex *lock, bool waiter)
{
     if (waiter) {
         return atomic_read(&lock->count) >= 0 &&
                atomic_xchg_acquire(&lock->count, -1) == 1;
     } else {
         return !need_yield_to_waiter(lock) &&
                !mutex_is_locked(lock) &&
                ((atomic_xchg_acquire(&lock->count, 0) == 1);
     }
}
>   /*
>    * Lock a mutex (possibly interruptible), slowpath:
>    */
> @@ -553,7 +610,7 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
>   	struct mutex_waiter waiter;
>   	unsigned long flags;
>   	bool  acquired = false;	/* True if the lock is acquired */
> -	int ret;
> +	int ret, wakeups = 0;
>
>   	if (use_ww_ctx) {
>   		struct ww_mutex *ww = container_of(lock, struct ww_mutex, base);
> @@ -576,7 +633,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;
>
> @@ -587,24 +644,18 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
>   	list_add_tail(&waiter.list,&lock->wait_list);
>   	waiter.task = task;
>
> +	/*
> +	 * If this is the first waiter, mark the lock as having pending
> +	 * waiters, if we happen to acquire it while doing so, yay!
> +	 */
> +	if (list_is_singular(&lock->wait_list)&&
> +	    __mutex_trylock_pending(lock))
> +		goto remove_waiter;
> +
>   	lock_contended(&lock->dep_map, ip);
>
>   	while (!acquired) {
>   		/*
> -		 * Lets try to take the lock again - this is needed even if
> -		 * we get here for the first time (shortly after failing to
> -		 * acquire the lock), to make sure that we get a wakeup once
> -		 * it's unlocked. Later on, if we sleep, this is the
> -		 * operation that gives us the lock. We xchg it to -1, so
> -		 * that when we release the lock, we properly wake up the
> -		 * 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&&
> -		    (atomic_xchg_acquire(&lock->count, -1) == 1))
> -			break;
> -
> -		/*
>   		 * got a signal? (This code gets eliminated in the
>   		 * TASK_UNINTERRUPTIBLE case.)
>   		 */
> @@ -631,9 +682,21 @@ __mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
>   		acquired = mutex_optimistic_spin(lock, ww_ctx, use_ww_ctx,
>   						 true);
>   		spin_lock_mutex(&lock->wait_lock, flags);
> +
> +		update_yield_to_waiter(lock,&wakeups);
> +
> +		/*
> +		 * Try-acquire now that we got woken at the head of the queue
> +		 * or we received a signal.
> +		 */
> +		if (__mutex_trylock_pending(lock))
> +			break;

That is not quite right. The lock may have been acquired in the 
optimistic spinning loop. You either have to move it back to the top or 
add a "!acquired" check before the trylock.

Cheers,
Longman

Re: [PATCH v4] locking/mutex: Prevent lock starvation when spinning is disabled

[Peter Zijlstra]

Please trim your emails..

On Fri, Aug 19, 2016 at 12:57:06PM -0400, Waiman Long wrote:

> >+static inline bool __mutex_trylock_pending(struct mutex *lock)
> >+{
> >+	return atomic_read(&lock->count)>= 0&&
> >+	       atomic_xchg_acquire(&lock->count, -1) == 1;
> >+}
> >+
> 
> Maybe you can make a more general __mutex_trylock function that is used in
> all three trylock attempts in the slowpath. For example,
> 
> static inline bool __mutex_trylock(struct mutex *lock, bool waiter)
> {
>     if (waiter) {
>         return atomic_read(&lock->count) >= 0 &&
>                atomic_xchg_acquire(&lock->count, -1) == 1;
>     } else {
>         return !need_yield_to_waiter(lock) &&
>                !mutex_is_locked(lock) &&
>                ((atomic_xchg_acquire(&lock->count, 0) == 1);
>     }
> }

That seems more messy to me..

Re: [PATCH v4] locking/mutex: Prevent lock starvation when spinning is disabled

[Linus Torvalds]

On Fri, Aug 19, 2016 at 12:33 PM, Peter Zijlstra <peterz@infradead.org> wrote:
>
> That seems more messy to me..

Can't we just say that we always spin? Are there any really valid
cases where spinning isn't ok?

We've historically disabled spinning when mutex debugging is enabled,
but since the spinning is limited anyway, couldn't we just spin even
with debugging enabled?

I hate how these patches are trying to solve a problem that doesn't
even happen under normal circumstances, and add special-case code for
something that is already a special-case condition. So rather than
adding even more special cases, could we look at _removing_ the
special cases that cause problems instead?

                Linus

Re: [PATCH v4] locking/mutex: Prevent lock starvation when spinning is disabled

[Peter Zijlstra]

On Fri, Aug 19, 2016 at 12:45:27PM -0700, Linus Torvalds wrote:
> On Fri, Aug 19, 2016 at 12:33 PM, Peter Zijlstra <peterz@infradead.org> wrote:
> >
> > That seems more messy to me..
> 
> Can't we just say that we always spin? Are there any really valid
> cases where spinning isn't ok?
> 
> We've historically disabled spinning when mutex debugging is enabled,
> but since the spinning is limited anyway, couldn't we just spin even
> with debugging enabled?
> 
> I hate how these patches are trying to solve a problem that doesn't
> even happen under normal circumstances, and add special-case code for
> something that is already a special-case condition. So rather than
> adding even more special cases, could we look at _removing_ the
> special cases that cause problems instead?

So I think I have a bunch of patches that solves the fundamental issues,
_However_... they're quite invasive and would need some serious
benchmarking.

I'll post them in a separate thread as RFC..