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

From: Waiman Long
Date: Fri Aug 19 2016 - 13:12:19 EST

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@xxxxxxxxx>
Signed-off-by: Jason Low<jason.low2@xxxxxxx>
---
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