[PATCH] sched/completion: Add fast path in __wait_for_common()

[Kirill Tkhai]

We may optimize __wait_for_common() in case of it's waken up
by complete{,_all}() if we do not take the spinlock.

New function completion_wake_function() is now used to wake
waiters. It a case of successful waking it deletes a waiter
from the task list. The waiter checks wait.task_list and skips
the locking if it's empty. In case of a single waiter this
prevents from unnecessary spin_{,un}_lock_irq().

In case of several waiters this improves parallelism in the
obvious way.

Signed-off-by: Kirill Tkhai <ktkhai@xxxxxxxxxxxxx>
---
 include/linux/wait.h      |    1 +
 kernel/sched/completion.c |   57 ++++++++++++++++++++++++++++++++-------------
 kernel/sched/wait.c       |    7 ++++++
 3 files changed, 48 insertions(+), 17 deletions(-)

diff --git a/include/linux/wait.h b/include/linux/wait.h
index 37423e0..506adfc 100644
--- a/include/linux/wait.h
+++ b/include/linux/wait.h
@@ -148,6 +148,7 @@ __remove_wait_queue(wait_queue_head_t *head, wait_queue_t *old)
 typedef int wait_bit_action_f(struct wait_bit_key *);
 void __wake_up(wait_queue_head_t *q, unsigned int mode, int nr, void *key);
 void __wake_up_locked_key(wait_queue_head_t *q, unsigned int mode, void *key);
+void __wake_up_locked_nr_key(wait_queue_head_t *q, unsigned int mode, int nr, void *key);
 void __wake_up_sync_key(wait_queue_head_t *q, unsigned int mode, int nr, void *key);
 void __wake_up_locked(wait_queue_head_t *q, unsigned int mode, int nr);
 void __wake_up_sync(wait_queue_head_t *q, unsigned int mode, int nr);
diff --git a/kernel/sched/completion.c b/kernel/sched/completion.c
index 607f852..b9959b8 100644
--- a/kernel/sched/completion.c
+++ b/kernel/sched/completion.c
@@ -32,7 +32,7 @@ void complete(struct completion *x)
 
 	spin_lock_irqsave(&x->wait.lock, flags);
 	x->done++;
-	__wake_up_locked(&x->wait, TASK_NORMAL, 1);
+	__wake_up_locked_nr_key(&x->wait, TASK_NORMAL, 1, x);
 	spin_unlock_irqrestore(&x->wait.lock, flags);
 }
 EXPORT_SYMBOL(complete);
@@ -52,17 +52,36 @@ void complete_all(struct completion *x)
 
 	spin_lock_irqsave(&x->wait.lock, flags);
 	x->done += UINT_MAX/2;
-	__wake_up_locked(&x->wait, TASK_NORMAL, 0);
+	__wake_up_locked_nr_key(&x->wait, TASK_NORMAL, 0, x);
 	spin_unlock_irqrestore(&x->wait.lock, flags);
 }
 EXPORT_SYMBOL(complete_all);
 
+static int completion_wake_function(wait_queue_t *wait, unsigned mode,
+				    int sync, void *key)
+{
+	int ret = default_wake_function(wait, mode, sync, key);
+	struct completion *x = key;
+
+	if (ret) {
+		list_del_init(&wait->task_list);
+		x->done--;
+	}
+	return ret;
+}
+
 static inline long __sched
-do_wait_for_common(struct completion *x,
-		   long (*action)(long), long timeout, int state)
+__wait_for_common(struct completion *x,
+		  long (*action)(long), long timeout, int state)
 {
+	might_sleep();
+
+	spin_lock_irq(&x->wait.lock);
 	if (!x->done) {
-		DECLARE_WAITQUEUE(wait, current);
+		wait_queue_t wait = {
+			.private	= current,
+			.func		= completion_wake_function,
+		};
 
 		__add_wait_queue_tail_exclusive(&x->wait, &wait);
 		do {
@@ -73,26 +92,30 @@ do_wait_for_common(struct completion *x,
 			__set_current_state(state);
 			spin_unlock_irq(&x->wait.lock);
 			timeout = action(timeout);
+			/*
+			 * This is the fast check whether we are woken up by
+			 * completion_wake_function(). No spinlock held here.
+			 */
+			if (list_empty(&wait.task_list))
+				goto out;
 			spin_lock_irq(&x->wait.lock);
+			/*
+			 * The above check is unlocked and racy with the wake
+			 * function. Test again to be sure, we haven't missed
+			 * the sign of its work.
+			 */
+			if (unlikely(list_empty(&wait.task_list)))
+				goto unlock;
 		} while (!x->done && timeout);
 		__remove_wait_queue(&x->wait, &wait);
 		if (!x->done)
 			return timeout;
 	}
 	x->done--;
-	return timeout ?: 1;
-}
-
-static inline long __sched
-__wait_for_common(struct completion *x,
-		  long (*action)(long), long timeout, int state)
-{
-	might_sleep();
-
-	spin_lock_irq(&x->wait.lock);
-	timeout = do_wait_for_common(x, action, timeout, state);
+unlock:
 	spin_unlock_irq(&x->wait.lock);
-	return timeout;
+out:
+	return timeout ?: 1;
 }
 
 static long __sched
diff --git a/kernel/sched/wait.c b/kernel/sched/wait.c
index 852143a..4846a57 100644
--- a/kernel/sched/wait.c
+++ b/kernel/sched/wait.c
@@ -112,6 +112,13 @@ void __wake_up_locked_key(wait_queue_head_t *q, unsigned int mode, void *key)
 }
 EXPORT_SYMBOL_GPL(__wake_up_locked_key);
 
+void __wake_up_locked_nr_key(wait_queue_head_t *q, unsigned int mode,
+			     int nr, void *key)
+{
+	__wake_up_common(q, mode, nr, 0, key);
+}
+EXPORT_SYMBOL_GPL(__wake_up_locked_nr_key);
+
 /**
  * __wake_up_sync_key - wake up threads blocked on a waitqueue.
  * @q: the waitqueue



--
To unsubscribe from this list: send the line "unsubscribe linux-kernel" in
the body of a message to majordomo@xxxxxxxxxxxxxxx
More majordomo info at  http://vger.kernel.org/majordomo-info.html
Please read the FAQ at  http://www.tux.org/lkml/