futex: Wake up waiter outside the hb->lock section

Wake ups of futex waiters happen with hb->lock held. This can lead to
unneccesary lock contention. Wake them when dropping hb->lock.
Signed-off-by: Peter Zijlstra <a.p.zijlstra@chello.nl>
Signed-off-by: Thomas Gleixner <tglx@linutronix.de>

include/linux/sched.h

@@ -1422,6 +1422,7 @@ struct task_struct {
 #endif
 	struct list_head pi_state_list;
 	struct futex_pi_state *pi_state_cache;
+	struct task_struct *futex_wakeup;
 #endif
 #ifdef CONFIG_PERF_COUNTERS
 	struct perf_counter_context *perf_counter_ctxp;

kernel/fork.c

@@ -1158,6 +1158,7 @@ static struct task_struct *copy_process(unsigned long clone_flags,
 #endif
 	INIT_LIST_HEAD(&p->pi_state_list);
 	p->pi_state_cache = NULL;
+	p->futex_wakeup = NULL;
 #endif
 	/*
 	 * sigaltstack should be cleared when sharing the same VM

kernel/futex.c

@@ -713,7 +713,7 @@ retry:
  * 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 task_struct **wake_list, struct futex_q *q)
 {
 	struct task_struct *p = q->task;
 
@@ -736,8 +736,51 @@ static void wake_futex(struct futex_q *q)
 	smp_wmb();
 	q->lock_ptr = NULL;
 
-	wake_up_state(p, TASK_NORMAL);
-	put_task_struct(p);
+	/*
+	 * Atomically grab the task, if ->futex_wakeup is !0 already it means
+	 * its already queued (either by us or someone else) and will get the
+	 * wakeup due to that.
+	 *
+	 * This cmpxchg() implies a full barrier, which pairs with the write
+	 * barrier implied by the wakeup in wake_futex_list().
+	 */
+	if (cmpxchg(&p->futex_wakeup, 0, p) != 0) {
+		/*
+		 * It was already queued, drop the extra ref and we're done.
+		 */
+		put_task_struct(p);
+		return;
+	}
+
+	/*
+	 * Put the task on our wakeup list by atomically switching it with
+	 * the list head. (XXX its a local list, no possible concurrency,
+	 * this could be written without cmpxchg).
+	 */
+	do {
+		p->futex_wakeup = *wake_list;
+	} while (cmpxchg(wake_list, p->futex_wakeup, p) != p->futex_wakeup);
+}
+
+/*
+ * For each task on the list, deliver the pending wakeup and release the
+ * task reference obtained in wake_futex().
+ */
+static void wake_futex_list(struct task_struct *head)
+{
+	while (head != &init_task) {
+		struct task_struct *next = head->futex_wakeup;
+
+		head->futex_wakeup = NULL;
+		/*
+		 * wake_up_state() implies a wmb() to pair with the queueing
+		 * in wake_futex() so as to not miss wakeups.
+		 */
+		wake_up_state(head, TASK_NORMAL);
+		put_task_struct(head);
+
+		head = next;
+	}
 }
 
 static int wake_futex_pi(u32 __user *uaddr, u32 uval, struct futex_q *this)
@@ -851,6 +894,7 @@ static int futex_wake(u32 __user *uaddr, int fshared, int nr_wake, u32 bitset)
 	struct futex_q *this, *next;
 	struct plist_head *head;
 	union futex_key key = FUTEX_KEY_INIT;
+	struct task_struct *wake_list = &init_task;
 	int ret;
 
 	if (!bitset)
@@ -875,7 +919,7 @@ static int futex_wake(u32 __user *uaddr, int fshared, int nr_wake, u32 bitset)
 			if (!(this->bitset & bitset))
 				continue;
 
-			wake_futex(this);
+			wake_futex(&wake_list, this);
 			if (++ret >= nr_wake)
 				break;
 		}
@@ -883,6 +927,8 @@ static int futex_wake(u32 __user *uaddr, int fshared, int nr_wake, u32 bitset)
 
 	spin_unlock(&hb->lock);
 	put_futex_key(fshared, &key);
+
+	wake_futex_list(wake_list);
 out:
 	return ret;
 }
@@ -899,6 +945,7 @@ futex_wake_op(u32 __user *uaddr1, int fshared, u32 __user *uaddr2,
 	struct futex_hash_bucket *hb1, *hb2;
 	struct plist_head *head;
 	struct futex_q *this, *next;
+	struct task_struct *wake_list = &init_task;
 	int ret, op_ret;
 
 retry:
@@ -949,7 +996,7 @@ retry_private:
 
 	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;
 		}
@@ -961,7 +1008,7 @@ retry_private:
 		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;
 			}
@@ -974,6 +1021,8 @@ out_put_keys:
 	put_futex_key(fshared, &key2);
 out_put_key1:
 	put_futex_key(fshared, &key1);
+
+	wake_futex_list(wake_list);
 out:
 	return ret;
 }
@@ -1119,6 +1168,7 @@ static int futex_requeue(u32 __user *uaddr1, int fshared, u32 __user *uaddr2,
 	struct futex_hash_bucket *hb1, *hb2;
 	struct plist_head *head1;
 	struct futex_q *this, *next;
+	struct task_struct *wake_list = &init_task;
 	u32 curval2;
 
 	if (requeue_pi) {
@@ -1256,7 +1306,7 @@ retry_private:
 		 * woken by futex_unlock_pi().
 		 */
 		if (++task_count <= nr_wake && !requeue_pi) {
-			wake_futex(this);
+			wake_futex(&wake_list, this);
 			continue;
 		}
 
@@ -1302,6 +1352,8 @@ out_put_keys:
 	put_futex_key(fshared, &key2);
 out_put_key1:
 	put_futex_key(fshared, &key1);
+
+	wake_futex_list(wake_list);
 out:
 	if (pi_state != NULL)
 		free_pi_state(pi_state);