Re: [RFC][PATCH 2/3] futex: Reduce hash bucket lock contention

Darren Hart
2011-09-14T15:46:16Z

On 09/14/2011 06:30 AM, Peter Zijlstra wrote:
> Use the brand spanking new wake_list to delay the futex wakeups until
> after we've released the hash bucket locks. This avoids the newly
> woken tasks from immediately getting stuck on the hb lock.
>
> This is esp. painful on -rt, where the hb lock is preemptible.

Nice!

Have you run this through the functional and performance tests from
futextest? Looks like I should also add a multiwake test to really
showcase this.

If you don't have it local I can setup a github repository for futextest
until korg is back.... or do the testing myself... right.

>
> Cc: Thomas Gleixner <tglx@linutronix.de>
> Cc: Darren Hart <dvhart@linux.intel.com>
> Signed-off-by: Peter Zijlstra <a.p.zijlstra@chello.nl>
> ---
> kernel/futex.c | 23 ++++++++++++++---------
> 1 file changed, 14 insertions(+), 9 deletions(-)
>
> Index: linux-2.6/kernel/futex.c
> ===================================================================
> --- linux-2.6.orig/kernel/futex.c
> +++ linux-2.6/kernel/futex.c
> @@ -823,7 +823,7 @@ static void __unqueue_futex(struct futex
> * The hash bucket lock must be held when this is called.
> * Afterwards, the futex_q must not be accessed.
> */
> -static void wake_futex(struct futex_q *q)
> +static void wake_futex(struct wake_list_head *wake_list, struct futex_q *q)

A good opportunity to add the proper kerneldoc to this function as well.

> {
> struct task_struct *p = q->task;
>
> @@ -834,7 +834,7 @@ static void wake_futex(struct futex_q *q
> * struct. Prevent this by holding a reference on p across the
> * wake up.
> */
> - get_task_struct(p);
> + wake_list_add(wake_list, p);
>
> __unqueue_futex(q);
> /*
> @@ -845,9 +845,6 @@ static void wake_futex(struct futex_q *q
> */
> smp_wmb();
> q->lock_ptr = NULL;
> -
> - wake_up_state(p, TASK_NORMAL);
> - put_task_struct(p);
> }
>
> static int wake_futex_pi(u32 __user *uaddr, u32 uval, struct futex_q *this)
> @@ -964,6 +961,7 @@ futex_wake(u32 __user *uaddr, unsigned i
> struct futex_q *this, *next;
> struct plist_head *head;
> union futex_key key = FUTEX_KEY_INIT;
> + WAKE_LIST(wake_list);
> int ret;
>
> if (!bitset)
> @@ -988,7 +986,7 @@ futex_wake(u32 __user *uaddr, unsigned i
> if (!(this->bitset & bitset))
> continue;
>
> - wake_futex(this);
> + wake_futex(&wake_list, this);

I guess this is OK. wake_futex_pi will always be one task I believe, so
the list syntax might confuse newcomers... Would it make sense to have a
wake_futex_list() call? Thinking outloud...

> if (++ret >= nr_wake)
> break;
> }
> @@ -996,6 +994,8 @@ futex_wake(u32 __user *uaddr, unsigned i
>
> spin_unlock(&hb->lock);
> put_futex_key(&key);
> +
> + wake_up_list(&wake_list, TASK_NORMAL);
> out:
> return ret;
> }
> @@ -1012,6 +1012,7 @@ futex_wake_op(u32 __user *uaddr1, unsign
> struct futex_hash_bucket *hb1, *hb2;
> struct plist_head *head;
> struct futex_q *this, *next;
> + WAKE_LIST(wake_list);
> int ret, op_ret;
>
> retry:
> @@ -1062,7 +1063,7 @@ futex_wake_op(u32 __user *uaddr1, unsign
>
> plist_for_each_entry_safe(this, next, head, list) {
> if (match_futex (&this->key, &key1)) {
> - wake_futex(this);
> + wake_futex(&wake_list, this);
> if (++ret >= nr_wake)
> break;
> }
> @@ -1074,7 +1075,7 @@ futex_wake_op(u32 __user *uaddr1, unsign
> op_ret = 0;
> plist_for_each_entry_safe(this, next, head, list) {
> if (match_futex (&this->key, &key2)) {
> - wake_futex(this);
> + wake_futex(&wake_list, this);
> if (++op_ret >= nr_wake2)
> break;
> }
> @@ -1087,6 +1088,8 @@ futex_wake_op(u32 __user *uaddr1, unsign
> put_futex_key(&key2);
> out_put_key1:
> put_futex_key(&key1);
> +
> + wake_up_list(&wake_list, TASK_NORMAL);
> out:
> return ret;
> }
> @@ -1239,6 +1242,7 @@ static int futex_requeue(u32 __user *uad
> struct futex_hash_bucket *hb1, *hb2;
> struct plist_head *head1;
> struct futex_q *this, *next;
> + WAKE_LIST(wake_list);
> u32 curval2;
>
> if (requeue_pi) {
> @@ -1384,7 +1388,7 @@ static int futex_requeue(u32 __user *uad
> * woken by futex_unlock_pi().
> */
> if (++task_count <= nr_wake && !requeue_pi) {
> - wake_futex(this);
> + wake_futex(&wake_list, this);
> continue;
> }
>
> @@ -1437,6 +1441,7 @@ static int futex_requeue(u32 __user *uad
> put_futex_key(&key2);
> out_put_key1:
> put_futex_key(&key1);
> + wake_up_list(&wake_list, TASK_NORMAL);
> out:
> if (pi_state != NULL)
> free_pi_state(pi_state);
>
>

I _think_ requeue_pi is in the clear here as it uses
requeue_pi_wake_futex, which calls wake_up_state directly. Still, some
testing with futextest functional/futex_requeue_pi is in order.

--
Darren Hart
Intel Open Source Technology Center
Yocto Project - Linux Kernel