ring_buffer: implement new locking

The old "lock always" scheme had issues with lockdep, and was not very
efficient anyways.

This patch does a new design to be partially lockless on writes.
Writes will add new entries to the per cpu pages by simply disabling
interrupts. When a write needs to go to another page than it will
grab the lock.

A new "read page" has been added so that the reader can pull out a page
from the ring buffer to read without worrying about the writer writing over
it. This allows us to not take the lock for all reads. The lock is
now only taken when a read needs to go to a new page.

This is far from lockless, and interrupts still need to be disabled,
but it is a step towards a more lockless solution, and it also
solves a lot of the issues that were noticed by the first conversion
of ftrace to the ring buffers.

Note: the ring_buffer_{un}lock API has been removed.

Signed-off-by: Steven Rostedt <srostedt@redhat.com>
Signed-off-by: Ingo Molnar <mingo@elte.hu>

2 files changed, 247 insertions, 164 deletions

diff --git a/kernel/trace/ring_buffer.c b/kernel/trace/ring_buffer.c
index 8e7392fd0db9..9631abf2ae29 100644
--- a/kernel/trace/ring_buffer.c
+++ b/kernel/trace/ring_buffer.c
@@ -161,8 +161,10 @@ struct ring_buffer_per_cpu {
         struct list_head                pages;
         unsigned long                   head;   /* read from head */
         unsigned long                   tail;   /* write to tail */
+        unsigned long                   reader;
         struct buffer_page              *head_page;
         struct buffer_page              *tail_page;
+        struct buffer_page              *reader_page;
         unsigned long                   overrun;
         unsigned long                   entries;
         u64                             write_stamp;
@@ -260,6 +262,7 @@ static struct ring_buffer_per_cpu *
 rb_allocate_cpu_buffer(struct ring_buffer *buffer, int cpu)
 {
         struct ring_buffer_per_cpu *cpu_buffer;
+        unsigned long addr;
         int ret;
 
         cpu_buffer = kzalloc_node(ALIGN(sizeof(*cpu_buffer), cache_line_size()),
@@ -272,9 +275,16 @@ rb_allocate_cpu_buffer(struct ring_buffer *buffer, int cpu)
         spin_lock_init(&cpu_buffer->lock);
         INIT_LIST_HEAD(&cpu_buffer->pages);
 
+        addr = __get_free_page(GFP_KERNEL);
+        if (!addr)
+                goto fail_free_buffer;
+        cpu_buffer->reader_page = (struct buffer_page *)virt_to_page(addr);
+        INIT_LIST_HEAD(&cpu_buffer->reader_page->list);
+        cpu_buffer->reader_page->size = 0;
+
         ret = rb_allocate_pages(cpu_buffer, buffer->pages);
         if (ret < 0)
-                goto fail_free_buffer;
+                goto fail_free_reader;
 
         cpu_buffer->head_page
                 = list_entry(cpu_buffer->pages.next, struct buffer_page, list);
@@ -283,6 +293,9 @@ rb_allocate_cpu_buffer(struct ring_buffer *buffer, int cpu)
 
         return cpu_buffer;
 
+ fail_free_reader:
+        free_buffer_page(cpu_buffer->reader_page);
+
  fail_free_buffer:
         kfree(cpu_buffer);
         return NULL;
@@ -293,6 +306,9 @@ static void rb_free_cpu_buffer(struct ring_buffer_per_cpu *cpu_buffer)
         struct list_head *head = &cpu_buffer->pages;
         struct buffer_page *page, *tmp;
 
+        list_del_init(&cpu_buffer->reader_page->list);
+        free_buffer_page(cpu_buffer->reader_page);
+
         list_for_each_entry_safe(page, tmp, head, list) {
                 list_del_init(&page->list);
                 free_buffer_page(page);
@@ -538,8 +554,10 @@ int ring_buffer_resize(struct ring_buffer *buffer, unsigned long size)
 
 static inline int rb_per_cpu_empty(struct ring_buffer_per_cpu *cpu_buffer)
 {
-        return cpu_buffer->head_page == cpu_buffer->tail_page &&
-                cpu_buffer->head == cpu_buffer->tail;
+        return (cpu_buffer->reader == cpu_buffer->reader_page->size &&
+                (cpu_buffer->tail_page == cpu_buffer->reader_page ||
+                 (cpu_buffer->tail_page == cpu_buffer->head_page &&
+                  cpu_buffer->head == cpu_buffer->tail)));
 }
 
 static inline int rb_null_event(struct ring_buffer_event *event)
@@ -555,10 +573,10 @@ static inline void *rb_page_index(struct buffer_page *page, unsigned index)
 }
 
 static inline struct ring_buffer_event *
-rb_head_event(struct ring_buffer_per_cpu *cpu_buffer)
+rb_reader_event(struct ring_buffer_per_cpu *cpu_buffer)
 {
-        return rb_page_index(cpu_buffer->head_page,
-                             cpu_buffer->head);
+        return rb_page_index(cpu_buffer->reader_page,
+                             cpu_buffer->reader);
 }
 
 static inline struct ring_buffer_event *
@@ -610,15 +628,32 @@ rb_add_stamp(struct ring_buffer_per_cpu *cpu_buffer, u64 *ts)
         cpu_buffer->write_stamp = *ts;
 }
 
-static void rb_reset_read_page(struct ring_buffer_per_cpu *cpu_buffer)
+static void rb_reset_head_page(struct ring_buffer_per_cpu *cpu_buffer)
 {
-        cpu_buffer->read_stamp = cpu_buffer->head_page->time_stamp;
         cpu_buffer->head = 0;
 }
 
-static void
-rb_reset_iter_read_page(struct ring_buffer_iter *iter)
+static void rb_reset_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
 {
+        cpu_buffer->read_stamp = cpu_buffer->reader_page->time_stamp;
+        cpu_buffer->reader = 0;
+}
+
+static inline void rb_inc_iter(struct ring_buffer_iter *iter)
+{
+        struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
+
+        /*
+         * The iterator could be on the reader page (it starts there).
+         * But the head could have moved, since the reader was
+         * found. Check for this case and assign the iterator
+         * to the head page instead of next.
+         */
+        if (iter->head_page == cpu_buffer->reader_page)
+                iter->head_page = cpu_buffer->head_page;
+        else
+                rb_inc_page(cpu_buffer, &iter->head_page);
+
         iter->read_stamp = iter->head_page->time_stamp;
         iter->head = 0;
 }
@@ -693,30 +728,39 @@ static struct ring_buffer_event *
 __rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer,
                   unsigned type, unsigned long length, u64 *ts)
 {
-        struct buffer_page *head_page, *tail_page;
+        struct buffer_page *tail_page, *head_page, *reader_page;
         unsigned long tail;
         struct ring_buffer *buffer = cpu_buffer->buffer;
         struct ring_buffer_event *event;
 
+        /* No locking needed for tail page */
         tail_page = cpu_buffer->tail_page;
-        head_page = cpu_buffer->head_page;
         tail = cpu_buffer->tail;
 
         if (tail + length > BUF_PAGE_SIZE) {
                 struct buffer_page *next_page = tail_page;
 
+                spin_lock(&cpu_buffer->lock);
                 rb_inc_page(cpu_buffer, &next_page);
 
+                head_page = cpu_buffer->head_page;
+                reader_page = cpu_buffer->reader_page;
+
+                /* we grabbed the lock before incrementing */
+                WARN_ON(next_page == reader_page);
+
                 if (next_page == head_page) {
-                        if (!(buffer->flags & RB_FL_OVERWRITE))
+                        if (!(buffer->flags & RB_FL_OVERWRITE)) {
+                                spin_unlock(&cpu_buffer->lock);