futex: Handle early deadlock return correctly

commit 56222b212e8e ("futex: Drop hb->lock before enqueueing on the rtmutex") changed the locking rules in the futex code so that the hash bucket lock is not longer held while the waiter is enqueued into the rtmutex wait list. This made the lock and the unlock path symmetric, but unfortunately the possible early exit from __rt_mutex_proxy_start() due to a detected deadlock was not updated accordingly. That allows a concurrent unlocker to observe inconsitent state which triggers the warning in the unlock path. futex_lock_pi() futex_unlock_pi() lock(hb->lock) queue(hb_waiter) lock(hb->lock) lock(rtmutex->wait_lock) unlock(hb->lock) // acquired hb->lock hb_waiter = futex_top_waiter() lock(rtmutex->wait_lock) __rt_mutex_proxy_start() ---> fail remove(rtmutex_waiter); ---> returns -EDEADLOCK unlock(rtmutex->wait_lock) // acquired wait_lock wake_futex_pi() rt_mutex_next_owner() --> returns NULL --> WARN lock(hb->lock) unqueue(hb_waiter) The problem is caused by the remove(rtmutex_waiter) in the failure case of __rt_mutex_proxy_start() as this lets the unlocker observe a waiter in the hash bucket but no waiter on the rtmutex, i.e. inconsistent state. The original commit handles this correctly for the other early return cases (timeout, signal) by delaying the removal of the rtmutex waiter until the returning task reacquired the hash bucket lock. Treat the failure case of __rt_mutex_proxy_start() in the same way and let the existing cleanup code handle the eventual handover of the rtmutex gracefully. The regular rt_mutex_proxy_start() gains the rtmutex waiter removal for the failure case, so that the other callsites are still operating correctly. Add proper comments to the code so all these details are fully documented. Thanks to Peter for helping with the analysis and writing the really valuable code comments.

Thomas Gleixner committed 7 years ago

Commit 1a1fb981 parent 6f568eb2 files · +50 −15Browse files

Showing the changes to kernel (2 of 2 files). Show all

Unified · Split

2 files

kernel/futex.c +18 −10
kernel/locking/rtmutex.c +32 −5

kernel/futex.c+18−10

@@ -2861,35 +2861,39 @@ retry_private: · 2860 unmodified lines
2861	* and BUG when futex_unlock_pi() interleaves with this.	2861	* and BUG when futex_unlock_pi() interleaves with this.
2862	*	2862	*
2863	* Therefore acquire wait_lock while holding hb->lock, but drop the	2863	* Therefore acquire wait_lock while holding hb->lock, but drop the
2864	* latter before calling rt_mutex_start_proxy_lock(). This still fully	2864	* latter before calling __rt_mutex_start_proxy_lock(). This
2865	* serializes against futex_unlock_pi() as that does the exact same	2865	* interleaves with futex_unlock_pi() -- which does a similar lock
2866	* lock handoff sequence.	2866	* handoff -- such that the latter can observe the futex_q::pi_state
		2867	* before __rt_mutex_start_proxy_lock() is done.
2867	*/	2868	*/
2868	raw_spin_lock_irq(&q.pi_state->pi_mutex.wait_lock);	2869	raw_spin_lock_irq(&q.pi_state->pi_mutex.wait_lock);
2869	spin_unlock(q.lock_ptr);	2870	spin_unlock(q.lock_ptr);
		2871	/*
		2872	* __rt_mutex_start_proxy_lock() unconditionally enqueues the @rt_waiter
		2873	* such that futex_unlock_pi() is guaranteed to observe the waiter when
		2874	* it sees the futex_q::pi_state.
		2875	*/
2870	ret = __rt_mutex_start_proxy_lock(&q.pi_state->pi_mutex, &rt_waiter, current);	2876	ret = __rt_mutex_start_proxy_lock(&q.pi_state->pi_mutex, &rt_waiter, current);
2871	raw_spin_unlock_irq(&q.pi_state->pi_mutex.wait_lock);	2877	raw_spin_unlock_irq(&q.pi_state->pi_mutex.wait_lock);
2872		2878
2873	if (ret) {	2879	if (ret) {
2874	if (ret == 1)	2880	if (ret == 1)
2875	ret = 0;	2881	ret = 0;
2876		2882	goto cleanup;
2877	spin_lock(q.lock_ptr);
2878	goto no_block;
2879	}	2883	}
2880		2884
2881
2882	if (unlikely(to))	2885	if (unlikely(to))
2883	hrtimer_start_expires(&to->timer, HRTIMER_MODE_ABS);	2886	hrtimer_start_expires(&to->timer, HRTIMER_MODE_ABS);
2884		2887
2885	ret = rt_mutex_wait_proxy_lock(&q.pi_state->pi_mutex, to, &rt_waiter);	2888	ret = rt_mutex_wait_proxy_lock(&q.pi_state->pi_mutex, to, &rt_waiter);
2886		2889
		2890	cleanup:
2887	spin_lock(q.lock_ptr);	2891	spin_lock(q.lock_ptr);
2888	/*	2892	/*
2889	* If we failed to acquire the lock (signal/timeout), we must	2893	* If we failed to acquire the lock (deadlock/signal/timeout), we must
2890	* first acquire the hb->lock before removing the lock from the	2894	* first acquire the hb->lock before removing the lock from the
2891	* rt_mutex waitqueue, such that we can keep the hb and rt_mutex	2895	* rt_mutex waitqueue, such that we can keep the hb and rt_mutex wait
2892	* wait lists consistent.	2896	* lists consistent.
2893	*	2897	*
2894	* In particular; it is important that futex_unlock_pi() can not	2898	* In particular; it is important that futex_unlock_pi() can not
2895	* observe this inconsistency.	2899	* observe this inconsistency.
@@ -3013,6 +3017,10 @@ retry: · 117 unmodified lines
3013	* there is no point where we hold neither; and therefore	3017	* there is no point where we hold neither; and therefore
3014	* wake_futex_pi() must observe a state consistent with what we	3018	* wake_futex_pi() must observe a state consistent with what we
3015	* observed.	3019	* observed.
		3020	*
		3021	* In particular; this forces __rt_mutex_start_proxy() to
		3022	* complete such that we're guaranteed to observe the
		3023	* rt_waiter. Also see the WARN in wake_futex_pi().
3016	*/	3024	*/
3017	raw_spin_lock_irq(&pi_state->pi_mutex.wait_lock);	3025	raw_spin_lock_irq(&pi_state->pi_mutex.wait_lock);
3018	spin_unlock(&hb->lock);	3026	spin_unlock(&hb->lock);

kernel/locking/rtmutex.c+32−5

 	rt_mutex_set_owner(lock, NULL);
 }
+/**
+ * __rt_mutex_start_proxy_lock() - Start lock acquisition for another task
+ * @lock:		the rt_mutex to take
+ * @waiter:		the pre-initialized rt_mutex_waiter
+ * @task:		the task to prepare
+ *
+ * Starts the rt_mutex acquire; it enqueues the @waiter and does deadlock
+ * detection. It does not wait, see rt_mutex_wait_proxy_lock() for that.
+ *
+ * NOTE: does _NOT_ remove the @waiter on failure; must either call
+ * rt_mutex_wait_proxy_lock() or rt_mutex_cleanup_proxy_lock() after this.
+ *
+ * Returns:
+ *  0 - task blocked on lock
+ *  1 - acquired the lock for task, caller should wake it up
+ * <0 - error
+ *
+ * Special API call for PI-futex support.
+ */
 int __rt_mutex_start_proxy_lock(struct rt_mutex *lock,
 			      struct rt_mutex_waiter *waiter,
 			      struct task_struct *task)
 {
 	int ret;
+	lockdep_assert_held(&lock->wait_lock);
 	if (try_to_take_rt_mutex(lock, task, NULL))
 		return 1;
 		ret = 0;
 	}
-	if (unlikely(ret))
-		remove_waiter(lock, waiter);
 	debug_rt_mutex_print_deadlock(waiter);
 	return ret;
  * @waiter:		the pre-initialized rt_mutex_waiter
  * @task:		the task to prepare
  *
+ * Starts the rt_mutex acquire; it enqueues the @waiter and does deadlock
+ * detection. It does not wait, see rt_mutex_wait_proxy_lock() for that.
+ *
+ * NOTE: unlike __rt_mutex_start_proxy_lock this _DOES_ remove the @waiter
+ * on failure.
+ *
  * Returns:
  *  0 - task blocked on lock
  *  1 - acquired the lock for task, caller should wake it up
  * <0 - error
  *
- * Special API call for FUTEX_REQUEUE_PI support.
+ * Special API call for PI-futex support.
  */
 int rt_mutex_start_proxy_lock(struct rt_mutex *lock,
 			      struct rt_mutex_waiter *waiter,
 	raw_spin_lock_irq(&lock->wait_lock);
 	ret = __rt_mutex_start_proxy_lock(lock, waiter, task);
+	if (unlikely(ret))
+		remove_waiter(lock, waiter);
 	raw_spin_unlock_irq(&lock->wait_lock);
 	return ret;
  * @lock:		the rt_mutex we were woken on
  * @waiter:		the pre-initialized rt_mutex_waiter
  *
- * Attempt to clean up after a failed rt_mutex_wait_proxy_lock().
+ * Attempt to clean up after a failed __rt_mutex_start_proxy_lock() or
+ * rt_mutex_wait_proxy_lock().
  *
  * Unless we acquired the lock; we're still enqueued on the wait-list and can
  * in fact still be granted ownership until we're removed. Therefore we can