[for-next][PATCH] ring-buffer: Do not wake up a splice waiter when page is not full

[for-next][PATCH] ring-buffer: Do not wake up a splice waiter when page is not full

[Steven Rostedt]

  git://git.kernel.org/pub/scm/linux/kernel/git/rostedt/linux-trace.git
for-next

Head SHA1: 1e0d6714aceb770b04161fbedd7765d0e1fc27bd


Steven Rostedt (Red Hat) (1):
      ring-buffer: Do not wake up a splice waiter when page is not full

----
 kernel/trace/ring_buffer.c | 40 +++++++++++++++++++++++++++++++++++-----
 1 file changed, 35 insertions(+), 5 deletions(-)
---------------------------
commit 1e0d6714aceb770b04161fbedd7765d0e1fc27bd
Author: Steven Rostedt (Red Hat) <rostedt@goodmis.org>
Date:   Tue Feb 10 22:14:53 2015 -0500

    ring-buffer: Do not wake up a splice waiter when page is not full
    
    When an application connects to the ring buffer via splice, it can only
    read full pages. Splice does not work with partial pages. If there is
    not enough data to fill a page, the splice command will either block
    or return -EAGAIN (if set to nonblock).
    
    Code was added where if the page is not full, to just sleep again.
    The problem is, it will get woken up again on the next event. That
    is, when something is written into the ring buffer, if there is a waiter
    it will wake it up. The waiter would then check the buffer, see that
    it still does not have enough data to fill a page and go back to sleep.
    To make matters worse, when the waiter goes back to sleep, it could
    cause another event, which would wake it back up again to see it
    doesn't have enough data and sleep again. This produces a tremendous
    overhead and fills the ring buffer with noise.
    
    For example, recording sched_switch on an idle system for 10 seconds
    produces 25,350,475 events!!!
    
    Create another wait queue for those waiters wanting full pages.
    When an event is written, it only wakes up waiters if there's a full
    page of data. It does not wake up the waiter if the page is not yet
    full.
    
    After this change, recording sched_switch on an idle system for 10
    seconds produces only 800 events. Getting rid of 25,349,675 useless
    events (99.9969% of events!!), is something to take seriously.
    
    Cc: stable@vger.kernel.org # 3.16+
    Cc: Rabin Vincent <rabin@rab.in>
    Fixes: e30f53aad220 "tracing: Do not busy wait in buffer splice"
    Signed-off-by: Steven Rostedt <rostedt@goodmis.org>

diff --git a/kernel/trace/ring_buffer.c b/kernel/trace/ring_buffer.c
index 96079180de3d..5040d44fe5a3 100644
--- a/kernel/trace/ring_buffer.c
+++ b/kernel/trace/ring_buffer.c
@@ -445,7 +445,10 @@ int ring_buffer_print_page_header(struct trace_seq *s)
 struct rb_irq_work {
 	struct irq_work			work;
 	wait_queue_head_t		waiters;
+	wait_queue_head_t		full_waiters;
 	bool				waiters_pending;
+	bool				full_waiters_pending;
+	bool				wakeup_full;
 };
 
 /*
@@ -527,6 +530,10 @@ static void rb_wake_up_waiters(struct irq_work *work)
 	struct rb_irq_work *rbwork = container_of(work, struct rb_irq_work, work);
 
 	wake_up_all(&rbwork->waiters);
+	if (rbwork->wakeup_full) {
+		rbwork->wakeup_full = false;
+		wake_up_all(&rbwork->full_waiters);
+	}
 }
 
 /**
@@ -551,9 +558,11 @@ int ring_buffer_wait(struct ring_buffer *buffer, int cpu, bool full)
 	 * data in any cpu buffer, or a specific buffer, put the
 	 * caller on the appropriate wait queue.
 	 */
-	if (cpu == RING_BUFFER_ALL_CPUS)
+	if (cpu == RING_BUFFER_ALL_CPUS) {
 		work = &buffer->irq_work;
-	else {
+		/* Full only makes sense on per cpu reads */
+		full = false;
+	} else {
 		if (!cpumask_test_cpu(cpu, buffer->cpumask))
 			return -ENODEV;
 		cpu_buffer = buffer->buffers[cpu];
@@ -562,7 +571,10 @@ int ring_buffer_wait(struct ring_buffer *buffer, int cpu, bool full)
 
 
 	while (true) {
-		prepare_to_wait(&work->waiters, &wait, TASK_INTERRUPTIBLE);
+		if (full)
+			prepare_to_wait(&work->full_waiters, &wait, TASK_INTERRUPTIBLE);
+		else
+			prepare_to_wait(&work->waiters, &wait, TASK_INTERRUPTIBLE);
 
 		/*
 		 * The events can happen in critical sections where
@@ -584,7 +596,10 @@ int ring_buffer_wait(struct ring_buffer *buffer, int cpu, bool full)
 		 * that is necessary is that the wake up happens after
 		 * a task has been queued. It's OK for spurious wake ups.
 		 */
-		work->waiters_pending = true;
+		if (full)
+			work->full_waiters_pending = true;
+		else
+			work->waiters_pending = true;
 
 		if (signal_pending(current)) {
 			ret = -EINTR;
@@ -613,7 +628,10 @@ int ring_buffer_wait(struct ring_buffer *buffer, int cpu, bool full)
 		schedule();
 	}
 
-	finish_wait(&work->waiters, &wait);
+	if (full)
+		finish_wait(&work->full_waiters, &wait);
+	else
+		finish_wait(&work->waiters, &wait);
 
 	return ret;
 }
@@ -1228,6 +1246,7 @@ rb_allocate_cpu_buffer(struct ring_buffer *buffer, int nr_pages, int cpu)
 	init_completion(&cpu_buffer->update_done);
 	init_irq_work(&cpu_buffer->irq_work.work, rb_wake_up_waiters);
 	init_waitqueue_head(&cpu_buffer->irq_work.waiters);
+	init_waitqueue_head(&cpu_buffer->irq_work.full_waiters);
 
 	bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
 			    GFP_KERNEL, cpu_to_node(cpu));
@@ -2799,6 +2818,8 @@ static void rb_commit(struct ring_buffer_per_cpu *cpu_buffer,
 static __always_inline void
 rb_wakeups(struct ring_buffer *buffer, struct ring_buffer_per_cpu *cpu_buffer)
 {
+	bool pagebusy;
+
 	if (buffer->irq_work.waiters_pending) {
 		buffer->irq_work.waiters_pending = false;
 		/* irq_work_queue() supplies it's own memory barriers */
@@ -2810,6 +2831,15 @@ rb_wakeups(struct ring_buffer *buffer, struct ring_buffer_per_cpu *cpu_buffer)
 		/* irq_work_queue() supplies it's own memory barriers */
 		irq_work_queue(&cpu_buffer->irq_work.work);
 	}
+
+	pagebusy = cpu_buffer->reader_page == cpu_buffer->commit_page;
+
+	if (!pagebusy && cpu_buffer->irq_work.full_waiters_pending) {
+		cpu_buffer->irq_work.wakeup_full = true;
+		cpu_buffer->irq_work.full_waiters_pending = false;
+		/* irq_work_queue() supplies it's own memory barriers */
+		irq_work_queue(&cpu_buffer->irq_work.work);
+	}
 }
 
 /**

Re: [for-next][PATCH] ring-buffer: Do not wake up a splice waiter when page is not full

[Steven Rostedt]

I added Rabin in the tags as Cc'd but as I send out single patches
manually, I forgot to add him to the real Cc.

-- Steve


On Wed, 11 Feb 2015 07:44:59 -0500
Steven Rostedt <rostedt@goodmis.org> wrote:

>   git://git.kernel.org/pub/scm/linux/kernel/git/rostedt/linux-trace.git
> for-next
> 
> Head SHA1: 1e0d6714aceb770b04161fbedd7765d0e1fc27bd
> 
> 
> Steven Rostedt (Red Hat) (1):
>       ring-buffer: Do not wake up a splice waiter when page is not full
> 
> ----
>  kernel/trace/ring_buffer.c | 40 +++++++++++++++++++++++++++++++++++-----
>  1 file changed, 35 insertions(+), 5 deletions(-)
> ---------------------------
> commit 1e0d6714aceb770b04161fbedd7765d0e1fc27bd
> Author: Steven Rostedt (Red Hat) <rostedt@goodmis.org>
> Date:   Tue Feb 10 22:14:53 2015 -0500
> 
>     ring-buffer: Do not wake up a splice waiter when page is not full
>     
>     When an application connects to the ring buffer via splice, it can only
>     read full pages. Splice does not work with partial pages. If there is
>     not enough data to fill a page, the splice command will either block
>     or return -EAGAIN (if set to nonblock).
>     
>     Code was added where if the page is not full, to just sleep again.
>     The problem is, it will get woken up again on the next event. That
>     is, when something is written into the ring buffer, if there is a waiter
>     it will wake it up. The waiter would then check the buffer, see that
>     it still does not have enough data to fill a page and go back to sleep.
>     To make matters worse, when the waiter goes back to sleep, it could
>     cause another event, which would wake it back up again to see it
>     doesn't have enough data and sleep again. This produces a tremendous
>     overhead and fills the ring buffer with noise.
>     
>     For example, recording sched_switch on an idle system for 10 seconds
>     produces 25,350,475 events!!!
>     
>     Create another wait queue for those waiters wanting full pages.
>     When an event is written, it only wakes up waiters if there's a full
>     page of data. It does not wake up the waiter if the page is not yet
>     full.
>     
>     After this change, recording sched_switch on an idle system for 10
>     seconds produces only 800 events. Getting rid of 25,349,675 useless
>     events (99.9969% of events!!), is something to take seriously.
>     
>     Cc: stable@vger.kernel.org # 3.16+
>     Cc: Rabin Vincent <rabin@rab.in>
>     Fixes: e30f53aad220 "tracing: Do not busy wait in buffer splice"
>     Signed-off-by: Steven Rostedt <rostedt@goodmis.org>
> 
> diff --git a/kernel/trace/ring_buffer.c b/kernel/trace/ring_buffer.c
> index 96079180de3d..5040d44fe5a3 100644
> --- a/kernel/trace/ring_buffer.c
> +++ b/kernel/trace/ring_buffer.c
> @@ -445,7 +445,10 @@ int ring_buffer_print_page_header(struct trace_seq *s)
>  struct rb_irq_work {
>  	struct irq_work			work;
>  	wait_queue_head_t		waiters;
> +	wait_queue_head_t		full_waiters;
>  	bool				waiters_pending;
> +	bool				full_waiters_pending;
> +	bool				wakeup_full;
>  };
>  
>  /*
> @@ -527,6 +530,10 @@ static void rb_wake_up_waiters(struct irq_work *work)
>  	struct rb_irq_work *rbwork = container_of(work, struct rb_irq_work, work);
>  
>  	wake_up_all(&rbwork->waiters);
> +	if (rbwork->wakeup_full) {
> +		rbwork->wakeup_full = false;
> +		wake_up_all(&rbwork->full_waiters);
> +	}
>  }
>  
>  /**
> @@ -551,9 +558,11 @@ int ring_buffer_wait(struct ring_buffer *buffer, int cpu, bool full)
>  	 * data in any cpu buffer, or a specific buffer, put the
>  	 * caller on the appropriate wait queue.
>  	 */
> -	if (cpu == RING_BUFFER_ALL_CPUS)
> +	if (cpu == RING_BUFFER_ALL_CPUS) {
>  		work = &buffer->irq_work;
> -	else {
> +		/* Full only makes sense on per cpu reads */
> +		full = false;
> +	} else {
>  		if (!cpumask_test_cpu(cpu, buffer->cpumask))
>  			return -ENODEV;
>  		cpu_buffer = buffer->buffers[cpu];
> @@ -562,7 +571,10 @@ int ring_buffer_wait(struct ring_buffer *buffer, int cpu, bool full)
>  
>  
>  	while (true) {
> -		prepare_to_wait(&work->waiters, &wait, TASK_INTERRUPTIBLE);
> +		if (full)
> +			prepare_to_wait(&work->full_waiters, &wait, TASK_INTERRUPTIBLE);
> +		else
> +			prepare_to_wait(&work->waiters, &wait, TASK_INTERRUPTIBLE);
>  
>  		/*
>  		 * The events can happen in critical sections where
> @@ -584,7 +596,10 @@ int ring_buffer_wait(struct ring_buffer *buffer, int cpu, bool full)
>  		 * that is necessary is that the wake up happens after
>  		 * a task has been queued. It's OK for spurious wake ups.
>  		 */
> -		work->waiters_pending = true;
> +		if (full)
> +			work->full_waiters_pending = true;
> +		else
> +			work->waiters_pending = true;
>  
>  		if (signal_pending(current)) {
>  			ret = -EINTR;
> @@ -613,7 +628,10 @@ int ring_buffer_wait(struct ring_buffer *buffer, int cpu, bool full)
>  		schedule();
>  	}
>  
> -	finish_wait(&work->waiters, &wait);
> +	if (full)
> +		finish_wait(&work->full_waiters, &wait);
> +	else
> +		finish_wait(&work->waiters, &wait);
>  
>  	return ret;
>  }
> @@ -1228,6 +1246,7 @@ rb_allocate_cpu_buffer(struct ring_buffer *buffer, int nr_pages, int cpu)
>  	init_completion(&cpu_buffer->update_done);
>  	init_irq_work(&cpu_buffer->irq_work.work, rb_wake_up_waiters);
>  	init_waitqueue_head(&cpu_buffer->irq_work.waiters);
> +	init_waitqueue_head(&cpu_buffer->irq_work.full_waiters);
>  
>  	bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
>  			    GFP_KERNEL, cpu_to_node(cpu));
> @@ -2799,6 +2818,8 @@ static void rb_commit(struct ring_buffer_per_cpu *cpu_buffer,
>  static __always_inline void
>  rb_wakeups(struct ring_buffer *buffer, struct ring_buffer_per_cpu *cpu_buffer)
>  {
> +	bool pagebusy;
> +
>  	if (buffer->irq_work.waiters_pending) {
>  		buffer->irq_work.waiters_pending = false;
>  		/* irq_work_queue() supplies it's own memory barriers */
> @@ -2810,6 +2831,15 @@ rb_wakeups(struct ring_buffer *buffer, struct ring_buffer_per_cpu *cpu_buffer)
>  		/* irq_work_queue() supplies it's own memory barriers */
>  		irq_work_queue(&cpu_buffer->irq_work.work);
>  	}
> +
> +	pagebusy = cpu_buffer->reader_page == cpu_buffer->commit_page;
> +
> +	if (!pagebusy && cpu_buffer->irq_work.full_waiters_pending) {
> +		cpu_buffer->irq_work.wakeup_full = true;
> +		cpu_buffer->irq_work.full_waiters_pending = false;
> +		/* irq_work_queue() supplies it's own memory barriers */
> +		irq_work_queue(&cpu_buffer->irq_work.work);
> +	}
>  }
>  
>  /**