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

[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@parallels.com>
---
 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

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

[Kirill Tkhai]

Please, ignore this patch.

This brings the problem with exiting threads like in
https://lkml.org/lkml/2015/2/4/761

В Ср, 04/02/2015 в 19:50 +0300, 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@parallels.com>
> ---
>  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
> 
>