tracing/ring-buffer: Fix wakeup of ring buffer waiters

[ Sorry for the spam, but I just noticed kernel-test-robot complained
about KernelDoc format, which this fixes ]

A patch was sent to "fix" the wait_index variable that is used to help with waking of waiters on the ring buffer. The patch was rejected, but I started looking at associated code. Discussing it on IRC with Mathieu Desnoyers we discovered a design flaw.

The waiter reads "wait_index" then enters a "wait loop". After adding itself to the wait queue, if the buffer is filled or a signal is pending it will exit out. If wait_index is different, it will also exit out. (Note, I noticed that the check for wait_index was after the schedule() call and should have been before it, but that's besides the point).

The race is what happens if the waker updates the wait_index, a new waiter comes in and reads the updated wait_index before it adds itself to the queue, it will miss the update!

These patches fix the design by converting the ring_buffer_wait() to use the wait_event_interruptible() interface.

Then the wait_to_pipe() caller will pass its own condition into the ring_buffer_wait() to be checked by the wait_event_interruptible().

- The first patch restructures the ring_buffer_wait() to use the wait_event_interruptible() logic. It does not change the interface, but adds a local "cond" condition callback function that will be what it defaults to in the second patch if NULL is passed in.

The default cond function is just a "wait once" function. That is, the first time it is called (before the wait_event_interruptible() calls schedule) will set the "once" variable to one and return false. The next time it is called (after wait_event_interruptible() wakes up) it will return true to break out of the loop.

- The second patch changes the ring_buffer_wait() interface to allow the caller to define its own "cond" callback. That will be checked in wait_event_interruptible() along with checking if the proper amount of data in the ring buffer has been hit.

Changes since v3: https://lore.kernel.org/all/[email protected]/

- I should have checked before sending v2, but after I did, I noticed that kernel-test-robot reported that the cond and data parameters added to ring_buffer_wait() were not added to the kerneldoc above it.

Changes since v2: https://lore.kernel.org/all/[email protected]/

- Patches 1-3 of v2 have been accepted.

- Instead of breaking ring_buffer_wait() into three functions that do a prepare_to_wait(), wait() and finish_wait(), have ring_buffer_wait() take a condition callback function and a data parameter.

The ring_buffer_wait() now uses a wait_event_interruptible() code, and added a helper function to do check its own conditions, and also to call the passed in condition function and allow the caller to specify other conditions to break out of the event loop.

Changes since v1: https://lore.kernel.org/lkml/[email protected]/

- My tests triggered a warning about calling a mutex_lock() after a prepare_to_wait() that changed the task's state. Convert the affected mutex over to a spinlock.

Steven Rostedt (Google) (2):
ring-buffer: Use wait_event_interruptible() in ring_buffer_wait() tracing/ring-buffer: Fix wait_on_pipe() race

----

 include/linux/ring_buffer.h  |   4 +-
 include/linux/trace_events.h |   5 +-
 kernel/trace/ring_buffer.c   | 119 ++++++++++++++++++++++++++-----------------
 kernel/trace/trace.c         |  43 +++++++++++-----
 4 files changed, 109 insertions(+), 62 deletions(-)

From: "Steven Rostedt (Google)" <[email protected]>

Convert ring_buffer_wait() over to wait_event_interruptible(). The default condition is to execute the wait loop inside __wait_event() just once.

This does not change the ring_buffer_wait() prototype yet, but restructures the code so that it can take a "cond" and "data" parameter and will call wait_event_interruptible() with a helper function as the condition.

The helper function (rb_wait_cond) takes the cond function and data parameters. It will first check if the buffer hit the watermark defined by the "full" parameter and then call the passed in condition parameter. If either are true, it returns true.

If rb_wait_cond() does not return true, it will set the appropriate "waiters_pending" flag and returns false.

---
 include/linux/ring_buffer.h |   1 +
 kernel/trace/ring_buffer.c  | 116 +++++++++++++++++++++---------------
 2 files changed, 69 insertions(+), 48 deletions(-)

index fa802db216f9..338a33db1577 100644
@@ -98,6 +98,7 @@ __ring_buffer_alloc(unsigned long size, unsigned flags, struct lock_class_key *k
 	__ring_buffer_alloc((size), (flags), &__key);	\
 })
 
+typedef bool (*ring_buffer_cond_fn)(void *data);
 int ring_buffer_wait(struct trace_buffer *buffer, int cpu, int full);
 __poll_t ring_buffer_poll_wait(struct trace_buffer *buffer, int cpu,
 			  struct file *filp, poll_table *poll_table, int full);

2.43.0

From: "Steven Rostedt (Google)" <[email protected]>

When the trace_pipe_raw file is closed, there should be no new readers on the file descriptor. This is mostly handled with the waking and wait_index fields of the iterator. But there's still a slight race.

CPU 0 CPU 1
----- -----
wait_index++;

   index = wait_index;
                                   ring_buffer_wake_waiters();
   wait_on_pipe()
     ring_buffer_wait();

The ring_buffer_wait() will miss the wakeup from CPU 1. The problem is that the ring_buffer_wait() needs the logic of:

        prepare_to_wait();
        if (!condition)
                schedule();

Where the missing condition check is the iter->wait_index update.

Have the ring_buffer_wait() take a conditional callback function and a data parameter that can be used within the wait_event_interruptible() of the ring_buffer_wait() function.

In wait_on_pipe(), pass a condition function that will check if the wait_index has been updated, if it has, it will return true to break out of the wait_event_interruptible() loop.

Create a new field "closed" in the trace_iterator and set it in the .flush() callback before calling ring_buffer_wake_waiters(). This will keep any new readers from waiting on a closed file descriptor.

Have the wait_on_pipe() condition callback also check the closed field.

Change the wait_index field of the trace_iterator to atomic_t. There's no reason it needs to be 'long' and making it atomic and using atomic_read_acquire() and atomic_fetch_inc_release() will provide the necessary memory barriers.

Add a "woken" flag to tracing_buffers_splice_read() to exit the loop after one more try to fetch data. That is, if it waited for data and something woke it up, it should try to collect any new data and then exit back to user space.

---
 include/linux/ring_buffer.h  |  3 ++-
 include/linux/trace_events.h |  5 ++++-
 kernel/trace/ring_buffer.c   | 13 ++++++-----
 kernel/trace/trace.c         | 43 ++++++++++++++++++++++++++----------
 4 files changed, 45 insertions(+), 19 deletions(-)

index 338a33db1577..dc5ae4e96aee 100644
@@ -99,7 +99,8 @@ __ring_buffer_alloc(unsigned long size, unsigned flags, struct lock_class_key *k
 })
 
 typedef bool (*ring_buffer_cond_fn)(void *data);
-int ring_buffer_wait(struct trace_buffer *buffer, int cpu, int full);
+int ring_buffer_wait(struct trace_buffer *buffer, int cpu, int full,
+		     ring_buffer_cond_fn cond, void *data);
 __poll_t ring_buffer_poll_wait(struct trace_buffer *buffer, int cpu,
 			  struct file *filp, poll_table *poll_table, int full);
 void ring_buffer_wake_waiters(struct trace_buffer *buffer, int cpu);

index d68ff9b1247f..fc6d0af56bb1 100644
@@ -103,13 +103,16 @@ struct trace_iterator {
 	unsigned int		temp_size;
 	char			*fmt;	/* modified format holder */
 	unsigned int		fmt_size;
-	long			wait_index;
+	atomic_t		wait_index;
 
 	/* trace_seq for __print_flags() and __print_symbolic() etc. */
 	struct trace_seq	tmp_seq;
 
 	cpumask_var_t		started;
 
+	/* Set when the file is closed to prevent new waiters */
+	bool			closed;
+
 	/* it's true when current open file is snapshot */
 	bool			snapshot;
 

index c198ba466853..67d8405f4451 100644
@@ -901,23 +901,26 @@ static bool rb_wait_once(void *data)
  * @buffer: buffer to wait on
  * @cpu: the cpu buffer to wait on
  * @full: wait until the percentage of pages are available, if @cpu != RING_BUFFER_ALL_CPUS
+ * @cond: condition function to break out of wait (NULL to run once)
+ * @data: the data to pass to @cond.
  *
  * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon
  * as data is added to any of the @buffer's cpu buffers. Otherwise
  * it will wait for data to be added to a specific cpu buffer.
  */
-int ring_buffer_wait(struct trace_buffer *buffer, int cpu, int full)
+int ring_buffer_wait(struct trace_buffer *buffer, int cpu, int full,
+		     ring_buffer_cond_fn cond, void *data)
 {
 	struct ring_buffer_per_cpu *cpu_buffer;
 	struct wait_queue_head *waitq;
-	ring_buffer_cond_fn cond;
 	struct rb_irq_work *rbwork;
-	void *data;
 	long once = 0;
 	int ret = 0;
 
-	cond = rb_wait_once;
-	data = &once;
+	if (!cond) {
+		cond = rb_wait_once;
+		data = &once;
+	}
 
 	/*
 	 * Depending on what the caller is waiting for, either any

2.43.0